diff --git a/branch_cleanup_guard.py b/branch_cleanup_guard.py index 64db84a..e91ea02 100644 --- a/branch_cleanup_guard.py +++ b/branch_cleanup_guard.py @@ -525,6 +525,90 @@ def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]: } +# Reviewer-lease reclaim is only reachable from a non-live (expired/stale) lease. +_RECLAIMABLE_REVIEWER_STATUSES = _EXPIRED_STATUSES | _STALE_STATUSES + + +def is_active_ownership_status(status: str | None) -> bool: + """True when *status* denotes live/active ownership of a branch (#855). + + Used to decide whether a *competing* active claimant still uses a branch + when weighing an expired reviewer lease for reclaim. Expired, stale, + released, and terminal statuses are not active. + """ + return _norm_str(status).lower() in _ACTIVE_OWNERSHIP_STATUSES + + +def assess_expired_reviewer_lease_reclaim( + *, + role: str, + status: str, + pr_merged: bool | None, + owner_pid_alive: bool | None, + competing_active_claimant: bool | None, +) -> dict[str, Any]: + """Decide, explicitly and fail-closed, whether an expired reviewer lease + may stop protecting an already-merged branch (#855 AC4). + + An expired reviewer lease should not protect a merged branch forever once + its work is done and no live claimant remains. Reclaim is permitted only + when **every** condition below is provably satisfied; any unknown + (``None``) or contrary value keeps the lease protective: + + - the lease is a ``reviewer`` lease (author/merger/controller/reconciler + leases are out of scope and always keep protecting); + - its status is expired or stale (never an active/live lease); + - the PR is proven merged (``pr_merged is True``); + - the lease owner process is proven dead (``owner_pid_alive is False``); + - no competing active claimant uses the branch + (``competing_active_claimant is False``). + + Returns a decision dict with ``reclaim_allowed`` and, when refused, the + fail-closed ``reasons``. The reasons never contain secrets — only the + role, the status, and which condition was unproven. + """ + reasons: list[str] = [] + normalized_role = _norm_str(role).lower() + normalized_status = _norm_str(status).lower() + + if normalized_role != "reviewer": + reasons.append( + f"lease role '{normalized_role or 'unknown'}' is not a reviewer " + "lease; expired-reviewer reclaim does not apply" + ) + if normalized_status not in _RECLAIMABLE_REVIEWER_STATUSES: + reasons.append( + f"lease status '{normalized_status or 'unknown'}' is not expired " + "or stale; only a non-live reviewer lease may be reclaimed" + ) + if pr_merged is not True: + reasons.append( + "PR merged state is not proven true; reclaim requires an " + "already-merged PR (fail closed)" + ) + if owner_pid_alive is not False: + reasons.append( + "lease owner process liveness is not proven dead; a live owner " + "still protects the branch (fail closed)" + ) + if competing_active_claimant is not False: + reasons.append( + "a competing active claimant may still use the branch; reclaim " + "requires no other active ownership (fail closed)" + ) + + allowed = not reasons + return { + "reclaim_allowed": allowed, + "role": normalized_role, + "status": normalized_status, + "decision": ( + "reclaim_expired_reviewer_lease" if allowed else "keep_protecting" + ), + "reasons": [] if allowed else reasons, + } + + def assess_active_branch_ownership( *, remote: str, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index fb50976..6472097 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -11131,6 +11131,9 @@ def _collect_branch_ownership_records( """ records: list[dict] = [] inventory_error = False + # #855 AC4: expired/stale reviewer-lease records eligible for an explicit + # reclaim decision, evaluated after the full ownership inventory is built. + reviewer_reclaim_candidates: list[tuple[dict, bool | None]] = [] target_branch = (branch or "").strip() if not target_branch: return {"records": records, "inventory_error": False} @@ -11279,15 +11282,28 @@ def _collect_branch_ownership_records( else: status = freshness_status reclaim_allowed = False - records.append( - _base_rec( - category=category, - status=status, - reclaim_allowed=reclaim_allowed, - role=role, - host=lease_host or host_n or host, - ) + rec = _base_rec( + category=category, + status=status, + reclaim_allowed=reclaim_allowed, + role=role, + host=lease_host or host_n or host, ) + records.append(rec) + # #855 AC4: a reviewer lease that is expired/stale (its owner + # gone) becomes a candidate for an explicit, fail-closed + # reclaim decision made once the full inventory is known. + if ( + role == "reviewer" + and status + in branch_cleanup_guard._RECLAIMABLE_REVIEWER_STATUSES + ): + owner_alive = ( + fr.get("owner_pid_alive") if isinstance(fr, dict) else None + ) + reviewer_reclaim_candidates.append( + (rec, owner_alive if isinstance(owner_alive, bool) else None) + ) except Exception: # O1: fail closed on control-plane inventory errors. inventory_error = True @@ -11358,6 +11374,44 @@ def _collect_branch_ownership_records( ) ) + # #855 AC4: decide, explicitly and fail-closed, whether any expired/stale + # reviewer lease may stop protecting an already-merged branch. This runs + # only after the full ownership inventory is built, so a competing active + # claimant (an active lease, author session, worktree binding, or active + # reviewer comment lease) is visible. An inventory failure keeps every + # reclaim candidate protective (reclaim_allowed stays False). + if reviewer_reclaim_candidates and not inventory_error: + pr_merged_state: bool | None = None + if pr_number is not None and auth and base_api: + try: + pr_live = api_request( + "GET", f"{base_api}/pulls/{int(pr_number)}", auth + ) + if isinstance(pr_live, dict) and pr_live: + pr_merged_state = bool( + pr_live.get("merged") or pr_live.get("merged_at") + ) + except Exception: + # Unknown merged state fails closed (candidate stays protective). + pr_merged_state = None + for cand_rec, owner_alive in reviewer_reclaim_candidates: + competing = any( + other is not cand_rec + and branch_cleanup_guard.is_active_ownership_status( + other.get("status") + ) + for other in records + ) + decision = branch_cleanup_guard.assess_expired_reviewer_lease_reclaim( + role=str(cand_rec.get("role")), + status=str(cand_rec.get("status")), + pr_merged=pr_merged_state, + owner_pid_alive=owner_alive, + competing_active_claimant=competing, + ) + cand_rec["reclaim_allowed"] = decision["reclaim_allowed"] + cand_rec["reclaim_decision"] = decision["decision"] + return {"records": records, "inventory_error": inventory_error} @@ -11420,6 +11474,7 @@ def gitea_reconcile_merged_cleanups( dry_run: bool = True, execute_confirmed: bool = False, limit: int = 50, + pr_number: int | None = None, remote: str = "dadeschools", host: str | None = None, org: str | None = None, @@ -11430,7 +11485,11 @@ def gitea_reconcile_merged_cleanups( Args: dry_run: Defaults to True. When True, only builds the reconciliation report. execute_confirmed: Must be True when dry_run=False. - limit: Max number of closed PRs to inspect. + limit: Max number of closed PRs to inspect (batch mode only; ignored when + ``pr_number`` is set). + pr_number: Optional exact merged PR selector (#855). When set, only that + PR is assessed/acted on (fail closed if missing, unmerged, or + ambiguous). When omitted, existing batch behaviour is preserved. remote: Known Gitea instance ('dadeschools' or 'prgs'). host: Override the Gitea host. org: Override the owner/organization. @@ -11465,11 +11524,120 @@ def gitea_reconcile_merged_cleanups( "audit_phase": audit_reconciliation_mode.current_phase(), } + # #855: optional exact PR pin. Fail closed before any inventory mutation. + exact_pr: int | None = None + if pr_number is not None: + try: + exact_pr = int(pr_number) + except (TypeError, ValueError): + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": "exact_pr", + "selected_pr_number": pr_number, + "reasons": [ + f"pr_number={pr_number!r} is not a valid integer " + "(fail closed; no mutation)" + ], + "blocker_kind": "invalid_pr_number", + } + if exact_pr <= 0: + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": "exact_pr", + "selected_pr_number": exact_pr, + "reasons": [ + f"pr_number={exact_pr} must be a positive integer " + "(fail closed; no mutation)" + ], + "blocker_kind": "invalid_pr_number", + } + h, o, r = _resolve(remote, host, org, repo) auth = _auth(h) base = repo_api_url(h, o, r) - closed_prs = api_get_all(f"{base}/pulls?state=closed", auth, limit=limit) - open_prs = api_get_all(f"{base}/pulls?state=open", auth) + + selection_mode = "batch" + closed_prs: list[dict] = [] + open_prs: list[dict] = [] + if exact_pr is not None: + selection_mode = "exact_pr" + try: + pr_live = api_request("GET", f"{base}/pulls/{exact_pr}", auth) + except Exception as exc: + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": selection_mode, + "selected_pr_number": exact_pr, + "reasons": [ + f"PR #{exact_pr} could not be uniquely resolved " + f"(fail closed; no mutation): {_redact(str(exc))}" + ], + "blocker_kind": "pr_unresolvable", + } + if not isinstance(pr_live, dict) or not pr_live: + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": selection_mode, + "selected_pr_number": exact_pr, + "reasons": [ + f"PR #{exact_pr} could not be uniquely resolved " + "(empty response; fail closed; no mutation)" + ], + "blocker_kind": "pr_unresolvable", + } + live_number = pr_live.get("number") + try: + live_number_int = int(live_number) if live_number is not None else None + except (TypeError, ValueError): + live_number_int = None + if live_number_int != exact_pr: + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": selection_mode, + "selected_pr_number": exact_pr, + "reasons": [ + f"PR #{exact_pr} resolution is ambiguous or mismatched " + f"(live number={live_number!r}; fail closed; no mutation)" + ], + "blocker_kind": "pr_ambiguous", + } + if not (pr_live.get("merged") or pr_live.get("merged_at")): + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": selection_mode, + "selected_pr_number": exact_pr, + "reasons": [ + f"PR #{exact_pr} is not merged " + "(exact-target cleanup requires a merged PR; " + "fail closed; no mutation)" + ], + "blocker_kind": "pr_not_merged", + } + closed_prs = [pr_live] + # Exact mode still needs open heads for remote-delete safety gates. + open_prs = api_get_all(f"{base}/pulls?state=open", auth) + else: + # Preserve historical call order (closed then open) for batch callers/tests. + closed_prs = api_get_all(f"{base}/pulls?state=closed", auth, limit=limit) + open_prs = api_get_all(f"{base}/pulls?state=open", auth) merged_closed: list[dict] = [] remote_branch_exists: dict[str, bool] = {} @@ -11496,6 +11664,13 @@ def gitea_reconcile_merged_cleanups( scratch_candidates = merged_cleanup_reconcile.discover_reviewer_scratch_worktrees( _canonical_local_git_root() ) + # #855: exact-target never inventories or mutates foreign PR scratch trees. + if exact_pr is not None: + scratch_candidates = [ + s + for s in scratch_candidates + if int(s.get("pr_number") or 0) == int(exact_pr) + ] active_reviewer_leases: dict[int, bool] = {} pr_states: dict[int, dict] = {} for scratch in scratch_candidates: @@ -11530,6 +11705,33 @@ def gitea_reconcile_merged_cleanups( active_reviewer_leases=active_reviewer_leases, pr_states=pr_states, ) + report["selection_mode"] = selection_mode + if exact_pr is not None: + report["selected_pr_number"] = exact_pr + # Fail closed if exact pin somehow produced other or zero entries. + entries = list(report.get("entries") or []) + entry_numbers = [] + for entry in entries: + try: + entry_numbers.append(int(entry.get("pr_number"))) + except (TypeError, ValueError): + entry_numbers.append(entry.get("pr_number")) + if entry_numbers != [exact_pr]: + return { + "success": False, + "performed": False, + "executed": False, + "dry_run": bool(dry_run), + "selection_mode": selection_mode, + "selected_pr_number": exact_pr, + "reasons": [ + f"exact PR #{exact_pr} selection produced unexpected " + f"candidate set {entry_numbers!r} " + "(fail closed; no mutation)" + ], + "blocker_kind": "exact_selection_mismatch", + "entries": entries, + } if dry_run: report["dry_run"] = True diff --git a/tests/test_branch_cleanup_guard.py b/tests/test_branch_cleanup_guard.py index 0bd1fda..fddb534 100644 --- a/tests/test_branch_cleanup_guard.py +++ b/tests/test_branch_cleanup_guard.py @@ -1639,6 +1639,358 @@ class TestSecondRemediationIntegration(unittest.TestCase): self.assertTrue(ownership_calls) +class TestIssue855ExactPrSelector(unittest.TestCase): + """#855: exact pr_number pin for reconcile_merged_cleanups (#851 lifecycle).""" + + def setUp(self): + self._remotes = patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.example.com", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + } + }, + ) + self._remotes.start() + patch("gitea_audit.audit_enabled", return_value=False).start() + self.mock_api = patch("mcp_server.api_request").start() + self.mock_all = patch("mcp_server.api_get_all", return_value=[]).start() + patch("mcp_server.get_auth_header", return_value=FAKE_AUTH).start() + patch( + "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", + return_value=True, + ).start() + patch( + "mcp_server.get_profile", + return_value=dict(RECONCILER_WITH_DELETE), + ).start() + patch( + "mcp_server._profile_operation_gate", + return_value=[], + ).start() + patch( + "mcp_server._collect_branch_ownership_records", + return_value={"records": [], "inventory_error": False}, + ).start() + patch( + "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees", + return_value=[], + ).start() + patch("mcp_server.verify_preflight_purity", return_value=None).start() + patch( + "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed", + return_value=(True, []), + ).start() + + def tearDown(self): + patch.stopall() + + def _merged_pr(self, number, branch, sha="c" * 40): + return { + "number": number, + "title": f"PR {number}", + "body": f"Closes #{number - 4}", + "merged": True, + "merged_at": "2026-07-23T12:00:00Z", + "merge_commit_sha": "f" * 40, + "state": "closed", + "head": {"ref": branch, "sha": sha}, + "base": {"ref": "master"}, + } + + def test_exact_pr_848_ignores_newer_852_in_batch_queue(self): + """pr_number=848 selects only #848 even when #852 is newer/first.""" + from mcp_server import gitea_reconcile_merged_cleanups + + pr_848 = self._merged_pr( + 848, "fix/issue-844-exclude-epic-containers", sha="c3f282ba" + "0" * 32 + ) + # Closed list would rank #852 first in batch mode; exact pin must ignore it. + closed_batch = [ + self._merged_pr(852, "fix/issue-851-cleanup-worktree-before-remote-delete"), + pr_848, + self._merged_pr(849, "fix/issue-849-other"), + self._merged_pr(846, "fix/issue-846-other"), + self._merged_pr(845, "fix/issue-845-other"), + ] + batch_fetch_calls = [] + + def fake_api(method, url, *args, **kwargs): + if method == "GET" and url.rstrip("/").endswith("/pulls/848"): + return dict(pr_848) + if method == "GET" and "/pulls/" in url: + raise AssertionError(f"unexpected PR fetch: {url}") + if method == "GET" and "/branches/" in url: + return {"name": "present"} + return {} + + def fake_all(url, auth, limit=None): + batch_fetch_calls.append((url, limit)) + if "state=open" in url: + return [] + if "state=closed" in url: + # Exact mode must not use the closed batch list. + raise AssertionError( + "exact pr_number mode must not page closed PRs: " + url + ) + return [] + + self.mock_api.side_effect = fake_api + self.mock_all.side_effect = fake_all + patch( + "mcp_server._remote_branch_exists", + return_value=True, + ).start() + patch( + "mcp_server.merged_cleanup_reconcile.build_reconciliation_report", + side_effect=lambda **kwargs: { + "entries": [ + { + "pr_number": int(pr["number"]), + "head_branch": (pr.get("head") or {}).get("ref"), + "issue_number": 844, + "remote_branch": { + "safe_to_delete_remote": True, + "head_branch": (pr.get("head") or {}).get("ref"), + }, + "local_worktree": { + "safe_to_remove_worktree": True, + "worktree_path": ( + "/tmp/branches/fix-issue-844-exclude-epic-containers" + ), + }, + "planned_execution_order": ( + mcp_server.merged_cleanup_reconcile.plan_cleanup_execution_order( + remote_assessment={"safe_to_delete_remote": True}, + local_assessment={"safe_to_remove_worktree": True}, + ) + ), + } + for pr in kwargs.get("closed_prs") or [] + if pr.get("merged_at") or pr.get("merged") + ], + "reviewer_scratch_entries": [], + "merged_pr_count": len(kwargs.get("closed_prs") or []), + }, + ).start() + + res = gitea_reconcile_merged_cleanups( + dry_run=True, + pr_number=848, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertTrue(res.get("success")) + self.assertFalse(res.get("performed")) + self.assertEqual(res.get("selection_mode"), "exact_pr") + self.assertEqual(res.get("selected_pr_number"), 848) + entries = res.get("entries") or [] + self.assertEqual(len(entries), 1, entries) + self.assertEqual(entries[0].get("pr_number"), 848) + self.assertEqual( + entries[0].get("head_branch"), + "fix/issue-844-exclude-epic-containers", + ) + # No other PR appears in plan. + self.assertEqual(list((res.get("planned_execution_orders") or {}).keys()), ["848"]) + plan = (res.get("planned_execution_orders") or {}).get("848") or [] + actions = [s.get("action") for s in plan] + self.assertEqual( + actions, + [ + "remove_local_worktree", + "reassess_branch_ownership", + "delete_remote_branch", + ], + ) + # Prove we never scanned the multi-PR closed batch. + self.assertFalse(any("state=closed" in (u or "") for u, _ in batch_fetch_calls)) + # closed_batch fixture must remain unused (sanity). + self.assertEqual(closed_batch[0]["number"], 852) + + def test_exact_pr_execute_only_mutates_selected_pr(self): + """Execute with pr_number must never touch #845/#846/#849/#852.""" + from mcp_server import gitea_reconcile_merged_cleanups + + pr_848 = self._merged_pr(848, "fix/issue-844-exclude-epic-containers") + worktree_path = "/tmp/branches/fix-issue-844-exclude-epic-containers" + remove_calls = [] + delete_api_calls = [] + ownership_branches = [] + + def fake_api(method, url, *args, **kwargs): + if method == "GET" and url.rstrip("/").endswith("/pulls/848"): + return dict(pr_848) + if method == "DELETE": + delete_api_calls.append(url) + # Forbid foreign PR branch deletion by URL content. + for forbidden in ("845", "846", "849", "852"): + self.assertNotIn(forbidden, url) + return {} + + def fake_remove(project_root, branch, worktree_path=None): + remove_calls.append({"branch": branch, "worktree_path": worktree_path}) + return { + "success": True, + "performed": True, + "message": f"removed {worktree_path}", + "worktree_path": worktree_path, + } + + def fake_collect(**kwargs): + ownership_branches.append(kwargs.get("branch")) + return {"records": [], "inventory_error": False} + + def fake_probe(h, o, r, auth, br): + return guard.classify_branch_readback_http_status( + 404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH + ) + + self.mock_api.side_effect = fake_api + self.mock_all.side_effect = lambda url, auth, limit=None: [] + patch("mcp_server._remote_branch_exists", return_value=True).start() + patch( + "mcp_server.merged_cleanup_reconcile.build_reconciliation_report", + return_value={ + "entries": [ + { + "pr_number": 848, + "head_branch": "fix/issue-844-exclude-epic-containers", + "remote_branch": {"safe_to_delete_remote": True}, + "local_worktree": { + "safe_to_remove_worktree": True, + "worktree_path": worktree_path, + }, + "planned_execution_order": [ + {"action": "remove_local_worktree", "phase": 1}, + {"action": "reassess_branch_ownership", "phase": 2}, + {"action": "delete_remote_branch", "phase": 3}, + ], + } + ], + "reviewer_scratch_entries": [ + # Foreign scratch must be filtered before report execute loop; + # if present here it would still be a test failure if acted on. + ], + "merged_pr_count": 1, + }, + ).start() + patch( + "mcp_server.merged_cleanup_reconcile.remove_local_worktree", + side_effect=fake_remove, + ).start() + patch( + "mcp_server._collect_branch_ownership_records", + side_effect=fake_collect, + ).start() + patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start() + + res = gitea_reconcile_merged_cleanups( + dry_run=False, + execute_confirmed=True, + pr_number=848, + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + ) + self.assertTrue(res.get("performed") or res.get("executed")) + self.assertEqual(res.get("selection_mode"), "exact_pr") + self.assertEqual(res.get("selected_pr_number"), 848) + actions = res.get("actions") or [] + pr_numbers_touched = { + a.get("pr_number") for a in actions if a.get("pr_number") is not None + } + self.assertTrue(pr_numbers_touched.issubset({None, 848}) or not pr_numbers_touched) + removes = [a for a in actions if a.get("action") == "remove_local_worktree"] + deletes = [a for a in actions if a.get("action") == "delete_remote_branch"] + self.assertEqual(len(removes), 1) + self.assertEqual(remove_calls[0]["branch"], "fix/issue-844-exclude-epic-containers") + self.assertEqual(len(deletes), 1) + self.assertTrue(deletes[0].get("success")) + self.assertTrue(deletes[0].get("after_worktree_removal")) + self.assertEqual(len(delete_api_calls), 1) + self.assertEqual( + ownership_branches, ["fix/issue-844-exclude-epic-containers"] + ) + + def test_exact_pr_unknown_fails_closed_without_mutation(self): + from mcp_server import gitea_reconcile_merged_cleanups + + def fake_api(method, url, *args, **kwargs): + if method == "GET" and "/pulls/99999" in url: + raise RuntimeError("HTTP 404 Not Found") + raise AssertionError(f"unexpected API call {method} {url}") + + self.mock_api.side_effect = fake_api + res = gitea_reconcile_merged_cleanups( + dry_run=True, + pr_number=99999, + remote="prgs", + ) + self.assertFalse(res.get("success")) + self.assertFalse(res.get("performed")) + self.assertEqual(res.get("blocker_kind"), "pr_unresolvable") + self.assertIn("99999", " ".join(res.get("reasons") or [])) + + def test_exact_pr_not_merged_fails_closed(self): + from mcp_server import gitea_reconcile_merged_cleanups + + def fake_api(method, url, *args, **kwargs): + if method == "GET" and url.rstrip("/").endswith("/pulls/900"): + return { + "number": 900, + "merged": False, + "merged_at": None, + "state": "open", + "head": {"ref": "feat/x", "sha": "a" * 40}, + } + raise AssertionError(f"unexpected {method} {url}") + + self.mock_api.side_effect = fake_api + res = gitea_reconcile_merged_cleanups( + dry_run=False, + execute_confirmed=True, + pr_number=900, + remote="prgs", + ) + self.assertFalse(res.get("success")) + self.assertFalse(res.get("performed")) + self.assertEqual(res.get("blocker_kind"), "pr_not_merged") + + def test_exact_pr_invalid_number_fails_closed(self): + from mcp_server import gitea_reconcile_merged_cleanups + + res = gitea_reconcile_merged_cleanups( + dry_run=True, + pr_number=0, + remote="prgs", + ) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("blocker_kind"), "invalid_pr_number") + self.mock_api.assert_not_called() + + def test_batch_mode_still_works_without_pr_number(self): + """Unfiltered batch path remains backward compatible.""" + from mcp_server import gitea_reconcile_merged_cleanups + + self.mock_all.side_effect = lambda url, auth, limit=None: [] + self.mock_api.side_effect = lambda *a, **k: {} + patch( + "mcp_server.merged_cleanup_reconcile.build_reconciliation_report", + return_value={ + "entries": [], + "reviewer_scratch_entries": [], + "merged_pr_count": 0, + }, + ).start() + res = gitea_reconcile_merged_cleanups(dry_run=True, remote="prgs", limit=10) + self.assertTrue(res.get("success")) + self.assertEqual(res.get("selection_mode"), "batch") + self.assertIsNone(res.get("selected_pr_number")) + if __name__ == "__main__": unittest.main() diff --git a/tests/test_issue_855_expired_reviewer_reclaim.py b/tests/test_issue_855_expired_reviewer_reclaim.py new file mode 100644 index 0000000..288b7f7 --- /dev/null +++ b/tests/test_issue_855_expired_reviewer_reclaim.py @@ -0,0 +1,244 @@ +"""#855 AC4: an expired reviewer lease must not indefinitely protect an +already-merged branch when no live claimant exists. + +Two layers are covered: + +* ``branch_cleanup_guard.assess_expired_reviewer_lease_reclaim`` — the pure, + fail-closed reclaim decision. Every condition must be provably satisfied or + the lease keeps protecting the branch. +* ``gitea_mcp_server._collect_branch_ownership_records`` — the wiring that + supplies authoritative evidence (PR merged state, owner-process liveness, + competing ownership) to that decision, and flips an expired reviewer lease + to reclaimable only under the full policy. + +All inputs are fabricated; no real repository, lease, or credential is used. +""" + +import importlib +import unittest +from unittest.mock import patch + +import branch_cleanup_guard + +mcp_server = importlib.import_module("gitea_mcp_server") + +FAKE_AUTH = "token fake" +REMOTE = "prgs" +ORG = "Scaled-Tech-Consulting" +REPO = "Gitea-Tools" +HOST = "gitea.prgs.cc" +BRANCH = "feat/issue-638-webui-app-shell-phase1" +PR_NUMBER = 818 + + +class TestAssessExpiredReviewerLeaseReclaim(unittest.TestCase): + """Pure fail-closed reclaim decision (#855 AC4).""" + + def _call(self, **overrides): + base = dict( + role="reviewer", + status="expired", + pr_merged=True, + owner_pid_alive=False, + competing_active_claimant=False, + ) + base.update(overrides) + return branch_cleanup_guard.assess_expired_reviewer_lease_reclaim(**base) + + def test_full_policy_satisfied_allows_reclaim(self): + out = self._call() + self.assertTrue(out["reclaim_allowed"]) + self.assertEqual(out["reasons"], []) + self.assertEqual(out["decision"], "reclaim_expired_reviewer_lease") + + def test_stale_dead_process_reviewer_also_reclaimable(self): + out = self._call(status="stale_dead_process") + self.assertTrue(out["reclaim_allowed"]) + + def test_non_reviewer_role_never_reclaims(self): + for role in ("author", "merger", "controller", "reconciler", "unknown"): + with self.subTest(role=role): + out = self._call(role=role) + self.assertFalse(out["reclaim_allowed"]) + self.assertTrue(out["reasons"]) + self.assertEqual(out["decision"], "keep_protecting") + + def test_active_status_never_reclaims(self): + out = self._call(status="active") + self.assertFalse(out["reclaim_allowed"]) + + def test_pr_not_merged_blocks_reclaim(self): + out = self._call(pr_merged=False) + self.assertFalse(out["reclaim_allowed"]) + + def test_pr_merged_unknown_fails_closed(self): + out = self._call(pr_merged=None) + self.assertFalse(out["reclaim_allowed"]) + + def test_owner_process_alive_blocks_reclaim(self): + out = self._call(owner_pid_alive=True) + self.assertFalse(out["reclaim_allowed"]) + + def test_owner_liveness_unknown_fails_closed(self): + out = self._call(owner_pid_alive=None) + self.assertFalse(out["reclaim_allowed"]) + + def test_competing_active_claimant_blocks_reclaim(self): + out = self._call(competing_active_claimant=True) + self.assertFalse(out["reclaim_allowed"]) + + def test_competing_claimant_unknown_fails_closed(self): + out = self._call(competing_active_claimant=None) + self.assertFalse(out["reclaim_allowed"]) + + def test_reasons_never_leak_secrets(self): + out = self._call(role="author") + blob = " ".join(out["reasons"]).lower() + self.assertNotIn("token", blob) + self.assertNotIn("password", blob) + + +class _FakeLease(dict): + pass + + +class TestCollectorExpiredReviewerReclaimWiring(unittest.TestCase): + """`_collect_branch_ownership_records` supplies authoritative evidence and + flips an expired reviewer lease to reclaimable only under the full policy.""" + + def _run( + self, + *, + lease_role="reviewer", + lease_freshness="stale_dead_process", + owner_pid_alive=False, + pr_merged=True, + extra_leases=None, + worktree_on_branch=False, + ): + lease = _FakeLease( + role=lease_role, + work_kind="pr", + work_number=PR_NUMBER, + branch=BRANCH, + status="active", + owner_pid=999999, + remote=REMOTE, + org=ORG, + repo=REPO, + host=HOST, + freshness={ + "freshness": lease_freshness, + "owner_pid": 999999, + "owner_pid_alive": owner_pid_alive, + "expired_by_time": lease_freshness == "expired", + }, + ) + leases = [lease] + list(extra_leases or []) + + pr_payload = { + "number": PR_NUMBER, + "merged": pr_merged, + "merged_at": "2026-07-23T00:00:00Z" if pr_merged else None, + "head": {"ref": BRANCH}, + } + + def fake_api_request(method, url, *a, **k): + if method == "GET" and f"/pulls/{PR_NUMBER}" in url: + return pr_payload + raise AssertionError(f"unexpected api_request {method} {url}") + + wt_entries = [] + if worktree_on_branch: + wt_entries = [{"branch": BRANCH, "path": f"/x/branches/{BRANCH}"}] + + with patch.object( + mcp_server.lease_lifecycle, + "list_active_leases", + return_value={"leases": leases}, + ), patch.object( + mcp_server.control_plane_db, "get_db", return_value=object(), create=True + ), patch.object( + mcp_server.issue_lock_store, "iter_lock_files", return_value=[] + ), patch.object( + mcp_server.worktree_cleanup_audit, + "list_worktrees", + return_value=wt_entries, + ), patch.object( + mcp_server, "api_get_all", return_value=[] + ), patch.object( + mcp_server, "api_request", side_effect=fake_api_request + ): + return mcp_server._collect_branch_ownership_records( + remote=REMOTE, + host=HOST, + org=ORG, + repo=REPO, + branch=BRANCH, + pr_number=PR_NUMBER, + project_root="/x", + auth=FAKE_AUTH, + base_api="https://gitea.prgs.cc/api/v1/repos/x/y", + ) + + def _reviewer_records(self, bundle): + return [ + rec + for rec in bundle["records"] + if rec.get("category") + == branch_cleanup_guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE + ] + + def test_merged_dead_uncontested_reviewer_lease_is_reclaimable(self): + bundle = self._run() + self.assertFalse(bundle["inventory_error"]) + recs = self._reviewer_records(bundle) + self.assertEqual(len(recs), 1) + self.assertTrue(recs[0]["reclaim_allowed"]) + # And the guard consequently does not block deletion on it. + ownership = branch_cleanup_guard.assess_active_branch_ownership( + remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST, + records=bundle["records"], + ) + self.assertFalse(ownership["block"]) + + def test_unmerged_pr_keeps_reviewer_lease_protective(self): + bundle = self._run(pr_merged=False) + recs = self._reviewer_records(bundle) + self.assertEqual(len(recs), 1) + self.assertFalse(recs[0]["reclaim_allowed"]) + ownership = branch_cleanup_guard.assess_active_branch_ownership( + remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST, + records=bundle["records"], + ) + self.assertTrue(ownership["block"]) + + def test_owner_process_alive_keeps_reviewer_lease_protective(self): + bundle = self._run(owner_pid_alive=True, lease_freshness="expired") + recs = self._reviewer_records(bundle) + self.assertFalse(recs[0]["reclaim_allowed"]) + + def test_competing_worktree_binding_keeps_reviewer_lease_protective(self): + bundle = self._run(worktree_on_branch=True) + recs = self._reviewer_records(bundle) + self.assertFalse(recs[0]["reclaim_allowed"]) + ownership = branch_cleanup_guard.assess_active_branch_ownership( + remote=REMOTE, org=ORG, repo=REPO, branch=BRANCH, host=HOST, + records=bundle["records"], + ) + self.assertTrue(ownership["block"]) + + def test_expired_author_lease_never_reclaimed_by_reviewer_policy(self): + bundle = self._run(lease_role="author") + author_recs = [ + rec + for rec in bundle["records"] + if rec.get("category") + == branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE + ] + self.assertEqual(len(author_recs), 1) + self.assertFalse(author_recs[0]["reclaim_allowed"]) + + +if __name__ == "__main__": + unittest.main()