diff --git a/branch_cleanup_guard.py b/branch_cleanup_guard.py index 519e61c..64db84a 100644 --- a/branch_cleanup_guard.py +++ b/branch_cleanup_guard.py @@ -126,6 +126,13 @@ READBACK_AUTHENTICATION = "authentication_error" READBACK_AUTHORIZATION = "authorization_error" READBACK_TRANSPORT = "transport_error" READBACK_UNEXPECTED = "unexpected_response" +READBACK_AMBIGUOUS_404 = "ambiguous_not_found" + +# Scope of a 404: only branch-scoped absence may set verified_absent. +NOT_FOUND_SCOPE_BRANCH = "branch" +NOT_FOUND_SCOPE_REPOSITORY = "repository" +NOT_FOUND_SCOPE_HOST = "host" +NOT_FOUND_SCOPE_UNKNOWN = "unknown" ERROR_CLASS_AUTHENTICATION = "authentication" ERROR_CLASS_AUTHORIZATION = "authorization" @@ -139,6 +146,7 @@ OWNERSHIP_CATEGORY_MERGER_LEASE = "merger_lease" OWNERSHIP_CATEGORY_CONTROLLER_LEASE = "controller_lease" OWNERSHIP_CATEGORY_RECONCILER_LEASE = "reconciler_lease" OWNERSHIP_CATEGORY_WORKTREE_BINDING = "worktree_binding" +OWNERSHIP_CATEGORY_INVENTORY_ERROR = "ownership_inventory_error" _ROLE_TO_OWNERSHIP_CATEGORY = { "author": OWNERSHIP_CATEGORY_AUTHOR_LEASE, @@ -162,21 +170,66 @@ def _norm_str(value: Any) -> str: return str(value or "").strip() +def normalize_host(host: str | None) -> str: + """Normalize a host identity for ownership matching (no credentials).""" + text = _norm_str(host).lower() + for prefix in ("https://", "http://"): + if text.startswith(prefix): + text = text[len(prefix) :] + # Drop path/query if a full URL slipped through. + text = text.split("/", 1)[0] + text = text.split("?", 1)[0] + return text.rstrip(".") + + 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.""" +def classify_branch_readback_http_status( + status_code: int | None, + *, + not_found_scope: str | None = None, +) -> dict[str, Any]: + """Classify a GET-branch HTTP status into a secret-free readback result. + + R1: A bare/generic/repository/wrong-host 404 never yields + ``verified_absent=True``. Only an authoritative *branch-scoped* not-found + (``not_found_scope='branch'``) may verify deletion. + """ if status_code == 404: + scope = _norm_str(not_found_scope).lower() or NOT_FOUND_SCOPE_UNKNOWN + if scope == NOT_FOUND_SCOPE_BRANCH: + return { + "status": READBACK_NOT_FOUND, + "error_class": None, + "verified_absent": True, + "branch_present": False, + "not_found_scope": NOT_FOUND_SCOPE_BRANCH, + "reasons": [], + } + # repository / host / unknown / generic 404 — not verified absence + reason = { + NOT_FOUND_SCOPE_REPOSITORY: ( + "post-delete readback 404 is repository-scoped, not branch absence" + ), + NOT_FOUND_SCOPE_HOST: ( + "post-delete readback 404 is host-scoped, not branch absence" + ), + }.get( + scope, + "post-delete readback 404 is ambiguous (not branch-scoped); " + "cannot verify absence", + ) return { - "status": READBACK_NOT_FOUND, - "error_class": None, - "verified_absent": True, - "branch_present": False, - "reasons": [], + "status": READBACK_AMBIGUOUS_404, + "error_class": ERROR_CLASS_UNEXPECTED, + "verified_absent": False, + "branch_present": None, + "not_found_scope": scope, + "reasons": [reason], } if status_code in (401, 407): return { @@ -220,70 +273,73 @@ def classify_branch_readback_http_status(status_code: int | None) -> dict[str, A } -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 +def _extract_http_status(exc: BaseException) -> int | None: + """Extract an HTTP status code from an exception chain (secret-free).""" 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 "" + for attr in ("code", "status", "status_code"): + value = getattr(current, attr, None) + if isinstance(value, int) and 100 <= value <= 599: + return value + text = str(current) if current 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" - ], - } + return int(match.group(1)) + current = current.__cause__ or current.__context__ + return None + + +def classify_branch_readback_exception( + exc: BaseException, + *, + not_found_scope: str | None = None, +) -> dict[str, Any]: + """Classify a GET-branch exception without leaking credentials or bodies. + + R1: substring ``404`` / ``not found`` alone never becomes verified_absent. + Callers must pass ``not_found_scope='branch'`` only after authoritative + proof that the repository/host is still reachable and the 404 is branch-level. + """ + status_code = _extract_http_status(exc) + if status_code is None: + lower = str(exc).lower() if exc is not None else "" + if 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 "unauthorized" in lower: + status_code = 401 + elif "forbidden" in lower: + status_code = 403 + elif "404" in lower or "not found" in lower: + # Ambiguous: do NOT treat as branch absence without scope proof. + status_code = 404 + if not_found_scope is None: + not_found_scope = NOT_FOUND_SCOPE_UNKNOWN if status_code is not None: - return classify_branch_readback_http_status(status_code) + return classify_branch_readback_http_status( + status_code, not_found_scope=not_found_scope + ) return { "status": READBACK_UNEXPECTED, @@ -297,21 +353,26 @@ def classify_branch_readback_exception(exc: BaseException) -> dict[str, Any]: 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. + Success requires authoritative branch-scoped not-found + (``verified_absent=True``). DELETE HTTP success alone is never enough. + Generic/repository/host 404 cannot verify absence (R1). """ 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: + # Only explicit verified_absent flag counts — never infer from status alone + # when status is a bare not_found without branch scope proof. + verified = bool(payload.get("verified_absent")) is True + if verified and status == READBACK_NOT_FOUND: return { "ok": True, "success": True, + "verified_absent": True, "readback": { "status": READBACK_NOT_FOUND, "verified_absent": True, "branch_present": False, "error_class": None, + "not_found_scope": NOT_FOUND_SCOPE_BRANCH, }, "reasons": [], } @@ -326,29 +387,72 @@ def assess_post_delete_readback(readback: dict[str, Any] | None) -> dict[str, An reasons = ["post-delete branch readback authorization failed"] elif status == READBACK_TRANSPORT: reasons = ["post-delete branch readback transport/upstream failure"] + elif status == READBACK_AMBIGUOUS_404: + reasons = [ + "post-delete readback 404 is not branch-scoped; " + "cannot verify absence" + ] else: reasons = ["post-delete branch readback could not verify deletion"] return { "ok": False, "success": False, + "verified_absent": False, "readback": { "status": status, "verified_absent": False, "branch_present": payload.get("branch_present"), "error_class": payload.get("error_class"), + "not_found_scope": payload.get("not_found_scope"), }, "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 cleanup_result_envelope( + *, + success: bool, + performed: bool, + delete_acknowledged: bool, + verified_absent: bool, + **extra: Any, +) -> dict[str, Any]: + """R2: consistent top-level cleanup result fields on every return path.""" + out: dict[str, Any] = { + "success": bool(success), + "performed": bool(performed), + "delete_acknowledged": bool(delete_acknowledged), + "verified_absent": bool(verified_absent), + } + out.update(extra) + return out + + +def _repo_matches( + record: dict[str, Any], + *, + remote: str, + org: str, + repo: str, + host: str | None = None, +) -> bool: + """Match ownership record to target remote/org/repo and normalized host.""" + if ( + _norm_str(record.get("remote")).lower() != _norm_str(remote).lower() + or _norm_str(record.get("org")).lower() != _norm_str(org).lower() + or _norm_str(record.get("repo")).lower() != _norm_str(repo).lower() + ): + return False + expected_host = normalize_host(host) + record_host = normalize_host(record.get("host") or record.get("host_name")) + # When both sides declare a host, they must agree after normalization. + # A record host that disagrees with the expected host is out of scope. + # Legacy records without host still match when remote/org/repo agree. + if expected_host and record_host and expected_host != record_host: + return False + return True def _branch_matches(record: dict[str, Any], branch: str) -> bool: @@ -359,13 +463,21 @@ 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). + Expired/stale records block unless reclaim_allowed is *explicitly* True + (O2: never treat missing/unknown reclaim as auto-allowed). """ status = _norm_str(record.get("status")).lower() category = _norm_str(record.get("category")) or "unknown" reclaim_allowed = record.get("reclaim_allowed") + if category == OWNERSHIP_CATEGORY_INVENTORY_ERROR: + return { + "blocks": True, + "status": status or "unknown", + "category": category, + "reason": "ownership inventory failed closed", + } + if status in _TERMINAL_OWNERSHIP_STATUSES: return { "blocks": False, @@ -381,8 +493,7 @@ def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]: "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. + # O2: only explicit reclaim_allowed=True skips the block. if reclaim_allowed is True: return { "blocks": False, @@ -393,24 +504,13 @@ def assess_ownership_record_activity(record: dict[str, Any]) -> dict[str, Any]: "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" + f"{status} {category} ownership still protects the target " + "branch (reclaim not proven; fail closed)" ), } # Unknown status → fail closed @@ -431,15 +531,16 @@ def assess_active_branch_ownership( org: str, repo: str, branch: str, + host: str | None = None, 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. + Matching requires remote/org/repo/branch and, when provided, normalized + host identity. Other repositories, hosts, or branches never false-block. """ target_branch = _norm_str(branch) + expected_host = normalize_host(host) considered: list[dict[str, Any]] = [] blocking: list[dict[str, Any]] = [] ignored: list[dict[str, Any]] = [] @@ -447,11 +548,13 @@ def assess_active_branch_ownership( for raw in records or []: if not isinstance(raw, dict): continue - if not _repo_matches(raw, remote=remote, org=org, repo=repo): + if not _repo_matches( + raw, remote=remote, org=org, repo=repo, host=expected_host or None + ): ignored.append( { "category": _norm_str(raw.get("category")) or "unknown", - "reason": "different repository scope", + "reason": "different repository or host scope", } ) continue @@ -488,6 +591,7 @@ def assess_active_branch_ownership( "remote": remote, "org": org, "repo": repo, + "host": expected_host or None, "branch": target_branch, "blocking_categories": categories, "blocking": blocking, diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index 7c26ca0..084dc1f 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -6026,13 +6026,15 @@ def gitea_cleanup_merged_pr_branch( """Delete a merged PR source branch through the guarded MCP path (#514).""" gate_reasons = _profile_operation_gate("gitea.branch.delete") if gate_reasons: - return { - "success": False, - "performed": False, - "required_permission": "gitea.branch.delete", - "reasons": gate_reasons, - "permission_report": _permission_block_report("gitea.branch.delete"), - } + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=False, + required_permission="gitea.branch.delete", + reasons=gate_reasons, + permission_report=_permission_block_report("gitea.branch.delete"), + ) profile = get_profile() active_role = _profile_role_kind(profile) @@ -6040,29 +6042,33 @@ def gitea_cleanup_merged_pr_branch( # Author/reviewer/merger must not reach this path even if they somehow # hold gitea.branch.delete. if active_role != "reconciler": - return { - "success": False, - "performed": False, - "required_permission": "gitea.branch.delete", - "required_role_kind": "reconciler", - "active_role_kind": active_role, - "reasons": [ + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=False, + required_permission="gitea.branch.delete", + required_role_kind="reconciler", + active_role_kind=active_role, + reasons=[ f"profile role '{active_role}' is not authorized for merged " "branch cleanup; required role is reconciler (fail closed)" ], - "permission_report": _permission_block_report("gitea.branch.delete"), - } + permission_report=_permission_block_report("gitea.branch.delete"), + ) if worktree_path is None or "/branches/" not in os.path.realpath(worktree_path): - return { - "success": False, - "performed": False, - "required_permission": "gitea.branch.delete", - "reasons": [ + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=False, + required_permission="gitea.branch.delete", + reasons=[ "merged branch cleanup requires an explicit branches/ worktree " "path; root checkout branch ref mutation is blocked (fail closed)" ], - } + ) verify_preflight_purity( remote, @@ -6080,16 +6086,18 @@ def gitea_cleanup_merged_pr_branch( target_branch = (pr.get("base") or {}).get("ref") or "master" if branch and branch != pr_head.get("ref"): - return { - "success": False, - "performed": False, - "pr_number": pr_number, - "branch": branch, - "reasons": [ + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=False, + pr_number=pr_number, + branch=branch, + reasons=[ f"requested branch '{branch}' does not match PR head " f"'{pr_head.get('ref')}'" ], - } + ) open_prs = api_get_all(f"{base}/pulls?state=open", auth) open_heads = { @@ -6119,50 +6127,74 @@ def gitea_cleanup_merged_pr_branch( confirmation=confirmation, ) if not assessment["safe_to_delete"]: - return { - "success": False, - "performed": False, - "pr_number": pr_number, - "branch": head_branch, - "assessment": assessment, - "reasons": assessment["block_reasons"], - } + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=False, + pr_number=pr_number, + branch=head_branch, + assessment=assessment, + 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( + ownership_bundle = _collect_branch_ownership_records( remote=remote, + host=h, org=o, repo=r, branch=head_branch, pr_number=pr_number, project_root=PROJECT_ROOT, + auth=auth, + base_api=base, ) + ownership_records = ownership_bundle.get("records") or [] + if ownership_bundle.get("inventory_error"): + # O1: control-plane / ownership inventory failure fails closed. + ownership_records = list(ownership_records) + [ + { + "category": branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, + "status": "unknown", + "remote": remote, + "host": h, + "org": o, + "repo": r, + "branch": head_branch, + "reclaim_allowed": False, + "role": "inventory", + } + ] ownership = branch_cleanup_guard.assess_active_branch_ownership( remote=remote, org=o, repo=r, branch=head_branch, + host=h, records=ownership_records, ) if ownership.get("block"): - return { - "success": False, - "performed": False, - "pr_number": pr_number, - "branch": head_branch, - "assessment": assessment, - "ownership": { + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=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"), + "checked": True, }, - "reasons": ownership.get("reasons") or [ - "active ownership protects the target branch" - ], - "blocker_kind": "active_branch_ownership", - } + reasons=ownership.get("reasons") + or ["active ownership protects the target branch"], + blocker_kind="active_branch_ownership", + ) import urllib.parse @@ -6188,19 +6220,20 @@ def gitea_cleanup_merged_pr_branch( api_request("DELETE", url, auth) # #687: authoritative post-delete readback. DELETE success alone is not - # enough — only an authoritative not-found proves deletion. + # enough — only branch-scoped not-found proves deletion (R1). readback = _probe_remote_branch(h, o, r, auth, head_branch) readback_assessment = branch_cleanup_guard.assess_post_delete_readback(readback) + verified_absent = bool(readback_assessment.get("verified_absent")) request_metadata["post_delete_readback"] = { "status": (readback_assessment.get("readback") or {}).get("status"), - "verified_absent": bool( - (readback_assessment.get("readback") or {}).get("verified_absent") - ), + "verified_absent": verified_absent, "error_class": (readback_assessment.get("readback") or {}).get( "error_class" ), + "not_found_scope": (readback_assessment.get("readback") or {}).get( + "not_found_scope" + ), } - # Emit a second audit row capturing readback evidence (no secrets). if gitea_audit.audit_enabled(): _audit( "cleanup_merged_pr_branch_readback", @@ -6220,53 +6253,60 @@ def gitea_cleanup_merged_pr_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": { + return branch_cleanup_guard.cleanup_result_envelope( + success=False, + performed=True, + delete_acknowledged=True, + verified_absent=False, + 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") + 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": ( + 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": ( + return branch_cleanup_guard.cleanup_result_envelope( + success=True, + performed=True, + delete_acknowledged=True, + verified_absent=True, + pr_number=pr_number, + branch=head_branch, + message=( f"Merged PR #{pr_number} source branch '{head_branch}' deleted " - "and verified absent via post-delete readback." + "and verified absent via branch-scoped post-delete readback." ), - "assessment": assessment, - "ownership": { + assessment=assessment, + ownership={ "block": False, "blocking_categories": [], "checked": True, }, - "readback": readback_assessment.get("readback"), - } + readback=readback_assessment.get("readback"), + ) 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.""" + """GET a remote branch and return a secret-free structured readback. + + R1: On HTTP 404, re-probe the repository endpoint. Only when the repo is + still reachable is the 404 treated as branch-scoped absence. Generic, + repository-scoped, or host-level 404 never sets verified_absent. + """ import urllib.parse encoded = urllib.parse.quote(branch, safe="") @@ -6275,23 +6315,56 @@ def _probe_remote_branch( api_request("GET", url, auth) return branch_cleanup_guard.classify_branch_readback_http_status(200) except Exception as exc: - return branch_cleanup_guard.classify_branch_readback_exception(exc) + status = branch_cleanup_guard._extract_http_status(exc) + if status != 404: + return branch_cleanup_guard.classify_branch_readback_exception(exc) + + # Distinguish branch vs repository/host 404 via repo reachability. + repo_url = repo_api_url(h, o, r) + try: + api_request("GET", repo_url, auth) + # Repo reachable → branch-scoped not-found. + return branch_cleanup_guard.classify_branch_readback_http_status( + 404, + not_found_scope=branch_cleanup_guard.NOT_FOUND_SCOPE_BRANCH, + ) + except Exception as repo_exc: + repo_status = branch_cleanup_guard._extract_http_status(repo_exc) + if repo_status == 404: + return branch_cleanup_guard.classify_branch_readback_http_status( + 404, + not_found_scope=branch_cleanup_guard.NOT_FOUND_SCOPE_REPOSITORY, + ) + if repo_status in (401, 407): + return branch_cleanup_guard.classify_branch_readback_http_status( + 401 + ) + if repo_status == 403: + return branch_cleanup_guard.classify_branch_readback_http_status( + 403 + ) + # Host/transport/unknown: not branch-verified. + return branch_cleanup_guard.classify_branch_readback_http_status( + 404, + not_found_scope=branch_cleanup_guard.NOT_FOUND_SCOPE_UNKNOWN, + ) 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. + """True when the remote branch exists; False on authoritative branch 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. + Authentication, authorization, transport, and ambiguous 404 failures are + raised 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: + if ( + status == branch_cleanup_guard.READBACK_NOT_FOUND + and probe.get("verified_absent") + ): 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})" @@ -6301,23 +6374,44 @@ def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> boo def _collect_branch_ownership_records( *, remote: str, + host: str, org: str, repo: str, branch: str, pr_number: int | None, project_root: str, -) -> list[dict]: + auth: str | None = None, + base_api: str | None = None, +) -> 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) + - comment-backed active reviewer leases (O3) - local worktree bindings checked out to the branch + + Returns ``{"records": [...], "inventory_error": bool}``. Control-plane + inventory errors set inventory_error=True so callers fail closed (O1). """ records: list[dict] = [] + inventory_error = False target_branch = (branch or "").strip() if not target_branch: - return records + return {"records": records, "inventory_error": False} + + host_n = branch_cleanup_guard.normalize_host(host) + + def _base_rec(**kwargs): + rec = { + "remote": remote, + "host": host_n or host, + "org": org, + "repo": repo, + "branch": target_branch, + } + rec.update(kwargs) + return rec # --- Author issue locks (session + lease) --- try: @@ -6331,6 +6425,12 @@ def _collect_branch_ownership_records( or str(lock.get("repo") or "") != str(repo) ): continue + # Host identity when present on the lock + lock_host = branch_cleanup_guard.normalize_host( + lock.get("host") or lock.get("host_name") + ) + if host_n and lock_host and host_n != lock_host: + continue if str(lock.get("branch_name") or "").strip() != target_branch: continue freshness = issue_lock_store.assess_lock_freshness(lock) @@ -6339,58 +6439,53 @@ def _collect_branch_ownership_records( 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", - } + _base_rec( + category=branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + status=status, + 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": ( + _base_rec( + category=( branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_SESSION ), - "status": "active", - "remote": remote, - "org": org, - "repo": repo, - "branch": target_branch, - "reclaim_allowed": False, - "role": "author", - } + status="active", + reclaim_allowed=False, + role="author", + ) ) except Exception: - # Fail closed: if lock inventory cannot be read, invent a sticky block. + inventory_error = True 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", - } + _base_rec( + category=branch_cleanup_guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + status="unknown", + 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 + db = None + if hasattr(control_plane_db, "get_db"): + try: + db = control_plane_db.get_db() + except Exception: + db = 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: + except Exception: db = None - if db is not None: + inventory_error = True + if db is None: + # O1: unavailable control-plane inventory fails closed. + inventory_error = True + else: listed = lease_lifecycle.list_active_leases( db, remote=remote, @@ -6416,11 +6511,14 @@ def _collect_branch_ownership_records( 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 + lease_host = branch_cleanup_guard.normalize_host( + lease.get("host") or lease.get("host_name") + ) + if host_n and lease_host and host_n != lease_host: + continue fr = lease.get("freshness") or lease_lifecycle.classify_lease_freshness( lease ) @@ -6431,11 +6529,7 @@ def _collect_branch_ownership_records( ) 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" + # O2: expired never auto-receives reclaim_allowed=True. if freshness_status == "active": status = "active" reclaim_allowed = False @@ -6446,27 +6540,61 @@ def _collect_branch_ownership_records( 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 + reclaim_allowed = False # O2 fail closed 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, - } + _base_rec( + category=category, + status=status, + reclaim_allowed=reclaim_allowed, + role=role, + host=lease_host or host_n or host, + ) ) except Exception: - # Soft: control-plane may be unavailable in unit tests / offline. - pass + # O1: fail closed on control-plane inventory errors. + inventory_error = True + records.append( + _base_rec( + category=branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, + status="unknown", + reclaim_allowed=False, + role="control_plane", + ) + ) + + # --- O3: comment-backed active reviewer leases --- + if pr_number is not None and auth and base_api: + try: + comments = api_get_all( + f"{base_api}/issues/{int(pr_number)}/comments", auth + ) + active = reviewer_pr_lease.find_active_reviewer_lease( + comments, pr_number=int(pr_number) + ) + if active: + records.append( + _base_rec( + category=( + branch_cleanup_guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE + ), + status="active", + reclaim_allowed=False, + role="reviewer", + ) + ) + except Exception: + inventory_error = True + records.append( + _base_rec( + category=branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, + status="unknown", + reclaim_allowed=False, + role="reviewer_comment_lease", + ) + ) # --- Worktree bindings checked out to the target branch --- try: @@ -6475,34 +6603,28 @@ def _collect_branch_ownership_records( if wt_branch != target_branch: continue records.append( - { - "category": ( + _base_rec( + category=( branch_cleanup_guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING ), - "status": "active", - "remote": remote, - "org": org, - "repo": repo, - "branch": target_branch, - "reclaim_allowed": False, - "role": "worktree", - } + status="active", + reclaim_allowed=False, + role="worktree", + ) ) except Exception: + inventory_error = True 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", - } + _base_rec( + category=branch_cleanup_guard.OWNERSHIP_CATEGORY_WORKTREE_BINDING, + status="unknown", + reclaim_allowed=False, + role="worktree", + ) ) - return records + return {"records": records, "inventory_error": inventory_error} + @mcp.tool() @@ -6689,6 +6811,66 @@ def gitea_reconcile_merged_cleanups( if remote_assessment.get("safe_to_delete_remote"): import urllib.parse + pr_num = entry.get("pr_number") + try: + pr_num_int = int(pr_num) if pr_num is not None else None + except (TypeError, ValueError): + pr_num_int = None + ownership_bundle = _collect_branch_ownership_records( + remote=remote, + host=h, + org=o, + repo=r, + branch=head_branch, + pr_number=pr_num_int, + project_root=PROJECT_ROOT, + auth=auth, + base_api=base, + ) + ownership_records = list(ownership_bundle.get("records") or []) + if ownership_bundle.get("inventory_error"): + ownership_records.append( + { + "category": ( + branch_cleanup_guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR + ), + "status": "unknown", + "remote": remote, + "host": h, + "org": o, + "repo": r, + "branch": head_branch, + "reclaim_allowed": False, + "role": "inventory", + } + ) + ownership = branch_cleanup_guard.assess_active_branch_ownership( + remote=remote, + org=o, + repo=r, + branch=head_branch, + host=h, + records=ownership_records, + ) + if ownership.get("block"): + actions.append( + { + "action": "delete_remote_branch", + "branch": head_branch, + "success": False, + "performed": False, + "delete_acknowledged": False, + "verified_absent": False, + "blocker_kind": "active_branch_ownership", + "reasons": ownership.get("reasons") or [], + "blocking_categories": ownership.get( + "blocking_categories" + ) + or [], + } + ) + continue + encoded = urllib.parse.quote(head_branch, safe="") url = f"{base}/branches/{encoded}" with _audited( @@ -6698,14 +6880,28 @@ def gitea_reconcile_merged_cleanups( org=o, repo=r, target_branch=head_branch, - request_metadata={"branch": head_branch, "source": "reconcile_merged_cleanups"}, + request_metadata={ + "branch": head_branch, + "source": "reconcile_merged_cleanups", + "ownership_checked": True, + }, ): api_request("DELETE", url, auth) + readback = _probe_remote_branch(h, o, r, auth, head_branch) + readback_assessment = branch_cleanup_guard.assess_post_delete_readback( + readback + ) + verified = bool(readback_assessment.get("verified_absent")) actions.append( { "action": "delete_remote_branch", "branch": head_branch, - "success": True, + "success": bool(readback_assessment.get("ok")), + "performed": True, + "delete_acknowledged": True, + "verified_absent": verified, + "readback": readback_assessment.get("readback"), + "reasons": readback_assessment.get("reasons") or [], } ) diff --git a/tests/test_branch_cleanup_guard.py b/tests/test_branch_cleanup_guard.py index d7f913a..ec62ca3 100644 --- a/tests/test_branch_cleanup_guard.py +++ b/tests/test_branch_cleanup_guard.py @@ -95,7 +95,7 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): # Default: no active ownership records (tests that need ownership patch this). patch( "mcp_server._collect_branch_ownership_records", - return_value=[], + return_value={"records": [], "inventory_error": False}, ).start() def tearDown(self): @@ -201,6 +201,9 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): if len(get_branch_calls) <= 1: return {"name": branch} raise RuntimeError("HTTP 404: not found") + if method == "GET" and url.rstrip("/").endswith("/Example-Repo"): + # Repo reachability probe after branch 404 (R1 branch-scoped). + return {"full_name": "Example-Org/Example-Repo"} if method == "DELETE": return {} raise AssertionError(f"unexpected {method} {url}") @@ -215,6 +218,8 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): ) self.assertTrue(res["success"]) self.assertTrue(res["performed"]) + self.assertTrue(res["delete_acknowledged"]) + self.assertTrue(res["verified_absent"]) 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" @@ -417,12 +422,37 @@ class TestMergedPrBranchCleanupTool(unittest.TestCase): class TestPostDeleteReadback(unittest.TestCase): - def test_not_found_is_verified_success(self): - readback = guard.classify_branch_readback_http_status(404) + def test_branch_scoped_not_found_is_verified_success(self): + readback = guard.classify_branch_readback_http_status( + 404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH + ) result = guard.assess_post_delete_readback(readback) self.assertTrue(result["ok"]) + self.assertTrue(result["verified_absent"]) self.assertTrue(result["readback"]["verified_absent"]) + def test_generic_404_not_verified_absent(self): + # R1: bare 404 must not verify absence + for scope in (None, guard.NOT_FOUND_SCOPE_UNKNOWN, + guard.NOT_FOUND_SCOPE_REPOSITORY, + guard.NOT_FOUND_SCOPE_HOST): + with self.subTest(scope=scope): + readback = guard.classify_branch_readback_http_status( + 404, not_found_scope=scope + ) + self.assertFalse(readback["verified_absent"]) + result = guard.assess_post_delete_readback(readback) + self.assertFalse(result["ok"]) + self.assertFalse(result["verified_absent"]) + + def test_exception_substring_404_not_verified(self): + # R1: substring/generic 404 without scope stays unverified + result = guard.classify_branch_readback_exception( + RuntimeError("HTTP 404: something not found") + ) + self.assertFalse(result["verified_absent"]) + self.assertNotEqual(result.get("not_found_scope"), guard.NOT_FOUND_SCOPE_BRANCH) + def test_exists_is_structured_failure(self): readback = guard.classify_branch_readback_http_status(200) result = guard.assess_post_delete_readback(readback) @@ -465,6 +495,7 @@ class TestActiveBranchOwnership(unittest.TestCase): "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, "status": "active", "remote": "prgs", + "host": "gitea.prgs.cc", "org": "Scaled-Tech-Consulting", "repo": "Gitea-Tools", "branch": "feat/target", @@ -479,6 +510,7 @@ class TestActiveBranchOwnership(unittest.TestCase): org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=[self._base()], ) self.assertTrue(result["block"]) @@ -490,6 +522,7 @@ class TestActiveBranchOwnership(unittest.TestCase): org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=[ self._base( category=guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, @@ -513,6 +546,7 @@ class TestActiveBranchOwnership(unittest.TestCase): org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=[self._base(category=cat, status="active")], ) self.assertTrue(result["block"]) @@ -532,6 +566,7 @@ class TestActiveBranchOwnership(unittest.TestCase): org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=records, ) self.assertFalse(result["block"]) @@ -542,26 +577,47 @@ class TestActiveBranchOwnership(unittest.TestCase): org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=[self._base(status="stale", reclaim_allowed=False)], ) self.assertTrue(result["block"]) - self.assertIn("sticky", " ".join(result["reasons"])) + self.assertTrue( + any( + token in " ".join(result["reasons"]) + for token in ("sticky", "reclaim not proven", "fail closed") + ) + ) 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"), + self._base(host="gitea.other.host", status="active"), ] result = guard.assess_active_branch_ownership( remote="prgs", org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=records, ) self.assertFalse(result["block"]) - self.assertEqual(len(result["ignored_out_of_scope"]), 3) + self.assertEqual(len(result["ignored_out_of_scope"]), 4) + + def test_normalized_host_matching(self): + # Host identity is normalized (scheme/path stripped) + records = [self._base(host="https://gitea.prgs.cc/")] + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Scaled-Tech-Consulting", + repo="Gitea-Tools", + branch="feat/target", + host="gitea.prgs.cc", + records=records, + ) + self.assertTrue(result["block"]) def test_denial_has_no_secrets(self): result = guard.assess_active_branch_ownership( @@ -569,6 +625,7 @@ class TestActiveBranchOwnership(unittest.TestCase): org="Scaled-Tech-Consulting", repo="Gitea-Tools", branch="feat/target", + host="gitea.prgs.cc", records=[ self._base( status="active", @@ -626,7 +683,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): branch = "feat/still-there" patch( "mcp_server._collect_branch_ownership_records", - return_value=[], + return_value={"records": [], "inventory_error": False}, ).start() def _api(method, url, *a, **k): @@ -649,13 +706,14 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): self.assertTrue(res.get("performed")) self.assertFalse(res.get("success")) self.assertTrue(res.get("delete_acknowledged")) + self.assertFalse(res.get("verified_absent")) 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=[], + return_value={"records": [], "inventory_error": False}, ).start() state = {"branch_gets": 0} @@ -687,7 +745,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): branch = "feat/transport-fail-readback" patch( "mcp_server._collect_branch_ownership_records", - return_value=[], + return_value={"records": [], "inventory_error": False}, ).start() state = {"branch_gets": 0} @@ -718,17 +776,21 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): 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, - } - ], + return_value={ + "records": [ + { + "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "status": "active", + "remote": "prgs", + "host": "gitea.example.com", + "org": "Example-Org", + "repo": "Example-Repo", + "branch": branch, + "reclaim_allowed": False, + } + ], + "inventory_error": False, + }, ).start() def _api(method, url, *a, **k): @@ -762,7 +824,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): branch = "feat/open-pr-head" patch( "mcp_server._collect_branch_ownership_records", - return_value=[], + return_value={"records": [], "inventory_error": False}, ).start() patch( "mcp_server.api_get_all", @@ -791,7 +853,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): branch = "feat/not-ancestor" patch( "mcp_server._collect_branch_ownership_records", - return_value=[], + return_value={"records": [], "inventory_error": False}, ).start() patch( "mcp_server.merged_cleanup_reconcile.is_head_ancestor_of_ref", @@ -820,7 +882,7 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): branch = "master" patch( "mcp_server._collect_branch_ownership_records", - return_value=[], + return_value={"records": [], "inventory_error": False}, ).start() def _api(method, url, *a, **k): @@ -843,5 +905,362 @@ class TestCleanupReadbackAndOwnershipIntegration(unittest.TestCase): + +class TestSecondRemediationR1R2(unittest.TestCase): + """R1/R2 second-remediation: branch-scoped 404 and top-level fields.""" + + def test_cleanup_envelope_always_has_top_level_fields(self): + env = guard.cleanup_result_envelope( + success=False, + performed=False, + delete_acknowledged=False, + verified_absent=False, + reasons=["x"], + ) + for key in ("success", "performed", "delete_acknowledged", "verified_absent"): + self.assertIn(key, env) + self.assertIsInstance(env[key], bool) + + def test_repo_scoped_404_never_verified(self): + rb = guard.classify_branch_readback_http_status( + 404, not_found_scope=guard.NOT_FOUND_SCOPE_REPOSITORY + ) + self.assertFalse(rb["verified_absent"]) + assessed = guard.assess_post_delete_readback(rb) + self.assertFalse(assessed["ok"]) + self.assertFalse(assessed["verified_absent"]) + + def test_wrong_host_404_never_verified(self): + rb = guard.classify_branch_readback_http_status( + 404, not_found_scope=guard.NOT_FOUND_SCOPE_HOST + ) + self.assertFalse(rb["verified_absent"]) + + +class TestSecondRemediationOwnership(unittest.TestCase): + """O1/O2/O3 ownership second-remediation.""" + + def test_inventory_error_category_blocks(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Org", + repo="Repo", + branch="feat/x", + host="gitea.example.com", + records=[ + { + "category": guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, + "status": "unknown", + "remote": "prgs", + "host": "gitea.example.com", + "org": "Org", + "repo": "Repo", + "branch": "feat/x", + "reclaim_allowed": False, + } + ], + ) + self.assertTrue(result["block"]) + self.assertIn( + guard.OWNERSHIP_CATEGORY_INVENTORY_ERROR, + result["blocking_categories"], + ) + + def test_expired_without_explicit_reclaim_blocks(self): + # O2: expired must not auto-allow + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Org", + repo="Repo", + branch="feat/x", + host="gitea.example.com", + records=[ + { + "category": guard.OWNERSHIP_CATEGORY_MERGER_LEASE, + "status": "expired", + "remote": "prgs", + "host": "gitea.example.com", + "org": "Org", + "repo": "Repo", + "branch": "feat/x", + # reclaim_allowed omitted / False + "reclaim_allowed": False, + } + ], + ) + self.assertTrue(result["block"]) + + def test_expired_with_explicit_reclaim_allowed_does_not_block(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Org", + repo="Repo", + branch="feat/x", + host="gitea.example.com", + records=[ + { + "category": guard.OWNERSHIP_CATEGORY_AUTHOR_LEASE, + "status": "expired", + "remote": "prgs", + "host": "gitea.example.com", + "org": "Org", + "repo": "Repo", + "branch": "feat/x", + "reclaim_allowed": True, + } + ], + ) + self.assertFalse(result["block"]) + + def test_active_reviewer_comment_lease_blocks(self): + result = guard.assess_active_branch_ownership( + remote="prgs", + org="Org", + repo="Repo", + branch="feat/x", + host="gitea.example.com", + records=[ + { + "category": guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, + "status": "active", + "remote": "prgs", + "host": "gitea.example.com", + "org": "Org", + "repo": "Repo", + "branch": "feat/x", + "reclaim_allowed": False, + "role": "reviewer", + } + ], + ) + self.assertTrue(result["block"]) + self.assertIn( + guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, + result["blocking_categories"], + ) + + +class TestSecondRemediationIntegration(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(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_r1_repo_404_after_delete_not_verified(self): + """After DELETE, branch 404 + repo 404 must not verify absence.""" + branch = "feat/r1-repo-404" + patch( + "mcp_server._collect_branch_ownership_records", + return_value={"records": [], "inventory_error": False}, + ).start() + state = {"branch_gets": 0} + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr(branch) + if method == "GET" and "/branches/" in url: + state["branch_gets"] += 1 + if state["branch_gets"] == 1: + return {"name": branch} + raise RuntimeError("HTTP 404: not found") + if method == "GET" and url.rstrip("/").endswith("/Example-Repo"): + raise RuntimeError("HTTP 404: repository not found") + if method == "DELETE": + return {} + raise AssertionError(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}", + branch=branch, + remote="prgs", + worktree_path="/tmp/repo/branches/cleanup", + ) + self.assertTrue(res["performed"]) + self.assertTrue(res["delete_acknowledged"]) + self.assertFalse(res["success"]) + self.assertFalse(res["verified_absent"]) + + def test_o1_inventory_error_blocks_before_delete(self): + branch = "feat/o1-inv" + patch( + "mcp_server._collect_branch_ownership_records", + return_value={"records": [], "inventory_error": True}, + ).start() + + def _api(method, url, *a, **k): + if method == "GET" and "/pulls/" in url: + return self._pr(branch) + if method == "GET" and "/branches/" in url: + return {"name": branch} + raise AssertionError(f"no mutation expected {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["performed"]) + self.assertFalse(res["delete_acknowledged"]) + self.assertFalse(res["verified_absent"]) + self.assertEqual(res.get("blocker_kind"), "active_branch_ownership") + + def test_o3_comment_reviewer_lease_in_collector(self): + """Collector includes active comment-backed reviewer leases.""" + active_lease = { + "pr_number": 10, + "phase": "claimed", + "session_id": "s1", + "expires_at": "2099-01-02T00:00:00Z", + } + with patch( + "mcp_server.api_get_all", return_value=[{"id": 1, "body": "x"}] + ), patch( + "mcp_server.reviewer_pr_lease.find_active_reviewer_lease", + return_value=active_lease, + ), patch( + "mcp_server.issue_lock_store.iter_lock_files", return_value=[] + ), patch( + "mcp_server.worktree_cleanup_audit.list_worktrees", return_value=[] + ), patch.object( + mcp_server.control_plane_db, + "ControlPlaneDB", + side_effect=RuntimeError("no cp"), + ): + bundle = mcp_server._collect_branch_ownership_records( + remote="prgs", + host="gitea.example.com", + org="Example-Org", + repo="Example-Repo", + branch="feat/x", + pr_number=10, + project_root="/tmp/repo", + auth=FAKE_AUTH, + base_api=( + "https://gitea.example.com/api/v1/repos/" + "Example-Org/Example-Repo" + ), + ) + cats = {r.get("category") for r in bundle.get("records") or []} + self.assertIn(guard.OWNERSHIP_CATEGORY_REVIEWER_LEASE, cats) + + def test_o4_reconcile_merged_cleanups_runs_ownership_and_readback(self): + from mcp_server import gitea_reconcile_merged_cleanups + + branch = "feat/reconcile-o4" + ownership_calls = [] + + def fake_collect(**kwargs): + ownership_calls.append(kwargs) + return {"records": [], "inventory_error": False} + + probe_calls = [] + + def fake_probe(h, o, r, auth, br): + probe_calls.append(br) + return guard.classify_branch_readback_http_status( + 404, not_found_scope=guard.NOT_FOUND_SCOPE_BRANCH + ) + + report = { + "entries": [ + { + "pr_number": 1, + "head_branch": branch, + "remote_branch": {"safe_to_delete_remote": True}, + "local_worktree": {"safe_to_remove_worktree": False}, + } + ], + "reviewer_scratch_entries": [], + } + patch( + "mcp_server.get_profile", + return_value={ + "profile_name": "prgs-reconciler", + "role": "reconciler", + "allowed_operations": [ + "gitea.read", + "gitea.branch.delete", + "gitea.pr.close", + ], + "forbidden_operations": [], + }, + ).start() + patch("mcp_server.api_get_all", return_value=[]).start() + patch( + "mcp_server.merged_cleanup_reconcile.build_reconciliation_report", + return_value=report, + ).start() + patch( + "mcp_server.merged_cleanup_reconcile.discover_reviewer_scratch_worktrees", + return_value=[], + ).start() + patch( + "mcp_server.audit_reconciliation_mode.check_cleanup_execution_allowed", + return_value=(True, []), + ).start() + patch("mcp_server.verify_preflight_purity", return_value=None).start() + patch( + "mcp_server._collect_branch_ownership_records", + side_effect=fake_collect, + ).start() + patch("mcp_server._probe_remote_branch", side_effect=fake_probe).start() + self.mock_api.side_effect = lambda *a, **k: {} + + res = gitea_reconcile_merged_cleanups( + dry_run=False, + execute_confirmed=True, + remote="prgs", + ) + self.assertTrue(res.get("performed") or res.get("executed")) + self.assertTrue(ownership_calls, "ownership must run before delete") + self.assertTrue(probe_calls, "post-delete readback must run") + actions = res.get("actions") or [] + delete_actions = [ + a for a in actions if a.get("action") == "delete_remote_branch" + ] + self.assertEqual(len(delete_actions), 1) + self.assertIn("verified_absent", delete_actions[0]) + self.assertIn("delete_acknowledged", delete_actions[0]) + self.assertTrue(delete_actions[0].get("verified_absent")) + + + if __name__ == "__main__": unittest.main()