feat: enforce capability stop terminal mode after reviewer denial (Issue #197)

Enter terminal mode when review_pr/merge_pr capability is denied. Block
reviewer queue tools (list_prs, eligibility checks) and add report-purity
validators for forbidden PR selection, fallback, and empty-queue claims.

Closes #197

Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>

capability_stop_terminal.py

@@ -0,0 +1,202 @@
+"""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",
+        ],
+    }

mcp_server.py

@@ -170,6 +170,7 @@ from gitea_auth import (  # noqa: E402
 )
 import gitea_audit  # noqa: E402
 import gitea_config  # noqa: E402
+import capability_stop_terminal  # noqa: E402
 import issue_duplicate_gate  # noqa: E402
 import role_session_router  # noqa: E402
 
@@ -571,6 +572,11 @@ def gitea_list_prs(
         'mergeable', 'updated_at' ('url' only with the reveal opt-in).
         The additional 'updated_at' aids stale/conflicting queue detection.
     """
+    allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
+        "list_prs"
+    )
+    if not allowed:
+        raise RuntimeError("; ".join(block_reasons))
     h, o, r = _resolve(remote, host, org, repo)
     auth = _auth(h)
     url = f"{repo_api_url(h, o, r)}/pulls?state={state}"
@@ -682,6 +688,17 @@ def gitea_check_pr_eligibility(
         'permission_report' (#142).
     """
     action = (action or "").strip().lower()
+    if action in ("review", "approve", "request_changes", "merge"):
+        allowed, block_reasons = capability_stop_terminal.check_reviewer_queue_tool(
+            "check_pr_eligibility"
+        )
+        if not allowed:
+            return {
+                "eligible": False,
+                "requested_action": action,
+                "reasons": block_reasons,
+                "terminal_mode": True,
+            }
     profile = get_profile()
     result = {
         "eligible": False,
@@ -3792,7 +3809,7 @@ def gitea_resolve_task_capability(
 
     record_mutation_authority(profile["profile_name"], username, remote if remote in REMOTES else None, task)
 
-    return {
+    result = {
         "requested_task": task,
         "required_operation_permission": required_permission,
         "required_role_kind": required_role,
@@ -3807,6 +3824,33 @@ def gitea_resolve_task_capability(
         "different_mcp_namespace_required": different_namespace_required,
         "exact_safe_next_action": next_safe_action,
     }
+    if stop_required:
+        terminal = capability_stop_terminal.enter_from_capability_result(result)
+        if terminal:
+            result["terminal_mode"] = True
+            result["terminal_report"] = (
+                capability_stop_terminal.build_terminal_report(result)
+            )
+    return result
+
+
+@mcp.tool()
+def gitea_capability_stop_terminal_report() -> dict:
+    """Read-only: terminal report template after reviewer capability denial (#197)."""
+    record = capability_stop_terminal.active_record()
+    if not record:
+        return {
+            "terminal_mode": False,
+            "reasons": ["capability stop terminal mode is not active"],
+        }
+    return capability_stop_terminal.build_terminal_report({
+        "requested_task": record.get("requested_task"),
+        "required_role_kind": record.get("required_role_kind"),
+        "active_profile": record.get("active_profile"),
+        "active_identity": record.get("active_identity"),
+        "stop_required": record.get("stop_required"),
+        "exact_safe_next_action": record.get("exact_safe_next_action"),
+    })
 
 
 # ── Entry point ───────────────────────────────────────────────────────────────

review_proofs.py

@@ -891,6 +891,22 @@ def assess_role_route_handoff(report_text, route_result=None):
     }
 
 
+def assess_capability_stop_terminal_report(
+    report_text,
+    *,
+    trust_gate_status=None,
+    capability_denied=True,
+):
+    """Issue #197: reports after reviewer capability denial must stay pure."""
+    from capability_stop_terminal import assess_capability_stop_report
+
+    return assess_capability_stop_report(
+        report_text,
+        trust_gate_status=trust_gate_status,
+        capability_denied=capability_denied,
+    )
+
+
 # ── PR Inventory Trust Gate (Issue #194) ──────────────────────────────────────
 #
 # A reviewer agent may not convert an empty PR list response into a definitive

tests/conftest.py

@@ -25,4 +25,14 @@ def _reset_mutation_authority(monkeypatch):
         return
     monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None)
     monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {})
+    try:
+        import capability_stop_terminal
+        capability_stop_terminal.clear()
+    except Exception:
+        pass
     yield
+    try:
+        import capability_stop_terminal
+        capability_stop_terminal.clear()
+    except Exception:
+        pass

tests/test_capability_stop_terminal.py

