feat: classify already-landed PRs as reconciliation-only (Closes #295)
Add reviewer_already_landed_classification verifier to block oldest/next eligible PR wording, review-eligible claims, and pinned reviewed head usage when a PR is already landed. Wire the check into build_final_report and tighten final_report_validator regex coverage. Co-Authored-By: Claude Opus 4.8 (1M context) <[email protected]>
This commit is contained in:
Claude Opus 4.8
@@ -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(
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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"
|
||||
),
|
||||
}
|
||||
@@ -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()
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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()
|
||||
Reference in New Issue
Block a user