Merge pull request 'feat: discover non-canonical worktrees by branch matching (Closes #528)' (#568) from feat/issue-528-reconcile-worktree-naming into master

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,

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()