@@ -0,0 +1,164 @@
+"""Tests for capability stop terminal mode (#197)."""
+import json
+import os
+import sys
+import tempfile
+import unittest
+from unittest.mock import patch
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import capability_stop_terminal
+import gitea_config
+import mcp_server
+from review_proofs import assess_capability_stop_terminal_report
+
+CONFIG = {
+    "version": 2,
+    "contexts": {
+        "ctx": {
+            "enabled": True,
+            "gitea": {"enabled": True, "base_url": "https://gitea.example.com"},
+        }
+    },
+    "profiles": {
+        "prgs-author": {
+            "enabled": True,
+            "context": "ctx",
+            "role": "author",
+            "username": "jcwalker3",
+            "auth": {"type": "env", "name": "GITEA_TOKEN_AUTHOR"},
+            "allowed_operations": [
+                "gitea.read", "gitea.issue.create", "gitea.pr.create",
+                "gitea.branch.push",
+            ],
+            "forbidden_operations": [
+                "gitea.pr.approve", "gitea.pr.merge", "gitea.pr.review",
+            ],
+            "execution_profile": "prgs-author",
+        },
+    },
+    "rules": {"allow_runtime_switching": False},
+}
+
+
+class TestCapabilityStopTerminal(unittest.TestCase):
+    def setUp(self):
+        capability_stop_terminal.clear()
+        self._remotes = patch.dict(mcp_server.REMOTES, {
+            "prgs": {
+                "host": "gitea.example.com",
+                "org": "Scaled-Tech-Consulting",
+                "repo": "Gitea-Tools",
+            },
+        })
+        self._remotes.start()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir = tempfile.TemporaryDirectory()
+        self.config_path = os.path.join(self._dir.name, "profiles.json")
+        with open(self.config_path, "w", encoding="utf-8") as fh:
+            fh.write(json.dumps(CONFIG))
+
+    def tearDown(self):
+        self._remotes.stop()
+        capability_stop_terminal.clear()
+        mcp_server._IDENTITY_CACHE.clear()
+        gitea_config._active_profile_override = None
+        self._dir.cleanup()
+
+    def _env(self):
+        return {
+            "GITEA_MCP_CONFIG": self.config_path,
+            "GITEA_MCP_PROFILE": "prgs-author",
+            "GITEA_TOKEN_AUTHOR": "author-pass",
+        }
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_review_pr_stop_enters_terminal_mode(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            res = mcp_server.gitea_resolve_task_capability(
+                task="review_pr", remote="prgs"
+            )
+        self.assertTrue(res["stop_required"])
+        self.assertTrue(res.get("terminal_mode"))
+        self.assertTrue(capability_stop_terminal.is_active())
+        self.assertIn(
+            capability_stop_terminal.TERMINAL_REPORT_HEADING,
+            res["terminal_report"]["heading"],
+        )
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_list_prs_blocked_after_capability_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            with self.assertRaises(RuntimeError) as ctx:
+                mcp_server.gitea_list_prs(remote="prgs")
+        self.assertIn("Cannot perform reviewer task", str(ctx.exception))
+
+    @patch("mcp_server.api_request", return_value={"login": "jcwalker3"})
+    @patch("mcp_server.get_auth_header", return_value="token author-pass")
+    def test_eligibility_check_blocked_after_stop(self, _auth, _api):
+        with patch.dict(os.environ, self._env()):
+            mcp_server.gitea_resolve_task_capability(task="review_pr", remote="prgs")
+            res = mcp_server.gitea_check_pr_eligibility(
+                pr_number=193, action="review", remote="prgs"
+            )
+        self.assertFalse(res["eligible"])
+        self.assertTrue(res.get("terminal_mode"))
+
+    def test_report_with_pr_selection_impure(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Selected PR #193 for review anyway."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+
+    def test_session_eligibility_wording_blocked(self):
+        ok, violations = capability_stop_terminal.validate_eligibility_wording(
+            "PR 193 is not authored by this session so it is eligible."
+        )
+        self.assertFalse(ok)
+        self.assertTrue(violations)
+
+    def test_rebase_fallback_blocked_in_report(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Or have me rebase conflicted PR 193."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertFalse(result["pure"])
+        self.assertTrue(
+            any("author fallback" in v for v in result["violations"])
+        )
+
+    def test_empty_queue_without_trusted_empty_blocked(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "The repo has 0 open PRs."
+        )
+        result = assess_capability_stop_terminal_report(
+            report, trust_gate_status="untrusted_empty"
+        )
+        self.assertFalse(result["pure"])
+
+    def test_pure_terminal_report_passes(self):
+        report = (
+            "Cannot perform reviewer task under current profile. "
+            "No reviewer mutations performed.\n"
+            "Identity: jcwalker3 / prgs-author.\n"
+            "Required: prgs-reviewer.\n"
+            "No mutations performed."
+        )
+        result = assess_capability_stop_terminal_report(report)
+        self.assertTrue(result["pure"])
+
+
+if __name__ == "__main__":
+    unittest.main()