diff --git a/final_report_validator.py b/final_report_validator.py index 492c15b..a807ff9 100644 --- a/final_report_validator.py +++ b/final_report_validator.py @@ -961,6 +961,69 @@ def _rule_reconcile_linked_issue_live( return [] +_PR_CLOSE_NEGATIVE = frozenset({"", "none", "n/a", "na", "0", "no", "not closed", "not performed"}) +_ANCESTOR_AFFIRMATIVE_RE = re.compile( + r"ancestor|passed|true|verified|confirmed", re.IGNORECASE +) + + +def _reconciler_pr_close_performed(fields: dict[str, str], lock: dict) -> bool: + """True when the report/session indicates a reconciler PR close happened.""" + if lock.get("pr_closed") is True: + return True + value = (fields.get("prs closed", "") or "").strip().lower() + if value in _PR_CLOSE_NEGATIVE: + return False + # A closed PR is reported by number (e.g. "#99") or an affirmative result. + return bool(re.search(r"#\s*\d+|\b(?:closed|success|done)\b", value)) + + +def _rule_reconcile_close_proof( + report_text: str, + *, + reconciler_close_lock: dict | None = None, +) -> list[dict[str, str]]: + """#306: a reconciler PR close must carry exact proof fields. + + Read-only/comment-only reconciliations are untouched. Once a PR close is + reported (via the ``PRs closed`` field or a session close lock), the + handoff must prove the close capability, ancestor landing, PR close + result, and the linked-issue result — narrative alone fails closed. + """ + fields = _handoff_fields(report_text) + lock = reconciler_close_lock or {} + if not _reconciler_pr_close_performed(fields, lock): + return [] + + missing: list[str] = [] + capabilities = fields.get("capabilities proven", "") + if "gitea.pr.close" not in capabilities.lower(): + missing.append("close capability proof (gitea.pr.close)") + ancestor = fields.get("ancestor proof", "") + if not _ANCESTOR_AFFIRMATIVE_RE.search(ancestor): + missing.append("ancestor proof") + prs_closed = (fields.get("prs closed", "") or "").strip().lower() + if prs_closed in _PR_CLOSE_NEGATIVE and lock.get("pr_closed") is True: + missing.append("PR close result") + linked = fields.get("linked issue live status", "") or fields.get("issues closed", "") + if not linked.strip(): + missing.append("linked issue result") + + if not missing: + return [] + return [ + validator_finding( + "reconcile.close_proof_fields", + "block", + "Reconciler close proof", + "reconciler PR close reported without required proof field(s): " + + ", ".join(missing), + "include close capability proof, ancestor proof, PR close result, " + "and linked issue result in the handoff", + ) + ] + + def _rule_reconcile_pagination_proof(report_text: str) -> list[dict[str, str]]: text = report_text or "" if not _INVENTORY_COMPLETE_RE.search(text): @@ -1253,6 +1316,7 @@ _RULES_BY_TASK: dict[str, list[Callable[..., list[dict[str, str]]]]] = { _rule_reconcile_stale_author_fields, _rule_reconcile_eligible_reviewed, _rule_reconcile_linked_issue_live, + _rule_reconcile_close_proof, _rule_reconcile_pagination_proof, _rule_reviewer_git_fetch_readonly, _rule_reviewer_legacy_workspace_mutations, @@ -1352,6 +1416,7 @@ def assess_final_report_validator( issue_filing_lock: dict | None = None, session_pr_opened: bool = False, validation_session: dict | None = None, + reconciler_close_lock: dict | None = None, ) -> dict[str, Any]: """Validate final-report text against task-specific proof rules (#327). @@ -1408,6 +1473,7 @@ def assess_final_report_validator( "local_edits": local_edits, "session_pr_opened": session_pr_opened, "validation_session": validation_session, + "reconciler_close_lock": reconciler_close_lock, } for rule in _RULES_BY_TASK.get(normalized_kind, ()): diff --git a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md index a9f4cd1..c58f3fe 100644 --- a/skills/llm-project-workflow/workflows/reconcile-landed-pr.md +++ b/skills/llm-project-workflow/workflows/reconcile-landed-pr.md @@ -375,6 +375,24 @@ Include: * confirmation that no normal review, approval, request-changes, or merge was performed +## 18A. Reconciler close proof is enforced (#306) + +When a reconciler run closes a PR, the final-report validator +(`final_report_validator` rule `reconcile.close_proof_fields`) **blocks** the +handoff unless it carries all four close proofs. The prompt is guidance; the +MCP validator is the authority. + +A close is detected from `PRs closed: #` (or a session close lock). Once a +close is reported, the handoff must include: + +* `Capabilities proven:` naming `gitea.pr.close` — the exact close capability +* `Ancestor proof:` — the landed/ancestor proof for the closed PR +* `PRs closed:` — the PR close result (the closed PR number) +* `Linked issue live status:` (or `Issues closed:`) — the linked-issue result + +Comment-only and blocked reconciliations (no PR close) are unaffected: the rule +returns no finding when nothing was closed. + ## 19. Local artifact and report consistency rule Do not create local walkthrough, notes, markdown, JSON, or report artifacts diff --git a/tests/test_final_report_validator.py b/tests/test_final_report_validator.py index 88c5b23..70121f4 100644 --- a/tests/test_final_report_validator.py +++ b/tests/test_final_report_validator.py @@ -498,6 +498,81 @@ class TestCanonicalReconcileSchema(unittest.TestCase): ) +class TestReconcilerCloseProof(unittest.TestCase): + """Issue #306: a reconciler PR close must carry its proof fields.""" + + def _closed_report(self, **kwargs): + # A report that claims PR #99 was closed via the reconciler path. + report = ( + _reconcile_handoff(**kwargs) + .replace( + "- Capabilities proven: gitea.read, gitea.pr.comment", + "- Capabilities proven: gitea.read, gitea.pr.comment, gitea.pr.close", + ) + .replace("- Missing capabilities: gitea.pr.close", "- Missing capabilities: none") + .replace("- PRs closed: none", "- PRs closed: #99") + .replace( + "- Blocker: missing gitea.pr.close capability", "- Blocker: none" + ) + ) + return report + + def test_full_proof_close_passes(self): + result = assess_final_report_validator( + self._closed_report(), "reconcile_already_landed" + ) + self.assertEqual(result["grade"], "A") + self.assertFalse( + any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) + ) + + def test_close_without_capability_proof_blocks(self): + # Claims PRs closed: #99 but never proves gitea.pr.close capability. + report = _reconcile_handoff().replace("- PRs closed: none", "- PRs closed: #99") + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) + ) + + def test_close_without_ancestor_proof_blocks(self): + report = self._closed_report(drop=("Ancestor proof",)) + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) + ) + + def test_close_without_linked_issue_result_blocks(self): + report = self._closed_report(drop=("Linked issue live status", "Issues closed")) + result = assess_final_report_validator(report, "reconcile_already_landed") + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) + ) + + def test_close_lock_requires_proof_even_if_text_says_none(self): + # Session lock proves a PR was closed; the report omits close proof. + result = assess_final_report_validator( + _reconcile_handoff(), + "reconcile_already_landed", + reconciler_close_lock={"pr_closed": True}, + ) + self.assertTrue(result["blocked"]) + self.assertTrue( + any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) + ) + + def test_comment_only_reconcile_needs_no_close_proof(self): + result = assess_final_report_validator( + _reconcile_handoff(), "reconcile_already_landed" + ) + self.assertEqual(result["grade"], "A") + self.assertFalse( + any(f["rule_id"] == "reconcile.close_proof_fields" for f in result["findings"]) + ) + + class TestEntryPoint(unittest.TestCase): def test_unknown_task_kind_blocks(self): result = assess_final_report_validator("report", "unknown_mode")