diff --git a/review_proofs.py b/review_proofs.py index 9c572c3..8dc2bff 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -4752,6 +4752,13 @@ def assess_validation_worktree_edit_report(report_text, **kwargs): 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/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 609906a..642ffb7 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -166,6 +166,12 @@ def test_validation_worktree_edit_verifier_exported(): assert callable(assess_validation_worktree_edit_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_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