final_report_validator.py

@@ -87,14 +87,34 @@ _ALREADY_LANDED_RE = re.compile(
     r"already[- ]landed|ancestor proof\s*:\s*passed|eligibility class\s*:\s*already_landed",
     re.IGNORECASE,
 )
+_ALREADY_LANDED_GATE_FIRED_RE = re.compile(
+    r"\b(?:fired|passed|true|yes|already_landed_reconcile_required)\b",
+    re.IGNORECASE,
+)
 _ELIGIBLE_REVIEW_RE = re.compile(
     r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr",
     re.IGNORECASE,
 )
+_APPROVED_STATE_RE = re.compile(r"\bapproved?\b", re.IGNORECASE)
+_MERGED_STATE_RE = re.compile(r"\bmerged\b", re.IGNORECASE)
+_READY_TO_MERGE_STATE_RE = re.compile(r"\bready_to_merge\b|\bready to merge\b", re.IGNORECASE)
+_NEGATED_STATE_RE = re.compile(
+    r"\b(?:not|no|none|n/a|never)\b|request_changes|not attempted",
+    re.IGNORECASE,
+)
 _REVIEWED_LANDED_RE = re.compile(
     r"(?:reviewed|approved).{0,40}already[- ]landed|already[- ]landed.{0,40}(?:reviewed|eligible)",
     re.IGNORECASE | re.DOTALL,
 )
+_TARGET_BRANCH_UP_TO_DATE_RE = re.compile(
+    r"\btarget branch (?:is )?up[- ]to[- ]date\b",
+    re.IGNORECASE,
+)
+_TARGET_BRANCH_SHA_RE = re.compile(
+    r"target branch sha\s*:\s*[0-9a-f]{40}",
+    re.IGNORECASE,
+)
+_FULL_SHA_RE = re.compile(r"\b[0-9a-f]{40}\b", re.IGNORECASE)
 _RECONCILE_STALE_FIELDS = (
     "pr number opened",
     "pinned reviewed head",
@@ -190,6 +210,54 @@ def _handoff_fields(report_text: str) -> dict[str, str]:
     return fields
 
 
+def _already_landed_gate_fired(report_text: str) -> bool:
+    text = report_text or ""
+    fields = _handoff_fields(text)
+
+    gate_value = fields.get("already-landed gate", "")
+    if gate_value and _ALREADY_LANDED_GATE_FIRED_RE.search(gate_value):
+        return True
+
+    ancestor_value = fields.get("ancestor proof", "")
+    if re.search(r"\bpassed\b", ancestor_value, re.IGNORECASE):
+        return True
+
+    eligibility_value = fields.get("eligibility class", "")
+    if "already_landed" in eligibility_value.lower():
+        return True
+
+    if re.search(
+        r"\bpr\b.{0,60}\balready[- ]landed\b|\balready[- ]landed\b.{0,60}\bpr\b",
+        text,
+        re.IGNORECASE | re.DOTALL,
+    ):
+        return True
+
+    return bool(_ALREADY_LANDED_RE.search(text))
+
+
+def _positive_review_state_terms(fields: dict[str, str]) -> list[str]:
+    terms: list[str] = []
+    for key, value in fields.items():
+        lowered_key = key.lower()
+        lowered_value = value.lower()
+        if _NEGATED_STATE_RE.search(value):
+            continue
+        if lowered_key == "review decision" and _APPROVED_STATE_RE.search(value):
+            terms.append("APPROVED")
+        elif lowered_key == "merge result" and _MERGED_STATE_RE.search(value):
+            terms.append("MERGED")
+        elif _READY_TO_MERGE_STATE_RE.search(value):
+            terms.append("READY_TO_MERGE")
+        elif lowered_value in {"approved", "approve"}:
+            terms.append("APPROVED")
+        elif lowered_value == "merged":
+            terms.append("MERGED")
+        elif lowered_value == "ready_to_merge":
+            terms.append("READY_TO_MERGE")
+    return sorted(set(terms))
+
+
 def _rule_shared_controller_handoff(
     report_text: str,
     task_kind: str,
@@ -492,7 +560,7 @@ def _rule_reviewer_main_checkout_path(report_text: str) -> list[dict[str, str]]:
 
 def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, str]]:
     text = report_text or ""
-    if not _ALREADY_LANDED_RE.search(text):
+    if not _already_landed_gate_fired(text):
         return []
     if not _ELIGIBLE_REVIEW_RE.search(text):
         return []
