From 6743cc11b92d1d6671b886a2b65a3838b4d1e2dc Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 09:11:15 -0400 Subject: [PATCH 1/2] feat: add native MCP preference gate and shell circuit breaker (Closes #270) Enforce native MCP-first paths for Gitea mutations with fail-closed assessment helpers, shell spawn circuit breaker tracking, and MCP tools for operation-path assessment and shell health reporting. Closes #270 --- gitea_mcp_server.py | 73 ++++++ native_mcp_preference.py | 369 ++++++++++++++++++++++++++++ tests/test_native_mcp_preference.py | 121 +++++++++ 3 files changed, 563 insertions(+) create mode 100644 native_mcp_preference.py create mode 100644 tests/test_native_mcp_preference.py diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index b5415eb..65bd6a5 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -477,6 +477,7 @@ import review_proofs # noqa: E402 import agent_temp_artifacts import issue_lock_worktree # noqa: E402 import merged_cleanup_reconcile # noqa: E402 +import native_mcp_preference # noqa: E402 # Fail-closed exact-issue-lock file (#204): written by gitea_lock_issue, @@ -4082,6 +4083,14 @@ _GUIDE_RULES = { "inventory, summarize) needs no explicit authorization. Enforced " "fail closed via subagent_gate.assess_subagent_delegation and " "validate_subagent_report (#266)."), + "native_mcp_preference": ( + "Gitea mutations must use native MCP tools first. When MCP is " + "available, block shell scripts, direct API calls, WebFetch, " + "Playwright, helper encoders, and profile-secret reads unless " + "explicit recovery-mode proof is complete. If MCP transport is " + "broken or the shell circuit breaker trips, emit a terminal recovery " + "report instead of improvising unsafe fallback. Reviewers must not " + "touch or restart MCP server files/processes (#270)."), } _COMMON_WORKFLOWS = [ @@ -4786,6 +4795,69 @@ def _build_runtime_task_capabilities( } +@mcp.tool() +@mcp.tool() +def gitea_assess_gitea_operation_path( + task: str, + path_kind: str | None = None, + command_or_detail: str | None = None, + mcp_available: bool = True, + mcp_tool_visible: bool = True, + recovery_mode: bool = False, + recovery_proof_complete: bool = False, + remote: str = "dadeschools", + host: str | None = None, +) -> dict: + """Read-only: assess whether a proposed Gitea mutation path is allowed (#270). + + Native MCP must be preferred when available. Shell scripts, direct API + calls, WebFetch, Playwright, and helper encoders are blocked unless + explicit recovery-mode proof is supplied. + """ + profile = get_profile() + role = _role_kind( + profile.get("allowed_operations") or [], + profile.get("forbidden_operations") or [], + ) + return native_mcp_preference.assess_gitea_operation_path( + task=task, + path_kind=path_kind, + command_or_detail=command_or_detail, + mcp_available=mcp_available, + mcp_tool_visible=mcp_tool_visible, + recovery_mode=recovery_mode, + recovery_proof_complete=recovery_proof_complete, + role=role, + active_profile=profile.get("profile_name"), + ) + + +@mcp.tool() +def gitea_record_shell_spawn_outcome( + exit_code: int | None = None, + stdout: str = "", + stderr: str = "", + spawn_failure: bool | None = None, + probe_attempted: bool = False, + probe_succeeded: bool | None = None, +) -> dict: + """Record shell spawn outcome and update the session circuit breaker (#270).""" + return native_mcp_preference.record_shell_spawn_outcome( + exit_code=exit_code, + stdout=stdout, + stderr=stderr, + spawn_failure=spawn_failure, + probe_attempted=probe_attempted, + probe_succeeded=probe_succeeded, + ) + + +@mcp.tool() +def gitea_get_shell_health() -> dict: + """Read-only: return shell spawn circuit-breaker state for this MCP session (#270).""" + return native_mcp_preference.shell_health_status() + + @mcp.tool() def gitea_get_runtime_context( remote: str = "dadeschools", @@ -4904,6 +4976,7 @@ def gitea_get_runtime_context( "preflight_block_reasons": preflight["preflight_block_reasons"], "preflight_workspace": preflight.get("preflight_workspace"), "session_capabilities": session_capabilities, + "shell_health": native_mcp_preference.shell_health_status(), } if reveal and h: diff --git a/native_mcp_preference.py b/native_mcp_preference.py new file mode 100644 index 0000000..872b800 --- /dev/null +++ b/native_mcp_preference.py @@ -0,0 +1,369 @@ +"""Native MCP preference gate and shell health circuit breaker (#270). + +Gitea mutations must use native MCP tools first. Shell scripts, direct API +calls, browser helpers, and improvised encoders are blocked when MCP is +available unless explicit recovery-mode proof is supplied. +""" + +from __future__ import annotations + +import re +from typing import Any + +from reviewer_fallback import LOCAL_GITEA_SCRIPT_NAMES + +# Tasks that must prefer native MCP over shell/API/helper fallbacks. +GITEA_MUTATION_TASKS = frozenset({ + "comment_issue", + "mark_issue", + "lock_issue", + "set_issue_labels", + "create_issue", + "close_issue", + "create_branch", + "push_branch", + "create_pr", + "close_pr", + "comment_pr", + "review_pr", + "merge_pr", + "approve_pr", + "request_changes_pr", + "commit_files", + "gitea_commit_files", + "delete_branch", + "address_pr_change_requests", +}) + +ALLOWED_PATH_KINDS = frozenset({ + "mcp_native", + "recovery_fallback", +}) + +BLOCKED_PATH_KINDS = frozenset({ + "shell_script", + "direct_api", + "webfetch", + "playwright", + "helper_script", + "unsafe_helper", + "mcp_server_touch", +}) + +SHELL_SPAWN_FAILURE_THRESHOLD = 2 + +TERMINAL_REPORT_HEADING = ( + "MCP transport unavailable or shell circuit breaker tripped. " + "No unsafe Gitea fallback performed." +) + +_RECOVERY_MODE_RE = re.compile( + r"\b(?:recovery mode|explicit recovery|mcp unavailable|mcp not available|" + r"mcp tools unavailable|no mcp path)\b", + re.IGNORECASE, +) +_LOCAL_SCRIPT_RE = re.compile( + r"(?:^|[\s\"'`/])(?:python3?|bash|sh)?\s*(?:" + + "|".join(re.escape(name) for name in LOCAL_GITEA_SCRIPT_NAMES) + + r")", + re.IGNORECASE, +) +_WEBFETCH_RE = re.compile(r"\b(?:webfetch|mcp_web_fetch|fetch\s+url)\b", re.IGNORECASE) +_PLAYWRIGHT_RE = re.compile(r"\b(?:playwright|browser_navigate|browser_click)\b", re.IGNORECASE) +_HELPER_SCRIPT_RE = re.compile( + r"(?:^|[\s\"'`/])(?:_encode_|_emit_|_inline_)[\w-]+\.py", + re.IGNORECASE, +) +_MANUAL_BASE64_RE = re.compile( + r"\b(?:manual\s+base64|llm-generated\s+base64|base64-encode\s+in\s+chat)\b", + re.IGNORECASE, +) +_DIRECT_API_RE = re.compile( + r"\b(?:api_request|urllib\.request|requests\.(?:post|patch|put)|curl\s+-X\s+POST)\b", + re.IGNORECASE, +) +_MCP_SERVER_TOUCH_RE = re.compile( + r"\b(?:kill|pkill|restart|reload|touch|edit|modify|write)\b.{0,40}\b(?:mcp_server|mcp-server|gitea_mcp_server)\b", + re.IGNORECASE, +) +_CLI_AUTH_DIVERGENCE_RE = re.compile( + r"GITEA_MCP_PROFILE\s*=\s*['\"]?([^\s'\";]+)", + re.IGNORECASE, +) + +_session_shell: dict[str, Any] = { + "consecutive_spawn_failures": 0, + "shell_unavailable": False, + "hard_stopped": False, + "last_exit_code": None, +} + + +def _clean(value: str | None) -> str: + return (value or "").strip() + + +def is_spawn_failure( + *, + exit_code: int | None = None, + stdout: str | None = None, + stderr: str | None = None, + spawn_failure: bool | None = None, +) -> bool: + """True for executor spawn failures (exit_code -1, empty output).""" + if spawn_failure is True: + return True + if spawn_failure is False: + return False + if exit_code != -1: + return False + return not (_clean(stdout) or _clean(stderr)) + + +def record_shell_spawn_outcome( + *, + exit_code: int | None = None, + stdout: str | None = None, + stderr: str | None = None, + spawn_failure: bool | None = None, + probe_attempted: bool = False, + probe_succeeded: bool | None = None, +) -> dict[str, Any]: + """Track shell spawn outcomes and trip the session circuit breaker (#270 AC4).""" + global _session_shell + failed = is_spawn_failure( + exit_code=exit_code, + stdout=stdout, + stderr=stderr, + spawn_failure=spawn_failure, + ) + + if failed: + _session_shell["consecutive_spawn_failures"] = ( + int(_session_shell.get("consecutive_spawn_failures") or 0) + 1 + ) + _session_shell["last_exit_code"] = exit_code + if probe_attempted and probe_succeeded is False: + _session_shell["shell_unavailable"] = True + else: + _session_shell["consecutive_spawn_failures"] = 0 + _session_shell["shell_unavailable"] = False + _session_shell["hard_stopped"] = False + _session_shell["last_exit_code"] = exit_code + + failures = int(_session_shell["consecutive_spawn_failures"]) + if failures >= SHELL_SPAWN_FAILURE_THRESHOLD: + _session_shell["shell_unavailable"] = True + _session_shell["hard_stopped"] = True + + return shell_health_status() + + +def shell_health_status() -> dict[str, Any]: + """Return the current shell health / circuit-breaker state.""" + failures = int(_session_shell.get("consecutive_spawn_failures") or 0) + hard_stopped = bool(_session_shell.get("hard_stopped")) + shell_unavailable = bool(_session_shell.get("shell_unavailable")) + return { + "consecutive_spawn_failures": failures, + "shell_unavailable": shell_unavailable, + "hard_stopped": hard_stopped, + "threshold": SHELL_SPAWN_FAILURE_THRESHOLD, + "shell_use_allowed": not hard_stopped, + "last_exit_code": _session_shell.get("last_exit_code"), + "safe_next_action": ( + "emit terminal recovery report; prefer native MCP for remaining Gitea mutations" + if hard_stopped + else ( + "probe shell once with echo/pwd; if probe fails mark shell unavailable" + if failures == 1 + else "shell healthy" + ) + ), + } + + +def clear_shell_health_for_tests() -> None: + """Reset session shell state (tests only).""" + global _session_shell + _session_shell = { + "consecutive_spawn_failures": 0, + "shell_unavailable": False, + "hard_stopped": False, + "last_exit_code": None, + } + + +def classify_command_path(command_or_detail: str | None) -> str: + """Classify a proposed command/detail into a path kind.""" + text = command_or_detail or "" + if _MCP_SERVER_TOUCH_RE.search(text): + return "mcp_server_touch" + if _WEBFETCH_RE.search(text): + return "webfetch" + if _PLAYWRIGHT_RE.search(text): + return "playwright" + if _HELPER_SCRIPT_RE.search(text) or _MANUAL_BASE64_RE.search(text): + return "unsafe_helper" + if _LOCAL_SCRIPT_RE.search(text): + return "shell_script" + if _DIRECT_API_RE.search(text): + return "direct_api" + return "mcp_native" + + +def detect_cli_auth_divergence( + command_or_detail: str | None, + *, + active_profile: str | None, +) -> list[str]: + """Flag shell commands that override GITEA_MCP_PROFILE away from the session.""" + reasons: list[str] = [] + text = command_or_detail or "" + active = _clean(active_profile) + match = _CLI_AUTH_DIVERGENCE_RE.search(text) + if not match or not active: + return reasons + requested = _clean(match.group(1)) + if requested and requested != active: + reasons.append( + f"CLI auth divergence: command sets GITEA_MCP_PROFILE='{requested}' " + f"but active session profile is '{active}' (fail closed)" + ) + return reasons + + +def format_mcp_unavailable_terminal_report( + *, + task: str | None = None, + reasons: list[str] | None = None, +) -> str: + """Terminal report when MCP is broken and unsafe recovery is forbidden (#270 AC3).""" + lines = [TERMINAL_REPORT_HEADING, ""] + if task: + lines.append(f"Blocked task: {task}") + if reasons: + lines.extend(["", "Reasons:"]) + lines.extend(f"- {reason}" for reason in reasons) + lines.extend([ + "", + "Required recovery (no improvised fallback):", + "- Restart the MCP session", + "- Kill hung background terminals holding the shell executor", + "- Retry the native MCP tool once after reconnect", + "- If MCP remains unavailable, stop and hand off to the operator", + "- Do not run local Gitea scripts, WebFetch, Playwright, or manual base64 encoders", + ]) + return "\n".join(lines) + + +def assess_gitea_operation_path( + *, + task: str, + path_kind: str | None = None, + command_or_detail: str | None = None, + mcp_available: bool = True, + mcp_tool_visible: bool = True, + recovery_mode: bool = False, + recovery_proof_complete: bool = False, + role: str | None = None, + active_profile: str | None = None, +) -> dict[str, Any]: + """Fail closed when a Gitea mutation would bypass native MCP (#270).""" + task_name = _clean(task) + resolved_kind = _clean(path_kind) or classify_command_path(command_or_detail) + reasons: list[str] = [] + + if task_name and task_name not in GITEA_MUTATION_TASKS: + reasons.append( + f"unknown or non-mutation Gitea task '{task_name}'; " + "native MCP preference gate applies only to Gitea mutations" + ) + + shell = shell_health_status() + if shell["hard_stopped"] and resolved_kind != "mcp_native": + reasons.append( + "shell circuit breaker is hard-stopped after consecutive spawn failures; " + "use native MCP or emit terminal recovery report" + ) + + recovery_declared = recovery_mode or bool( + _RECOVERY_MODE_RE.search(command_or_detail or "") + ) + recovery_allowed = ( + recovery_declared + and recovery_proof_complete + and not mcp_available + and resolved_kind in BLOCKED_PATH_KINDS - {"mcp_server_touch"} + ) + + if resolved_kind in BLOCKED_PATH_KINDS and not recovery_allowed: + reasons.append(f"path kind '{resolved_kind}' is not an approved Gitea mutation path") + + if resolved_kind == "mcp_server_touch": + if _clean(role) == "reviewer" or ( + active_profile and "reviewer" in active_profile.lower() + ): + reasons.append( + "reviewer workflows must not touch, restart, or kill MCP server files/processes" + ) + else: + reasons.append( + "MCP server files/processes must not be modified during normal Gitea workflows" + ) + + reasons.extend( + detect_cli_auth_divergence(command_or_detail, active_profile=active_profile) + ) + + if ( + mcp_available + and mcp_tool_visible + and resolved_kind != "mcp_native" + and not recovery_allowed + ): + if not recovery_declared: + reasons.append( + "native MCP tools are available; shell/API/helper fallback is forbidden" + ) + elif not recovery_proof_complete: + reasons.append( + "recovery-mode fallback requires complete recovery proof before proceeding" + ) + + if not mcp_available and resolved_kind != "mcp_native" and not recovery_allowed: + if not recovery_declared: + reasons.append( + "MCP transport unavailable; produce terminal recovery report instead of " + "improvising unsafe fallback" + ) + + block = bool(reasons) + terminal_report = None + if block and (not mcp_available or shell["hard_stopped"]): + terminal_report = format_mcp_unavailable_terminal_report( + task=task_name or None, + reasons=reasons, + ) + + return { + "task": task_name or None, + "path_kind": resolved_kind, + "mcp_available": mcp_available, + "mcp_tool_visible": mcp_tool_visible, + "recovery_mode": recovery_declared, + "shell_health": shell, + "block": block, + "allowed": not block, + "reasons": reasons, + "terminal_report": terminal_report, + "safe_next_action": ( + "use the native MCP tool for this task" + if mcp_available and mcp_tool_visible and block + else ( + terminal_report or "proceed with native MCP" + if block + else "proceed with native MCP" + ) + ), + } \ No newline at end of file diff --git a/tests/test_native_mcp_preference.py b/tests/test_native_mcp_preference.py new file mode 100644 index 0000000..7607814 --- /dev/null +++ b/tests/test_native_mcp_preference.py @@ -0,0 +1,121 @@ +"""Tests for native MCP preference gate and shell circuit breaker (#270).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +import native_mcp_preference as nmp # noqa: E402 + + +class TestShellHealthCircuitBreaker(unittest.TestCase): + def setUp(self): + nmp.clear_shell_health_for_tests() + + def test_two_spawn_failures_hard_stop(self): + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + status = nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + self.assertTrue(status["hard_stopped"]) + self.assertFalse(status["shell_use_allowed"]) + self.assertEqual(status["consecutive_spawn_failures"], 2) + + def test_success_resets_breaker(self): + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + status = nmp.record_shell_spawn_outcome(exit_code=0, stdout="ok", stderr="") + self.assertFalse(status["hard_stopped"]) + self.assertEqual(status["consecutive_spawn_failures"], 0) + + +class TestNativeMcpPreferenceGate(unittest.TestCase): + def setUp(self): + nmp.clear_shell_health_for_tests() + + def test_mcp_available_blocks_local_script(self): + result = nmp.assess_gitea_operation_path( + task="create_pr", + command_or_detail="python create_pr.py --remote prgs --title T --head h", + mcp_available=True, + mcp_tool_visible=True, + ) + self.assertTrue(result["block"]) + self.assertEqual(result["path_kind"], "shell_script") + + def test_mcp_native_path_allowed(self): + result = nmp.assess_gitea_operation_path( + task="comment_issue", + path_kind="mcp_native", + mcp_available=True, + mcp_tool_visible=True, + ) + self.assertFalse(result["block"]) + self.assertTrue(result["allowed"]) + + def test_mcp_unavailable_requires_terminal_report(self): + result = nmp.assess_gitea_operation_path( + task="merge_pr", + path_kind="shell_script", + mcp_available=False, + mcp_tool_visible=False, + ) + self.assertTrue(result["block"]) + self.assertTrue( + any("terminal recovery report" in r for r in result["reasons"]) + ) + self.assertIn(nmp.TERMINAL_REPORT_HEADING, result["terminal_report"]) + + def test_recovery_fallback_allowed_with_proof(self): + result = nmp.assess_gitea_operation_path( + task="merge_pr", + path_kind="shell_script", + command_or_detail="Recovery mode: MCP tools unavailable in this session.", + mcp_available=False, + recovery_mode=True, + recovery_proof_complete=True, + ) + self.assertFalse(result["block"]) + + def test_cli_auth_divergence_blocked(self): + result = nmp.assess_gitea_operation_path( + task="create_pr", + command_or_detail="GITEA_MCP_PROFILE=prgs-reviewer python create_pr.py", + mcp_available=True, + active_profile="prgs-author", + ) + self.assertTrue(result["block"]) + self.assertTrue(any("CLI auth divergence" in r for r in result["reasons"])) + + def test_unsafe_helper_fallback_denied(self): + result = nmp.assess_gitea_operation_path( + task="commit_files", + command_or_detail="python _encode_payload.py", + mcp_available=True, + mcp_tool_visible=True, + ) + self.assertTrue(result["block"]) + self.assertEqual(result["path_kind"], "unsafe_helper") + + def test_reviewer_mcp_server_touch_blocked(self): + result = nmp.assess_gitea_operation_path( + task="review_pr", + command_or_detail="pkill -f gitea_mcp_server.py", + mcp_available=True, + role="reviewer", + active_profile="prgs-reviewer", + ) + self.assertTrue(result["block"]) + self.assertEqual(result["path_kind"], "mcp_server_touch") + + def test_hard_stopped_shell_blocks_non_mcp_path(self): + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + nmp.record_shell_spawn_outcome(exit_code=-1, stdout="", stderr="") + result = nmp.assess_gitea_operation_path( + task="lock_issue", + path_kind="shell_script", + mcp_available=True, + ) + self.assertTrue(result["block"]) + self.assertTrue(any("circuit breaker" in r for r in result["reasons"])) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file From 958d470f9810efd46e63943a1ce04767ca23e410 Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Tue, 7 Jul 2026 09:15:13 -0400 Subject: [PATCH 2/2] feat: dedicated reconciler capability for closing landed PRs (Closes #309) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Already-landed open PRs block the review queue: the already-landed gate correctly stops approval/merge, but no task path could close the redundant PR. Add the reconciler close path: - task_capability_map: new reconcile_close_landed_pr (gitea.pr.close) and reconcile_close_landed_issue (gitea.issue.close) tasks under a dedicated "reconciler" role. Exact close capabilities only — the role grants no review, approve, request-changes, or merge authority, and normal reviewer profiles keep no broad PR-close authority. - role_session_router: reconciler tasks route wrong_role_stop in author/reviewer sessions with a dedicated recovery message; matching reconciler sessions route allowed_current_session. - review_proofs.assess_reconciler_close_gate: PR close allowed only when every proof passes — PR live-verified open, candidate head SHA pinned and matching the live head, target branch freshly fetched with a full SHA, head an ancestor of the target, exact gitea.pr.close proven. Non-landed PRs return NOT_LANDED_CLOSE_BLOCKED; missing close capability returns RECOVERY_HANDOFF_REQUIRED; incomplete proof fails closed as GATE_NOT_PROVEN. Linked-issue close is skipped when the issue is already closed and allowed only for an open issue resolved by the landed commits with exact gitea.issue.close capability. The gate also returns the required final-report field list. 17 new tests cover the capability map, wrong-role routing, and every gate path (landed close, non-landed block, stale target, changed head, missing capabilities, linked-issue variants, review-mutation denial). Co-Authored-By: Claude Opus 4.8 (1M context) --- review_proofs.py | 163 ++++++++++++++++++++++++ role_session_router.py | 28 +++++ task_capability_map.py | 10 ++ tests/test_reconciler_close_gate.py | 188 ++++++++++++++++++++++++++++ 4 files changed, 389 insertions(+) create mode 100644 tests/test_reconciler_close_gate.py diff --git a/review_proofs.py b/review_proofs.py index e0e948a..bf126b2 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -3759,6 +3759,169 @@ def assess_full_suite_failure_approval_gate( } +# --------------------------------------------------------------------------- +# Reconciler close gate (#309) +# --------------------------------------------------------------------------- + +RECONCILER_CLOSE_REPORT_FIELDS = ( + "identity/profile", + "close capability proof", + "PR live state", + "candidate head SHA", + "target branch SHA", + "ancestor proof", + "linked issue status", + "PR close result", + "issue close result", + "no review/merge confirmation", +) + + +def assess_reconciler_close_gate( + *, + pr_number: int | None, + pr_state: str | None, + candidate_head_sha: str | None, + live_head_sha: str | None, + target_branch: str | None, + target_branch_sha: str | None, + head_is_ancestor_of_target: bool | None, + close_pr_capability: bool, + close_issue_capability: bool = False, + linked_issue_state: str | None = None, + issue_resolved_by_landing: bool = False, +) -> dict: + """#309: gate the dedicated reconciler close path for landed PRs. + + The reconciler path may close a PR only with exact ``gitea.pr.close`` + capability and full already-landed proof: live open PR, pinned head, + freshly fetched target branch SHA, and ancestry of the head in the + target. Non-landed PRs are never closable here, and the gate never + grants review, approval, request-changes, or merge. + + Linked-issue closure: skipped when the issue is already closed; + allowed only when the issue is open, resolved by the landed commits, + and exact ``gitea.issue.close`` capability is proven. + """ + reasons: list[str] = [] + + if not isinstance(pr_number, int) or pr_number <= 0: + reasons.append("PR number missing or invalid (#309)") + if (pr_state or "").strip().lower() != "open": + reasons.append( + "PR is not live-verified open at mutation time; re-fetch the " + "PR before any reconciler close (#309)" + ) + if not _FULL_SHA.match((candidate_head_sha or "").strip()): + reasons.append( + "candidate head SHA is not a full 40-hex commit SHA (#309)" + ) + if live_head_sha is not None and candidate_head_sha and ( + live_head_sha.strip() != candidate_head_sha.strip() + ): + reasons.append( + "PR head changed since the ancestry proof was pinned; re-run " + "the already-landed check (#309)" + ) + if not (target_branch or "").strip(): + reasons.append("target branch missing (#309)") + if not _FULL_SHA.match((target_branch_sha or "").strip()): + reasons.append( + "target branch SHA missing or not a full 40-hex SHA; fetch the " + "target branch freshly before the ancestry check (#309)" + ) + if head_is_ancestor_of_target is None: + reasons.append( + "ancestry of the PR head against the target branch was not " + "checked (#309)" + ) + + never_allowed = { + "review_allowed": False, + "approve_allowed": False, + "request_changes_allowed": False, + "merge_allowed": False, + } + + if reasons: + return { + "outcome": "GATE_NOT_PROVEN", + "pr_close_allowed": False, + "issue_close_allowed": False, + "reasons": reasons, + "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS, + "safe_next_action": ( + "re-fetch the PR and target branch, pin SHAs, run the " + "ancestry check, then re-run this gate" + ), + **never_allowed, + } + + if head_is_ancestor_of_target is False: + return { + "outcome": "NOT_LANDED_CLOSE_BLOCKED", + "pr_close_allowed": False, + "issue_close_allowed": False, + "reasons": [ + f"PR #{pr_number} head is not an ancestor of " + f"{target_branch}; non-landed PRs cannot be closed through " + "the reconciler path (#309)" + ], + "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS, + "safe_next_action": ( + "route the PR through the normal review workflow; the " + "reconciler path only closes already-landed PRs" + ), + **never_allowed, + } + + if close_pr_capability is not True: + return { + "outcome": "RECOVERY_HANDOFF_REQUIRED", + "pr_close_allowed": False, + "issue_close_allowed": False, + "reasons": [ + "exact gitea.pr.close capability not proven; produce a " + "recovery handoff instead of closing (#309)" + ], + "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS, + "safe_next_action": ( + "produce a recovery handoff naming the missing " + "gitea.pr.close capability and the completed ancestor proof" + ), + **never_allowed, + } + + issue_close_allowed = ( + (linked_issue_state or "").strip().lower() == "open" + and issue_resolved_by_landing is True + and close_issue_capability is True + ) + issue_close_reasons = [] + if (linked_issue_state or "").strip().lower() == "closed": + issue_close_reasons.append( + "linked issue already closed; no issue close attempted (#309)" + ) + elif not issue_close_allowed: + issue_close_reasons.append( + "issue close requires an open linked issue resolved by the " + "landed commits and exact gitea.issue.close capability (#309)" + ) + + return { + "outcome": "CLOSE_ALLOWED", + "pr_close_allowed": True, + "issue_close_allowed": issue_close_allowed, + "reasons": issue_close_reasons, + "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS, + "safe_next_action": ( + "close the already-landed PR, report the close result, and " + "handle the linked issue per the proven capability" + ), + **never_allowed, + } + + # --------------------------------------------------------------------------- # Identity disclosure (#305) # --------------------------------------------------------------------------- diff --git a/role_session_router.py b/role_session_router.py index a7db807..ff75025 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -90,12 +90,20 @@ TASK_REQUIRED_ROLE = { "blind_pr_queue_review": "reviewer", "request_changes_pr": "reviewer", "approve_pr": "reviewer", + # #309: reconciler tasks close already-landed PRs/issues only. + "reconcile_close_landed_pr": "reconciler", + "reconcile_close_landed_issue": "reconciler", } WRONG_ROLE_REVIEWER_MSG = ( "Wrong role/session for reviewer task. Launch reviewer MCP namespace." ) +WRONG_ROLE_RECONCILER_MSG = ( + "Wrong role/session for reconciler task. Launch a reconciler-capable " + "MCP namespace/profile with exact close capability." +) + _session_last_route: dict | None = None @@ -191,6 +199,26 @@ def route_task_session( _record_route(result) return result + if required_role == "reconciler": + result = { + "task_type": task_type, + "required_role": required_role, + "active_role": active_role_kind, + "active_profile": active_profile, + "route_result": ROUTE_WRONG_ROLE, + "downstream_allowed": False, + "reasons": [ + WRONG_ROLE_RECONCILER_MSG, + "Reconciler tasks cannot run in author or reviewer " + "sessions without exact close capability.", + ], + "message": WRONG_ROLE_RECONCILER_MSG, + "runtime_switching_supported": runtime_switching_supported, + "profile_switch_blocked": not runtime_switching_supported, + } + _record_route(result) + return result + if required_role == "author": route = ROUTE_TO_AUTHOR message = ( diff --git a/task_capability_map.py b/task_capability_map.py index 9d162c1..7e252a8 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -96,6 +96,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.read", "role": "author", }, + # #309: dedicated reconciler path for already-landed open PRs. Exact + # close capabilities only — never review/approve/request_changes/merge. + "reconcile_close_landed_pr": { + "permission": "gitea.pr.close", + "role": "reconciler", + }, + "reconcile_close_landed_issue": { + "permission": "gitea.issue.close", + "role": "reconciler", + }, } # Issue-mutating MCP tools and their resolver task keys. diff --git a/tests/test_reconciler_close_gate.py b/tests/test_reconciler_close_gate.py new file mode 100644 index 0000000..bae8438 --- /dev/null +++ b/tests/test_reconciler_close_gate.py @@ -0,0 +1,188 @@ +"""Issue #309: dedicated reconciler capability to close already-landed PRs. + +The reconciler path must prove exact ``gitea.pr.close`` capability, must +never imply review/approve/request-changes/merge capability, and may close +a PR only after full already-landed proof. +""" + +import sys +import unittest + +sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) + +import role_session_router # noqa: E402 +import task_capability_map # noqa: E402 +from review_proofs import assess_reconciler_close_gate # noqa: E402 + +PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9" +OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9" + + +class TestReconcilerCapabilityMap(unittest.TestCase): + def test_reconcile_close_pr_task_maps_to_pr_close(self): + self.assertEqual( + task_capability_map.required_permission( + "reconcile_close_landed_pr"), + "gitea.pr.close", + ) + self.assertEqual( + task_capability_map.required_role("reconcile_close_landed_pr"), + "reconciler", + ) + + def test_reconcile_close_issue_task_maps_to_issue_close(self): + self.assertEqual( + task_capability_map.required_permission( + "reconcile_close_landed_issue"), + "gitea.issue.close", + ) + self.assertEqual( + task_capability_map.required_role( + "reconcile_close_landed_issue"), + "reconciler", + ) + + def test_reconciler_tasks_do_not_grant_review_permissions(self): + for task in ("reconcile_close_landed_pr", + "reconcile_close_landed_issue"): + permission = task_capability_map.required_permission(task) + self.assertNotIn("review", permission) + self.assertNotIn("approve", permission) + self.assertNotIn("merge", permission) + self.assertNotIn("request_changes", permission) + + +class TestReconcilerRouting(unittest.TestCase): + def test_reconciler_task_requires_reconciler_role(self): + self.assertEqual( + role_session_router.required_role_for_task( + "reconcile_close_landed_pr"), + "reconciler", + ) + + def test_reconciler_task_blocked_in_author_session(self): + result = role_session_router.route_task_session( + "reconcile_close_landed_pr", + active_profile="prgs-author", + active_role_kind="author", + allowed_in_current_session=False, + ) + self.assertEqual(result["route_result"], "wrong_role_stop") + self.assertFalse(result["downstream_allowed"]) + + def test_reconciler_task_allowed_in_matching_session(self): + result = role_session_router.route_task_session( + "reconcile_close_landed_pr", + active_profile="prgs-reconciler", + active_role_kind="reconciler", + allowed_in_current_session=True, + ) + self.assertEqual(result["route_result"], "allowed_current_session") + self.assertTrue(result["downstream_allowed"]) + + +class TestReconcilerCloseGate(unittest.TestCase): + def _gate(self, **overrides): + kwargs = { + "pr_number": 278, + "pr_state": "open", + "candidate_head_sha": PINNED, + "live_head_sha": PINNED, + "target_branch": "master", + "target_branch_sha": OTHER, + "head_is_ancestor_of_target": True, + "close_pr_capability": True, + "close_issue_capability": False, + "linked_issue_state": "closed", + } + kwargs.update(overrides) + return assess_reconciler_close_gate(**kwargs) + + def test_already_landed_open_pr_close_allowed(self): + result = self._gate() + self.assertEqual(result["outcome"], "CLOSE_ALLOWED") + self.assertTrue(result["pr_close_allowed"]) + + def test_not_landed_pr_cannot_be_closed(self): + result = self._gate(head_is_ancestor_of_target=False) + self.assertEqual(result["outcome"], "NOT_LANDED_CLOSE_BLOCKED") + self.assertFalse(result["pr_close_allowed"]) + + def test_unchecked_ancestry_blocks_close(self): + result = self._gate(head_is_ancestor_of_target=None) + self.assertEqual(result["outcome"], "GATE_NOT_PROVEN") + self.assertFalse(result["pr_close_allowed"]) + + def test_stale_target_branch_sha_blocks_close(self): + result = self._gate(target_branch_sha=None) + self.assertEqual(result["outcome"], "GATE_NOT_PROVEN") + self.assertFalse(result["pr_close_allowed"]) + + def test_closed_pr_state_blocks_close(self): + result = self._gate(pr_state="closed") + self.assertEqual(result["outcome"], "GATE_NOT_PROVEN") + self.assertFalse(result["pr_close_allowed"]) + + def test_changed_head_blocks_close(self): + result = self._gate(live_head_sha=OTHER) + self.assertEqual(result["outcome"], "GATE_NOT_PROVEN") + self.assertFalse(result["pr_close_allowed"]) + + def test_missing_close_capability_produces_recovery_handoff(self): + result = self._gate(close_pr_capability=False) + self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED") + self.assertFalse(result["pr_close_allowed"]) + self.assertTrue(any( + "gitea.pr.close" in reason for reason in result["reasons"] + )) + + def test_linked_issue_already_closed_skips_issue_close(self): + result = self._gate(linked_issue_state="closed", + close_issue_capability=True) + self.assertFalse(result["issue_close_allowed"]) + + def test_linked_open_issue_close_requires_capability(self): + allowed = self._gate( + linked_issue_state="open", + issue_resolved_by_landing=True, + close_issue_capability=True, + ) + self.assertTrue(allowed["issue_close_allowed"]) + + blocked = self._gate( + linked_issue_state="open", + issue_resolved_by_landing=True, + close_issue_capability=False, + ) + self.assertFalse(blocked["issue_close_allowed"]) + + def test_gate_never_allows_review_mutations(self): + for result in ( + self._gate(), + self._gate(head_is_ancestor_of_target=False), + self._gate(close_pr_capability=False), + ): + self.assertFalse(result["review_allowed"]) + self.assertFalse(result["approve_allowed"]) + self.assertFalse(result["request_changes_allowed"]) + self.assertFalse(result["merge_allowed"]) + + def test_report_fields_listed_for_close(self): + result = self._gate() + for field in ( + "identity/profile", + "close capability proof", + "PR live state", + "candidate head SHA", + "target branch SHA", + "ancestor proof", + "linked issue status", + "PR close result", + "issue close result", + "no review/merge confirmation", + ): + self.assertIn(field, result["required_report_fields"]) + + +if __name__ == "__main__": + unittest.main()