373 lines
12 KiB
Python
373 lines
12 KiB
Python
"""Merged PR branch and worktree cleanup reconciliation (#269).
|
|
|
|
Builds dry-run reports for merged pull requests and optionally executes
|
|
remote branch deletion and local worktree removal when every safety gate passes.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
from typing import Any
|
|
|
|
from reviewer_worktree import parse_dirty_tracked_files
|
|
import issue_lock_store
|
|
|
|
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
|
|
CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
|
|
|
|
|
|
def extract_linked_issue(title: str, body: str) -> int | None:
|
|
"""Return the first Closes/Fixes issue number from PR metadata."""
|
|
for text in (title or "", body or ""):
|
|
match = CLOSES_FIXES_RE.search(text)
|
|
if match:
|
|
return int(match.group(1))
|
|
return None
|
|
|
|
|
|
def branch_worktree_folder(branch: str) -> str:
|
|
return (branch or "").replace("/", "-")
|
|
|
|
|
|
def resolve_worktree_path(project_root: str, branch: str) -> str:
|
|
return os.path.join(project_root, "branches", branch_worktree_folder(branch))
|
|
|
|
|
|
def discover_local_worktrees(project_root: str) -> dict[str, str]:
|
|
"""Return a mapping from branch name to absolute worktree path (#528)."""
|
|
res = subprocess.run(
|
|
["git", "-C", project_root, "worktree", "list", "--porcelain"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if res.returncode != 0:
|
|
return {}
|
|
|
|
mapping: dict[str, str] = {}
|
|
current_worktree = None
|
|
for line in res.stdout.splitlines():
|
|
line = line.strip()
|
|
if not line:
|
|
current_worktree = None
|
|
continue
|
|
if line.startswith("worktree "):
|
|
current_worktree = line[9:].strip()
|
|
elif line.startswith("branch ") and current_worktree:
|
|
ref = line[7:].strip()
|
|
if ref.startswith("refs/heads/"):
|
|
branch_name = ref[11:]
|
|
mapping[branch_name] = current_worktree
|
|
return mapping
|
|
|
|
|
|
def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
|
|
if path:
|
|
return issue_lock_store.read_lock_file(path.strip())
|
|
return issue_lock_store.read_session_issue_lock()
|
|
|
|
|
|
def has_active_issue_lock(branch: str, lock_path: str | None = None) -> bool:
|
|
if lock_path:
|
|
lock = issue_lock_store.read_lock_file(lock_path.strip())
|
|
if not lock:
|
|
return False
|
|
return (lock.get("branch_name") or "").strip() == (branch or "").strip()
|
|
return issue_lock_store.has_active_issue_lock(branch)
|
|
|
|
|
|
def collect_open_pr_heads(open_prs: list[dict[str, Any]]) -> set[str]:
|
|
heads: set[str] = set()
|
|
for pr in open_prs or []:
|
|
head = pr.get("head")
|
|
if isinstance(head, dict):
|
|
ref = head.get("ref")
|
|
else:
|
|
ref = head
|
|
if ref:
|
|
heads.add(str(ref))
|
|
return heads
|
|
|
|
|
|
def read_local_worktree_state(worktree_path: str) -> dict[str, Any]:
|
|
path = (worktree_path or "").strip()
|
|
if not path or not os.path.isdir(path):
|
|
return {
|
|
"exists": False,
|
|
"clean": None,
|
|
"current_branch": None,
|
|
"head_sha": None,
|
|
"dirty_files": [],
|
|
}
|
|
|
|
branch_res = subprocess.run(
|
|
["git", "-C", path, "branch", "--show-current"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
current_branch = (branch_res.stdout or "").strip() or None
|
|
|
|
status_res = subprocess.run(
|
|
["git", "-C", path, "status", "--porcelain"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
dirty_files = parse_dirty_tracked_files(status_res.stdout or "")
|
|
|
|
head_res = subprocess.run(
|
|
["git", "-C", path, "rev-parse", "HEAD"],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
head_sha = (head_res.stdout or "").strip() if head_res.returncode == 0 else None
|
|
|
|
return {
|
|
"exists": True,
|
|
"clean": not dirty_files,
|
|
"current_branch": current_branch,
|
|
"head_sha": head_sha,
|
|
"dirty_files": dirty_files,
|
|
}
|
|
|
|
|
|
def is_head_ancestor_of_ref(project_root: str, head_sha: str | None, base_ref: str) -> bool | None:
|
|
if not head_sha:
|
|
return None
|
|
res = subprocess.run(
|
|
["git", "-C", project_root, "merge-base", "--is-ancestor", head_sha, base_ref],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if res.returncode == 0:
|
|
return True
|
|
if res.returncode == 1:
|
|
return False
|
|
return None
|
|
|
|
|
|
def assess_remote_branch_cleanup(
|
|
*,
|
|
pr_number: int,
|
|
head_branch: str,
|
|
merged: bool,
|
|
remote_branch_exists: bool,
|
|
open_pr_heads: set[str],
|
|
protected_branches: frozenset[str] | None = None,
|
|
head_on_master: bool | None,
|
|
delete_capability_allowed: bool,
|
|
active_lock: bool,
|
|
) -> dict[str, Any]:
|
|
protected = protected_branches or PROTECTED_BRANCHES
|
|
reasons: list[str] = []
|
|
if not merged:
|
|
reasons.append("PR is not merged")
|
|
if not remote_branch_exists:
|
|
reasons.append("remote branch already absent")
|
|
if head_branch in protected:
|
|
reasons.append(f"branch '{head_branch}' is protected")
|
|
if head_branch in open_pr_heads:
|
|
reasons.append("an open PR still references this head branch")
|
|
if active_lock:
|
|
reasons.append("active issue lock references this branch")
|
|
if head_on_master is False:
|
|
reasons.append("PR head is not an ancestor of master")
|
|
if not delete_capability_allowed:
|
|
reasons.append("delete_branch capability is not allowed in the active profile")
|
|
|
|
safe = merged and remote_branch_exists and not reasons
|
|
return {
|
|
"pr_number": pr_number,
|
|
"head_branch": head_branch,
|
|
"remote_branch_exists": remote_branch_exists,
|
|
"safe_to_delete_remote": safe,
|
|
"block_reasons": reasons,
|
|
"recommended_action": "delete_remote_branch" if safe else "keep_remote_branch",
|
|
}
|
|
|
|
|
|
def assess_local_worktree_cleanup(
|
|
*,
|
|
pr_number: int,
|
|
head_branch: str,
|
|
merged: bool,
|
|
worktree_state: dict[str, Any],
|
|
active_lock: bool,
|
|
) -> dict[str, Any]:
|
|
reasons: list[str] = []
|
|
exists = bool(worktree_state.get("exists"))
|
|
if not merged:
|
|
reasons.append("PR is not merged")
|
|
if not exists:
|
|
reasons.append("local worktree not present")
|
|
if exists and worktree_state.get("clean") is False:
|
|
dirty = worktree_state.get("dirty_files") or []
|
|
reasons.append(
|
|
"local worktree has tracked edits"
|
|
+ (f" ({', '.join(dirty)})" if dirty else "")
|
|
)
|
|
if exists:
|
|
current_branch = worktree_state.get("current_branch")
|
|
if current_branch and current_branch != head_branch:
|
|
reasons.append(
|
|
f"worktree branch '{current_branch}' does not match PR head '{head_branch}'"
|
|
)
|
|
if active_lock:
|
|
reasons.append("active issue lock references this branch")
|
|
|
|
safe = merged and exists and not reasons
|
|
return {
|
|
"pr_number": pr_number,
|
|
"head_branch": head_branch,
|
|
"worktree_path": worktree_state.get("worktree_path"),
|
|
"worktree_exists": exists,
|
|
"worktree_clean": worktree_state.get("clean"),
|
|
"safe_to_remove_worktree": safe,
|
|
"block_reasons": reasons,
|
|
"recommended_action": "remove_local_worktree" if safe else "keep_local_worktree",
|
|
}
|
|
|
|
|
|
def build_pr_cleanup_entry(
|
|
*,
|
|
pr: dict[str, Any],
|
|
project_root: str,
|
|
open_pr_heads: set[str],
|
|
remote_branch_exists: bool,
|
|
head_on_master: bool | None,
|
|
delete_capability_allowed: bool,
|
|
issue_lock_path: str | None = None,
|
|
protected_branches: frozenset[str] | None = None,
|
|
worktree_map: dict[str, str] | None = None,
|
|
) -> dict[str, Any]:
|
|
pr_number = int(pr["number"])
|
|
head_branch = pr.get("head") or ""
|
|
if isinstance(head_branch, dict):
|
|
head_branch = head_branch.get("ref") or ""
|
|
title = pr.get("title") or ""
|
|
body = pr.get("body") or ""
|
|
merged = bool(pr.get("merged_at"))
|
|
|
|
discovered_path = worktree_map.get(head_branch) if worktree_map else None
|
|
derived_path = resolve_worktree_path(project_root, head_branch)
|
|
if discovered_path:
|
|
worktree_path = discovered_path
|
|
match_type = "branch"
|
|
else:
|
|
worktree_path = derived_path
|
|
match_type = "path"
|
|
|
|
worktree_state = read_local_worktree_state(worktree_path)
|
|
worktree_state["worktree_path"] = worktree_path
|
|
active_lock = has_active_issue_lock(head_branch, issue_lock_path)
|
|
|
|
remote = assess_remote_branch_cleanup(
|
|
pr_number=pr_number,
|
|
head_branch=head_branch,
|
|
merged=merged,
|
|
remote_branch_exists=remote_branch_exists,
|
|
open_pr_heads=open_pr_heads,
|
|
protected_branches=protected_branches,
|
|
head_on_master=head_on_master,
|
|
delete_capability_allowed=delete_capability_allowed,
|
|
active_lock=active_lock,
|
|
)
|
|
local = assess_local_worktree_cleanup(
|
|
pr_number=pr_number,
|
|
head_branch=head_branch,
|
|
merged=merged,
|
|
worktree_state=worktree_state,
|
|
active_lock=active_lock,
|
|
)
|
|
local["match_type"] = match_type
|
|
|
|
return {
|
|
"pr_number": pr_number,
|
|
"issue_number": extract_linked_issue(title, body),
|
|
"title": title,
|
|
"head_branch": head_branch,
|
|
"merge_commit_sha": pr.get("merge_commit_sha"),
|
|
"merged_at": pr.get("merged_at"),
|
|
"merged": merged,
|
|
"remote_branch": remote,
|
|
"local_worktree": local,
|
|
}
|
|
|
|
|
|
def build_reconciliation_report(
|
|
*,
|
|
project_root: str,
|
|
closed_prs: list[dict[str, Any]],
|
|
open_prs: list[dict[str, Any]],
|
|
remote_branch_exists: dict[str, bool],
|
|
head_on_master: dict[int, bool | None],
|
|
delete_capability_allowed: bool,
|
|
issue_lock_path: str | None = None,
|
|
protected_branches: frozenset[str] | None = None,
|
|
) -> dict[str, Any]:
|
|
open_heads = collect_open_pr_heads(open_prs)
|
|
worktree_map = discover_local_worktrees(project_root)
|
|
entries: list[dict[str, Any]] = []
|
|
for pr in closed_prs or []:
|
|
if not pr.get("merged_at"):
|
|
continue
|
|
pr_number = int(pr["number"])
|
|
head = pr.get("head") or ""
|
|
if isinstance(head, dict):
|
|
head = head.get("ref") or ""
|
|
entries.append(
|
|
build_pr_cleanup_entry(
|
|
pr=pr,
|
|
project_root=project_root,
|
|
open_pr_heads=open_heads,
|
|
remote_branch_exists=bool(remote_branch_exists.get(head)),
|
|
head_on_master=head_on_master.get(pr_number),
|
|
delete_capability_allowed=delete_capability_allowed,
|
|
issue_lock_path=issue_lock_path,
|
|
protected_branches=protected_branches,
|
|
worktree_map=worktree_map,
|
|
)
|
|
)
|
|
return {
|
|
"project_root": os.path.realpath(project_root),
|
|
"merged_pr_count": len(entries),
|
|
"entries": entries,
|
|
"dry_run": True,
|
|
"executed": False,
|
|
}
|
|
|
|
|
|
def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
|
|
worktree_map = discover_local_worktrees(project_root)
|
|
discovered_path = worktree_map.get(branch)
|
|
worktree_path = discovered_path or resolve_worktree_path(project_root, branch)
|
|
if not os.path.isdir(worktree_path):
|
|
return {
|
|
"success": False,
|
|
"performed": False,
|
|
"message": f"worktree not found: {worktree_path}",
|
|
}
|
|
res = subprocess.run(
|
|
["git", "-C", project_root, "worktree", "remove", worktree_path],
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if res.returncode != 0:
|
|
return {
|
|
"success": False,
|
|
"performed": False,
|
|
"message": (res.stderr or res.stdout or "worktree remove failed").strip(),
|
|
}
|
|
return {
|
|
"success": True,
|
|
"performed": True,
|
|
"message": f"removed worktree {worktree_path}",
|
|
"worktree_path": worktree_path,
|
|
} |