Merge pull request 'feat: enforce reconciler PR-close proof fields in final reports (Closes #306)' (#423) from feat/issue-306-reconciler-close-proof into master

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, ()):

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: #<n>` (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

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")