feat: guard merged branch cleanup path

Closes #514
This commit is contained in:
2026-07-08 04:18:09 -04:00
parent 4d24c34548
commit 3cb05284b3
7 changed files with 460 additions and 1 deletions
+144
View File
@@ -615,6 +615,7 @@ import reconciliation_workflow # noqa: E402
import review_merge_state_machine # noqa: E402
import pr_work_lease # noqa: E402
import native_mcp_preference # noqa: E402
import branch_cleanup_guard # noqa: E402
# Keyed issue-lock storage (#443): per remote/org/repo/issue files under
@@ -4116,6 +4117,149 @@ def gitea_delete_branch(
return {"success": True, "message": f"Remote branch '{branch}' deleted."}
@mcp.tool()
def gitea_cleanup_merged_pr_branch(
pr_number: int,
confirmation: str,
branch: str | None = None,
remote: str = "dadeschools",
host: str | None = None,
org: str | None = None,
repo: str | None = None,
worktree_path: str | None = None,
) -> dict:
"""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"),
}
profile = get_profile()
active_role = _role_kind(
profile.get("allowed_operations", []),
profile.get("forbidden_operations", []),
)
if active_role == "reviewer":
return {
"success": False,
"performed": False,
"required_permission": "gitea.branch.delete",
"reasons": [
"reviewer profile is not authorized for merged branch cleanup "
"(fail closed)"
],
"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": [
"merged branch cleanup requires an explicit branches/ worktree "
"path; root checkout branch ref mutation is blocked (fail closed)"
],
}
verify_preflight_purity(
remote,
worktree_path=worktree_path,
task="cleanup_merged_pr_branch",
)
h, o, r = _resolve(remote, host, org, repo)
auth = _auth(h)
base = repo_api_url(h, o, r)
pr = api_request("GET", f"{base}/pulls/{pr_number}", auth)
pr_head = pr.get("head") or {}
head_branch = branch or pr_head.get("ref") or ""
head_sha = pr_head.get("sha")
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": [
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 = {
str((open_pr.get("head") or {}).get("ref"))
for open_pr in open_prs
if (open_pr.get("head") or {}).get("ref")
}
remote_exists = _remote_branch_exists(h, o, r, auth, head_branch)
target_ref = (
f"{remote}/{target_branch}"
if remote in REMOTES
else f"origin/{target_branch}"
)
head_on_target = merged_cleanup_reconcile.is_head_ancestor_of_ref(
PROJECT_ROOT,
head_sha,
target_ref,
)
assessment = branch_cleanup_guard.assess_merged_pr_branch_cleanup(
pr_number=pr_number,
head_branch=head_branch,
merged=bool(pr.get("merged") or pr.get("merged_at")),
remote_branch_exists=remote_exists,
open_pr_heads=open_heads,
head_on_target=head_on_target,
delete_capability_allowed=True,
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"],
}
import urllib.parse
encoded_branch = urllib.parse.quote(head_branch, safe="")
url = f"{base}/branches/{encoded_branch}"
with _audited(
"cleanup_merged_pr_branch",
host=h,
remote=remote,
org=o,
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",
},
):
api_request("DELETE", url, auth)
return {
"success": True,
"performed": True,
"pr_number": pr_number,
"branch": head_branch,
"message": f"Merged PR #{pr_number} source branch '{head_branch}' deleted.",
"assessment": assessment,
}
def _remote_branch_exists(h: str, o: str, r: str, auth: str, branch: str) -> bool:
import urllib.parse