Gitea-Tools/capability_stop_terminal.py

"""Hard-stop terminal mode after reviewer capability denial (#197)."""

from __future__ import annotations

import re

TERMINAL_REPORT_HEADING = (
    "Cannot perform reviewer task under current profile. "
    "No reviewer mutations performed."
)

REVIEWER_CAPABILITY_TASKS = frozenset({
    "review_pr",
    "merge_pr",
    "blind_pr_queue_review",
    "request_changes_pr",
    "approve_pr",
})

BLOCKED_QUEUE_TOOLS = frozenset({
    "list_prs",
    "check_pr_eligibility",
    "view_pr",
    "submit_pr_review",
    "dry_run_pr_review",
    "merge_pr",
    "review_pr",
})

_session_terminal: dict | None = None


def enter_from_capability_result(capability: dict) -> dict | None:
    """Enter terminal mode when a reviewer/merge task is denied."""
    global _session_terminal
    task = (capability or {}).get("requested_task", "")
    required_role = (capability or {}).get("required_role_kind")
    if not capability.get("stop_required"):
        return None
    if required_role != "reviewer" and task not in REVIEWER_CAPABILITY_TASKS:
        return None
    record = {
        "active": True,
        "requested_task": task,
        "required_role_kind": required_role,
        "active_profile": capability.get("active_profile"),
        "active_identity": capability.get("active_identity"),
        "stop_required": True,
        "exact_safe_next_action": capability.get("exact_safe_next_action"),
        "terminal_message": TERMINAL_REPORT_HEADING,
    }
    _session_terminal = record
    return dict(record)


def enter_from_route_result(route: dict) -> dict | None:
    """Enter terminal mode from a role router wrong_role_stop (#206 compat)."""
    if (route or {}).get("route_result") != "wrong_role_stop":
        return None
    if route.get("required_role") != "reviewer":
        return None
    return enter_from_capability_result({
        "requested_task": route.get("task_type"),
        "required_role_kind": "reviewer",
        "stop_required": True,
        "active_profile": route.get("active_profile"),
        "active_identity": None,
        "exact_safe_next_action": route.get("message"),
    })


def is_active() -> bool:
    return bool(_session_terminal and _session_terminal.get("active"))


def active_record() -> dict | None:
    if not is_active():
        return None
    return dict(_session_terminal)


def clear():
    global _session_terminal
    _session_terminal = None


def check_reviewer_queue_tool(tool_name: str) -> tuple[bool, list[str]]:
    """Return (allowed, reasons). False when terminal mode blocks queue work."""
    if not is_active():
        return True, []
    name = (tool_name or "").strip().lower().removeprefix("gitea_")
    if name in BLOCKED_QUEUE_TOOLS:
        return False, [
            TERMINAL_REPORT_HEADING,
            f"Reviewer queue tool '{tool_name}' is blocked after "
            "capability denial (fail closed).",
            "Relaunch a reviewer MCP namespace to perform reviewer work.",
        ]
    return True, []


def validate_eligibility_wording(text: str) -> tuple[bool, list[str]]:
    """Reject session-based eligibility reasoning (#197)."""
    lower = (text or "").lower()
    violations = []
    if "not authored by this session" in lower:
        violations.append(
            "eligibility must use authenticated account identity, not "
            "'this session' wording"
        )
    if re.search(r"not (?:self-)?authored by (?:the )?session", lower):
        violations.append("session-based eligibility reasoning is invalid")
    return (len(violations) == 0), violations


def assess_capability_stop_report(
    report_text: str,
    *,
    trust_gate_status: str | None = None,
    capability_denied: bool = True,
) -> dict:
    """Validate final report purity after reviewer capability denial."""
    text = report_text or ""
    lower = text.lower()
    violations = []

    if capability_denied and TERMINAL_REPORT_HEADING.lower() not in lower:
        violations.append("missing required terminal report heading")

    forbidden_patterns = [
        ("pr selection", re.compile(
            r"selected pr|pr #\d+ (?:to review|selected)|eligible pr|"
            r"next pr to review", re.I)),
        ("sibling repo inventory", re.compile(
            r"sibling repo|other repo|mcp-control-plane|gitea-tools and", re.I)),
        ("author fallback", re.compile(
            r"rebase conflicted|author-side fallback|have me rebase|"
            r"implement the fix|push a branch|open a pr for", re.I)),
        ("invalid session eligibility", re.compile(
            r"not authored by this session", re.I)),
    ]
    for label, pattern in forbidden_patterns:
        if pattern.search(text):
            violations.append(f"forbidden after hard stop: {label}")

    empty_queue_patterns = re.compile(
        r"\b0 open pr|\bno open pr|\bno eligible pr|\bempty (?:review )?queue|"
        r"inventory empty",
        re.I,
    )
    if empty_queue_patterns.search(text):
        if trust_gate_status != "trusted_empty":
            violations.append(
                "empty-queue claim after capability stop without "
                "pr_inventory_trust_gate.status == trusted_empty"
            )

    ok, elig_violations = validate_eligibility_wording(text)
    violations.extend(elig_violations)

    if violations:
        return {
            "pure": False,
            "downgraded": True,
            "violations": violations,
            "reasons": violations,
        }
    return {
        "pure": True,
        "downgraded": False,
        "violations": [],
        "reasons": [],
    }


def build_terminal_report(capability: dict) -> dict:
    """Minimal allowed report fields after hard stop."""
    return {
        "terminal_mode": True,
        "heading": TERMINAL_REPORT_HEADING,
        "authenticated_profile": capability.get("active_profile"),
        "authenticated_identity": capability.get("active_identity"),
        "denied_task": capability.get("requested_task"),
        "required_role_kind": capability.get("required_role_kind"),
        "stop_required": capability.get("stop_required"),
        "required_action": capability.get("exact_safe_next_action"),
        "mutations_performed": False,
        "allowed_sections": [
            "authenticated identity/profile",
            "denied capability result",
            "reason task cannot proceed",
            "required reviewer profile/identity",
            "mutation confirmation (none)",
        ],
        "forbidden_sections": [
            "PR selection",
            "sibling-repo queue recommendations",
            "author-side fallback suggestions",
            "empty-queue claims without trusted_empty",
            "session-based eligibility wording",
        ],
    }