Merge pull request 'feat: require infra_stop repair handoff for blocked reviewers (Closes #289)' (#366) from feat/issue-289-infra-stop-handoff into master

2026-07-07 08:39:35 -05:00
parent b4c79fdbbc 93230457d8
commit a17e8561d0
4 changed files with 281 additions and 0 deletions
@@ -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

@@ -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()