feat: reject contradictory reviewer handoff ledgers (Closes #501)

Add reviewer_handoff_consistency validation for narrative vs mutation ledger contradictions, wire it into review_pr final-report rules, and document the blocked-review handoff template. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
2026-07-08 03:33:31 -04:00
parent 48cf3a334b
commit 429faccd08
6 changed files with 385 additions and 0 deletions
@@ -0,0 +1,120 @@
+"""Tests for reviewer handoff consistency validation (#501)."""
+import sys
+import unittest
+from pathlib import Path
+
+sys.path.insert(0, str(Path(__file__).resolve().parent.parent))
+
+import reviewer_handoff_consistency  # noqa: E402
+from final_report_validator import assess_final_report_validator  # noqa: E402
+
+
+def _base_handoff(**overrides):
+    lines = [
+        "## Controller Handoff",
+        "",
+        "- Task: review PR #472",
+        "- Review decision: request_changes",
+        "- Review mutations: submitted request_changes review",
+        "- Merge mutations: none",
+        "- MCP/Gitea mutations: reviewer lease acquired",
+    ]
+    text = "\n".join(lines)
+    for key, value in overrides.items():
+        token = f"- {key}:"
+        if token in text:
+            before, _, after = text.partition(token)
+            rest = after.split("\n", 1)
+            suffix = rest[1] if len(rest) > 1 else ""
+            text = f"{before}{token} {value}" + (f"\n{suffix}" if suffix else "")
+        else:
+            text += f"\n- {key}: {value}"
+    return text
+
+
+class TestReviewerHandoffConsistency(unittest.TestCase):
+    def test_consistent_handoff_passes(self):
+        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(
+            _base_handoff()
+        )
+        self.assertTrue(result["proven"], result["reasons"])
+
+    def test_claimed_merge_missing_from_ledger(self):
+        report = _base_handoff(
+            **{
+                "Current status": "Merged PR #411 earlier; reviewing PR #472",
+                "Merge mutations": "none",
+                "MCP/Gitea mutations": "reviewer lease acquired",
+            }
+        )
+        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("merge" in r.lower() for r in result["reasons"]))
+
+    def test_terminal_budget_consumed_without_merge_in_ledger(self):
+        report = _base_handoff(
+            **{
+                "Current status": "Terminal mutation budget already consumed by merge",
+                "Merge mutations": "none",
+                "MCP/Gitea mutations": "reviewer lease acquired",
+            }
+        )
+        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("terminal mutation budget" in r for r in result["reasons"]))
+
+    def test_review_blocked_but_final_decision_marked(self):
+        report = _base_handoff(
+            **{
+                "Current status": "gitea_submit_pr_review blocked",
+                "Review mutations": "none",
+            }
+        )
+        report += "\nServer-side final decision marked via gitea_mark_final_review_decision."
+        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("blocked" in r and "final decision" in r for r in result["reasons"]))
+
+    def test_lease_without_review_decision(self):
+        report = _base_handoff(**{"Review decision": "none"})
+        report += "\nReviewer lease acquired for PR #472."
+        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("Review decision" in r for r in result["reasons"]))
+
+    def test_rejected_mutation_without_proof(self):
+        report = _base_handoff(
+            **{
+                "Review mutations": "none",
+                "MCP/Gitea mutations": "none",
+            }
+        )
+        report += "\nReview submission blocked before mutation."
+        result = reviewer_handoff_consistency.assess_reviewer_handoff_consistency(report)
+        self.assertFalse(result["proven"])
+        self.assertTrue(any("proof fields" in r for r in result["reasons"]))
+
+    def test_blocked_template_includes_required_fields(self):
+        template = reviewer_handoff_consistency.render_blocked_review_handoff_template()
+        for field in reviewer_handoff_consistency._BLOCKED_REVIEW_PROOF_FIELDS:
+            self.assertIn(field, template.lower())
+
+    def test_final_report_validator_integration(self):
+        report = _base_handoff(
+            **{
+                "Current status": "Merged PR #411; terminal mutation budget consumed",
+                "Merge mutations": "none",
+            }
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.handoff_consistency"
+                for f in result["findings"]
+            ),
+            result["findings"],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()