Gitea-Tools/review_workflow_boundary.py

"""Reviewer session boundary tracking for workflow-load gate (#403).

Pre-review commands executed before ``gitea_load_review_workflow`` must be
classified. Boundary violations block downstream reviewer mutations even when
workflow hash proof is present.
"""

from __future__ import annotations

import os
import re
from typing import Any

CLASSIFICATION_READ_ONLY_INVENTORY = "read_only_inventory"
CLASSIFICATION_DIAGNOSTIC = "diagnostic"
CLASSIFICATION_BOUNDARY_VIOLATION = "boundary_violation"
CLASSIFICATION_UNCLASSIFIED = "unclassified"

ALLOWED_CLASSIFICATIONS = frozenset({
    CLASSIFICATION_READ_ONLY_INVENTORY,
    CLASSIFICATION_DIAGNOSTIC,
    CLASSIFICATION_BOUNDARY_VIOLATION,
    CLASSIFICATION_UNCLASSIFIED,
})

_PRE_REVIEW_COMMANDS: list[dict[str, Any]] = []

_READ_ONLY_INVENTORY_PATTERNS = (
    re.compile(
        r"\bgitea[_-](?:list|view|whoami|get[-_]|resolve[-_]task|check[-_]pr|route[-_]task)",
        re.I,
    ),
    re.compile(r"\bgit\s+(?:fetch|remote\s+update|branch\s+-a|log|show|rev-parse)\b", re.I),
    re.compile(r"\bgit\s+status\b", re.I),
    re.compile(r"\bgit\s+worktree\s+list\b", re.I),
)

_DIAGNOSTIC_PATTERNS = (
    re.compile(r"\bgit\s+diff(?:\s+--stat)?\b", re.I),
    re.compile(r"\bwhich\s+pytest\b", re.I),
    re.compile(r"\bpytest\s+--version\b", re.I),
)

_BOUNDARY_VIOLATION_PATTERNS: tuple[tuple[re.Pattern[str], str], ...] = (
    (re.compile(r"\b(?:pytest|python\s+-m\s+pytest|python\s+-m\s+unittest)\b", re.I),
     "validation command before workflow load"),
    (re.compile(r"\bprofiles\.json\b", re.I), "local profile config inspection"),
    (re.compile(r"\bgitea-mcp(?:\.v2-contexts)?\.json\b", re.I),
     "local Gitea MCP config inspection"),
    (re.compile(r"\b\.env(?:\.|$|\b)", re.I), "credential file inspection"),
    (re.compile(r"\bkeychain\b", re.I), "credential store inspection"),
    (re.compile(r"\bpkill\b", re.I), "MCP repair activity"),
    (re.compile(r"\b(?:edit|write|modify).{0,40}\bmcp\b", re.I),
     "MCP config exploration"),
    (re.compile(r"\bgit\s+(?:add|commit|reset|clean|checkout|merge|rebase|push)\b", re.I),
     "git mutation before workflow load"),
)


def clear_pre_review_commands() -> None:
    """Test helper and session reset."""
    global _PRE_REVIEW_COMMANDS
    _PRE_REVIEW_COMMANDS = []


def pre_review_commands() -> list[dict[str, Any]]:
    """Return a shallow copy of recorded pre-review commands."""
    return [dict(entry) for entry in _PRE_REVIEW_COMMANDS]


def _normalize_path(path: str | None) -> str:
    return os.path.realpath(os.path.abspath((path or "").strip() or os.getcwd()))


def is_main_checkout_path(cwd: str | None, project_root: str | None) -> bool:
    """True when *cwd* is the stable control checkout (not under branches/)."""
    if not project_root:
        return False
    root = _normalize_path(project_root)
    path = _normalize_path(cwd)
    if path != root:
        return False
    marker = f"{os.sep}branches{os.sep}"
    return marker not in path


