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",
|
||||
],
|
||||
}
|
||||
Reference in New Issue
Block a user