fix(kanban): isolate and protect linked worktree anchors

hermes_cli/kanban_db.py

@@ -5959,11 +5959,14 @@ def _cleanup_worktree_workspace(
     """Remove a finished task's linked git worktree when it holds no work.
 
     Mirrors the safety judgment of the CLI startup pruner
-    (``cli._prune_stale_worktrees``): removal requires a clean working tree
-    AND every commit reachable from a remote-tracking ref. Any doubt — dirty
-    files, unpushed commits, unresolvable repo, failing git — preserves the
-    worktree. The task's auto-generated ``wt/<task-id>`` branch is deleted
-    with it; custom branches are kept. Best-effort like the scratch path.
+    (``cli._prune_stale_worktrees``), but first requires deterministic task
+    ownership: the resolved path must be ``.worktrees/<task-id>`` and its
+    checked-out branch must match the task row. Removal then requires a clean
+    working tree AND every commit reachable from a remote-tracking ref. Any
+    doubt — ambiguous ownership, dirty files, unpushed commits, unresolvable
+    repo, failing git — preserves the worktree. The task's auto-generated
+    ``wt/<task-id>`` branch is deleted with it; custom branches are kept.
+    Best-effort like the scratch path.
     """
     try:
         from cli import _worktree_has_unpushed_commits, _worktree_is_dirty
@@ -5973,11 +5976,28 @@ def _cleanup_worktree_workspace(
         wp = Path(path).expanduser()
         if not wp.is_dir():
             return
+        wp = wp.resolve(strict=False)
+        if wp.name != task_id or wp.parent.name != ".worktrees":
+            _log.warning(
+                "Preserving worktree for task %s: path is not the task-owned "
+                ".worktrees/%s checkout: %s",
+                task_id, task_id, wp,
+            )
+            return
+        branch = (branch_name or "").strip() or f"wt/{task_id}"
+        actual_branch = _git_current_branch(wp)
+        if actual_branch != branch:
+            _log.warning(
+                "Preserving worktree for task %s: branch ownership is ambiguous "
+                "at %s (expected %s, found %s)",
+                task_id, wp, branch, actual_branch or "detached/unknown",
+            )
+            return
         common = _git_common_dir(wp)
         if common is None or common.name != ".git":
             return  # not a linked worktree of a normal repo — never guess
         repo_root = common.parent
-        if wp.resolve(strict=False) == repo_root.resolve(strict=False):
+        if wp == repo_root.resolve(strict=False):
             return  # never remove the main checkout
         if _worktree_is_dirty(str(wp)) or _worktree_has_unpushed_commits(str(wp)):
             _log.info(
@@ -6003,7 +6023,6 @@ def _cleanup_worktree_workspace(
             )
             return
         _log.debug("Removed worktree workspace: %s", wp)
-        branch = (branch_name or "").strip() or f"wt/{task_id}"
         if branch.startswith("wt/"):
             subprocess.run(
                 ["git", "-C", str(repo_root), "branch", "-D", branch],
@@ -7714,10 +7733,25 @@ def _ensure_git_worktree(repo_root: Path, target: Path, branch_name: str) -> Non
     """Materialize ``target`` as a linked git worktree under ``repo_root``."""
     target = target.expanduser()
     repo_common = _git_common_dir(repo_root)
-    if target.exists() and repo_common is not None:
+    if target.exists():
         target_common = _git_common_dir(target)
-        if target_common == repo_common:
+        actual_branch = _git_current_branch(target)
+        if (
+            repo_common is not None
+            and target_common == repo_common
+            and actual_branch == branch_name
+        ):
             return
+        if repo_common is None or target_common != repo_common:
+            raise RuntimeError(
+                f"refusing to reuse existing worktree target {target}: "
+                "repository ownership is ambiguous"
+            )
+        raise RuntimeError(
+            f"refusing to reuse existing worktree target {target}: branch ownership "
+            f"is ambiguous (expected {branch_name}, found "
+            f"{actual_branch or 'detached/unknown'})"
+        )
     target.parent.mkdir(parents=True, exist_ok=True)
     if _git_branch_exists(repo_root, branch_name):
         cmd = ["git", "-C", str(repo_root), "worktree", "add", str(target), branch_name]
@@ -7792,26 +7826,34 @@ def _resolve_worktree_workspace(
     requested_resolved = requested.resolve(strict=False)
 
     if requested.exists() and _is_linked_worktree_checkout(requested):
-        actual_branch = _git_current_branch(requested)
-        if actual_branch == branch_name:
-            return requested_resolved, actual_branch
-        # The requested path is an existing checkout of a DIFFERENT
-        # task's branch. Decompose children inherit the root's
-        # workspace_path verbatim, so siblings all point here; reusing
-        # the checkout as-is would run this task on the other task's
-        # branch — silent cross-task provenance corruption, and unsafe
-        # when siblings run concurrently. Fall back to a fresh worktree
-        # of our own under the same repo.
-        fallback_root = _repo_root_for_worktree_target(requested.parent)
-        if fallback_root is not None:
+        common = _git_common_dir(requested_resolved)
+        if common is None or common.name != ".git":
+            raise RuntimeError(
+                f"task {task.id} linked worktree {requested_resolved} has "
+                "ambiguous repository ownership"
+            )
+        is_task_owned_target = (
+            requested_resolved.name == task.id
+            and requested_resolved.parent.name == ".worktrees"
+        )
+        if is_task_owned_target:
+            _ensure_git_worktree(common.parent, requested_resolved, branch_name)
+            return requested_resolved, branch_name
+        # A concrete target below a conventional .worktrees directory belongs
+        # to another task. Heal stale/shared rows by resolving a sibling target
+        # from the linked checkout's verified common Git directory — never from
+        # an unrelated repository that merely surrounds the filesystem path.
+        if requested_resolved.parent.name == ".worktrees":
+            fallback_root = common.parent
             fallback = fallback_root / ".worktrees" / task.id
-            if fallback.resolve(strict=False) != requested_resolved:
-                _ensure_git_worktree(fallback_root, fallback, branch_name)
-                return fallback.resolve(strict=False), branch_name
-        # No repo to anchor a fallback on (or the occupied path IS this
-        # task's own canonical worktree): keep the legacy reuse rather
-        # than failing dispatch.
-        return requested_resolved, actual_branch or branch_name
+            _ensure_git_worktree(fallback_root, fallback, branch_name)
+            return fallback.resolve(strict=False), branch_name
+        # Otherwise the linked checkout itself is a persistent repo-root anchor.
+        # Materialize below that exact anchor so filesystem nesting cannot switch
+        # repository identity when the anchor lives inside another checkout.
+        target = requested_resolved / ".worktrees" / task.id
+        _ensure_git_worktree(requested_resolved, target, branch_name)
+        return target, branch_name
 
     repo_root = _git_toplevel(requested)
     if repo_root is not None and requested_resolved == repo_root:

tests/hermes_cli/test_kanban_worktree_isolation.py

@@ -48,8 +48,8 @@ def _git(cwd: Path, *args: str) -> None:
     )
 
 
-def _make_repo(tmp_path: Path) -> Path:
-    repo = tmp_path / "repo"
+def _make_repo(tmp_path: Path, name: str = "repo") -> Path:
+    repo = tmp_path / name
     repo.mkdir()
     subprocess.run(
         ["git", "init", "-b", "main", str(repo)],
@@ -66,6 +66,118 @@ def _add_worktree(repo: Path, target: Path, branch: str) -> Path:
     return target
 
 
+def test_resolve_linked_repo_root_anchor_materializes_task_worktree(
+    kanban_home, tmp_path
+):
+    repo = _make_repo(tmp_path)
+    linked_root = _add_worktree(repo, tmp_path / "linked-root", "anchor/main")
+    branch = "wt/linked-anchor-task"
+
+    with kb.connect() as conn:
+        tid = kb.create_task(
+            conn,
+            title="linked anchor",
+            workspace_kind="worktree",
+            workspace_path=str(linked_root),
+            branch_name=branch,
+        )
+        task = kb.get_task(conn, tid)
+
+    assert task is not None
+    workspace, resolved_branch = kb._resolve_worktree_workspace(task)
+    expected = linked_root / ".worktrees" / tid
+    assert workspace == expected
+    assert resolved_branch == branch
+    head = subprocess.run(
+        ["git", "-C", str(workspace), "branch", "--show-current"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    assert head == branch
+
+
+def test_linked_anchor_inside_source_repo_stays_nested_under_anchor(
+    kanban_home, tmp_path
+):
+    repo = _make_repo(tmp_path)
+    linked_root = _add_worktree(repo, repo / "canonical", "anchor/main")
+
+    with kb.connect() as conn:
+        tid = kb.create_task(
+            conn,
+            title="nested linked anchor",
+            workspace_kind="worktree",
+            workspace_path=str(linked_root),
+        )
+        task = kb.get_task(conn, tid)
+
+    assert task is not None
+    workspace, _branch = kb._resolve_worktree_workspace(task)
+    assert workspace == linked_root / ".worktrees" / tid
+    assert kb._git_common_dir(workspace) == kb._git_common_dir(repo)
+
+
+def test_linked_anchor_inside_unrelated_repo_uses_anchor_repository(
+    kanban_home, tmp_path
+):
+    source = _make_repo(tmp_path, "source")
+    unrelated = _make_repo(tmp_path, "unrelated")
+    linked_root = _add_worktree(source, unrelated / "canonical", "anchor/main")
+
+    with kb.connect() as conn:
+        tid = kb.create_task(
+            conn,
+            title="cross-repo linked anchor",
+            workspace_kind="worktree",
+            workspace_path=str(linked_root),
+        )
+        task = kb.get_task(conn, tid)
+
+    assert task is not None
+    workspace, _branch = kb._resolve_worktree_workspace(task)
+    assert workspace == linked_root / ".worktrees" / tid
+    assert kb._git_common_dir(workspace) == kb._git_common_dir(source)
+    assert kb._git_common_dir(workspace) != kb._git_common_dir(unrelated)
+
+
+def test_linked_repo_root_matching_branch_is_never_reused(kanban_home, tmp_path):
+    repo = _make_repo(tmp_path)
+    linked_root = _add_worktree(repo, tmp_path / "linked-root", "anchor/main")
+
+    with kb.connect() as conn:
+        tid = kb.create_task(
+            conn,
+            title="matching linked anchor",
+            workspace_kind="worktree",
+            workspace_path=str(linked_root),
+            branch_name="anchor/main",
+        )
+        task = kb.get_task(conn, tid)
+
+    assert task is not None
+    with pytest.raises(RuntimeError):
+        kb._resolve_worktree_workspace(task)
+    assert linked_root.is_dir()
+    assert not (linked_root / ".worktrees" / tid).exists()
+
+
+def test_existing_task_target_on_wrong_branch_fails_closed(kanban_home, tmp_path):
+    repo = _make_repo(tmp_path)
+    target = repo / ".worktrees" / "t_existing"
+    _add_worktree(repo, target, "wt/other-task")
+
+    with pytest.raises(RuntimeError, match="branch ownership"):
+        kb._ensure_git_worktree(repo, target, "wt/t_existing")
+    head = subprocess.run(
+        ["git", "-C", str(target), "branch", "--show-current"],
+        capture_output=True,
+        text=True,
+        check=True,
+    ).stdout.strip()
+    assert head == "wt/other-task"
+
+
 def test_decompose_worktree_children_get_own_workspace(kanban_home):
     with kb.connect() as conn:
         root = kb.create_task(conn, title="build the feature", triage=True)

tests/hermes_cli/test_kanban_worktree_teardown.py

@@ -218,3 +218,82 @@ def test_parent_worktree_deferred_until_children_done(
         assert kb.complete_task(conn, child, summary="child done")
     # last child terminal -> deferred parent worktree reaped
     assert not parent_wt.exists()
+
+
+def test_linked_anchor_deferred_cleanup_removes_only_nested_task_worktree(
+    kanban_home: Path, repo: Path, tmp_path: Path
+) -> None:
+    canonical = tmp_path / "canonical-production-main"
+    unrelated = tmp_path / "unrelated-worktree"
+    kb._ensure_git_worktree(repo, canonical, "canonical/main")
+    kb._ensure_git_worktree(repo, unrelated, "unrelated/branch")
+
+    with kb.connect_closing() as conn:
+        parent = kb.create_task(
+            conn,
+            title="anchored parent",
+            assignee="worker",
+            workspace_kind="worktree",
+            workspace_path=str(canonical),
+        )
+        task = kb.get_task(conn, parent)
+        assert task is not None
+        workspace, branch = kb._resolve_worktree_workspace(task)
+        assert workspace == canonical / ".worktrees" / parent
+        kb.set_workspace_path(conn, parent, workspace)
+        kb.set_branch_name(conn, parent, branch)
+
+        child = kb.create_task(conn, title="review child", assignee="worker")
+        kb.link_tasks(conn, parent, child)
+        with kb.write_txn(conn):
+            conn.execute("UPDATE tasks SET status='ready' WHERE id=?", (parent,))
+
+        assert kb.claim_task(conn, parent, claimer="worker") is not None
+        assert kb.complete_task(conn, parent, summary="parent done")
+        assert workspace.is_dir()
+        assert canonical.is_dir()
+
+        with kb.write_txn(conn):
+            conn.execute("UPDATE tasks SET status='ready' WHERE id=?", (child,))
+        assert kb.claim_task(conn, child, claimer="worker") is not None
+        assert kb.complete_task(conn, child, summary="child done")
+
+    assert not workspace.exists()
+    assert canonical.is_dir()
+    assert (canonical / "README.md").is_file()
+    assert unrelated.is_dir()
+    assert (unrelated / "README.md").is_file()
+
+
+def test_deferred_cleanup_preserves_unowned_canonical_linked_checkout(
+    kanban_home: Path, repo: Path, tmp_path: Path
+) -> None:
+    canonical = tmp_path / "canonical-production-main"
+    unrelated = tmp_path / "unrelated-worktree"
+    kb._ensure_git_worktree(repo, canonical, "canonical/main")
+    kb._ensure_git_worktree(repo, unrelated, "unrelated/branch")
+
+    with kb.connect_closing() as conn:
+        parent = kb.create_task(conn, title="legacy parent", assignee="worker")
+        child = kb.create_task(conn, title="review child", assignee="worker")
+        kb.link_tasks(conn, parent, child)
+        with kb.write_txn(conn):
+            conn.execute(
+                "UPDATE tasks SET workspace_kind='worktree', workspace_path=?, "
+                "branch_name='canonical/main', status='ready' WHERE id=?",
+                (str(canonical), parent),
+            )
+
+        assert kb.claim_task(conn, parent, claimer="worker") is not None
+        assert kb.complete_task(conn, parent, summary="parent done")
+        assert canonical.is_dir()
+
+        with kb.write_txn(conn):
+            conn.execute("UPDATE tasks SET status='ready' WHERE id=?", (child,))
+        assert kb.claim_task(conn, child, claimer="worker") is not None
+        assert kb.complete_task(conn, child, summary="child done")
+
+    assert canonical.is_dir()
+    assert (canonical / "README.md").is_file()
+    assert unrelated.is_dir()
+    assert (unrelated / "README.md").is_file()