feat: discover non-canonical worktrees by branch matching (Closes #528)

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,