diff --git a/capability_stop_terminal.py b/capability_stop_terminal.py new file mode 100644 index 0000000..fdead79 --- /dev/null +++ b/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", + ], + } \ No newline at end of file diff --git a/mcp_server.py b/mcp_server.py index 8fe3009..155ec06 100644 --- a/mcp_server.py +++ b/mcp_server.py @@ -171,6 +171,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 @@ -690,6 +691,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}" @@ -801,6 +807,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, @@ -4269,7 +4286,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, @@ -4284,6 +4301,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 ─────────────────────────────────────────────────────────────── diff --git a/review_proofs.py b/review_proofs.py index da79188..8da7369 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -993,6 +993,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 diff --git a/tests/conftest.py b/tests/conftest.py index 41c72c4..7bbf78c 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -26,4 +26,14 @@ def _reset_mutation_authority(monkeypatch): monkeypatch.setattr(mcp_server, "_MUTATION_AUTHORITY", None) monkeypatch.setattr(mcp_server, "_IDENTITY_CACHE", {}) monkeypatch.setattr(mcp_server, "_REVIEW_DECISION_LOCK", None) + 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 diff --git a/tests/test_capability_stop_terminal.py b/tests/test_capability_stop_terminal.py new file mode 100644 index 0000000..ef0cd50 --- /dev/null +++ b/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() \ No newline at end of file