diff --git a/review_proofs.py b/review_proofs.py index 70214d5..17f1527 100644 --- a/review_proofs.py +++ b/review_proofs.py @@ -4531,6 +4531,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) # --------------------------------------------------------------------------- diff --git a/role_session_router.py b/role_session_router.py index a7db807..ff75025 100644 --- a/role_session_router.py +++ b/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 = ( diff --git a/task_capability_map.py b/task_capability_map.py index 2b794e7..8453255 100644 --- a/task_capability_map.py +++ b/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", diff --git a/tests/test_reconciler_close_gate.py b/tests/test_reconciler_close_gate.py new file mode 100644 index 0000000..bae8438 --- /dev/null +++ b/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()