review_proofs.py

@@ -4237,6 +4237,169 @@ def assess_full_suite_failure_approval_gate(
     }
 
 
+# ---------------------------------------------------------------------------
+# Reconciler close gate (#309)
+# ---------------------------------------------------------------------------
+
+RECONCILER_CLOSE_REPORT_FIELDS = (
+    "identity/profile",
+    "close capability proof",
+    "PR live state",
+    "candidate head SHA",
+    "target branch SHA",
+    "ancestor proof",
+    "linked issue status",
+    "PR close result",
+    "issue close result",
+    "no review/merge confirmation",
+)
+
+
+def assess_reconciler_close_gate(
+    *,
+    pr_number: int | None,
+    pr_state: str | None,
+    candidate_head_sha: str | None,
+    live_head_sha: str | None,
+    target_branch: str | None,
+    target_branch_sha: str | None,
+    head_is_ancestor_of_target: bool | None,
+    close_pr_capability: bool,
+    close_issue_capability: bool = False,
+    linked_issue_state: str | None = None,
+    issue_resolved_by_landing: bool = False,
+) -> dict:
+    """#309: gate the dedicated reconciler close path for landed PRs.
+
+    The reconciler path may close a PR only with exact ``gitea.pr.close``
+    capability and full already-landed proof: live open PR, pinned head,
+    freshly fetched target branch SHA, and ancestry of the head in the
+    target. Non-landed PRs are never closable here, and the gate never
+    grants review, approval, request-changes, or merge.
+
+    Linked-issue closure: skipped when the issue is already closed;
+    allowed only when the issue is open, resolved by the landed commits,
+    and exact ``gitea.issue.close`` capability is proven.
+    """
+    reasons: list[str] = []
+
+    if not isinstance(pr_number, int) or pr_number <= 0:
+        reasons.append("PR number missing or invalid (#309)")
+    if (pr_state or "").strip().lower() != "open":
+        reasons.append(
+            "PR is not live-verified open at mutation time; re-fetch the "
+            "PR before any reconciler close (#309)"
+        )
+    if not _FULL_SHA.match((candidate_head_sha or "").strip()):
+        reasons.append(
+            "candidate head SHA is not a full 40-hex commit SHA (#309)"
+        )
+    if live_head_sha is not None and candidate_head_sha and (
+        live_head_sha.strip() != candidate_head_sha.strip()
+    ):
+        reasons.append(
+            "PR head changed since the ancestry proof was pinned; re-run "
+            "the already-landed check (#309)"
+        )
+    if not (target_branch or "").strip():
+        reasons.append("target branch missing (#309)")
+    if not _FULL_SHA.match((target_branch_sha or "").strip()):
+        reasons.append(
+            "target branch SHA missing or not a full 40-hex SHA; fetch the "
+            "target branch freshly before the ancestry check (#309)"
+        )
+    if head_is_ancestor_of_target is None:
+        reasons.append(
+            "ancestry of the PR head against the target branch was not "
+            "checked (#309)"
+        )
+
+    never_allowed = {
+        "review_allowed": False,
+        "approve_allowed": False,
+        "request_changes_allowed": False,
+        "merge_allowed": False,
+    }
+
+    if reasons:
+        return {
+            "outcome": "GATE_NOT_PROVEN",
+            "pr_close_allowed": False,
+            "issue_close_allowed": False,
+            "reasons": reasons,
+            "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
+            "safe_next_action": (
+                "re-fetch the PR and target branch, pin SHAs, run the "
+                "ancestry check, then re-run this gate"
+            ),
+            **never_allowed,
+        }
+
+    if head_is_ancestor_of_target is False:
+        return {
+            "outcome": "NOT_LANDED_CLOSE_BLOCKED",
+            "pr_close_allowed": False,
+            "issue_close_allowed": False,
+            "reasons": [
+                f"PR #{pr_number} head is not an ancestor of "
+                f"{target_branch}; non-landed PRs cannot be closed through "
+                "the reconciler path (#309)"
+            ],
+            "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
+            "safe_next_action": (
+                "route the PR through the normal review workflow; the "
+                "reconciler path only closes already-landed PRs"
+            ),
+            **never_allowed,
+        }
+
+    if close_pr_capability is not True:
+        return {
+            "outcome": "RECOVERY_HANDOFF_REQUIRED",
+            "pr_close_allowed": False,
+            "issue_close_allowed": False,
+            "reasons": [
+                "exact gitea.pr.close capability not proven; produce a "
+                "recovery handoff instead of closing (#309)"
+            ],
+            "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
+            "safe_next_action": (
+                "produce a recovery handoff naming the missing "
+                "gitea.pr.close capability and the completed ancestor proof"
+            ),
+            **never_allowed,
+        }
+
+    issue_close_allowed = (
+        (linked_issue_state or "").strip().lower() == "open"
+        and issue_resolved_by_landing is True
+        and close_issue_capability is True
+    )
+    issue_close_reasons = []
+    if (linked_issue_state or "").strip().lower() == "closed":
+        issue_close_reasons.append(
+            "linked issue already closed; no issue close attempted (#309)"
+        )
+    elif not issue_close_allowed:
+        issue_close_reasons.append(
+            "issue close requires an open linked issue resolved by the "
+            "landed commits and exact gitea.issue.close capability (#309)"
+        )
+
+    return {
+        "outcome": "CLOSE_ALLOWED",
+        "pr_close_allowed": True,
+        "issue_close_allowed": issue_close_allowed,
+        "reasons": issue_close_reasons,
+        "required_report_fields": RECONCILER_CLOSE_REPORT_FIELDS,
+        "safe_next_action": (
+            "close the already-landed PR, report the close result, and "
+            "handle the linked issue per the proven capability"
+        ),
+        **never_allowed,
+    }
+
+
 # ---------------------------------------------------------------------------
 # Identity disclosure (#305)
 # ---------------------------------------------------------------------------

