diff --git a/final_report_validator.py b/final_report_validator.py index bff9467..1c82ab3 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -88,7 +88,8 @@ _ALREADY_LANDED_RE = re.compile( re.IGNORECASE, ) _ELIGIBLE_REVIEW_RE = re.compile( - r"eligible for (?:review|merge)|ready for (?:review|merge)|next eligible pr", + r"eligible for (?:review|merge)|ready for (?:review|merge)|" + r"(?:next|oldest)\s+eligible\s+pr", re.IGNORECASE, ) _REVIEWED_LANDED_RE = re.compile( diff --git a/review_proofs.py b/review_proofs.py index 43dfbd3..b667665 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -1505,6 +1505,17 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "reasons": [], "violations": [], } + if report_text: + already_landed_classification = assess_already_landed_classification_report( + report_text + ) + else: + already_landed_classification = { + "proven": True, + "block": False, + "reasons": [], + "violations": [], + } contamination_status = contamination.get("status", "unknown") checkout_proven = bool(checkout_proof.get("proven")) @@ -1659,6 +1670,11 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "reviewer baseline validation proof missing or failed (#325)" ) downgrade_reasons.extend(baseline_validation.get("reasons", [])) + if not already_landed_classification.get("proven"): + downgrade_reasons.append( + "already-landed classification wording missing or invalid (#295)" + ) + downgrade_reasons.extend(already_landed_classification.get("reasons", [])) merge_allowed = ( identity_eligible @@ -1751,6 +1767,12 @@ def build_final_report(checkout_proof, inventory, validation, contamination, "baseline_validation_violations": list( baseline_validation.get("violations") or [] ), + "already_landed_classification_proven": bool( + already_landed_classification.get("proven") + ), + "already_landed_classification_violations": list( + already_landed_classification.get("reasons") or [] + ), } @@ -4460,6 +4482,15 @@ def assess_merge_simulation_report(report_text, **kwargs): return _assess(report_text, **kwargs) +def assess_already_landed_classification_report(report_text, **kwargs): + """#295: already-landed PRs are reconciliation-only, not review eligible.""" + from reviewer_already_landed_classification import ( + assess_already_landed_classification_report as _assess, + ) + + return _assess(report_text, **kwargs) + + def assess_prior_blocker_skip_proof(report_text, **kwargs): """#318: require live blocker proof before skipping earlier open PRs.""" from reviewer_blocker_skip import assess_prior_blocker_skip_proof as _assess diff --git a/reviewer_already_landed_classification.py b/reviewer_already_landed_classification.py new file mode 100644 index 0000000..b5c5f10 --- /dev/null +++ b/reviewer_already_landed_classification.py @@ -0,0 +1,143 @@ +"""Already-landed PR classification verifier for reviewer reports (#295).""" + +from __future__ import annotations + +import re +from typing import Any + +ALREADY_LANDED_ELIGIBILITY_CLASS = "ALREADY_LANDED_RECONCILE_REQUIRED" + +_LANDED_CONTEXT_RE = re.compile( + r"already[- ]landed|ancestor proof\s*:\s*passed|" + r"eligibility class\s*:\s*already_landed", + re.IGNORECASE, +) +_ELIGIBILITY_CLASS_LINE_RE = re.compile( + r"eligibility class\s*:\s*(.+)$", + re.IGNORECASE | re.MULTILINE, +) +_INELIGIBLE_SELECTION_RE = re.compile( + r"\b(?:next|oldest)\s+eligible\s+pr\b", + re.IGNORECASE, +) +_REVIEW_ELIGIBLE_RE = re.compile( + r"\beligible for (?:review|merge)\b|\bready for (?:review|merge)\b", + re.IGNORECASE, +) +_PINNED_REVIEWED_HEAD_RE = re.compile( + r"\bpinned reviewed head\s*:", + re.IGNORECASE, +) +_CANDIDATE_HEAD_RE = re.compile( + r"\bcandidate head sha\s*:", + re.IGNORECASE, +) +_REVIEWED_HEAD_NONE_RE = re.compile( + r"\breviewed head sha\s*:\s*none\b", + re.IGNORECASE, +) +_VALIDATION_PROOF_RE = re.compile( + r"(?:validation passed|pytest.*passed|diff review passed|" + r"validated on pinned head)", + re.IGNORECASE, +) +_QUEUE_BLOCKED_RE = re.compile( + r"queue blocked by already[- ]landed|" + r"reconciliation(?:-only)?\s+(?:next step|required)", + re.IGNORECASE, +) + + +def _landed_context(report_text: str, eligibility_class: str | None) -> bool: + if (eligibility_class or "").strip().upper() == ALREADY_LANDED_ELIGIBILITY_CLASS: + return True + return bool(_LANDED_CONTEXT_RE.search(report_text or "")) + + +def assess_already_landed_classification_report( + report_text: str, + *, + eligibility_class: str | None = None, + selected_pr_already_landed: bool | None = None, +) -> dict[str, Any]: + """#295: already-landed PRs are reconciliation-only, not review eligible.""" + text = report_text or "" + reasons: list[str] = [] + + landed = _landed_context(text, eligibility_class) + if selected_pr_already_landed is True: + landed = True + if not landed: + return { + "proven": True, + "block": False, + "landed_context": False, + "reasons": [], + "safe_next_action": "proceed", + } + + if _INELIGIBLE_SELECTION_RE.search(text): + reasons.append( + "already-landed PR must not be described as oldest/next eligible PR; " + "use 'Oldest open PR requiring action' and " + f"'Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}'" + ) + + if _REVIEW_ELIGIBLE_RE.search(text): + reasons.append( + "already-landed PR must not be described as eligible for review or merge" + ) + + class_match = _ELIGIBILITY_CLASS_LINE_RE.search(text) + if class_match: + declared = class_match.group(1).strip().upper() + if declared != ALREADY_LANDED_ELIGIBILITY_CLASS: + reasons.append( + "already-landed PR eligibility class must be " + f"{ALREADY_LANDED_ELIGIBILITY_CLASS}" + ) + elif selected_pr_already_landed is True: + reasons.append( + f"already-landed selected PR must declare Eligibility class: " + f"{ALREADY_LANDED_ELIGIBILITY_CLASS}" + ) + + if _PINNED_REVIEWED_HEAD_RE.search(text): + if not _VALIDATION_PROOF_RE.search(text): + reasons.append( + "already-landed pre-review report must use Candidate head SHA, " + "not Pinned reviewed head, unless validation and diff review passed" + ) + + if selected_pr_already_landed is True and not _CANDIDATE_HEAD_RE.search(text): + if _PINNED_REVIEWED_HEAD_RE.search(text) or "head sha" in text.lower(): + reasons.append( + "already-landed ancestry check must report Candidate head SHA" + ) + + if selected_pr_already_landed is True and not _REVIEWED_HEAD_NONE_RE.search(text): + if _PINNED_REVIEWED_HEAD_RE.search(text) and not _VALIDATION_PROOF_RE.search(text): + reasons.append( + "already-landed report must state 'Reviewed head SHA: none' " + "when validation did not run" + ) + + if selected_pr_already_landed is True and not _QUEUE_BLOCKED_RE.search(text): + reasons.append( + "already-landed queue blocker must state reconciliation requirement " + "or queue blocked wording" + ) + + proven = not reasons + return { + "proven": proven, + "block": not proven, + "landed_context": True, + "reasons": reasons, + "safe_next_action": ( + "classify as ALREADY_LANDED_RECONCILE_REQUIRED, use Candidate head SHA, " + "and stop normal review" + if not proven + else "proceed" + ), + } \ No newline at end of file diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py index f8e8848..0a0124d 100644 --- a/tests/test_final_report_validator.py +++ b/tests/test_final_report_validator.py @@ -190,6 +190,20 @@ class TestReviewPrRules(unittest.TestCase): ) ) + def test_already_landed_oldest_eligible_pr_blocks(self): + report = ( + _review_handoff() + + "\nAlready landed.\nOldest eligible PR: PR #278." + ) + result = assess_final_report_validator(report, "review_pr") + self.assertTrue(result["blocked"]) + self.assertTrue( + any( + f["rule_id"] == "reviewer.already_landed_eligible_wording" + for f in result["findings"] + ) + ) + def test_git_fetch_must_not_be_readonly_only(self): report = ( _review_handoff() diff --git a/tests/test_llm_workflow_split.py b/tests/test_llm_workflow_split.py index 0c2b89a..f27c5b5 100644 --- a/tests/test_llm_workflow_split.py +++ b/tests/test_llm_workflow_split.py @@ -158,3 +158,9 @@ def test_worktree_ownership_verifier_exported(): from review_proofs import assess_worktree_ownership_report assert callable(assess_worktree_ownership_report) + + +def test_already_landed_classification_verifier_exported(): + from review_proofs import assess_already_landed_classification_report + + assert callable(assess_already_landed_classification_report) diff --git a/tests/test_reviewer_already_landed_classification.py b/tests/test_reviewer_already_landed_classification.py new file mode 100644 index 0000000..7506eeb --- /dev/null +++ b/tests/test_reviewer_already_landed_classification.py @@ -0,0 +1,81 @@ +import unittest + +from reviewer_already_landed_classification import ( # noqa: E402 + ALREADY_LANDED_ELIGIBILITY_CLASS, + assess_already_landed_classification_report, +) + + +class TestAlreadyLandedClassification(unittest.TestCase): + def test_non_landed_report_passes(self): + result = assess_already_landed_classification_report( + "Selected the next eligible PR: PR #286." + ) + self.assertTrue(result["proven"]) + self.assertFalse(result["landed_context"]) + + def test_oldest_eligible_pr_on_landed_blocks(self): + result = assess_already_landed_classification_report( + "PR is already landed.\n" + "Oldest eligible PR: PR #278." + ) + self.assertFalse(result["proven"]) + self.assertTrue(result["block"]) + + def test_next_eligible_pr_on_landed_blocks(self): + result = assess_already_landed_classification_report( + "Ancestor proof: passed\n" + "Selected the next eligible PR: PR #278." + ) + self.assertFalse(result["proven"]) + + def test_eligible_for_review_on_landed_blocks(self): + result = assess_already_landed_classification_report( + "Already landed on master and eligible for review next." + ) + self.assertFalse(result["proven"]) + + def test_proper_landed_wording_passes(self): + result = assess_already_landed_classification_report( + "Oldest open PR requiring action: PR #278\n" + f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n" + "Candidate head SHA: abc123\n" + "Reviewed head SHA: none\n" + "Queue blocked by already-landed PR requiring reconciliation", + selected_pr_already_landed=True, + ) + self.assertTrue(result["proven"]) + + def test_pinned_reviewed_head_without_validation_blocks(self): + result = assess_already_landed_classification_report( + "Already landed.\n" + "Pinned reviewed head: abc123def456", + selected_pr_already_landed=True, + ) + self.assertFalse(result["proven"]) + + def test_wrong_eligibility_class_blocks(self): + result = assess_already_landed_classification_report( + "Already landed.\n" + "Eligibility class: REVIEW_ELIGIBLE", + selected_pr_already_landed=True, + ) + self.assertFalse(result["proven"]) + + def test_missing_queue_blocker_blocks(self): + result = assess_already_landed_classification_report( + f"Eligibility class: {ALREADY_LANDED_ELIGIBILITY_CLASS}\n" + "Candidate head SHA: abc123\n" + "Reviewed head SHA: none", + selected_pr_already_landed=True, + ) + self.assertFalse(result["proven"]) + + def test_exported_from_review_proofs(self): + from review_proofs import assess_already_landed_classification_report as exported + + self.assertTrue(callable(exported)) + + +if __name__ == "__main__": + unittest.main() \ No newline at end of file