From 137426f7ad6562c9e4ae09451d55491b6a4ccc7d Mon Sep 17 00:00:00 2001 From: Jason Walker <913443@dadeschools.net> Date: Thu, 16 Jul 2026 00:57:30 -0400 Subject: [PATCH] fix(cleanup): post-delete readback and active ownership gates (#687) Require authoritative not-found readback after merged-PR branch DELETE, and block cleanup when active author/reviewer/merger/controller/reconciler session, lease, or worktree-binding ownership still uses the target branch. --- branch_cleanup_guard.py | 383 +++++++++++++++++++++++ gitea_mcp_server.py | 357 +++++++++++++++++++++- tests/test_branch_cleanup_guard.py | 471 ++++++++++++++++++++++++++++- 3 files changed, 1188 insertions(+), 23 deletions(-) diff --git a/branch_cleanup_guard.py b/branch_cleanup_guard.py index 6e3bc3e..519e61c 100644 --- a/branch_cleanup_guard.py +++ b/branch_cleanup_guard.py @@ -114,3 +114,386 @@ def assess_merged_pr_branch_cleanup( "block_reasons": reasons, "recommended_action": "delete_remote_branch" if safe else "keep_remote_branch", } + + +# --------------------------------------------------------------------------- +# #687 remediation: post-delete readback + active ownership protection +# --------------------------------------------------------------------------- + +READBACK_NOT_FOUND = "not_found" +READBACK_EXISTS = "exists" +READBACK_AUTHENTICATION = "authentication_error" +READBACK_AUTHORIZATION = "authorization_error" +READBACK_TRANSPORT = "transport_error" +READBACK_UNEXPECTED = "unexpected_response" + +ERROR_CLASS_AUTHENTICATION = "authentication" +ERROR_CLASS_AUTHORIZATION = "authorization" +ERROR_CLASS_TRANSPORT = "transport" +ERROR_CLASS_UNEXPECTED = "unexpected" + +OWNERSHIP_CATEGORY_AUTHOR_SESSION = "author_session" +OWNERSHIP_CATEGORY_AUTHOR_LEASE = "author_lease" +OWNERSHIP_CATEGORY_REVIEWER_LEASE = "reviewer_lease" +OWNERSHIP_CATEGORY_MERGER_LEASE = "merger_lease" +OWNERSHIP_CATEGORY_CONTROLLER_LEASE = "controller_lease" +OWNERSHIP_CATEGORY_RECONCILER_LEASE = "reconciler_lease" +OWNERSHIP_CATEGORY_WORKTREE_BINDING = "worktree_binding" + +_ROLE_TO_OWNERSHIP_CATEGORY = { + "author": OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "reviewer": OWNERSHIP_CATEGORY_REVIEWER_LEASE, + "merger": OWNERSHIP_CATEGORY_MERGER_LEASE, + "controller": OWNERSHIP_CATEGORY_CONTROLLER_LEASE, + "reconciler": OWNERSHIP_CATEGORY_RECONCILER_LEASE, +} + +_ACTIVE_OWNERSHIP_STATUSES = frozenset( + {"active", "live", "claimed", "in_progress", "working", "pushing", "pushed"} +) +_TERMINAL_OWNERSHIP_STATUSES = frozenset( + {"released", "abandoned", "done", "blocked", "terminal", "closed"} +) +_EXPIRED_STATUSES = frozenset({"expired"}) +_STALE_STATUSES = frozenset({"stale", "stale_dead_process", "stale_missing_worktree"}) + + +def _norm_str(value: Any) -> str: + return str(value or "").strip() + + +def ownership_category_for_role(role: str | None) -> str: + """Map a role kind to a non-secret ownership category label.""" + key = _norm_str(role).lower() + return _ROLE_TO_OWNERSHIP_CATEGORY.get(key, f"{key or 'unknown'}_lease") + + +def classify_branch_readback_http_status(status_code: int | None) -> dict[str, Any]: + """Classify a GET-branch HTTP status into a secret-free readback result.""" + if status_code == 404: + return { + "status": READBACK_NOT_FOUND, + "error_class": None, + "verified_absent": True, + "branch_present": False, + "reasons": [], + } + if status_code in (401, 407): + return { + "status": READBACK_AUTHENTICATION, + "error_class": ERROR_CLASS_AUTHENTICATION, + "verified_absent": False, + "branch_present": None, + "reasons": ["post-delete branch readback authentication failed"], + } + if status_code == 403: + return { + "status": READBACK_AUTHORIZATION, + "error_class": ERROR_CLASS_AUTHORIZATION, + "verified_absent": False, + "branch_present": None, + "reasons": ["post-delete branch readback authorization failed"], + } + if status_code is not None and 200 <= int(status_code) < 300: + return { + "status": READBACK_EXISTS, + "error_class": None, + "verified_absent": False, + "branch_present": True, + "reasons": ["post-delete readback found branch still present"], + } + if status_code is not None and int(status_code) >= 500: + return { + "status": READBACK_TRANSPORT, + "error_class": ERROR_CLASS_TRANSPORT, + "verified_absent": False, + "branch_present": None, + "reasons": ["post-delete branch readback transport/upstream failure"], + } + return { + "status": READBACK_UNEXPECTED, + "error_class": ERROR_CLASS_UNEXPECTED, + "verified_absent": False, + "branch_present": None, + "reasons": ["post-delete branch readback returned an unexpected response"], + "http_status": status_code, + } + + +def classify_branch_readback_exception(exc: BaseException) -> dict[str, Any]: + """Classify a GET-branch exception without leaking credentials or bodies. + + Prefer typed HTTP status when present on the exception chain; fall back to + a narrow ``HTTP `` prefix parse. Never includes response bodies, + tokens, or raw exception text in the returned payload. + """ + status_code: int | None = None + seen: set[int] = set() + current: BaseException | None = exc + while current is not None and id(current) not in seen: + seen.add(id(current)) + code = getattr(current, "code", None) + if isinstance(code, int): + status_code = code + break + status = getattr(current, "status", None) + if isinstance(status, int): + status_code = status + break + status_code_attr = getattr(current, "status_code", None) + if isinstance(status_code_attr, int): + status_code = status_code_attr + break + current = current.__cause__ or current.__context__ + + if status_code is None: + # Narrow, non-secret parse of our own RuntimeError shape: "HTTP 404: ..." + text = str(exc) if exc is not None else "" + match = re.match(r"HTTP\s+(\d{3})\b", text) + if match: + status_code = int(match.group(1)) + else: + lower = text.lower() + if "not found" in lower or "404" in lower: + status_code = 404 + elif "unauthorized" in lower or "401" in lower: + status_code = 401 + elif "forbidden" in lower or "403" in lower: + status_code = 403 + elif any( + token in lower + for token in ( + "timed out", + "timeout", + "connection", + "network", + "temporarily unavailable", + "name or service not known", + "nodename nor servname", + ) + ): + return { + "status": READBACK_TRANSPORT, + "error_class": ERROR_CLASS_TRANSPORT, + "verified_absent": False, + "branch_present": None, + "reasons": [ + "post-delete branch readback transport/upstream failure" + ], + } + + if status_code is not None: + return classify_branch_readback_http_status(status_code) + + return { + "status": READBACK_UNEXPECTED, + "error_class": ERROR_CLASS_UNEXPECTED, + "verified_absent": False, + "branch_present": None, + "reasons": ["post-delete branch readback returned an unexpected response"], + } + + +def assess_post_delete_readback(readback: dict[str, Any] | None) -> dict[str, Any]: + """Decide whether DELETE success may be reported after branch readback. + + Success requires an authoritative not-found readback. DELETE HTTP success + alone is never enough. + """ + payload = dict(readback or {}) + status = _norm_str(payload.get("status")) or READBACK_UNEXPECTED + verified = bool(payload.get("verified_absent")) or status == READBACK_NOT_FOUND + if status == READBACK_NOT_FOUND and verified: + return { + "ok": True, + "success": True, + "readback": { + "status": READBACK_NOT_FOUND, + "verified_absent": True, + "branch_present": False, + "error_class": None, + }, + "reasons": [], + } + + reasons = list(payload.get("reasons") or []) + if not reasons: + if status == READBACK_EXISTS: + reasons = ["post-delete readback found branch still present"] + elif status == READBACK_AUTHENTICATION: + reasons = ["post-delete branch readback authentication failed"] + elif status == READBACK_AUTHORIZATION: + reasons = ["post-delete branch readback authorization failed"] + elif status == READBACK_TRANSPORT: + reasons = ["post-delete branch readback transport/upstream failure"] + else: + reasons = ["post-delete branch readback could not verify deletion"] + + return { + "ok": False, + "success": False, + "readback": { + "status": status, + "verified_absent": False, + "branch_present": payload.get("branch_present"), + "error_class": payload.get("error_class"), + }, + "reasons": reasons, + "blocker_kind": "post_delete_readback_failed", + } + + +def _repo_matches(record: dict[str, Any], *, remote: str, org: str, repo: str) -> bool: + return ( + _norm_str(record.get("remote")).lower() == _norm_str(remote).lower() + and _norm_str(record.get("org")).lower() == _norm_str(org).lower() + and _norm_str(record.get("repo")).lower() == _norm_str(repo).lower() + ) + + +def _branch_matches(record: dict[str, Any], branch: str) -> bool: + return _norm_str(record.get("branch")) == _norm_str(branch) + + +def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]: + """Classify one ownership record as blocking or non-blocking. + + Distinguishes active ownership from expired/released/terminal/stale records. + Stale/expired records block only when reclaim is not allowed (sticky foreign + ownership with live owner + present worktree). + """ + status = _norm_str(record.get("status")).lower() + category = _norm_str(record.get("category")) or "unknown" + reclaim_allowed = record.get("reclaim_allowed") + + if status in _TERMINAL_OWNERSHIP_STATUSES: + return { + "blocks": False, + "status": status, + "category": category, + "reason": f"{category} ownership is terminal/released ({status})", + } + if status in _ACTIVE_OWNERSHIP_STATUSES: + return { + "blocks": True, + "status": status, + "category": category, + "reason": f"active {category} ownership still uses the target branch", + } + if status in _EXPIRED_STATUSES or status in _STALE_STATUSES: + # Canonical recovery policy (#601): reclaimable expired/stale does not + # block; sticky expired/stale (live pid + present worktree) does. + if reclaim_allowed is True: + return { + "blocks": False, + "status": status, + "category": category, + "reason": ( + f"{category} ownership is {status} and reclaimable; " + "does not block deletion" + ), + } + if reclaim_allowed is False: + return { + "blocks": True, + "status": status, + "category": category, + "reason": ( + f"sticky {status} {category} ownership still protects the " + "target branch (recovery review required)" + ), + } + # Unknown reclaimability → fail closed + return { + "blocks": True, + "status": status, + "category": category, + "reason": ( + f"{status} {category} ownership reclaimability unknown; " + "fail closed" + ), + } + # Unknown status → fail closed + return { + "blocks": True, + "status": status or "unknown", + "category": category, + "reason": ( + f"unclassified {category} ownership status " + f"'{status or 'unknown'}'; fail closed" + ), + } + + +def assess_active_branch_ownership( + *, + remote: str, + org: str, + repo: str, + branch: str, + records: list[dict[str, Any]] | None = None, +) -> dict[str, Any]: + """Assess whether any active ownership still uses *branch* in *repo*. + + Only records matching the exact remote/org/repo/branch quadruple are + considered. Other repositories or branches never produce a false block. + Returned denial identifies ownership category only — never secrets. + """ + target_branch = _norm_str(branch) + considered: list[dict[str, Any]] = [] + blocking: list[dict[str, Any]] = [] + ignored: list[dict[str, Any]] = [] + + for raw in records or []: + if not isinstance(raw, dict): + continue + if not _repo_matches(raw, remote=remote, org=org, repo=repo): + ignored.append( + { + "category": _norm_str(raw.get("category")) or "unknown", + "reason": "different repository scope", + } + ) + continue + if not _branch_matches(raw, target_branch): + ignored.append( + { + "category": _norm_str(raw.get("category")) or "unknown", + "reason": "different branch", + } + ) + continue + activity = assess_ownership_record_activity(raw) + entry = { + "category": activity["category"], + "status": activity["status"], + "blocks": activity["blocks"], + "reason": activity["reason"], + } + considered.append(entry) + if activity["blocks"]: + blocking.append(entry) + + block = bool(blocking) + categories = sorted({b["category"] for b in blocking}) + reasons = [ + ( + "active ownership protects branch " + f"'{target_branch}': " + "; ".join(b["reason"] for b in blocking) + ) + ] if block else [] + return { + "block": block, + "safe_to_delete": not block, + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "blocking_categories": categories, + "blocking": blocking, + "considered": considered, + "ignored_out_of_scope": ignored, + "reasons": reasons, + "blocker_kind": "active_branch_ownership" if block else None, + "recommended_action": "keep_remote_branch" if block else "delete_remote_branch", + } diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 2a5af99..7c26ca0 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -6128,10 +6128,53 @@ def gitea_cleanup_merged_pr_branch( "reasons": assessment["block_reasons"], } + # #687: active ownership protection (sessions/leases/worktree bindings). + # Do not rely only on the caller worktree living under branches/. + ownership_records = _collect_branch_ownership_records( + remote=remote, + org=o, + repo=r, + branch=head_branch, + pr_number=pr_number, + project_root=PROJECT_ROOT, + ) + ownership = branch_cleanup_guard.assess_active_branch_ownership( + remote=remote, + org=o, + repo=r, + branch=head_branch, + records=ownership_records, + ) + if ownership.get("block"): + return { + "success": False, + "performed": False, + "pr_number": pr_number, + "branch": head_branch, + "assessment": assessment, + "ownership": { + "block": True, + "blocking_categories": ownership.get("blocking_categories") or [], + "reasons": ownership.get("reasons") or [], + "blocker_kind": ownership.get("blocker_kind"), + }, + "reasons": ownership.get("reasons") or [ + "active ownership protects the target branch" + ], + "blocker_kind": "active_branch_ownership", + } + import urllib.parse encoded_branch = urllib.parse.quote(head_branch, safe="") url = f"{base}/branches/{encoded_branch}" + request_metadata = { + "branch": head_branch, + "required_permission": "gitea.branch.delete", + "cleanup_path": "gitea_cleanup_merged_pr_branch", + "ownership_checked": True, + "ownership_blocking_categories": [], + } with _audited( "cleanup_merged_pr_branch", host=h, @@ -6140,36 +6183,326 @@ def gitea_cleanup_merged_pr_branch( repo=r, pr_number=pr_number, target_branch=head_branch, - request_metadata={ - "branch": head_branch, - "required_permission": "gitea.branch.delete", - "cleanup_path": "gitea_cleanup_merged_pr_branch", - }, + request_metadata=request_metadata, ): api_request("DELETE", url, auth) + + # #687: authoritative post-delete readback. DELETE success alone is not + # enough — only an authoritative not-found proves deletion. + readback = _probe_remote_branch(h, o, r, auth, head_branch) + readback_assessment = branch_cleanup_guard.assess_post_delete_readback(readback) + request_metadata["post_delete_readback"] = { + "status": (readback_assessment.get("readback") or {}).get("status"), + "verified_absent": bool( + (readback_assessment.get("readback") or {}).get("verified_absent") + ), + "error_class": (readback_assessment.get("readback") or {}).get( + "error_class" + ), + } + # Emit a second audit row capturing readback evidence (no secrets). + if gitea_audit.audit_enabled(): + _audit( + "cleanup_merged_pr_branch_readback", + host=h, + remote=remote, + org=o, + repo=r, + result=( + gitea_audit.SUCCEEDED + if readback_assessment.get("ok") + else gitea_audit.FAILED + ), + reason="; ".join(readback_assessment.get("reasons") or []) or None, + request_metadata=request_metadata, + pr_number=pr_number, + target_branch=head_branch, + ) + + if not readback_assessment.get("ok"): + return { + "success": False, + "performed": True, + "delete_acknowledged": True, + "pr_number": pr_number, + "branch": head_branch, + "assessment": assessment, + "ownership": { + "block": False, + "blocking_categories": [], + "checked": True, + }, + "readback": readback_assessment.get("readback"), + "reasons": readback_assessment.get("reasons") or [ + "post-delete branch readback could not verify deletion" + ], + "blocker_kind": readback_assessment.get("blocker_kind") + or "post_delete_readback_failed", + "message": ( + f"DELETE accepted for '{head_branch}' but post-delete readback " + "did not verify absence" + ), + } + return { "success": True, "performed": True, "pr_number": pr_number, "branch": head_branch, - "message": f"Merged PR #{pr_number} source branch '{head_branch}' deleted.", + "message": ( + f"Merged PR #{pr_number} source branch '{head_branch}' deleted " + "and verified absent via post-delete readback." + ), "assessment": assessment, + "ownership": { + "block": False, + "blocking_categories": [], + "checked": True, + }, + "readback": readback_assessment.get("readback"), } -def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool: +def _probe_remote_branch( + h: str, o: str, r: str, auth: str, branch: str +) -> dict: + """GET a remote branch and return a secret-free structured readback.""" import urllib.parse encoded = urllib.parse.quote(branch, safe="") url = f"{repo_api_url(h, o, r)}/branches/{encoded}" try: api_request("GET", url, auth) - return True + return branch_cleanup_guard.classify_branch_readback_http_status(200) except Exception as exc: - message = str(exc).lower() - if "404" in message or "not found" in message: - return False - raise + return branch_cleanup_guard.classify_branch_readback_exception(exc) + + +def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool: + """True when the remote branch exists; False on authoritative 404. + + Authentication, authorization, transport, and unexpected failures are + re-raised (or returned as structured failures by callers that use + ``_probe_remote_branch``) rather than treated as absence. + """ + probe = _probe_remote_branch(h, o, r, auth, branch) + status = probe.get("status") + if status == branch_cleanup_guard.READBACK_NOT_FOUND: + return False + if status == branch_cleanup_guard.READBACK_EXISTS: + return True + # Preserve structured failure modes for callers that need them. + error_class = probe.get("error_class") or "unexpected" + raise RuntimeError( + f"remote branch probe failed ({error_class}/{status})" + ) + + +def _collect_branch_ownership_records( + *, + remote: str, + org: str, + repo: str, + branch: str, + pr_number: int | None, + project_root: str, +) -> list[dict]: + """Gather canonical ownership records for *branch* (secret-free). + + Sources: + - author issue locks (task-session / lease files) + - control-plane leases (author/reviewer/merger/controller/reconciler) + - local worktree bindings checked out to the branch + """ + records: list[dict] = [] + target_branch = (branch or "").strip() + if not target_branch: + return records + + # --- Author issue locks (session + lease) --- + try: + for path in issue_lock_store.iter_lock_files(): + lock = issue_lock_store.read_lock_file(path) + if not isinstance(lock, dict): + continue + if ( + str(lock.get("remote") or "") != str(remote) + or str(lock.get("org") or "") != str(org) + or str(lock.get("repo") or "") != str(repo) + ): + continue + if str(lock.get("branch_name") or "").strip() != target_branch: + continue + freshness = issue_lock_store.assess_lock_freshness(lock) + reclaim = issue_lock_store.assess_expired_lock_reclaim(lock) + status = str(freshness.get("status") or "unknown") + if freshness.get("live"): + status = "active" + records.append( + { + "category": branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "status": status, + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "reclaim_allowed": bool(reclaim.get("reclaim_allowed")), + "role": "author", + } + ) + # Session pointer binding (same record, distinct category when live) + if freshness.get("live"): + records.append( + { + "category": ( + branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_SESSION + ), + "status": "active", + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "reclaim_allowed": False, + "role": "author", + } + ) + except Exception: + # Fail closed: if lock inventory cannot be read, invent a sticky block. + records.append( + { + "category": branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "status": "unknown", + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "reclaim_allowed": False, + "role": "author", + } + ) + + # --- Control-plane leases (role-tagged) --- + try: + db = control_plane_db.get_db() if hasattr(control_plane_db, "get_db") else None + if db is None and hasattr(control_plane_db, "ControlPlaneDB"): + # Best-effort default path used by runtime tools. + try: + db = control_plane_db.ControlPlaneDB() + except TypeError: + db = None + if db is not None: + listed = lease_lifecycle.list_active_leases( + db, + remote=remote, + org=org, + repo=repo, + include_non_active=True, + limit=200, + ) + for lease in listed.get("leases") or []: + work_kind = str(lease.get("work_kind") or "") + work_number = lease.get("work_number") + lease_branch = str( + lease.get("branch") + or lease.get("branch_name") + or "" + ).strip() + matches_branch = lease_branch == target_branch + matches_pr = ( + work_kind == "pr" + and pr_number is not None + and int(work_number or 0) == int(pr_number) + ) + if not (matches_branch or matches_pr): + continue + if matches_pr and not lease_branch: + # PR-scoped lease without explicit branch still protects + # the merged PR source branch being cleaned. + lease_branch = target_branch + if lease_branch != target_branch: + continue + fr = lease.get("freshness") or lease_lifecycle.classify_lease_freshness( + lease + ) + freshness_status = str( + (fr.get("freshness") if isinstance(fr, dict) else None) + or lease.get("status") + or "unknown" + ) + role = str(lease.get("role") or "unknown") + category = branch_cleanup_guard.ownership_category_for_role(role) + reclaim_allowed = freshness_status in { + "released", + "abandoned", + "expired", + } and freshness_status != "active" + if freshness_status == "active": + status = "active" + reclaim_allowed = False + elif freshness_status in {"released", "abandoned"}: + status = freshness_status + reclaim_allowed = True + elif freshness_status == "expired" or ( + isinstance(fr, dict) and fr.get("expired_by_time") + ): + status = "expired" + # Sticky if session still alive with worktree — unknown here + # defaults to reclaimable when status is expired by time. + reclaim_allowed = True + else: + status = freshness_status + reclaim_allowed = False + records.append( + { + "category": category, + "status": status, + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "reclaim_allowed": reclaim_allowed, + "role": role, + } + ) + except Exception: + # Soft: control-plane may be unavailable in unit tests / offline. + pass + + # --- Worktree bindings checked out to the target branch --- + try: + for entry in worktree_cleanup_audit.list_worktrees(project_root): + wt_branch = str(entry.get("branch") or "").strip() + if wt_branch != target_branch: + continue + records.append( + { + "category": ( + branch_cleanup_guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING + ), + "status": "active", + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "reclaim_allowed": False, + "role": "worktree", + } + ) + except Exception: + records.append( + { + "category": branch_cleanup_guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING, + "status": "unknown", + "remote": remote, + "org": org, + "repo": repo, + "branch": target_branch, + "reclaim_allowed": False, + "role": "worktree", + } + ) + + return records @mcp.tool() diff --git a/tests/test_branch_cleanup_guard.py b/tests/test_branch_cleanup_guard.py index c916067..d7f913a 100644 --- a/tests/test_branch_cleanup_guard.py +++ b/tests/test_branch_cleanup_guard.py @@ -92,6 +92,11 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", return_value=True, ).start() + # Default: no active ownership records (tests that need ownership patch this). + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() def tearDown(self): patch.stopall() @@ -176,17 +181,31 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): "mcp_server.get_profile", return_value=dict(RECONCILER_WITH_DELETE), ).start() - self.mock_api.side_effect = [ - { - "number": 487, - "merged": True, - "merged_at": "2026-07-08T01:00:00Z", - "head": {"ref": branch, "sha": "a" * 40}, - "base": {"ref": "master"}, - }, - {}, - {}, - ] + + def _api(method, url, *args, **kwargs): + if method == "GET" and "/pulls/" in url: + return { + "number": 487, + "merged": True, + "merged_at": "2026-07-08T01:00:00Z", + "head": {"ref": branch, "sha": "a" * 40}, + "base": {"ref": "master"}, + } + if method == "GET" and "/branches/" in url: + # First pre-delete probe: present. Post-delete: not found. + get_branch_calls = [ + c + for c in self.mock_api.call_args_list + if c.args and c.args[0] == "GET" and "/branches/" in c.args[1] + ] + if len(get_branch_calls) <= 1: + return {"name": branch} + raise RuntimeError("HTTP 404: not found") + if method == "DELETE": + return {} + raise AssertionError(f"unexpected {method} {url}") + + self.mock_api.side_effect = _api res = gitea_cleanup_merged_pr_branch( pr_number=487, confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", @@ -194,7 +213,9 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): remote="prgs", worktree_path="/tmp/repo/branches/cleanup", ) + self.assertTrue(res["success"]) self.assertTrue(res["performed"]) + self.assertTrue((res.get("readback") or {}).get("verified_absent")) delete_calls = [ call for call in self.mock_api.call_args_list if call.args[0] == "DELETE" ] @@ -394,5 +415,433 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): ) + +class TestPostDeleteReadback(unittest.TestCase): + def test_not_found_is_verified_success(self): + readback = guard.classify_branch_readback_http_status(404) + result = guard.assess_post_delete_readback(readback) + self.assertTrue(result["ok"]) + self.assertTrue(result["readback"]["verified_absent"]) + + def test_exists_is_structured_failure(self): + readback = guard.classify_branch_readback_http_status(200) + result = guard.assess_post_delete_readback(readback) + self.assertFalse(result["ok"]) + self.assertFalse(result["readback"]["verified_absent"]) + self.assertTrue(result["readback"]["branch_present"]) + self.assertIn("still present", " ".join(result["reasons"])) + + def test_auth_failure_preserved(self): + readback = guard.classify_branch_readback_http_status(401) + result = guard.assess_post_delete_readback(readback) + self.assertFalse(result["ok"]) + self.assertEqual(result["readback"]["error_class"], "authentication") + self.assertIn("authentication", " ".join(result["reasons"])) + + def test_authz_and_transport_failures(self): + for code, err in ((403, "authorization"), (503, "transport")): + with self.subTest(code=code): + readback = guard.classify_branch_readback_http_status(code) + result = guard.assess_post_delete_readback(readback) + self.assertFalse(result["ok"]) + self.assertEqual(result["readback"]["error_class"], err) + + def test_exception_classifier_no_secret_leak(self): + class FakeHTTPError(Exception): + def __init__(self): + super().__init__("HTTP 401: token=super-secret-value Authorization: Bearer xyz") + self.code = 401 + + result = guard.classify_branch_readback_exception(FakeHTTPError()) + blob = str(result) + self.assertNotIn("super-secret", blob) + self.assertNotIn("Bearer", blob) + self.assertEqual(result["error_class"], "authentication") + + +class TestActiveBranchOwnership(unittest.TestCase): + def _base(self, **overrides): + rec = { + "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "status": "active", + "remote": "prgs", + "org": "Scaled-Tech-Consulting", + "repo": "Gitea-Tools", + "branch": "feat/target", + "reclaim_allowed": False, + } + rec.update(overrides) + return rec + + def test_active_author_blocks(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=[self._base()], + ) + self.assertTrue(result["block"]) + self.assertIn(guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, result["blocking_categories"]) + + def test_active_reviewer_lease_blocks(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=[ + self._base( + category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, + status="active", + ) + ], + ) + self.assertTrue(result["block"]) + self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, result["blocking_categories"]) + + def test_merger_controller_reconciler_binding_blocks(self): + for cat in ( + guard.OWNERSHIP_CATEGORY_MERGER_LEASE, + guard.OWNERSHIP_CATEGORY_CONTROLLER_LEASE, + guard.OWNERSHIP_CATEGORY_RECONCILER_LEASE, + guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING, + ): + with self.subTest(category=cat): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=[self._base(category=cat, status="active")], + ) + self.assertTrue(result["block"]) + self.assertIn(cat, result["blocking_categories"]) + + def test_released_and_expired_reclaimable_do_not_block(self): + records = [ + self._base(status="released", reclaim_allowed=True), + self._base( + category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, + status="expired", + reclaim_allowed=True, + ), + ] + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=records, + ) + self.assertFalse(result["block"]) + + def test_sticky_stale_blocks_per_recovery_policy(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=[self._base(status="stale", reclaim_allowed=False)], + ) + self.assertTrue(result["block"]) + self.assertIn("sticky", " ".join(result["reasons"])) + + def test_other_repo_or_branch_no_false_block(self): + records = [ + self._base(repo="Other-Repo", status="active"), + self._base(branch="feat/other", status="active"), + self._base(remote="dadeschools", status="active"), + ] + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=records, + ) + self.assertFalse(result["block"]) + self.assertEqual(len(result["ignored_out_of_scope"]), 3) + + def test_denial_has_no_secrets(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + records=[ + self._base( + status="active", + # Poison fields that must never be echoed as secrets + token="sekrit-token", + authorization="Bearer abc", + ) + ], + ) + blob = str(result) + self.assertNotIn("sekrit", blob) + self.assertNotIn("Bearer", blob) + self.assertIn("author_lease", blob) + + +class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): + def setUp(self): + self._remotes = patch.dict( + mcp_server.REMOTES, + { + "prgs": { + "host": "gitea.example.com", + "org": "Example-Org", + "repo": "Example-Repo", + } + }, + ) + self._remotes.start() + patch("gitea_audit.audit_enabled", return_value=False).start() + self.mock_api = patch("mcp_server.api_request").start() + 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() + + def tearDown(self): + patch.stopall() + + def _pr_payload(self, branch, number=487): + return { + "number": number, + "merged": True, + "merged_at": "2026-07-08T01:00:00Z", + "head": {"ref": branch, "sha": "a" * 40}, + "base": {"ref": "master"}, + } + + def test_delete_success_but_branch_remains(self): + branch = "feat/still-there" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch) + if method == "GET" and "/branches/" in url: + return {"name": branch} # present before and after + if method == "DELETE": + return {} + raise AssertionError(method) + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertTrue(res.get("performed")) + self.assertFalse(res.get("success")) + self.assertTrue(res.get("delete_acknowledged")) + self.assertIn("still present", " ".join(res.get("reasons") or [])) + + def test_readback_authentication_failure(self): + branch = "feat/auth-fail-readback" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() + state = {"branch_gets": 0} + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch) + if method == "GET" and "/branches/" in url: + state["branch_gets"] += 1 + if state["branch_gets"] == 1: + return {"name": branch} + raise RuntimeError("HTTP 401: unauthorized") + if method == "DELETE": + return {} + raise AssertionError(method) + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertTrue(res.get("performed")) + self.assertFalse(res.get("success")) + self.assertEqual((res.get("readback") or {}).get("error_class"), "authentication") + + def test_readback_transport_failure(self): + branch = "feat/transport-fail-readback" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() + state = {"branch_gets": 0} + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch) + if method == "GET" and "/branches/" in url: + state["branch_gets"] += 1 + if state["branch_gets"] == 1: + return {"name": branch} + raise RuntimeError("HTTP 503: temporarily unavailable") + if method == "DELETE": + return {} + raise AssertionError(method) + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res.get("success")) + self.assertEqual((res.get("readback") or {}).get("error_class"), "transport") + + def test_active_author_ownership_blocks_before_delete(self): + branch = "feat/owned" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[ + { + "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "status": "active", + "remote": "prgs", + "org": "Example-Org", + "repo": "Example-Repo", + "branch": branch, + "reclaim_allowed": False, + } + ], + ).start() + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch) + if method == "GET" and "/branches/" in url: + return {"name": branch} + raise AssertionError(f"unexpected mutation {method}") + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res.get("performed")) + self.assertFalse(res.get("success")) + self.assertEqual(res.get("blocker_kind"), "active_branch_ownership") + self.assertIn( + guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + (res.get("ownership") or {}).get("blocking_categories") or [], + ) + delete_calls = [ + c for c in self.mock_api.call_args_list if c.args and c.args[0] == "DELETE" + ] + self.assertFalse(delete_calls) + + def test_open_pr_guard_still_blocks(self): + branch = "feat/open-pr-head" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() + patch( + "mcp_server.api_get_all", + return_value=[{"head": {"ref": branch}, "number": 999}], + ).start() + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch, number=487) + if method == "GET" and "/branches/" in url: + return {"name": branch} + raise AssertionError(method) + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res.get("performed")) + self.assertTrue(any("open PR" in r for r in (res.get("reasons") or []))) + + def test_non_ancestor_guard_still_blocks(self): + branch = "feat/not-ancestor" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() + patch( + "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", + return_value=False, + ).start() + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch) + if method == "GET" and "/branches/" in url: + return {"name": branch} + raise AssertionError(method) + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res.get("performed")) + self.assertTrue(any("ancestor" in r for r in (res.get("reasons") or []))) + + def test_protected_default_branch_guard(self): + branch = "master" + patch( + "mcp_server._collect_branch_ownership_records", + return_value=[], + ).start() + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr_payload(branch) + if method == "GET" and "/branches/" in url: + return {"name": branch} + raise AssertionError(method) + + self.mock_api.side_effect = _api + res = gitea_cleanup_merged_pr_branch( + pr_number=487, + confirmation=f"CLEANUP MERGED PR 487 BRANCH {branch}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertFalse(res.get("performed")) + self.assertTrue(any("protected" in r for r in (res.get("reasons") or []))) + + + if __name__ == "__main__": unittest.main()