@@ -507,6 +575,65 @@ def _rule_reviewer_already_landed_eligible(report_text: str) -> list[dict[str, s
     ]
 
 
+def _rule_reviewer_already_landed_state(report_text: str) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _already_landed_gate_fired(text):
+        return []
+
+    terms = _positive_review_state_terms(_handoff_fields(text))
+    if not terms:
+        return []
+
+    return [
+        validator_finding(
+            "reviewer.already_landed_review_state",
+            "block",
+            "Review decision",
+            "already-landed gate fired but final report claims "
+            f"{', '.join(terms)} state",
+            "stop normal review/merge; classify as ALREADY_LANDED_RECONCILE_REQUIRED",
+        )
+    ]
+
+
+def _rule_reviewer_target_branch_freshness(
+    report_text: str,
+    *,
+    action_log: list[dict] | None = None,
+) -> list[dict[str, str]]:
+    text = report_text or ""
+    if not _TARGET_BRANCH_UP_TO_DATE_RE.search(text):
+        return []
+
+    fields = _handoff_fields(text)
+    fetch_reported = bool(_GIT_FETCH_RE.search(text)) or any(
+        _GIT_FETCH_RE.search(str(e.get("command") or e.get("action") or ""))
+        for e in (action_log or [])
+    )
+    target_sha_reported = bool(_TARGET_BRANCH_SHA_RE.search(text)) or any(
+        "target branch" in key and "sha" in key and _FULL_SHA_RE.search(value)
+        for key, value in fields.items()
+    )
+    if fetch_reported and target_sha_reported:
+        return []
+
+    missing = []
+    if not fetch_reported:
+        missing.append("fresh git fetch")
+    if not target_sha_reported:
+        missing.append("target branch SHA")
+    return [
+        validator_finding(
+            "reviewer.target_branch_freshness_proof",
+            "block",
+            "Target branch SHA",
+            "target branch up-to-date claim lacks " + " and ".join(missing),
+            "fetch the target branch and report the exact target branch SHA before "
+            "claiming it is up to date",
+        )
+    ]
+
+
 def _rule_reconcile_controller_handoff(report_text: str) -> list[dict[str, str]]:
     from review_proofs import _handoff_section_lines
 
@@ -706,6 +833,8 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = {
         _rule_reviewer_main_checkout_baseline,
         _rule_reviewer_main_checkout_path,
         _rule_reviewer_already_landed_eligible,
+        _rule_reviewer_already_landed_state,
+        _rule_reviewer_target_branch_freshness,
         _rule_reviewer_mutation_ledger,
         _rule_reviewer_review_mutation,
     ],
@@ -875,4 +1004,4 @@ def assess_final_report_validator(
         "safe_next_action": _primary_safe_next_action(findings),
         "complete": grade == "A",
         "handoff_heading": HANDOFF_HEADING,
-    }
+    }

tests/test_final_report_validator.py

@@ -25,11 +25,15 @@ def _review_handoff(**overrides):
         "- Files changed: review_proofs.py",
         "- Validation: pytest tests/test_review_proofs.py -q in branches/review-203",
         "- Mutations: review only",
-        "- Workspace mutations: none",
         "- File edits by reviewer: none",
         "- Worktree/index mutations: none",
         "- Git ref mutations: git fetch prgs master",
         "- MCP/Gitea mutations: review submitted",
+        "- Review mutations: submitted request_changes review",
+        "- Merge mutations: none",
+        "- Cleanup mutations: none",
+        "- External-state mutations: none",
+        "- Read-only diagnostics: issue and PR metadata inspected",
         "- Current status: PR open",
         "- Blockers: none",
         "- Next: await author",
@@ -190,11 +194,63 @@ class TestReviewPrRules(unittest.TestCase):
             )
         )
 
+    def test_already_landed_gate_blocks_review_terminal_states(self):
+        cases = {
+            "APPROVED": ("- Review decision: request_changes", "- Review decision: APPROVED"),
+            "MERGED": ("- Merge result: not attempted", "- Merge result: MERGED"),
+            "READY_TO_MERGE": ("- Current status: PR open", "- Current status: READY_TO_MERGE"),
+        }
+        for state, (old, new) in cases.items():
+            with self.subTest(state=state):
+                report = (
+                    _review_handoff()
+                    .replace("- Reviewer eligibility: eligible", "- Reviewer eligibility: blocked")
+                    .replace(old, new)
+                    + "\n- Already-landed gate: fired"
+                    + "\n- Ancestor proof: passed"
+                    + "\n- Eligibility class: ALREADY_LANDED_RECONCILE_REQUIRED"
+                )
+                result = assess_final_report_validator(report, "review_pr")
+                self.assertTrue(result["blocked"])
+                self.assertTrue(
+                    any(
+                        f["rule_id"] == "reviewer.already_landed_review_state"
+                        for f in result["findings"]
+                    )
+                )
+
+    def test_target_branch_up_to_date_requires_fetch_and_sha(self):
+        report = (
+            _review_handoff()
+            .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
+            + "\nTarget branch up to date."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertTrue(result["blocked"])
+        self.assertTrue(
+            any(
+                f["rule_id"] == "reviewer.target_branch_freshness_proof"
+                for f in result["findings"]
+            )
+        )
+
+    def test_target_branch_up_to_date_with_fetch_and_sha_passes(self):
+        report = (
+            _review_handoff()
+            + "\n- Target branch SHA: 0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+            + "\nTarget branch up to date."
+        )
+        result = assess_final_report_validator(report, "review_pr")
+        self.assertEqual(result["grade"], "A")
+
     def test_git_fetch_must_not_be_readonly_only(self):
         report = (
             _review_handoff()
             .replace("- Git ref mutations: git fetch prgs master", "- Git ref mutations: none")
-            + "\n- Read-only diagnostics: git fetch prgs master"
+            .replace(
+                "- Read-only diagnostics: issue and PR metadata inspected",
+                "- Read-only diagnostics: git fetch prgs master",
+            )
         )
         result = assess_final_report_validator(
             report,
@@ -270,4 +326,4 @@ class TestEntryPoint(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()