role_session_router.py

@@ -90,12 +90,20 @@ TASK_REQUIRED_ROLE = {
     "blind_pr_queue_review": "reviewer",
     "request_changes_pr": "reviewer",
     "approve_pr": "reviewer",
+    # #309: reconciler tasks close already-landed PRs/issues only.
+    "reconcile_close_landed_pr": "reconciler",
+    "reconcile_close_landed_issue": "reconciler",
 }
 
 WRONG_ROLE_REVIEWER_MSG = (
     "Wrong role/session for reviewer task. Launch reviewer MCP namespace."
 )
 
+WRONG_ROLE_RECONCILER_MSG = (
+    "Wrong role/session for reconciler task. Launch a reconciler-capable "
+    "MCP namespace/profile with exact close capability."
+)
+
 _session_last_route: dict | None = None
 
 
@@ -191,6 +199,26 @@ def route_task_session(
         _record_route(result)
         return result
 
+    if required_role == "reconciler":
+        result = {
+            "task_type": task_type,
+            "required_role": required_role,
+            "active_role": active_role_kind,
+            "active_profile": active_profile,
+            "route_result": ROUTE_WRONG_ROLE,
+            "downstream_allowed": False,
+            "reasons": [
+                WRONG_ROLE_RECONCILER_MSG,
+                "Reconciler tasks cannot run in author or reviewer "
+                "sessions without exact close capability.",
+            ],
+            "message": WRONG_ROLE_RECONCILER_MSG,
+            "runtime_switching_supported": runtime_switching_supported,
+            "profile_switch_blocked": not runtime_switching_supported,
+        }
+        _record_route(result)
+        return result
+
     if required_role == "author":
         route = ROUTE_TO_AUTHOR
         message = (

task_capability_map.py

@@ -96,6 +96,16 @@ TASK_CAPABILITY_MAP: dict[str, dict[str, str]] = {
         "permission": "gitea.read",
         "role": "author",
     },
+    # #309: dedicated reconciler path for already-landed open PRs. Exact
+    # close capabilities only — never review/approve/request_changes/merge.
+    "reconcile_close_landed_pr": {
+        "permission": "gitea.pr.close",
+        "role": "reconciler",
+    },
+    "reconcile_close_landed_issue": {
+        "permission": "gitea.issue.close",
+        "role": "reconciler",
+    },
     "post_heartbeat": {
         "permission": "gitea.issue.comment",
         "role": "author",

tests/test_reconciler_close_gate.py

@@ -0,0 +1,188 @@
+"""Issue #309: dedicated reconciler capability to close already-landed PRs.
+
+The reconciler path must prove exact ``gitea.pr.close`` capability, must
+never imply review/approve/request-changes/merge capability, and may close
+a PR only after full already-landed proof.
+"""
+
+import sys
+import unittest
+
+sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
+
+import role_session_router  # noqa: E402
+import task_capability_map  # noqa: E402
+from review_proofs import assess_reconciler_close_gate  # noqa: E402
+
+PINNED = "0fdc8f582026b72a229d59a172c0a63ac4aaeaf9"
+OTHER = "a4060c5de00f2b1c9e88f4f6f0f3f9a7b2c1d0e9"
+
+
+class TestReconcilerCapabilityMap(unittest.TestCase):
+    def test_reconcile_close_pr_task_maps_to_pr_close(self):
+        self.assertEqual(
+            task_capability_map.required_permission(
+                "reconcile_close_landed_pr"),
+            "gitea.pr.close",
+        )
+        self.assertEqual(
+            task_capability_map.required_role("reconcile_close_landed_pr"),
+            "reconciler",
+        )
+
+    def test_reconcile_close_issue_task_maps_to_issue_close(self):
+        self.assertEqual(
+            task_capability_map.required_permission(
+                "reconcile_close_landed_issue"),
+            "gitea.issue.close",
+        )
+        self.assertEqual(
+            task_capability_map.required_role(
+                "reconcile_close_landed_issue"),
+            "reconciler",
+        )
+
+    def test_reconciler_tasks_do_not_grant_review_permissions(self):
+        for task in ("reconcile_close_landed_pr",
+                     "reconcile_close_landed_issue"):
+            permission = task_capability_map.required_permission(task)
+            self.assertNotIn("review", permission)
+            self.assertNotIn("approve", permission)
+            self.assertNotIn("merge", permission)
+            self.assertNotIn("request_changes", permission)
+
+
+class TestReconcilerRouting(unittest.TestCase):
+    def test_reconciler_task_requires_reconciler_role(self):
+        self.assertEqual(
+            role_session_router.required_role_for_task(
+                "reconcile_close_landed_pr"),
+            "reconciler",
+        )
+
+    def test_reconciler_task_blocked_in_author_session(self):
+        result = role_session_router.route_task_session(
+            "reconcile_close_landed_pr",
+            active_profile="prgs-author",
+            active_role_kind="author",
+            allowed_in_current_session=False,
+        )
+        self.assertEqual(result["route_result"], "wrong_role_stop")
+        self.assertFalse(result["downstream_allowed"])
+
+    def test_reconciler_task_allowed_in_matching_session(self):
+        result = role_session_router.route_task_session(
+            "reconcile_close_landed_pr",
+            active_profile="prgs-reconciler",
+            active_role_kind="reconciler",
+            allowed_in_current_session=True,
+        )
+        self.assertEqual(result["route_result"], "allowed_current_session")
+        self.assertTrue(result["downstream_allowed"])
+
+
+class TestReconcilerCloseGate(unittest.TestCase):
+    def _gate(self, **overrides):
+        kwargs = {
+            "pr_number": 278,
+            "pr_state": "open",
+            "candidate_head_sha": PINNED,
+            "live_head_sha": PINNED,
+            "target_branch": "master",
+            "target_branch_sha": OTHER,
+            "head_is_ancestor_of_target": True,
+            "close_pr_capability": True,
+            "close_issue_capability": False,
+            "linked_issue_state": "closed",
+        }
+        kwargs.update(overrides)
+        return assess_reconciler_close_gate(**kwargs)
+
+    def test_already_landed_open_pr_close_allowed(self):
+        result = self._gate()
+        self.assertEqual(result["outcome"], "CLOSE_ALLOWED")
+        self.assertTrue(result["pr_close_allowed"])
+
+    def test_not_landed_pr_cannot_be_closed(self):
+        result = self._gate(head_is_ancestor_of_target=False)
+        self.assertEqual(result["outcome"], "NOT_LANDED_CLOSE_BLOCKED")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_unchecked_ancestry_blocks_close(self):
+        result = self._gate(head_is_ancestor_of_target=None)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_stale_target_branch_sha_blocks_close(self):
+        result = self._gate(target_branch_sha=None)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_closed_pr_state_blocks_close(self):
+        result = self._gate(pr_state="closed")
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_changed_head_blocks_close(self):
+        result = self._gate(live_head_sha=OTHER)
+        self.assertEqual(result["outcome"], "GATE_NOT_PROVEN")
+        self.assertFalse(result["pr_close_allowed"])
+
+    def test_missing_close_capability_produces_recovery_handoff(self):
+        result = self._gate(close_pr_capability=False)
+        self.assertEqual(result["outcome"], "RECOVERY_HANDOFF_REQUIRED")
+        self.assertFalse(result["pr_close_allowed"])
+        self.assertTrue(any(
+            "gitea.pr.close" in reason for reason in result["reasons"]
+        ))
+
+    def test_linked_issue_already_closed_skips_issue_close(self):
+        result = self._gate(linked_issue_state="closed",
+                            close_issue_capability=True)
+        self.assertFalse(result["issue_close_allowed"])
+
+    def test_linked_open_issue_close_requires_capability(self):
+        allowed = self._gate(
+            linked_issue_state="open",
+            issue_resolved_by_landing=True,
+            close_issue_capability=True,
+        )
+        self.assertTrue(allowed["issue_close_allowed"])
+
+        blocked = self._gate(
+            linked_issue_state="open",
+            issue_resolved_by_landing=True,
+            close_issue_capability=False,
+        )
+        self.assertFalse(blocked["issue_close_allowed"])
+
+    def test_gate_never_allows_review_mutations(self):
+        for result in (
+            self._gate(),
+            self._gate(head_is_ancestor_of_target=False),
+            self._gate(close_pr_capability=False),
+        ):
+            self.assertFalse(result["review_allowed"])
+            self.assertFalse(result["approve_allowed"])
+            self.assertFalse(result["request_changes_allowed"])
+            self.assertFalse(result["merge_allowed"])
+
+    def test_report_fields_listed_for_close(self):
+        result = self._gate()
+        for field in (
+            "identity/profile",
+            "close capability proof",
+            "PR live state",
+            "candidate head SHA",
+            "target branch SHA",
+            "ancestor proof",
+            "linked issue status",
+            "PR close result",
+            "issue close result",
+            "no review/merge confirmation",
+        ):
+            self.assertIn(field, result["required_report_fields"])
+
+
+if __name__ == "__main__":
+    unittest.main()