def classify_pre_review_command(
    command: str,
    *,
    cwd: str | None = None,
    project_root: str | None = None,
) -> dict[str, Any]:
    """Classify a command executed before workflow load."""
    text = (command or "").strip()
    path = _normalize_path(cwd)
    root = _normalize_path(project_root) if project_root else None
    reasons: list[str] = []

    for pattern, label in _BOUNDARY_VIOLATION_PATTERNS:
        if pattern.search(text):
            if label.startswith("validation") and root and not is_main_checkout_path(path, root):
                continue
            if label.startswith("git mutation") and root and not is_main_checkout_path(path, root):
                continue
            reasons.append(label)
            return {
                "command": text,
                "cwd": path,
                "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
                "reasons": reasons,
            }

    for pattern in _READ_ONLY_INVENTORY_PATTERNS:
        if pattern.search(text):
            return {
                "command": text,
                "cwd": path,
                "classification": CLASSIFICATION_READ_ONLY_INVENTORY,
                "reasons": [],
            }

    for pattern in _DIAGNOSTIC_PATTERNS:
        if pattern.search(text):
            return {
                "command": text,
                "cwd": path,
                "classification": CLASSIFICATION_DIAGNOSTIC,
                "reasons": [],
            }

    if root and is_main_checkout_path(path, root):
        if re.search(r"\b(?:cat|head|less|read)\b", text, re.I):
            if re.search(r"workflow|skill|runbook", text, re.I):
                return {
                    "command": text,
                    "cwd": path,
                    "classification": CLASSIFICATION_BOUNDARY_VIOLATION,
                    "reasons": [
                        "canonical workflow viewed as local file without "
                        "gitea_load_review_workflow (narrative load is not proof)"
                    ],
                }

    return {
        "command": text,
        "cwd": path,
        "classification": CLASSIFICATION_UNCLASSIFIED,
        "reasons": [
            "pre-review command not classified; record via "
            "gitea_record_pre_review_command before workflow load"
        ],
    }


def record_pre_review_command(
    command: str,
    *,
    cwd: str | None = None,
    project_root: str | None = None,
    classification: str | None = None,
) -> dict[str, Any]:
    """Record and classify a pre-review command for the current session."""
    assessed = classify_pre_review_command(
        command, cwd=cwd, project_root=project_root)
    if classification:
        if classification not in ALLOWED_CLASSIFICATIONS:
            assessed["classification"] = CLASSIFICATION_UNCLASSIFIED
            assessed["reasons"] = [
                f"unknown classification '{classification}'; fail closed"
            ]
        else:
            assessed["classification"] = classification
            assessed["reasons"] = []
    entry = {
        **assessed,
        "session_pid": os.getpid(),
    }
    _PRE_REVIEW_COMMANDS.append(entry)
    return dict(entry)


def assess_boundary_status(project_root: str | None = None) -> dict[str, Any]:
    """Summarize pre-review boundary state for session proof and reports."""
    violations = [
        entry for entry in _PRE_REVIEW_COMMANDS
        if entry.get("classification") == CLASSIFICATION_BOUNDARY_VIOLATION
    ]
    unclassified = [
        entry for entry in _PRE_REVIEW_COMMANDS
        if entry.get("classification") == CLASSIFICATION_UNCLASSIFIED
    ]
    reasons: list[str] = []
    for entry in violations:
        reasons.extend(entry.get("reasons") or [
            f"boundary violation: {entry.get('command', '')[:80]}"
        ])
    for entry in unclassified:
        reasons.extend(entry.get("reasons") or [
            "unclassified pre-review command blocks reviewer mutations"
        ])

    clean = not reasons
    return {
        "boundary_status": "clean" if clean else "violation",
        "boundary_clean": clean,
        "pre_review_command_count": len(_PRE_REVIEW_COMMANDS),
        "boundary_violation_count": len(violations),
        "unclassified_command_count": len(unclassified),
        "violations": [
            {
                "command": v.get("command"),
                "cwd": v.get("cwd"),
                "reasons": list(v.get("reasons") or []),
            }
            for v in violations
        ],
        "reasons": reasons,
    }


def boundary_blockers(project_root: str | None = None) -> list[str]:
    """Reasons reviewer mutations must fail closed due to boundary state."""
    status = assess_boundary_status(project_root)
    if status.get("boundary_clean"):
        return []
    return list(status.get("reasons") or [
        "reviewer session boundary violation before workflow load"
    ])


def workflow_load_helper_result(
    load: dict | None,
    project_root: str | None = None,
) -> dict[str, Any]:
    """Structured helper result for final reports (#403)."""
    boundary = assess_boundary_status(project_root)
    if load is None:
        return {
            "workflow_load_proof_present": False,
            "workflow_source": None,
            "workflow_hash": None,
            "final_report_schema_hash": None,
            "boundary_status": boundary.get("boundary_status"),
            "boundary_clean": False,
            "reasons": [
                "gitea_load_review_workflow helper result missing from report"
            ],
        }
    return {
        "workflow_load_proof_present": True,
        "workflow_source": load.get("workflow_source"),
        "workflow_hash": load.get("workflow_hash"),
        "final_report_schema_path": load.get("final_report_schema_path"),
        "final_report_schema_hash": load.get("final_report_schema_hash"),
        "boundary_status": load.get("boundary_status", boundary.get("boundary_status")),
        "boundary_clean": bool(load.get("boundary_clean", boundary.get("boundary_clean"))),
        "pre_review_command_count": boundary.get("pre_review_command_count"),
        "reasons": [],
    }