diff --git a/merged_cleanup_reconcile.py b/merged_cleanup_reconcile.py index bf81258..92490ac 100644 --- a/merged_cleanup_reconcile.py +++ b/merged_cleanup_reconcile.py @@ -36,6 +36,34 @@ 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()) @@ -216,6 +244,7 @@ def build_pr_cleanup_entry( 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 "" @@ -224,7 +253,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) + + 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) @@ -247,6 +285,8 @@ def build_pr_cleanup_entry( 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), @@ -272,6 +312,7 @@ def build_reconciliation_report( 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"): @@ -290,6 +331,7 @@ def build_reconciliation_report( delete_capability_allowed=delete_capability_allowed, issue_lock_path=issue_lock_path, protected_branches=protected_branches, + worktree_map=worktree_map, ) ) return { @@ -302,7 +344,9 @@ def build_reconciliation_report( def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]: - worktree_path = resolve_worktree_path(project_root, branch) + 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, diff --git a/tests/test_merged_cleanup_reconcile.py b/tests/test_merged_cleanup_reconcile.py index 779df64..ad4a2ee 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 patch, MagicMock sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent)) @@ -154,6 +154,89 @@ class TestMergedCleanupReport(unittest.TestCase): finally: os.remove(lock_path) + @patch("subprocess.run") + def test_discover_local_worktrees(self, mock_run): + mock_output = ( + "worktree /repo/Gitea-Tools\n" + "bare\n\n" + "worktree /repo/Gitea-Tools/branches/custom-name\n" + "branch refs/heads/feat/issue-11-x\n" + "HEAD cafebabe\n\n" + ) + mock_run.return_value = MagicMock(returncode=0, stdout=mock_output) + res = mcr.discover_local_worktrees("/repo/Gitea-Tools") + self.assertEqual(res, {"feat/issue-11-x": "/repo/Gitea-Tools/branches/custom-name"}) + + @patch("subprocess.run") + @patch("merged_cleanup_reconcile.read_local_worktree_state") + def test_build_report_discovers_non_canonical_worktree_by_branch(self, mock_state, mock_run): + mock_output = ( + "worktree /repo/Gitea-Tools\n" + "bare\n\n" + "worktree /repo/Gitea-Tools/branches/custom-name\n" + "branch refs/heads/feat/issue-11-x\n" + "HEAD cafebabe\n\n" + ) + mock_run.return_value = MagicMock(returncode=0, stdout=mock_output) + mock_state.return_value = { + "exists": True, + "clean": True, + "current_branch": "feat/issue-11-x", + "head_sha": "cafebabe", + "dirty_files": [], + } + + report = mcr.build_reconciliation_report( + project_root="/repo/Gitea-Tools", + closed_prs=[ + { + "number": 11, + "title": "merged", + "body": "Closes #11", + "head": {"ref": "feat/issue-11-x"}, + "merged_at": "2026-07-06T12:00:00Z", + "merge_commit_sha": "abc123", + } + ], + open_prs=[], + remote_branch_exists={"feat/issue-11-x": True}, + head_on_master={11: True}, + delete_capability_allowed=True, + ) + self.assertEqual(report["merged_pr_count"], 1) + entry = report["entries"][0] + local = entry["local_worktree"] + self.assertEqual(local["worktree_path"], "/repo/Gitea-Tools/branches/custom-name") + self.assertEqual(local["match_type"], "branch") + + @patch("subprocess.run") + def test_remove_local_worktree_uses_discovered_path(self, mock_run): + mock_output = ( + "worktree /repo/Gitea-Tools\n" + "bare\n\n" + "worktree /repo/Gitea-Tools/branches/custom-name\n" + "branch refs/heads/feat/issue-11-x\n" + "HEAD cafebabe\n\n" + ) + # First call is discover_local_worktrees, second is git worktree remove + mock_run.side_effect = [ + MagicMock(returncode=0, stdout=mock_output), + MagicMock(returncode=0, stdout="", stderr=""), + ] + + with patch("os.path.isdir", return_value=True): + result = mcr.remove_local_worktree("/repo/Gitea-Tools", "feat/issue-11-x") + self.assertTrue(result["success"]) + self.assertTrue(result["performed"]) + self.assertEqual(result["worktree_path"], "/repo/Gitea-Tools/branches/custom-name") + # Assert git worktree remove was called with the custom path + mock_run.assert_any_call( + ["git", "-C", "/repo/Gitea-Tools", "worktree", "remove", "/repo/Gitea-Tools/branches/custom-name"], + capture_output=True, + text=True, + check=False, + ) + if __name__ == "__main__": unittest.main() \ No newline at end of file