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]>
This commit is contained in:
Claude Opus 4.8
@@ -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",
|
||||
],
|
||||
}
|
||||
+45
-1
@@ -45,6 +45,7 @@ from gitea_auth import ( # noqa: E402
|
||||
)
|
||||
import gitea_audit # noqa: E402
|
||||
import gitea_config # noqa: E402
|
||||
import capability_stop_terminal # noqa: E402
|
||||
|
||||
|
||||
def _reveal_endpoints() -> bool:
|
||||
@@ -406,6 +407,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}"
|
||||
@@ -517,6 +523,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,
|
||||
@@ -3476,7 +3493,7 @@ def gitea_resolve_task_capability(
|
||||
"STOP: the active profile cannot perform the requested task; "
|
||||
"follow exact_safe_next_action instead of improvising.")
|
||||
|
||||
return {
|
||||
result = {
|
||||
"requested_task": task,
|
||||
"required_operation_permission": required_permission,
|
||||
"required_role_kind": required_role,
|
||||
@@ -3491,6 +3508,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 ───────────────────────────────────────────────────────────────
|
||||
|
||||
@@ -852,6 +852,22 @@ def assess_controller_handoff(report_text, role=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
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user