[hotfix] worktree: recover clone root left on the task branch (be-pm needs_revision wedge)

Live NAS (v0.14.0): after pr_reviewer correctly pr_fail'd task d3dab0fc to
needs_revision, the revision_coordination dispatcher tried to spawn be-pm and
re-attach its per-task worktree. git worktree add failed FATAL:
  'feature/backend/fb836f80--03f80432--d3dab0fc' is already checked out at
  '/data/workspaces/guard-core-saas-backend/backend/be-pm'
releasing the claim to pending and re-dispatching into the same collision
every tick — be-pm never spawned.

F123 invariant: the clone root parks on the default branch (or detached);
the task branch lives in the worktree. The clone root had drifted onto the
task branch (a pre-F123 leftover / a missed checkout path), so worktree add
refused. ensure_worktree + ensure_worktree_for_resume now restore the
invariant before the add: if the clone root's HEAD is the task branch, move
it back to the default branch (read from origin/HEAD); if that is
unresolvable or the checkout is blocked, detach HEAD at the same commit —
either frees the branch ref for the worktree. The clone root never carries
task work under F123, so nothing is lost; the branch ref (and pushed work)
survives unchanged.

TDD: 3 new tests (default-branch restore, detach fallback, self_heal recovery)
+ 36-test worktree regression green; ruff/mypy clean.

roboco/services/workspace.py

@@ -482,6 +482,54 @@ class WorkspaceService:
             check=check,
         )
 
+    @staticmethod
+    def _clone_root_default_branch(clone_root: Path) -> str:
+        # origin/HEAD points at the repo's default branch (set on clone).
+        # Returns the bare branch name ("main"), or "" when unresolvable (no
+        # remote, e.g. a test/local clone).
+        res = WorkspaceService._worktree_git(
+            clone_root,
+            ["symbolic-ref", "--short", "refs/remotes/origin/HEAD"],
+            check=False,
+        )
+        if res.returncode != 0:
+            return ""
+        val = res.stdout.strip()
+        return val.split("/", 1)[1] if val.startswith("origin/") else val
+
+    @staticmethod
+    def _park_clone_root_off_branch(clone_root: Path, branch: str) -> None:
+        """Restore the F123 invariant before ``git worktree add <branch>``.
+
+        The clone root parks on the default branch (or detached); the task
+        branch lives in the worktree. A re-dispatch after the clone root drifted
+        onto the task branch (a pre-F123 leftover / a missed checkout path) makes
+        ``worktree add <branch>`` fatal ("already checked out at '<clone_root>'"),
+        releasing the claim and re-dispatching into the same collision every
+        tick. If the clone root's HEAD is the task branch, move it back to the
+        default branch (via origin/HEAD); if that is unresolvable or the checkout
+        is blocked, detach HEAD at the same commit — either frees the branch ref
+        for the worktree. The clone root never carries task work under F123, so
+        nothing is lost.
+        """
+        cur = WorkspaceService._worktree_git(
+            clone_root, ["branch", "--show-current"], check=False
+        )
+        if cur.returncode != 0 or cur.stdout.strip() != branch:
+            return
+        default = WorkspaceService._clone_root_default_branch(clone_root)
+        if default:
+            moved = WorkspaceService._worktree_git(
+                clone_root, ["checkout", default], check=False
+            )
+            if moved.returncode == 0:
+                return
+        # No resolvable default, or checkout blocked by a dirty tree: detach
+        # at the same commit (no working-tree change) so the branch ref is free.
+        WorkspaceService._worktree_git(
+            clone_root, ["checkout", "--detach"], check=False
+        )
+
     @staticmethod
     def _link_shared_venv(worktree: Path, clone_root: Path) -> None:
         """Symlink ``worktree/.venv -> ../../.venv`` (the clone-root venv).
@@ -518,6 +566,7 @@ class WorkspaceService:
         ``reset --hard`` + ``checkout -b`` that clobbered a still-active root.
         """
         if not (worktree.exists() and (worktree / ".git").is_file()):
