Scaled-Tech-Consulting/Gitea-Tools
2
0
Fork 0
Code Issues 73 Pull Requests 9 Actions Packages Projects Releases Wiki Activity

feat: discover merged cleanup worktree aliases

Browse Source
This commit is contained in:
sysadmin
2026-07-08 23:04:57 -04:00
parent 719de6fb7a
commit 8d9f1f2940
3 changed files with 376 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gitea_mcp_server.py
+4 -1
View File
@@ -4744,6 +4744,7 @@ def gitea_reconcile_merged_cleanups(
remote_branch_exists=remote_branch_exists,
head_on_master=head_on_master,
delete_capability_allowed=delete_capability_allowed,
target_ref=master_ref,
)
if dry_run:
@@ -4783,7 +4784,9 @@ def gitea_reconcile_merged_cleanups(
if local_assessment.get("safe_to_remove_worktree"):
result = merged_cleanup_reconcile.remove_local_worktree(
PROJECT_ROOT, head_branch
PROJECT_ROOT,
head_branch,
worktree_path=local_assessment.get("worktree_path"),
)
actions.append({"action": "remove_local_worktree", **result})
merged_cleanup_reconcile.py
+181 -8
View File
@@ -17,6 +17,7 @@ import issue_lock_store
PROTECTED_BRANCHES = frozenset({"master", "main", "dev"})
CLOSES_FIXES_RE = re.compile(r"\b(?:closes|fixes)\s+#(\d+)\b", re.IGNORECASE)
ISSUE_MARKER_RE = re.compile(r"(?:^|[-_/])issue-(\d+)(?:[-_/]|$)", re.IGNORECASE)
def extract_linked_issue(title: str, body: str) -> int | None:
@@ -36,6 +37,155 @@ def resolve_worktree_path(project_root: str, branch: str) -> str:
return os.path.join(project_root, "branches", branch_worktree_folder(branch))
def extract_issue_marker(*values: str | None) -> int | None:
"""Return the first issue marker found in branch/path-like values."""
for value in values:
match = ISSUE_MARKER_RE.search(value or "")
if match:
return int(match.group(1))
return None
def list_local_worktrees(project_root: str) -> list[dict[str, Any]]:
"""Parse ``git worktree list --porcelain`` into structured entries."""
res = subprocess.run(
["git", "-C", project_root, "worktree", "list", "--porcelain"],
capture_output=True,
text=True,
check=False,
)
if res.returncode != 0:
return []
entries: list[dict[str, Any]] = []
current: dict[str, Any] | None = None
def flush() -> None:
nonlocal current
if current:
entries.append(current)
current = None
for raw_line in (res.stdout or "").splitlines():
line = raw_line.strip()
if not line:
flush()
continue
if line.startswith("worktree "):
flush()
current = {"path": line[9:].strip()}
elif current is not None and line.startswith("HEAD "):
current["head_sha"] = line[5:].strip()
elif current is not None and line.startswith("branch "):
ref = line[7:].strip()
current["branch_ref"] = ref
current["branch"] = (
ref[len("refs/heads/"):]
if ref.startswith("refs/heads/")
else ref
)
elif current is not None and line == "detached":
current["detached"] = True
flush()
return entries
def _is_under_branches(project_root: str, path: str) -> bool:
branches_root = os.path.realpath(os.path.join(project_root, "branches"))
real_path = os.path.realpath(path)
return real_path == branches_root or real_path.startswith(branches_root + os.sep)
def _head_is_safe_for_cleanup(
*,
project_root: str,
candidate_head: str | None,
pr_head_sha: str | None,
target_ref: str | None,
) -> bool:
if not candidate_head:
return False
if pr_head_sha and candidate_head == pr_head_sha:
return True
if target_ref:
return is_head_ancestor_of_ref(project_root, candidate_head, target_ref) is True
return False
def resolve_cleanup_worktree_state(
*,
project_root: str,
head_branch: str,
issue_number: int | None,
pr_head_sha: str | None = None,
target_ref: str | None = None,
) -> dict[str, Any]:
"""Find the exact or safe alias worktree used for local cleanup (#532)."""
expected_path = resolve_worktree_path(project_root, head_branch)
state = read_local_worktree_state(expected_path)
state["worktree_path"] = expected_path
state["expected_worktree_path"] = expected_path
state["discovered_worktree_path"] = expected_path if state.get("exists") else None
state["match_type"] = "derived_path" if state.get("exists") else "none"
state["candidate_paths"] = []
state["alias_block_reasons"] = []
if state.get("exists"):
return state
branch_issue = extract_issue_marker(head_branch)
wanted_issue = issue_number or branch_issue
candidates: list[dict[str, Any]] = []
unsafe_matches: list[str] = []
for entry in list_local_worktrees(project_root):
path = entry.get("path") or ""
if not path or not _is_under_branches(project_root, path):
continue
branch = entry.get("branch") or ""
path_issue = extract_issue_marker(os.path.basename(path), path)
entry_issue = extract_issue_marker(branch) or path_issue
branch_match = branch == head_branch
issue_match = wanted_issue is not None and entry_issue == wanted_issue
if not (branch_match or issue_match):
continue
if not _head_is_safe_for_cleanup(
project_root=project_root,
candidate_head=entry.get("head_sha"),
pr_head_sha=pr_head_sha,
target_ref=target_ref,
):
unsafe_matches.append(path)
continue
candidates.append(entry)
state["candidate_paths"] = [entry.get("path") for entry in candidates]
if len(candidates) == 1:
path = candidates[0]["path"]
alias_state = read_local_worktree_state(path)
alias_state["worktree_path"] = path
alias_state["expected_worktree_path"] = expected_path
alias_state["discovered_worktree_path"] = path
alias_state["match_type"] = "alias"
alias_state["candidate_paths"] = [path]
alias_state["alias_block_reasons"] = []
alias_state["alias_verified"] = True
return alias_state
if len(candidates) > 1:
state["alias_block_reasons"].append(
"ambiguous local worktree aliases: " + ", ".join(
sorted(str(entry.get("path")) for entry in candidates)
)
)
state["match_type"] = "ambiguous_alias"
elif unsafe_matches:
state["alias_block_reasons"].append(
"matching local worktree aliases did not point to the PR head or target branch: "
+ ", ".join(sorted(unsafe_matches))
)
state["match_type"] = "unsafe_alias"
return state
def read_issue_lock(path: str | None = None) -> dict[str, Any] | None:
if path:
return issue_lock_store.read_lock_file(path.strip())
@@ -176,6 +326,7 @@ def assess_local_worktree_cleanup(
exists = bool(worktree_state.get("exists"))
if not merged:
reasons.append("PR is not merged")
reasons.extend(worktree_state.get("alias_block_reasons") or [])
if not exists:
reasons.append("local worktree not present")
if exists and worktree_state.get("clean") is False:
@@ -186,7 +337,11 @@ def assess_local_worktree_cleanup(
)
if exists:
current_branch = worktree_state.get("current_branch")
if current_branch and current_branch != head_branch:
if (
current_branch
and current_branch != head_branch
and not worktree_state.get("alias_verified")
):
reasons.append(
f"worktree branch '{current_branch}' does not match PR head '{head_branch}'"
)
@@ -198,6 +353,10 @@ def assess_local_worktree_cleanup(
"pr_number": pr_number,
"head_branch": head_branch,
"worktree_path": worktree_state.get("worktree_path"),
"expected_worktree_path": worktree_state.get("expected_worktree_path"),
"discovered_worktree_path": worktree_state.get("discovered_worktree_path"),
"match_type": worktree_state.get("match_type"),
"candidate_paths": worktree_state.get("candidate_paths") or [],
"worktree_exists": exists,
"worktree_clean": worktree_state.get("clean"),
"safe_to_remove_worktree": safe,
@@ -216,6 +375,7 @@ def build_pr_cleanup_entry(
delete_capability_allowed: bool,
issue_lock_path: str | None = None,
protected_branches: frozenset[str] | None = None,
target_ref: str | None = None,
) -> dict[str, Any]:
pr_number = int(pr["number"])
head_branch = pr.get("head") or ""
@@ -224,9 +384,16 @@ def build_pr_cleanup_entry(
title = pr.get("title") or ""
body = pr.get("body") or ""
merged = bool(pr.get("merged_at"))
worktree_path = resolve_worktree_path(project_root, head_branch)
worktree_state = read_local_worktree_state(worktree_path)
worktree_state["worktree_path"] = worktree_path
issue_number = extract_linked_issue(title, body) or extract_issue_marker(head_branch)
head_payload = pr.get("head") if isinstance(pr.get("head"), dict) else {}
pr_head_sha = head_payload.get("sha") if isinstance(head_payload, dict) else None
worktree_state = resolve_cleanup_worktree_state(
project_root=project_root,
head_branch=head_branch,
issue_number=issue_number,
pr_head_sha=pr_head_sha,
target_ref=target_ref,
)
active_lock = has_active_issue_lock(head_branch, issue_lock_path)
remote = assess_remote_branch_cleanup(
@@ -249,7 +416,7 @@ def build_pr_cleanup_entry(
)
return {
"pr_number": pr_number,
"issue_number": extract_linked_issue(title, body),
"issue_number": issue_number,
"title": title,
"head_branch": head_branch,
"merge_commit_sha": pr.get("merge_commit_sha"),
@@ -270,6 +437,7 @@ def build_reconciliation_report(
delete_capability_allowed: bool,
issue_lock_path: str | None = None,
protected_branches: frozenset[str] | None = None,
target_ref: str | None = None,
) -> dict[str, Any]:
open_heads = collect_open_pr_heads(open_prs)
entries: list[dict[str, Any]] = []
@@ -290,6 +458,7 @@ def build_reconciliation_report(
delete_capability_allowed=delete_capability_allowed,
issue_lock_path=issue_lock_path,
protected_branches=protected_branches,
target_ref=target_ref,
)
)
return {
@@ -301,8 +470,12 @@ def build_reconciliation_report(
}
def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
worktree_path = resolve_worktree_path(project_root, branch)
def remove_local_worktree(
project_root: str,
branch: str,
worktree_path: str | None = None,
) -> dict[str, Any]:
worktree_path = worktree_path or resolve_worktree_path(project_root, branch)
if not os.path.isdir(worktree_path):
return {
"success": False,
@@ -326,4 +499,4 @@ def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
"performed": True,
"message": f"removed worktree {worktree_path}",
"worktree_path": worktree_path,
}
}
tests/test_merged_cleanup_reconcile.py
+191 -2
View File
@@ -4,7 +4,7 @@ import os
import sys
import tempfile
import unittest
from unittest.mock import patch
from unittest.mock import Mock, patch
sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
@@ -154,6 +154,195 @@ class TestMergedCleanupReport(unittest.TestCase):
finally:
os.remove(lock_path)
@patch("merged_cleanup_reconcile.read_local_worktree_state")
@patch("subprocess.run")
def test_build_report_discovers_issue_alias_worktree(self, mock_run, mock_state):
mock_run.return_value = Mock(
returncode=0,
stdout=(
"worktree /repo/Gitea-Tools\n"
"HEAD aaaa\n"
"branch refs/heads/master\n\n"
"worktree /repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation\n"
"HEAD cafebabe\n"
"branch refs/heads/feat/issue-510-workspace-binding-isolation\n\n"
),
)
mock_state.side_effect = [
{
"exists": False,
"clean": None,
"current_branch": None,
"head_sha": None,
"dirty_files": [],
},
{
"exists": True,
"clean": True,
"current_branch": "feat/issue-510-workspace-binding-isolation",
"head_sha": "cafebabe",
"dirty_files": [],
},
]
report = mcr.build_reconciliation_report(
project_root="/repo/Gitea-Tools",
closed_prs=[
{
"number": 512,
"title": "merged cleanup",
"body": "Closes #510",
"head": {
"ref": "feat/issue-510-workspace-binding-isolation",
"sha": "cafebabe",
},
"merged_at": "2026-07-06T12:00:00Z",
"merge_commit_sha": "abc123",
}
],
open_prs=[],
remote_branch_exists={"feat/issue-510-workspace-binding-isolation": True},
head_on_master={512: True},
delete_capability_allowed=True,
target_ref="prgs/master",
)
local = report["entries"][0]["local_worktree"]
self.assertTrue(local["safe_to_remove_worktree"])
self.assertEqual(local["match_type"], "alias")
self.assertEqual(
local["expected_worktree_path"],
"/repo/Gitea-Tools/branches/feat-issue-510-workspace-binding-isolation",
)
self.assertEqual(
local["discovered_worktree_path"],
"/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation",
)
@patch("merged_cleanup_reconcile.read_local_worktree_state")
@patch("subprocess.run")
def test_ambiguous_issue_alias_worktrees_fail_closed(self, mock_run, mock_state):
mock_run.return_value = Mock(
returncode=0,
stdout=(
"worktree /repo/Gitea-Tools/branches/issue-510-one\n"
"HEAD cafebabe\n"
"branch refs/heads/feat/issue-510-workspace-binding-isolation\n\n"
"worktree /repo/Gitea-Tools/branches/issue-510-two\n"
"HEAD cafebabe\n"
"branch refs/heads/feat/issue-510-workspace-binding-isolation\n\n"
),
)
mock_state.return_value = {
"exists": False,
"clean": None,
"current_branch": None,
"head_sha": None,
"dirty_files": [],
}
report = mcr.build_reconciliation_report(
project_root="/repo/Gitea-Tools",
closed_prs=[
{
"number": 512,
"title": "merged cleanup",
"body": "Closes #510",
"head": {
"ref": "feat/issue-510-workspace-binding-isolation",
"sha": "cafebabe",
},
"merged_at": "2026-07-06T12:00:00Z",
"merge_commit_sha": "abc123",
}
],
open_prs=[],
remote_branch_exists={"feat/issue-510-workspace-binding-isolation": True},
head_on_master={512: True},
delete_capability_allowed=True,
target_ref="prgs/master",
)
local = report["entries"][0]["local_worktree"]
self.assertFalse(local["safe_to_remove_worktree"])
self.assertEqual(local["match_type"], "ambiguous_alias")
self.assertIn("ambiguous local worktree aliases", local["block_reasons"][0])
@patch("merged_cleanup_reconcile.is_head_ancestor_of_ref", return_value=True)
@patch("merged_cleanup_reconcile.read_local_worktree_state")
@patch("subprocess.run")
def test_alias_head_on_target_branch_is_safe_cleanup_commit(
self, mock_run, mock_state, _ancestor
):
mock_run.return_value = Mock(
returncode=0,
stdout=(
"worktree /repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation\n"
"HEAD deadbeef\n"
"branch refs/heads/feat/issue-510-workspace-binding-isolation\n\n"
),
)
mock_state.side_effect = [
{
"exists": False,
"clean": None,
"current_branch": None,
"head_sha": None,
"dirty_files": [],
},
{
"exists": True,
"clean": True,
"current_branch": "feat/issue-510-workspace-binding-isolation",
"head_sha": "deadbeef",
"dirty_files": [],
},
]
state = mcr.resolve_cleanup_worktree_state(
project_root="/repo/Gitea-Tools",
head_branch="feat/issue-510-workspace-binding-isolation",
issue_number=510,
pr_head_sha="cafebabe",
target_ref="prgs/master",
)
self.assertTrue(state["exists"])
self.assertEqual(state["match_type"], "alias")
self.assertEqual(
state["discovered_worktree_path"],
"/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation",
)
@patch("os.path.isdir", return_value=True)
@patch("subprocess.run")
def test_remove_local_worktree_uses_assessed_path(self, mock_run, _isdir):
mock_run.return_value = Mock(returncode=0, stdout="", stderr="")
result = mcr.remove_local_worktree(
"/repo/Gitea-Tools",
"feat/issue-510-workspace-binding-isolation",
worktree_path="/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation",
)
self.assertTrue(result["success"])
self.assertEqual(
result["worktree_path"],
"/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation",
)
mock_run.assert_called_once_with(
[
"git",
"-C",
"/repo/Gitea-Tools",
"worktree",
"remove",
"/repo/Gitea-Tools/branches/issue-510-workspace-binding-isolation",
],
capture_output=True,
text=True,
check=False,
)
if __name__ == "__main__":
unittest.main()
unittest.main()