feat: discover merged cleanup worktree aliases

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

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,8 +37,17 @@ 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)."""
+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,
@@ -45,25 +55,153 @@ def discover_local_worktrees(project_root: str) -> dict[str, str]:
         check=False,
     )
     if res.returncode != 0:
-        return {}
+        return []
 
-    mapping: dict[str, str] = {}
-    current_worktree = None
-    for line in res.stdout.splitlines():
-        line = line.strip()
+    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:
-            current_worktree = None
+            flush()
             continue
         if line.startswith("worktree "):
-            current_worktree = line[9:].strip()
-        elif line.startswith("branch ") and current_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()
-            if ref.startswith("refs/heads/"):
-                branch_name = ref[11:]
-                mapping[branch_name] = current_worktree
+            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 discover_local_worktrees(project_root: str) -> dict[str, str]:
+    """Return a mapping from branch name to absolute worktree path (#528)."""
+    mapping: dict[str, str] = {}
+    for entry in list_local_worktrees(project_root):
+        branch = entry.get("branch")
+        path = entry.get("path")
+        if branch and path:
+            mapping[str(branch)] = str(path)
     return mapping
 
 
+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
+        safe_head = _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,
+        )
+        if not safe_head and branch_match and not pr_head_sha and not target_ref:
+            safe_head = True
+        if not safe_head:
+            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"] = (
+            "branch" if candidates[0].get("branch") == head_branch else "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())
@@ -204,6 +342,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:
@@ -214,7 +353,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}'"
             )
@@ -226,6 +369,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,
@@ -244,7 +391,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,
+    target_ref: str | None = None,
 ) -> dict[str, Any]:
     pr_number = int(pr["number"])
     head_branch = pr.get("head") or ""
@@ -253,18 +400,16 @@ def build_pr_cleanup_entry(
     title = pr.get("title") or ""
     body = pr.get("body") or ""
     merged = bool(pr.get("merged_at"))
-
-    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
+    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(
@@ -285,11 +430,9 @@ 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),
+        "issue_number": issue_number,
         "title": title,
         "head_branch": head_branch,
         "merge_commit_sha": pr.get("merge_commit_sha"),
@@ -310,9 +453,9 @@ 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)
-    worktree_map = discover_local_worktrees(project_root)
     entries: list[dict[str, Any]] = []
     for pr in closed_prs or []:
         if not pr.get("merged_at"):
@@ -331,7 +474,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,
+                target_ref=target_ref,
             )
         )
     return {
@@ -343,10 +486,12 @@ def build_reconciliation_report(
     }
 
 
-def remove_local_worktree(project_root: str, branch: str) -> dict[str, Any]:
-    worktree_map = discover_local_worktrees(project_root)
-    discovered_path = worktree_map.get(branch)
-    worktree_path = discovered_path or 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,
@@ -370,4 +515,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

@@ -4,7 +4,7 @@ import os
 import sys
 import tempfile
 import unittest
-from unittest.mock import patch, MagicMock
+from unittest.mock import Mock, patch
 
 sys.path.insert(0, str(__import__("pathlib").Path(__file__).resolve().parent.parent))
 
@@ -163,7 +163,7 @@ class TestMergedCleanupReport(unittest.TestCase):
             "branch refs/heads/feat/issue-11-x\n"
             "HEAD cafebabe\n\n"
         )
-        mock_run.return_value = MagicMock(returncode=0, stdout=mock_output)
+        mock_run.return_value = Mock(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"})
 
@@ -177,14 +177,23 @@ class TestMergedCleanupReport(unittest.TestCase):
             "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": [],
-        }
+        mock_run.return_value = Mock(returncode=0, stdout=mock_output)
+        mock_state.side_effect = [
+            {
+                "exists": False,
+                "clean": None,
+                "current_branch": None,
+                "head_sha": None,
+                "dirty_files": [],
+            },
+            {
+                "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",
@@ -209,29 +218,190 @@ class TestMergedCleanupReport(unittest.TestCase):
         self.assertEqual(local["worktree_path"], "/repo/Gitea-Tools/branches/custom-name")
         self.assertEqual(local["match_type"], "branch")
 
+    @patch("merged_cleanup_reconcile.read_local_worktree_state")
     @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"
+    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"
+            ),
         )
-        # 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=""),
+        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": [],
+            },
         ]
 
-        with patch("os.path.isdir", return_value=True):
-            result = mcr.remove_local_worktree("/repo/Gitea-Tools", "feat/issue-11-x")
+        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"], "branch")
+        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"], "branch")
+        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.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"],
+        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,
@@ -239,4 +409,4 @@ class TestMergedCleanupReport(unittest.TestCase):
 
 
 if __name__ == "__main__":
-    unittest.main()
+    unittest.main()