diff --git a/gitea_mcp_server.py b/gitea_mcp_server.py index ed650b5..d9bdbf1 100644 --- a/gitea_mcp_server.py +++ b/gitea_mcp_server.py @@ -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}) diff --git a/merged_cleanup_reconcile.py b/merged_cleanup_reconcile.py index bf81258..fb06c31 100644 --- a/merged_cleanup_reconcile.py +++ b/merged_cleanup_reconcile.py @@ -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, - } \ No newline at end of file + } diff --git a/tests/test_merged_cleanup_reconcile.py b/tests/test_merged_cleanup_reconcile.py index 779df64..5ce1e2a 100644 --- a/tests/test_merged_cleanup_reconcile.py +++ b/tests/test_merged_cleanup_reconcile.py @@ -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() \ No newline at end of file + unittest.main()