+            self._park_clone_root_off_branch(clone_root, branch)
             branch_exists = (
                 self._worktree_git(
                     clone_root,
@@ -549,6 +598,7 @@ class WorkspaceService:
         worktree is a no-op.
         """
         if not (worktree.exists() and (worktree / ".git").is_file()):
+            self._park_clone_root_off_branch(clone_root, branch)
             res = self._worktree_git(
                 clone_root, ["worktree", "add", str(worktree), branch], check=False
             )

tests/unit/services/test_workspace_worktree_lifecycle.py

@@ -399,3 +399,89 @@ async def test_self_heal_falls_back_to_origin_head_when_branch_not_pushed(
     assert fetch.await_count == 1, "missing local ref still attempts a fetch"
     assert wt.exists(), "fallback -b from origin/HEAD must break the loop"
     assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == branch
+
+
+# ---------------------------------------------------------------------------
+# Clone-root-left-on-task-branch recovery (live be-pm needs_revision wedge,
+# 2026-06-30). F123 invariant: the clone root parks on the default branch (or
+# detached); the task branch lives in the worktree. A re-dispatch after the clone
+# root drifted onto the task branch (a pre-F123 leftover / missed checkout)
+# made `git worktree add <branch>` fatal ("already checked out at '<clone>'"),
+# releasing the claim and re-dispatching into the same collision every tick.
+# ensure_worktree must restore the invariant before the add: move the clone root
+# back to the default branch (via origin/HEAD), detaching as a fallback so the
+# branch ref is free for the worktree either way.
+# ---------------------------------------------------------------------------
+
+
+def _clone_on_branch(clone: Path, branch: str) -> None:
+    _git(clone, "branch", branch)
+    _git(clone, "checkout", branch)
+
+
+async def test_ensure_worktree_restores_clone_root_left_on_task_branch(
+    clone: Path,
+) -> None:
+    svc = _service()
+    branch = "feature/d3dab0fc"
+    # Set up a resolvable origin/HEAD (a real clone has this) so the default
+    # branch is "main", then drift the clone root onto the task branch.
+    main_sha = _git(clone, "rev-parse", "main").strip()
+    _git(clone, "remote", "add", "origin", str(clone))
+    _git(clone, "update-ref", "refs/remotes/origin/main", main_sha)
+    _git(clone, "symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main")
+    _clone_on_branch(clone, branch)
+    assert _git(clone, "branch", "--show-current").strip() == branch
+
+    wt = clone / ".worktrees" / "d3dab0fc"
+    with patch("roboco.services.workspace._ensure_agent_owned"):
+        await svc.ensure_worktree(clone, wt, branch, "origin/HEAD")
+
+    assert (wt / ".git").is_file(), "worktree must be created despite the collision"
+    assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == branch
+    # Clone root restored to the default branch (F123 invariant), not still on
+    # the task branch, and not left dangling mid-recovery.
+    assert _git(clone, "rev-parse", "--abbrev-ref", "HEAD").strip() == "main"
+
+
+async def test_ensure_worktree_detaches_when_default_branch_unresolvable(
+    clone: Path,
+) -> None:
+    # No origin remote (e.g. a test/local clone): origin/HEAD can't resolve, so
+    # the recovery detaches the clone root to free the branch for the worktree.
+    svc = _service()
+    branch = "feature/d3dab0fc"
+    _clone_on_branch(clone, branch)
+    assert _git(clone, "branch", "--show-current").strip() == branch
+
+    wt = clone / ".worktrees" / "d3dab0fc"
+    with patch("roboco.services.workspace._ensure_agent_owned"):
+        await svc.ensure_worktree(clone, wt, branch, "origin/HEAD")
+
+    assert (wt / ".git").is_file(), "worktree must be created (branch freed via detach)"
+    assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == branch
+    # Detached HEAD (abbrev-ref is HEAD), the task branch ref no longer checked
+    # out at the clone root.
+    assert _git(clone, "rev-parse", "--abbrev-ref", "HEAD").strip() == "HEAD"
+    assert _ref_exists(clone, f"refs/heads/{branch}"), "branch ref preserved"
+
+
+async def test_self_heal_recovers_clone_root_left_on_task_branch(clone: Path) -> None:
+    # The live failure path: spawn -> ensure_worktree_self_heal -> worktree add.
+    # With the clone root parked on the task branch, the self-heal re-add must
+    # restore the invariant and re-attach the worktree instead of fatal-looping.
+    svc = _service()
+    branch = "feature/d3dab0fc"
+    main_sha = _git(clone, "rev-parse", "main").strip()
+    _git(clone, "remote", "add", "origin", str(clone))
+    _git(clone, "update-ref", "refs/remotes/origin/main", main_sha)
+    _git(clone, "symbolic-ref", "refs/remotes/origin/HEAD", "refs/remotes/origin/main")
+    _clone_on_branch(clone, branch)
+
+    wt = clone / ".worktrees" / "d3dab0fc"
+    with patch("roboco.services.workspace._ensure_agent_owned"):
+        await svc.ensure_worktree_self_heal(clone, wt, branch, "proj")
+
+    assert (wt / ".git").is_file()
+    assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == branch
+    assert _git(clone, "rev-parse", "--abbrev-ref", "HEAD").strip() == "main"