diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 6983c9a..6cbb0c5 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -479,6 +479,7 @@ import issue_lock_worktree # noqa: E402 import already_landed_reconcile # noqa: E402 import issue_claim_heartbeat # 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, @@ -4273,6 +4274,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 = [ @@ -4979,6 +4988,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", @@ -5097,6 +5169,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/review_proofs.py b/review_proofs.py index 9c572c3..17f1527 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1242,6 +1242,121 @@ def assess_workspace_mutation_consistency(report_text, observed_commands=None): } +# ── Reconciliation controller-handoff schema (Issue #303) ──────────────────── +# +# Already-landed PR reconciliation is neither author issue work nor +# reviewer merge work; its handoff needs its own schema. Stale +# author/reviewer fields fail validation, and observed git ref +# mutations (fetch) must be reported under the precise category. + +RECONCILIATION_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED" + +RECONCILIATION_HANDOFF_FIELDS = ( + ("Task", ("task",)), + ("Repo", ("repo",)), + ("Role/profile", ("role/profile", "role", "profile")), + ("Identity", ("identity",)), + ("Selected PR", ("selected pr",)), + ("PR live state", ("pr live state",)), + ("Candidate head SHA", ("candidate head sha",)), + ("Target branch", ("target branch",)), + ("Target branch SHA", ("target branch sha",)), + ("Ancestor proof", ("ancestor proof",)), + ("Linked issue", ("linked issue",)), + ("Linked issue live status", ("linked issue live status",)), + ("Eligibility class", ("eligibility class",)), + ("Capabilities proven", ("capabilities proven",)), + ("Missing capabilities", ("missing capabilities",)), + ("PR comments posted", ("pr comments posted",)), + ("Issue comments posted", ("issue comments posted",)), + ("PRs closed", ("prs closed",)), + ("Issues closed", ("issues closed",)), + ("Git ref mutations", ("git ref mutations",)), + ("Worktree mutations", ("worktree mutations",)), + ("MCP/Gitea mutations", ("mcp/gitea mutations",)), + ("External-state mutations", ("external-state mutations",)), + ("Read-only diagnostics", ("read-only diagnostics",)), + ("Blocker", ("blocker",)), + ("Safe next action", ("safe next action",)), + ("No review/merge confirmation", ("no review/merge",)), +) + +RECONCILIATION_FORBIDDEN_FIELDS = ( + "pr number opened", + "pinned reviewed head", + "scratch worktree used", + "workspace mutations", + "issue lock proof", + "claim/comment status", +) + + +def assess_reconciliation_handoff(report_text, observed_commands=None): + """#303: validate the dedicated reconciliation handoff schema. + + Requires a Controller Handoff section carrying every reconciliation + field, ``Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED``, and + no stale author/reviewer fields. *observed_commands* is the git + command log; an observed fetch/pull must be reported under ``Git ref + mutations`` (never claimed none). + + Returns {'complete', 'downgraded', 'reasons'}; fails closed. + """ + section = _handoff_section_lines(report_text) + if section is None: + return { + "complete": False, + "downgraded": True, + "reasons": [ + "reconciliation report has no section titled exactly " + f"'{HANDOFF_HEADING}'" + ], + } + + fields = {} + for line in section: + stripped = line.strip().lstrip("-*").strip() + if ":" in stripped: + k, v = stripped.split(":", 1) + fields[k.strip().lower()] = v.strip() + + reasons = [] + + for name, aliases in RECONCILIATION_HANDOFF_FIELDS: + if not any(label.startswith(alias) + for label in fields for alias in aliases): + reasons.append( + f"reconciliation handoff missing required field: {name}" + ) + + for stale in RECONCILIATION_FORBIDDEN_FIELDS: + if any(label.startswith(stale) for label in fields): + reasons.append( + f"reconciliation handoff must not include stale field " + f"'{stale}'" + ) + + eligibility = fields.get("eligibility class", "") + if eligibility and RECONCILIATION_ELIGIBILITY_CLASS not in eligibility.upper(): + reasons.append( + "reconciliation handoff eligibility class must be " + f"'{RECONCILIATION_ELIGIBILITY_CLASS}' (got '{eligibility}')" + ) + + _, ref_cmds = _classify_git_commands(observed_commands) + if ref_cmds and _field_claims_none(fields, "git ref mutations"): + reasons.append( + "observed git ref mutation not reported under 'Git ref " + f"mutations': {ref_cmds[0]}" + ) + + return { + "complete": not reasons, + "downgraded": bool(reasons), + "reasons": reasons, + } + + # ── Already-landed report wording (Issue #298) ─────────────────────────────── # # An already-landed PR is reconciliation-only: it must never be called @@ -1589,6 +1704,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, if report_text else {"proven": True, "block": False, "reasons": [], "violations": []} ) + inventory_worktree = ( + assess_inventory_worktree_report(report_text) + if report_text + else {"proven": True, "block": False, "reasons": []} + ) if baseline_validation is None and report_text: baseline_validation = assess_reviewer_baseline_validation_proof( report_text=report_text, @@ -1750,6 +1870,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "queue-status report violates loaded workflow proof gates (#339)" ) downgrade_reasons.extend(queue_status_report.get("reasons", [])) + if not inventory_worktree.get("proven"): + downgrade_reasons.append( + "inventory pagination or worktree fields inconsistent (#293)" + ) + downgrade_reasons.extend(inventory_worktree.get("reasons", [])) if not baseline_validation.get("proven"): downgrade_reasons.append( "reviewer baseline validation proof missing or failed (#325)" @@ -1843,6 +1968,10 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "queue_status_violations": list( queue_status_report.get("violations") or [] ), + "inventory_worktree_proven": bool(inventory_worktree.get("proven")), + "inventory_worktree_violations": list( + inventory_worktree.get("reasons") or [] + ), "baseline_validation_proven": bool(baseline_validation.get("proven")), "baseline_validation_violations": list( baseline_validation.get("violations") or [] @@ -3965,6 +4094,168 @@ def assess_reviewer_baseline_validation_proof( # --------------------------------------------------------------------------- +# Partial reconciliation policy (#302) +# --------------------------------------------------------------------------- + +# Durable policy choice for already-landed PR reconciliation when +# ``gitea.pr.close`` is unavailable: Option B, comment-then-stop. +PARTIAL_RECONCILIATION_POLICY = "comment_then_stop" + +PARTIAL_RECONCILIATION_COMMENT_FIELDS = ( + "PR head SHA", + "target branch SHA", + "ancestor proof", + "linked issue status", + "required missing capability", +) + + +def resolve_partial_reconciliation_plan( + *, + ancestor_proven: bool, + capabilities: dict | None, +) -> dict: + """#302: decide reconciliation mutations from proven capabilities. + + Implements the comment-then-stop policy (Option B): with ancestor + proof but no ``close_pr`` capability, a reconciliation comment with + the full proof is posted when ``comment_pr`` is proven, then the + workflow stops for an authorized close. Missing comment capability + degrades to a recovery handoff with no Gitea mutation. Unproven + ancestry blocks every mutation regardless of capability. + + *capabilities* maps ``close_pr``/``comment_pr``/``close_issue``/ + ``comment_issue`` to proven booleans; ``None`` or missing keys fail + closed. + """ + caps = capabilities or {} + if not ancestor_proven: + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "GATE_NOT_PROVEN", + "allowed_mutations": (), + "required_comment_fields": (), + "report_requirements": ( + "state that ancestry was not proven and no Gitea mutation " + "was performed", + ), + "reasons": [ + "ancestor proof missing; no reconciliation mutation may " + "run (#302)" + ], + "safe_next_action": ( + "run the already-landed ancestry check before any " + "reconciliation mutation" + ), + } + + if caps.get("close_pr") is True: + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "FULL_RECONCILE_CLOSE_ALLOWED", + "allowed_mutations": ("close_pr", "comment_pr"), + "required_comment_fields": (), + "report_requirements": ( + "report the PR close result", + ), + "reasons": [], + "safe_next_action": ( + "close the already-landed PR with the proven close_pr " + "capability and report the close result" + ), + } + + if caps.get("comment_pr") is True: + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "PARTIAL_RECONCILE_COMMENT_THEN_STOP", + "allowed_mutations": ("comment_pr",), + "required_comment_fields": PARTIAL_RECONCILIATION_COMMENT_FIELDS, + "report_requirements": ( + "report that the reconciliation comment was posted", + "report the missing close capability that prevented the " + "PR close", + ), + "reasons": [ + "close_pr capability missing; policy is comment-then-stop " + "(#302)" + ], + "safe_next_action": ( + "post one reconciliation comment carrying the ancestor " + "proof, then stop for a human or authorized close" + ), + } + + return { + "policy": PARTIAL_RECONCILIATION_POLICY, + "outcome": "RECOVERY_HANDOFF_ONLY", + "allowed_mutations": (), + "required_comment_fields": (), + "report_requirements": ( + "report that no reconciliation comment was posted and name " + "the missing capability", + "report that no Gitea mutation was performed", + ), + "reasons": [ + "close_pr and comment_pr capabilities missing; recovery " + "handoff only (#302)" + ], + "safe_next_action": ( + "produce a recovery handoff recording the intended comment " + "and required capabilities; perform no Gitea mutation" + ), + } + + +def assess_partial_reconciliation_report( + report_text: str | None, + *, + plan: dict, +) -> dict: + """#302: final reports must explain why comments were or were not posted.""" + text = (report_text or "").lower() + outcome = (plan or {}).get("outcome", "") + reasons: list[str] = [] + + if outcome == "FULL_RECONCILE_CLOSE_ALLOWED": + if "close" not in text or "result" not in text: + reasons.append( + "full reconciliation report must state the PR close " + "result (#302)" + ) + elif outcome == "PARTIAL_RECONCILE_COMMENT_THEN_STOP": + if "comment" not in text: + reasons.append( + "partial reconciliation report must state that the " + "reconciliation comment was posted (#302)" + ) + if "capability" not in text and "close_pr" not in text: + reasons.append( + "partial reconciliation report must name the missing " + "close capability (#302)" + ) + elif outcome == "RECOVERY_HANDOFF_ONLY": + if "comment" not in text or "capability" not in text: + reasons.append( + "recovery handoff report must explain that no comment was " + "posted and name the missing capability (#302)" + ) + if "no gitea mutation" not in text and "no mutation" not in text: + reasons.append( + "recovery handoff report must state that no Gitea " + "mutation was performed (#302)" + ) + + return { + "complete": not reasons, + "block": bool(reasons), + "reasons": reasons, + } + + +# --------------------------------------------------------------------------- + + # Already-landed review gate (#292) # --------------------------------------------------------------------------- @@ -4130,6 +4421,9 @@ def assess_already_landed_report_state( } +# --------------------------------------------------------------------------- + + # --------------------------------------------------------------------------- # Full-suite failure approval gate (#323) # --------------------------------------------------------------------------- @@ -4237,6 +4531,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) # --------------------------------------------------------------------------- @@ -4752,6 +5209,20 @@ def assess_validation_worktree_edit_report(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_inventory_worktree_report(report_text, **kwargs): + """#293: prove inventory pagination and consistent worktree reporting.""" + from reviewer_inventory_worktree import assess_inventory_worktree_report as _assess + + return _assess(report_text, **kwargs) + + +def assess_infra_stop_handoff_report(report_text, **kwargs): + """#289: repair handoff purity when infra_stop blocks reviewer capability.""" + from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report as _assess + + return _assess(report_text, **kwargs) + + def assess_mutation_categories_report(report_text, **kwargs): """#319: require precise mutation categories in reviewer controller handoffs.""" from reviewer_mutation_categories import assess_mutation_categories_report as _assess diff --git a/reviewer_infra_stop_handoff.py b/reviewer_infra_stop_handoff.py new file mode 100644 index 0000000..f39147d --- /dev/null +++ b/reviewer_infra_stop_handoff.py @@ -0,0 +1,162 @@ +"""Infra-stop repair handoff verifier for blocked reviewer workflows (#289).""" + +from __future__ import annotations + +import re +from typing import Any + +from capability_stop_terminal import ( + TERMINAL_REPORT_HEADING, + assess_capability_stop_report, +) + +_REPAIR_HANDOFF_RE = re.compile( + r"repair handoff|control-checkout repair mode", + re.IGNORECASE, +) +_CONTROL_REPAIR_RE = re.compile(r"control-checkout repair mode", re.IGNORECASE) +_PR_QUEUE_ADVANCE_RE = re.compile( + r"(?:selected pr|pr #\d+ (?:to review|selected)|eligible pr|" + r"next (?:eligible )?pr(?: to review)?|oldest eligible pr|pinned (?:review )?head)", + re.IGNORECASE, +) +_REVIEW_MUTATION_RE = re.compile( + r"(?:gitea_review_pr|gitea_merge_pr|request_changes|submitted\s+approve|" + r"merge result|review decision\s*:\s*approve)", + re.IGNORECASE, +) +_BACKGROUND_TOOL_RE = re.compile(r"\b(?:schedule|manage_task)\b", re.IGNORECASE) +_PR_REVIEW_REPORT_RE = re.compile( + r"(?:review summary|merge recommendation|validation result\s*:\s*pass|" + r"queue status report with selected pr)", + re.IGNORECASE, +) + +_DIAGNOSTIC_FIELDS = { + "mcp process root": re.compile(r"mcp process root\s*:", re.IGNORECASE), + "inspected git root": re.compile(r"inspected git root\s*:", re.IGNORECASE), + "conflict marker path": re.compile( + r"(?:conflict marker path|exact conflict marker path)\s*:", + re.IGNORECASE, + ), + "merge/rebase control path": re.compile( + r"(?:merge/rebase control path|exact merge/rebase control path)\s*:", + re.IGNORECASE, + ), + "safe next repair action": re.compile( + r"safe next (?:repair )?action\s*:", + re.IGNORECASE, + ), +} + + +def assess_infra_stop_handoff_report( + report_text: str, + *, + stop_session: dict | None = None, +) -> dict[str, Any]: + """Validate repair handoff purity when infra_stop blocks reviewer capability (#289).""" + text = report_text or "" + session = dict(stop_session or {}) + reasons: list[str] = [] + + infra_stop = bool( + session.get("infra_stop") + or session.get("infra_stop_blocked") + or session.get("route_result") == "infra_stop" + ) + if not infra_stop: + return { + "proven": True, + "block": False, + "reasons": [], + "infra_stop": False, + "safe_next_action": "proceed", + } + + lower = text.lower() + repair_handoff = bool( + _REPAIR_HANDOFF_RE.search(text) + or TERMINAL_REPORT_HEADING.lower() in lower + ) + if not repair_handoff: + reasons.append( + "infra_stop requires a repair handoff, not a PR review report" + ) + + if _PR_REVIEW_REPORT_RE.search(text) and not _REPAIR_HANDOFF_RE.search(text): + reasons.append( + "final output must be a repair handoff instead of stale PR review state" + ) + + if _PR_QUEUE_ADVANCE_RE.search(text): + reasons.append( + "infra_stop blocks PR queue advancement; do not select or advance next PR" + ) + + if _REVIEW_MUTATION_RE.search(text): + reasons.append( + "review, approval, request-changes, merge, or comment mutations " + "forbidden while infra_stop is active" + ) + + if session.get("pinned_head_sha") or re.search( + r"pinned (?:review )?head sha\s*:", text, re.IGNORECASE + ): + if session.get("capability_cleared") is not True: + reasons.append( + "cannot pin PR head SHA while review capability never cleared" + ) + + if session.get("main_checkout_diagnostics") and not _CONTROL_REPAIR_RE.search(text): + reasons.append( + "main-checkout diagnostics must be labeled CONTROL-CHECKOUT REPAIR MODE" + ) + + if _BACKGROUND_TOOL_RE.search(text): + reasons.append( + "background schedule/manage_task tools must not be used during " + "blocked infra_stop recovery" + ) + + assessment = session.get("infra_stop_assessment") or {} + if assessment.get("infra_stop") or infra_stop: + missing = [ + label + for label, pattern in _DIAGNOSTIC_FIELDS.items() + if not pattern.search(text) + ] + if missing and session.get("require_infra_diagnostics", True): + reasons.append( + "infra_stop repair handoff missing diagnostic fields: " + + ", ".join(missing) + ) + conflict_file = assessment.get("conflict_file") + if conflict_file and conflict_file.lower() not in lower: + reasons.append( + f"infra_stop assessment conflict file {conflict_file!r} " + "not reflected in repair handoff" + ) + + capability = assess_capability_stop_report( + text, + trust_gate_status=session.get("trust_gate_status"), + capability_denied=True, + ) + if not capability.get("pure"): + reasons.extend(capability.get("reasons") or []) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": list(dict.fromkeys(reasons)), + "infra_stop": True, + "repair_handoff": repair_handoff, + "safe_next_action": ( + "stop PR queue work; emit CONTROL-CHECKOUT REPAIR MODE handoff " + "with infra diagnostics and safe next repair action" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/reviewer_inventory_worktree.py b/reviewer_inventory_worktree.py new file mode 100644 index 0000000..0ca95b3 --- /dev/null +++ b/reviewer_inventory_worktree.py @@ -0,0 +1,308 @@ +"""Reviewer inventory completeness and worktree state verifier (#293).""" + +from __future__ import annotations + +import re +from typing import Any + +_PAGINATION_FINALITY_EVIDENCE = re.compile( + r"pagination_complete\s*:\s*true|inventory_complete\s*:\s*true|" + r"is_final_page\s*:\s*true|pages_fetched|has_more\s*:\s*false|" + r"no next page|final[- ]page|pr_inventory_trust_gate\.status|" + r"total_count\s*:|pagination.*(?:final|complete)", + re.I, +) + +_INCOMPLETE_PAGINATION_PROOF = re.compile( + r"first page only|partial page|page 1 only|truncated|" + r"returned \d+ open prs?(?:\s*$|\s*[,;])", + re.I, +) + +_DEFAULT_PAGE_SIZE_ASSUMPTION = re.compile( + r"(?:less|fewer) than (?:the )?(?:default )?(?:gitea )?page[- ]?(?:size|limit)|" + r"default (?:gitea )?page[- ]?size|under (?:the )?50[- ]?(?:item )?limit|" + r"(?:complete|exhaustive).*(?:default )?page[- ]?size|" + r"page[- ]?size assumption|assumed complete", + re.I, +) + +_ELIGIBILITY_CLAIM_RE = re.compile( + r"\b(?:oldest eligible pr|next eligible pr|next pr to review|" + r"inventory (?:is )?complete|inventory exhaustive|exhaustive inventory)\b", + re.I, +) + +_EXACT_PAGE_LIMIT_RE = re.compile( + r"(?:returned|listed|fetched)\s+(\d+)\s+open prs?|" + r"open pr count\s*:\s*(\d+)|" + r"page[- ]?size\s*(?:=|:)\s*(\d+)", + re.I, +) + +_LEGACY_SCRATCH_FALSE_RE = re.compile( + r"scratch worktree used\s*:\s*false", + re.I, +) + +_BRANCHES_WORKTREE_RE = re.compile(r"\bbranches/", re.I) + +_CONTRADICTORY_HEAD_RE = re.compile( + r"detached head\s*/\s*branch\s+master|" + r"branch\s+master\s*/\s*detached head|" + r"detached head.*branch\s+(?:master|main|dev)\b.*detached|" + r"checkout\s*:\s*detached head\s*/\s*branch", + re.I, +) + +_MAIN_CHECKOUT_BRANCH_RE = re.compile( + r"main checkout branch\s*:\s*(\S+)", + re.I, +) + +_REVIEW_HEAD_STATE_RE = re.compile( + r"review worktree head state\s*:\s*(\S+)", + re.I, +) + +_HANDOFF_SECTION_RE = re.compile( + r"^##\s*Controller Handoff\s*$", + re.I | re.M, +) + + +def _handoff_field_map(report_text: str) -> dict[str, str]: + """Parse ``- Field: value`` lines from the Controller Handoff section.""" + text = report_text or "" + match = _HANDOFF_SECTION_RE.search(text) + if not match: + return {} + section = text[match.end() :] + fields: dict[str, str] = {} + for line in section.splitlines(): + stripped = line.strip().lstrip("-*").strip() + if ":" not in stripped: + continue + key, value = stripped.split(":", 1) + fields[key.strip().lower()] = value.strip() + return fields + + +def _truthy_field(value: str) -> bool: + lowered = (value or "").strip().lower() + if not lowered: + return False + if lowered in {"false", "no", "none", "not applicable", "n/a", "—", "-"}: + return False + return True + + +def _field_proves_pagination(value: str) -> bool: + proof = (value or "").strip() + if not proof or proof.lower() in {"none", "n/a", "not applicable", "unknown"}: + return False + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(proof): + return False + if _INCOMPLETE_PAGINATION_PROOF.search(proof): + return False + return bool(_PAGINATION_FINALITY_EVIDENCE.search(proof)) + + +def _pagination_proven(text: str, session: dict, *, pagination_field: str = "") -> bool: + if session.get("pagination_complete") or session.get("inventory_complete"): + return True + session_proof = session.get("inventory_pagination_proof") or "" + if _field_proves_pagination(str(session_proof)): + return True + if _field_proves_pagination(pagination_field): + return True + if _PAGINATION_FINALITY_EVIDENCE.search(text): + return True + return False + + +def _exact_page_limit_unproven( + text: str, session: dict, *, pagination_proven: bool = False +) -> bool: + """True when a full page was returned without final-page proof.""" + if pagination_proven: + return False + requested = session.get("requested_page_size") + returned = session.get("returned_page_size") + if isinstance(requested, int) and isinstance(returned, int): + if returned >= requested > 0: + return True + for match in _EXACT_PAGE_LIMIT_RE.finditer(text): + count = next((g for g in match.groups() if g), None) + if not count: + continue + try: + n = int(count) + except ValueError: + continue + if n in {10, 20, 50}: + return True + return False + + +def assess_inventory_worktree_report( + report_text: str, + *, + inventory_session: dict | None = None, +) -> dict[str, Any]: + """Validate PR inventory pagination proof and worktree field consistency (#293).""" + text = report_text or "" + session = dict(inventory_session or {}) + reasons: list[str] = [] + + fields = _handoff_field_map(text) + pagination_proof_field = fields.get("inventory pagination proof", "") + + pagination_proven = _pagination_proven( + text, session, pagination_field=pagination_proof_field + ) + + if _ELIGIBILITY_CLAIM_RE.search(text): + if not pagination_proven: + reasons.append( + "oldest/next eligible PR or inventory-complete claim requires " + "final-page/no-next-page/pagination_complete proof" + ) + + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(text): + if not pagination_proven: + reasons.append( + "inventory pagination assumed from default page size without " + "final-page/no-next-page/total-count/traversal proof" + ) + + if pagination_proof_field: + if _DEFAULT_PAGE_SIZE_ASSUMPTION.search(pagination_proof_field): + reasons.append( + "Inventory pagination proof field relies on page-size assumption" + ) + elif _INCOMPLETE_PAGINATION_PROOF.search(pagination_proof_field): + reasons.append( + "Inventory pagination proof field does not prove final page" + ) + elif not _field_proves_pagination(pagination_proof_field): + if _ELIGIBILITY_CLAIM_RE.search(text) or _EXACT_PAGE_LIMIT_RE.search(text): + reasons.append( + "Inventory pagination proof field missing final-page metadata" + ) + + if _exact_page_limit_unproven(text, session, pagination_proven=pagination_proven): + reasons.append( + "exactly-full first page returned without final-page or " + "no-next-page proof" + ) + + review_worktree_used = fields.get("review worktree used", "") + review_worktree_path = fields.get("review worktree path", "") + scratch_used = fields.get("scratch worktree used", "") + inside_branches = fields.get("review worktree inside branches:", "") or fields.get( + "review worktree inside branches", "" + ) + + branches_path_in_text = bool( + _BRANCHES_WORKTREE_RE.search(review_worktree_path) + or _BRANCHES_WORKTREE_RE.search(text) + ) + + if branches_path_in_text: + if review_worktree_used and not _truthy_field(review_worktree_used): + reasons.append( + "reports using a branches/ review worktree must set " + "Review worktree used: true" + ) + if scratch_used and re.search(r"\bfalse\b", scratch_used, re.I): + reasons.append( + "Scratch worktree used: false rejected when a branches/ " + "review worktree was created or used" + ) + if _LEGACY_SCRATCH_FALSE_RE.search(text) and not _truthy_field( + review_worktree_used + ): + reasons.append( + "legacy Scratch worktree used: false contradicts branches/ " + "review worktree usage" + ) + + if review_worktree_path and _truthy_field(review_worktree_used): + if "branches/" not in review_worktree_path.replace("\\", "/").lower(): + reasons.append( + "Review worktree path must be under branches/ when " + "Review worktree used is true" + ) + if inside_branches and re.search(r"\bfalse\b", inside_branches, re.I): + reasons.append( + "Review worktree inside branches must be true when path is " + "under branches/" + ) + + main_branch = ( + fields.get("main checkout branch", "") + or _MAIN_CHECKOUT_BRANCH_RE.search(text).group(1) + if _MAIN_CHECKOUT_BRANCH_RE.search(text) + else "" + ) + head_state = ( + fields.get("review worktree head state", "") + or ( + _REVIEW_HEAD_STATE_RE.search(text).group(1) + if _REVIEW_HEAD_STATE_RE.search(text) + else "" + ) + ) + if main_branch and head_state: + main_norm = main_branch.strip().lower() + head_norm = head_state.strip().lower() + if head_norm in {"branch", "on branch"} and main_norm in { + "master", + "main", + "dev", + }: + if "detached" in text.lower() and "review worktree" in text.lower(): + reasons.append( + "review worktree HEAD state must not reuse main checkout " + "branch wording" + ) + + if _CONTRADICTORY_HEAD_RE.search(text): + reasons.append( + "contradictory checkout wording such as 'Detached HEAD / Branch master'" + ) + + if review_worktree_used and _truthy_field(review_worktree_used): + if not review_worktree_path or review_worktree_path.lower() in { + "none", + "n/a", + "not applicable", + }: + reasons.append( + "Review worktree used: true requires Review worktree path" + ) + if not head_state or head_state.lower() in { + "none", + "n/a", + "not applicable", + "unknown", + }: + reasons.append( + "Review worktree used: true requires Review worktree HEAD state" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "reasons": list(dict.fromkeys(reasons)), + "pagination_proven": pagination_proven, + "branches_worktree_used": branches_path_in_text, + "safe_next_action": ( + "prove inventory pagination with final-page metadata; align " + "Review worktree used/path/HEAD state with branches/ usage" + if reasons + else "proceed" + ), + } \ No newline at end of file diff --git a/role_session_router.py b/role_session_router.py index 1d0c125..5456fba 100644 --- a/role_session_router.py +++ b/role_session_router.py @@ -103,6 +103,9 @@ TASK_REQUIRED_ROLE = { "reconcile_already_landed_pr": "reconciler", "reconcile_already_landed": "reconciler", "reconcile-landed-pr": "reconciler", + # #309: reconciler tasks close already-landed PRs/issues only. + "reconcile_close_landed_pr": "reconciler", + "reconcile_close_landed_issue": "reconciler", } WRONG_ROLE_REVIEWER_MSG = ( @@ -110,7 +113,8 @@ WRONG_ROLE_REVIEWER_MSG = ( ) WRONG_ROLE_RECONCILER_MSG = ( - "Wrong role/session for reconciler task. Launch reconciler MCP namespace." + "Wrong role/session for reconciler task. Launch a reconciler-capable " + "MCP namespace/profile with exact close capability." ) _session_last_route: dict | None = None @@ -209,12 +213,6 @@ def route_task_session( return result if required_role == "reconciler": - message = WRONG_ROLE_RECONCILER_MSG - if active_role_kind == "reconciler": - message = ( - "Reconciler task requires gitea.pr.close on a dedicated " - "reconciler profile." - ) result = { "task_type": task_type, "required_role": required_role, @@ -223,9 +221,9 @@ def route_task_session( "route_result": ROUTE_WRONG_ROLE, "downstream_allowed": False, "reasons": [ - message, - "Reconciler tasks cannot run in author or reviewer sessions.", - "Static-profile mode does not permit in-place role switching.", + 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, diff --git a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md index 80a7c4b..10ba60d 100644 --- a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md +++ b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md @@ -248,6 +248,28 @@ Post a reconciliation comment only if: If comment capability is missing, record the intended comment in the final handoff only. +## 12A. Partial reconciliation policy (#302) + +The durable policy when `close_pr` capability is missing is +**comment-then-stop** (`resolve_partial_reconciliation_plan` in +`review_proofs.py` enforces it): + +* `close_pr` proven → full reconciliation: close the PR and report the + close result (`FULL_RECONCILE_CLOSE_ALLOWED`). +* `close_pr` missing, `comment_pr` proven → post exactly one + reconciliation comment carrying PR head SHA, target branch SHA, + ancestor proof, linked issue status, and the required missing + capability, then stop for a human or authorized close + (`PARTIAL_RECONCILE_COMMENT_THEN_STOP`). +* `comment_pr` also missing → no Gitea mutation; produce a recovery + handoff recording the intended comment and required capabilities + (`RECOVERY_HANDOFF_ONLY`). +* Ancestry not proven → no mutation regardless of capability + (`GATE_NOT_PROVEN`). + +Final reports must explain why the comment was or was not posted and +name the missing capability (`assess_partial_reconciliation_report`). + ## 13. PR close rules Close a PR only if: diff --git a/task_capability_map.py b/task_capability_map.py index f0fc7dd..9a366c0 100644 --- a/task_capability_map.py +++ b/task_capability_map.py @@ -100,6 +100,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = { "permission": "gitea.pr.close", "role": "reconciler", }, + # #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", + }, "post_heartbeat": { "permission": "gitea.issue.comment", "role": "author", diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 609906a..6d191c0 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -78,6 +78,12 @@ def test_reconcile_landed_workflow_contract(): assert "canonical: true" in text assert "Do not review or merge normal PRs" in text assert "Already-landed proof" in text + # Issue #302: partial reconciliation policy must stay documented. + assert "## 12A. Partial reconciliation policy (#302)" in text + assert "comment-then-stop" in text + assert "PARTIAL_RECONCILE_COMMENT_THEN_STOP" in text + assert "RECOVERY_HANDOFF_ONLY" in text + assert "resolve_partial_reconciliation_plan" in text def test_create_issue_workflow_contract(): @@ -166,6 +172,18 @@ def test_validation_worktree_edit_verifier_exported(): assert callable(assess_validation_worktree_edit_report) +def test_inventory_worktree_verifier_exported(): + from review_proofs import assess_inventory_worktree_report + + assert callable(assess_inventory_worktree_report) + + +def test_infra_stop_handoff_verifier_exported(): + from review_proofs import assess_infra_stop_handoff_report + + assert callable(assess_infra_stop_handoff_report) + + def test_mutation_categories_verifier_exported(): from review_proofs import assess_mutation_categories_report 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 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() diff --git a/tests/test_reconciliation_handoff.py b/tests/test_reconciliation_handoff.py new file mode 100644 index 0000000..baa2787 --- /dev/null +++ b/tests/test_reconciliation_handoff.py @@ -0,0 +1,162 @@ +"""Tests for the dedicated reconciliation controller-handoff schema (#303). + +Already-landed PR reconciliation runs must emit the reconciliation +schema, not stale author/reviewer handoff fields; observed git ref +mutations (fetch) must be reported, and legacy fields fail validation. +""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from review_proofs import assess_reconciliation_handoff # noqa: E402 + + +def _good_handoff(**overrides): + fields = { + "Task": "Reconcile already-landed open PRs", + "Repo": "prgs/Scaled-Tech-Consulting/Gitea-Tools", + "Role/profile": "prgs-reconciler", + "Identity": "jcwalker3", + "Selected PR": "278", + "PR live state": "open", + "Candidate head SHA": "2a544b7", + "Target branch": "master", + "Target branch SHA": "fa0cb12", + "Ancestor proof": "candidate head is ancestor of target branch SHA", + "Linked issue": "263", + "Linked issue live status": "open (fetched live this session)", + "Eligibility class": "ALREADY_LANDED_RECONCILE_REQUIRED", + "Capabilities proven": "gitea.pr.comment", + "Missing capabilities": "gitea.pr.close", + "PR comments posted": "1 (PR #278 reconciliation notice)", + "Issue comments posted": "none", + "PRs closed": "none", + "Issues closed": "none", + "Git ref mutations": "git fetch prgs master", + "Worktree mutations": "none", + "MCP/Gitea mutations": "PR comment on #278", + "External-state mutations": "none", + "Read-only diagnostics": "gitea_view_pr 278, gitea_view_issue 263", + "Blocker": "PR close capability missing", + "Safe next action": "operator closes PR #278 or grants close capability", + "No review/merge confirmation": "no review or merge mutations performed", + } + fields.update(overrides) + lines = ["Controller Handoff"] + lines += [f"- {k}: {v}" for k, v in fields.items() if v is not None] + return "\n".join(lines) + "\n" + + +class TestReconciliationSchemaPasses(unittest.TestCase): + def test_blocked_reconciliation_passes(self): + result = assess_reconciliation_handoff( + _good_handoff(), + observed_commands=["git fetch prgs master"], + ) + self.assertTrue(result["complete"]) + self.assertEqual(result["reasons"], []) + + def test_successful_pr_close_passes(self): + report = _good_handoff(**{ + "Missing capabilities": "none", + "PRs closed": "PR #278", + "Blocker": "none", + "Safe next action": "none; reconciliation complete", + }) + result = assess_reconciliation_handoff( + report, observed_commands=["git fetch prgs master"] + ) + self.assertTrue(result["complete"]) + + def test_comment_only_reconciliation_passes(self): + report = _good_handoff(**{ + "Git ref mutations": "none", + }) + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertTrue(result["complete"]) + + def test_noop_already_closed_passes(self): + report = _good_handoff(**{ + "PR live state": "closed", + "PR comments posted": "none", + "MCP/Gitea mutations": "none", + "Git ref mutations": "none", + "Blocker": "none", + "Safe next action": "none; PR already closed", + }) + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertTrue(result["complete"]) + + +class TestSchemaViolationsFail(unittest.TestCase): + def test_missing_linked_issue_proof_fails(self): + report = _good_handoff(**{"Linked issue live status": None}) + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + self.assertTrue( + any("linked issue live status" in r.lower() + for r in result["reasons"]) + ) + + def test_wrong_eligibility_class_fails(self): + report = _good_handoff(**{"Eligibility class": "REVIEW_ELIGIBLE"}) + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + + def test_missing_handoff_section_fails(self): + result = assess_reconciliation_handoff( + "no handoff here", observed_commands=[] + ) + self.assertFalse(result["complete"]) + + +class TestStaleFieldsFail(unittest.TestCase): + def test_pr_number_opened_fails(self): + report = _good_handoff() + "- PR number opened: 278\n" + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + self.assertTrue( + any("pr number opened" in r.lower() for r in result["reasons"]) + ) + + def test_pinned_reviewed_head_fails(self): + report = _good_handoff() + "- Pinned reviewed head: 2a544b7\n" + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + + def test_scratch_worktree_used_fails(self): + report = _good_handoff() + "- Scratch worktree used: False\n" + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + + def test_workspace_mutations_fails(self): + report = _good_handoff() + "- Workspace mutations: None\n" + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + + def test_author_lock_fields_fail(self): + report = ( + _good_handoff() + + "- Issue lock proof: locked before diff\n" + + "- Claim/comment status: claimed\n" + ) + result = assess_reconciliation_handoff(report, observed_commands=[]) + self.assertFalse(result["complete"]) + + +class TestObservedFetchMustBeReported(unittest.TestCase): + def test_fetch_with_ref_mutations_none_fails(self): + report = _good_handoff(**{"Git ref mutations": "none"}) + result = assess_reconciliation_handoff( + report, observed_commands=["git fetch prgs master"] + ) + self.assertFalse(result["complete"]) + self.assertTrue( + any("git ref mutation" in r.lower() for r in result["reasons"]) + ) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_review_proofs.py b/tests/test_review_proofs.py index 0d42359..0558517 100644 --- a/tests/test_review_proofs.py +++ b/tests/test_review_proofs.py @@ -26,6 +26,8 @@ from review_proofs import ( # noqa: E402 assess_already_landed_report_state, assess_already_landed_review_gate, assess_author_pr_report, + assess_partial_reconciliation_report, + resolve_partial_reconciliation_plan, assess_validation_environment_proof, assess_full_suite_failure_approval_gate, assess_queue_ordering_proof, @@ -2641,6 +2643,133 @@ class TestReviewerBaselineValidationProof(unittest.TestCase): self.assertFalse(report["baseline_validation_proven"]) +class TestPartialReconciliationPlan(unittest.TestCase): + """Issue #302: policy when PR-close capability is missing (Option B).""" + + def _plan(self, **overrides): + kwargs = { + "ancestor_proven": True, + "capabilities": { + "close_pr": False, + "comment_pr": True, + "close_issue": True, + "comment_issue": True, + }, + } + kwargs.update(overrides) + return resolve_partial_reconciliation_plan(**kwargs) + + def test_close_capability_present_allows_full_reconcile(self): + plan = self._plan(capabilities={ + "close_pr": True, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }) + self.assertEqual(plan["outcome"], "FULL_RECONCILE_CLOSE_ALLOWED") + self.assertIn("close_pr", plan["allowed_mutations"]) + + def test_close_missing_with_comment_posts_comment_then_stops(self): + plan = self._plan() + self.assertEqual( + plan["outcome"], "PARTIAL_RECONCILE_COMMENT_THEN_STOP") + self.assertEqual(plan["allowed_mutations"], ("comment_pr",)) + for field in ( + "PR head SHA", + "target branch SHA", + "ancestor proof", + "linked issue status", + "required missing capability", + ): + self.assertIn(field, plan["required_comment_fields"]) + + def test_comment_capability_missing_produces_handoff_only(self): + plan = self._plan(capabilities={ + "close_pr": False, "comment_pr": False, + "close_issue": True, "comment_issue": True, + }) + self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") + self.assertEqual(plan["allowed_mutations"], ()) + + def test_all_capabilities_missing_produces_handoff_only(self): + plan = self._plan(capabilities={ + "close_pr": False, "comment_pr": False, + "close_issue": False, "comment_issue": False, + }) + self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") + self.assertEqual(plan["allowed_mutations"], ()) + + def test_unproven_ancestry_blocks_all_mutations(self): + plan = self._plan(ancestor_proven=False, capabilities={ + "close_pr": True, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }) + self.assertEqual(plan["outcome"], "GATE_NOT_PROVEN") + self.assertEqual(plan["allowed_mutations"], ()) + + def test_missing_capability_dict_fails_closed(self): + plan = self._plan(capabilities=None) + self.assertEqual(plan["outcome"], "RECOVERY_HANDOFF_ONLY") + self.assertEqual(plan["allowed_mutations"], ()) + + +class TestPartialReconciliationReport(unittest.TestCase): + """Issue #302: reports must explain why comments were or were not posted.""" + + def _plan(self, outcome): + capabilities = { + "FULL_RECONCILE_CLOSE_ALLOWED": { + "close_pr": True, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }, + "PARTIAL_RECONCILE_COMMENT_THEN_STOP": { + "close_pr": False, "comment_pr": True, + "close_issue": True, "comment_issue": True, + }, + "RECOVERY_HANDOFF_ONLY": { + "close_pr": False, "comment_pr": False, + "close_issue": False, "comment_issue": False, + }, + }[outcome] + return resolve_partial_reconciliation_plan( + ancestor_proven=True, capabilities=capabilities) + + def test_partial_report_requires_comment_and_missing_capability(self): + plan = self._plan("PARTIAL_RECONCILE_COMMENT_THEN_STOP") + good = ( + "Reconciliation comment posted on PR #278 with ancestor proof. " + "PR close skipped: close capability gitea.pr.close missing; " + "stopped for authorized close." + ) + result = assess_partial_reconciliation_report(good, plan=plan) + self.assertTrue(result["complete"]) + + bad = "Stopped. Nothing done." + result = assess_partial_reconciliation_report(bad, plan=plan) + self.assertTrue(result["block"]) + + def test_handoff_only_report_requires_no_comment_explanation(self): + plan = self._plan("RECOVERY_HANDOFF_ONLY") + good = ( + "No reconciliation comment posted: comment capability missing. " + "No Gitea mutation performed; recovery handoff produced." + ) + result = assess_partial_reconciliation_report(good, plan=plan) + self.assertTrue(result["complete"]) + + bad = "Recovery handoff produced." + result = assess_partial_reconciliation_report(bad, plan=plan) + self.assertTrue(result["block"]) + + def test_full_reconcile_report_requires_close_result(self): + plan = self._plan("FULL_RECONCILE_CLOSE_ALLOWED") + good = "PR close result: closed PR #278 after ancestor proof." + result = assess_partial_reconciliation_report(good, plan=plan) + self.assertTrue(result["complete"]) + + bad = "Reconciliation done." + result = assess_partial_reconciliation_report(bad, plan=plan) + self.assertTrue(result["block"]) + + class TestAlreadyLandedReviewGate(unittest.TestCase): """Issue #292: PR head already on target blocks approval and merge.""" diff --git a/tests/test_reviewer_infra_stop_handoff.py b/tests/test_reviewer_infra_stop_handoff.py new file mode 100644 index 0000000..6d49dc3 --- /dev/null +++ b/tests/test_reviewer_infra_stop_handoff.py @@ -0,0 +1,106 @@ +"""Tests for infra-stop repair handoff verifier (#289).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from capability_stop_terminal import TERMINAL_REPORT_HEADING +from reviewer_infra_stop_handoff import assess_infra_stop_handoff_report # noqa: E402 + + +def _repair_handoff() -> str: + return "\n".join([ + TERMINAL_REPORT_HEADING, + "Repair handoff for infra_stop blocked reviewer workflow.", + "CONTROL-CHECKOUT REPAIR MODE", + "MCP process root: /Users/dev/Gitea-Tools", + "Inspected git root: /Users/dev/Gitea-Tools", + "Conflict marker path: gitea_mcp_server.py", + "Merge/rebase control path: none", + "Safe next repair action: resolve conflict markers and restart MCP", + "infra_stop: true", + ]) + + +class TestInfraStopHandoff(unittest.TestCase): + def test_repair_handoff_passes(self): + result = assess_infra_stop_handoff_report( + _repair_handoff(), + stop_session={ + "infra_stop": True, + "infra_stop_assessment": { + "infra_stop": True, + "conflict_file": "gitea_mcp_server.py", + }, + }, + ) + self.assertTrue(result["proven"], result["reasons"]) + + def test_next_pr_selection_blocks(self): + report = "\n".join([ + TERMINAL_REPORT_HEADING, + "Next eligible PR to review: PR #276", + "Pinned review head SHA: abc123", + ]) + result = assess_infra_stop_handoff_report( + report, + stop_session={"infra_stop": True, "require_infra_diagnostics": False}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("queue" in r.lower() for r in result["reasons"])) + + def test_missing_repair_handoff_blocks(self): + result = assess_infra_stop_handoff_report( + "Validation result: pass\nReview summary for PR #276", + stop_session={"infra_stop": True}, + ) + self.assertFalse(result["proven"]) + + def test_main_checkout_diagnostics_without_label_blocks(self): + report = "\n".join([ + TERMINAL_REPORT_HEADING, + "Repair handoff", + "Ran git status in main checkout for MCP diagnostics.", + ]) + result = assess_infra_stop_handoff_report( + report, + stop_session={ + "infra_stop": True, + "main_checkout_diagnostics": True, + "require_infra_diagnostics": False, + }, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("control-checkout" in r.lower() for r in result["reasons"])) + + def test_background_scheduling_blocks(self): + report = "\n".join([ + TERMINAL_REPORT_HEADING, + "Repair handoff", + "Used schedule/manage_task while waiting for MCP recovery.", + ]) + result = assess_infra_stop_handoff_report( + report, + stop_session={"infra_stop": True, "require_infra_diagnostics": False}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("schedule" in r.lower() for r in result["reasons"])) + + def test_non_infra_stop_skips(self): + result = assess_infra_stop_handoff_report( + "Selected PR #1 for review", + stop_session={"infra_stop": False}, + ) + self.assertTrue(result["proven"]) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_infra_stop_handoff_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file diff --git a/tests/test_reviewer_inventory_worktree.py b/tests/test_reviewer_inventory_worktree.py new file mode 100644 index 0000000..15f77f7 --- /dev/null +++ b/tests/test_reviewer_inventory_worktree.py @@ -0,0 +1,126 @@ +"""Tests for reviewer inventory/worktree verifier (#293).""" +import sys +import unittest +from pathlib import Path + +sys.path.insert(0, str(Path(__file__).resolve().parent.parent)) + +from reviewer_inventory_worktree import assess_inventory_worktree_report # noqa: E402 + + +def _good_handoff() -> str: + return "\n".join([ + "## Controller Handoff", + "- Selected PR: 280", + "- Inventory pagination proof: is_final_page: true, has_more: false, total_count: 6", + "- Review worktree used: true", + "- Review worktree path: branches/review-pr280-feat-issue-275-claim", + "- Review worktree inside branches: true", + "- Review worktree HEAD state: detached", + "- Main checkout branch: master", + "- Current status: reviewed PR #280", + ]) + + +class TestInventoryPaginationProof(unittest.TestCase): + def test_no_next_page_proof_passes(self): + result = assess_inventory_worktree_report(_good_handoff()) + self.assertTrue(result["proven"], result["reasons"]) + self.assertTrue(result["pagination_proven"]) + + def test_rejects_partial_first_page_eligibility_claim(self): + report = "\n".join([ + "Oldest eligible PR is #280.", + "gitea_list_prs returned 10 open PRs on the first page.", + "## Controller Handoff", + "- Inventory pagination proof: first page only", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + self.assertTrue(any("eligible" in r.lower() for r in result["reasons"])) + + def test_rejects_default_page_size_assumption(self): + report = ( + "Inventory exhaustive because fewer than the default Gitea page size " + "were returned." + ) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("page size" in r.lower() for r in result["reasons"])) + + def test_rejects_exactly_full_first_page_without_finality(self): + report = "\n".join([ + "Next eligible PR: #290 based on complete inventory.", + "gitea_list_prs returned 50 open PRs.", + "## Controller Handoff", + "- Inventory pagination proof: returned 50 open PRs", + ]) + result = assess_inventory_worktree_report( + report, + inventory_session={"requested_page_size": 50, "returned_page_size": 50}, + ) + self.assertFalse(result["proven"]) + self.assertTrue(any("full first page" in r.lower() for r in result["reasons"])) + + def test_allows_exact_page_with_session_pagination_complete(self): + report = "Oldest eligible PR: #12 after queue inventory." + result = assess_inventory_worktree_report( + report, + inventory_session={"pagination_complete": True}, + ) + self.assertTrue(result["proven"], result["reasons"]) + + +class TestWorktreeConsistency(unittest.TestCase): + def test_branches_worktree_requires_review_worktree_used_true(self): + report = "\n".join([ + "## Controller Handoff", + "- Inventory pagination proof: is_final_page: true", + "- Review worktree used: false", + "- Review worktree path: branches/review-pr280-feat", + "- Scratch worktree used: false", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + joined = " ".join(result["reasons"]).lower() + self.assertIn("review worktree used", joined) + self.assertIn("scratch worktree", joined) + + def test_rejects_contradictory_detached_and_branch_wording(self): + report = "\n".join([ + "Checkout: Detached HEAD / Branch master", + "## Controller Handoff", + "- Inventory pagination proof: is_final_page: true", + "- Review worktree used: true", + "- Review worktree path: branches/review-pr280-feat", + "- Review worktree HEAD state: detached", + "- Main checkout branch: master", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue( + any("contradictory" in r.lower() for r in result["reasons"]) + ) + + def test_requires_head_state_when_review_worktree_used(self): + report = "\n".join([ + "## Controller Handoff", + "- Inventory pagination proof: pagination_complete: true", + "- Review worktree used: true", + "- Review worktree path: branches/review-pr280-feat", + ]) + result = assess_inventory_worktree_report(report) + self.assertFalse(result["proven"]) + self.assertTrue(any("head state" in r.lower() for r in result["reasons"])) + + +class TestExport(unittest.TestCase): + def test_review_proofs_reexport(self): + from review_proofs import assess_inventory_worktree_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file