fix(workspace): role-aware worktree refresh at every spawn (#692)

* fix(workspace): refresh a present per-task worktree at every respawn

ensure_worktree_self_heal treated an already-present worktree as a pure
no-op (venv-link + chown only), so a worktree created once at first claim
or first claim_review stayed frozen at that commit across every later
respawn even as new commits landed on origin — the root mechanism behind
a live multi-round QA/PR-gate bounce loop, where the reviewer kept
re-examining its own stale round-1 checkout.

_ensure_worktree_before_spawn now classifies the caller's role
(WORKTREE_AUTHOR_ROLES: developer/documenter, mirroring the gateway
commit tool's RBAC) and _refresh_present_worktree compares local HEAD
against origin/<branch>: behind-or-equal fast-forwards for every role
(never discarding an author's uncommitted edits to do it); strictly
ahead is always left alone; diverged only resets for a pure reader,
whose local history can never be anything but a stale prior-round
checkout.

conventions_check_for_task's list-vs-content gap (list from git objects,
content from the physical worktree) is closed as a side effect: the
reviewer's worktree is now current as of spawn, and the branch under
review gains no further commits while it sits in awaiting_pr_review.

* fix(workspace): refresh re-added worktrees; fail the dirty guard toward preservation

- A pruned worktree re-added from a surviving local ref now runs the same
  fetch-and-classify refresh as a present one, so an evicted reviewer
  worktree cannot resurrect a stale checkout.
- A failing git status reads as dirty, never clean: the guard that
  protects an author's uncommitted edits fails toward preservation.
- The hard reset verifies the worktree is actually on the task branch
  first; a detached or drifted worktree is left alone with a warning.
- The conventions-check docstring states the remaining second-claim
  ceiling instead of claiming full closure.

---------

Co-authored-by: Renn F <rennf93@users.noreply.github.com>

roboco/foundation/identity.py

@@ -328,6 +328,19 @@ def is_human_only_role(role: Role | None) -> bool:
     return role in _HUMAN_ONLY_ROLES
 
 
+# Roles whose task worktree may legitimately hold committed-unpushed work —
+# mirrors the gateway commit tool's RBAC (content_actions._COMMIT_ALLOWED_ROLES).
+# Every other role (qa, pr_reviewer, cell_pm, main_pm, board) only ever reads a
+# task's worktree, so a stale respawn checkout there is always safe to reset to
+# origin — it can never be discarding real unpushed work.
+WORKTREE_AUTHOR_ROLES: frozenset[Role] = frozenset({Role.DEVELOPER, Role.DOCUMENTER})
+
+
+def is_worktree_author_role(role: str | None) -> bool:
+    """True for developer/documenter. See :data:`WORKTREE_AUTHOR_ROLES`."""
+    return role in WORKTREE_AUTHOR_ROLES
+
+
 def is_spawnable_agent_slug(slug: str) -> bool:
     """True only when ``slug`` resolves to a known non-human agent role.
 

roboco/runtime/orchestrator.py

@@ -53,6 +53,7 @@ from roboco.foundation.identity import (
     CELL_TEAMS,
     is_human_only_role,
     is_spawnable_agent_slug,
+    is_worktree_author_role,
     role_for_slug_or_none,
 )
 from roboco.foundation.policy.agent_loop import DEFAULT_BUDGET as _AGENT_LOOP_BUDGET
@@ -2535,10 +2536,14 @@ class AgentOrchestrator:
         worktree (reaper, disk pressure, manual cleanup while the agent was
         down) — or a vanished clone root (disk loss, a redeploy that wiped
         ``/data/workspaces``) — would start the agent in a missing directory.
-        Idempotent: a present worktree is a no-op; a pruned worktree is re-added
-        from the surviving branch ref; a missing clone is re-cloned and the
-        branch ref recovered from origin (``create_branch`` pushes at claim
-        time) so the pushed work survives. No-op for branchless / no-task spawns.
+        Idempotent: a pruned worktree is re-added from the surviving branch
+        ref; a missing clone is re-cloned and the branch ref recovered from
+        origin (``create_branch`` pushes at claim time) so the pushed work
+        survives. A PRESENT worktree is refreshed against origin, role-aware
+        (see ``ensure_worktree_self_heal``) rather than left frozen at
+        whatever commit it was created on — the role comes from
+        ``get_agent_role``, the same lookup ``_prepare_agent_spawn`` already
+        makes for this agent. No-op for branchless / no-task spawns.
 
         The reaper-style claim release preserves ownership + ``branch_name``, so
         a re-dispatch is a RESUME, not a fresh claim — ``create_branch`` never
@@ -2559,6 +2564,7 @@ class AgentOrchestrator:
                 project_slug, team, agent_id, git_context.task_short_id
             )
         )
+        can_author = is_worktree_author_role(get_agent_role(agent_id))
         from roboco.db.base import get_db_context
         from roboco.services.workspace import WorkspaceError, WorkspaceService
 
@@ -2574,7 +2580,11 @@ class AgentOrchestrator:
                 if not WorkspaceService._is_workspace_healthy(clone_root):
                     await ws.ensure_workspace(project_slug, agent_id)
                 await ws.ensure_worktree_self_heal(
-                    clone_root, worktree, git_context.branch_name, project_slug
+                    clone_root,
+                    worktree,
+                    git_context.branch_name,
+                    project_slug,
+                    can_author=can_author,
                 )
         except WorkspaceError as e:
             # Fatal git state (clone won't re-clone, token missing, branch ref

roboco/services/git.py

@@ -6147,6 +6147,28 @@ class GitService(BaseService):
 
         ``preferred_parent`` threads to ``list_changed_files`` — the in-path
         PR-review gate's cross-team parent (see ``diff``'s docstring).
+
+        The changed-file LIST above comes from git objects (``list_changed_files``
+        fetches + diffs ``origin/<branch>``); the validator below reads CONTENT
+        off the physical worktree, which only ``_ensure_worktree_for_commit``
+        touches here (re-add if pruned, no refresh — no fetch, no classify).
+        These could disagree on a worktree that predates the branch's current
+        tip. For the FIRST claim of a review session, they don't:
+        ``ensure_worktree_self_heal`` (the spawn chokepoint, run once before
+        this agent's session started — including its re-add-from-a-surviving-
+        local-ref path) already classified the worktree against origin, and
+        the assembled PR under review can gain no further commits while it
+        sits in this task's own review state, so list and content are the
+        same origin tip by the time this runs.
+
+        Narrow accepted ceiling: a reviewer session that claims a SECOND task
+        mid-session (same container, no respawn) never gets another spawn-time
+        refresh — that only runs once, before the session started. A fresh
+        worktree this session creates for that claim has nothing to disagree
+        with (both list and content start at its own branch tip), but a
+        worktree this session INHERITS from an earlier, still-open claim of
+        that same task could be stale by whatever origin gained since. No
+        claim-time refresh exists to close this; it is accepted, not fixed.
         """
         try:
             branch = task.branch_name

roboco/services/workspace.py

@@ -716,6 +716,8 @@ class WorkspaceService:
         worktree: Path,
         branch: str,
         project_slug: str,
+        *,
+        can_author: bool,
     ) -> None:
         """Re-attach a per-task worktree, self-healing a vanished clone + ref.
 
@@ -732,21 +734,43 @@ class WorkspaceService:
         unhealthy (re-clone from default). This then recovers the task branch
         ref so the worktree re-attaches with the pushed work intact:
 
-        1. A present, registered worktree is a no-op (just venv + ownership).
+        1. A present, registered worktree is refreshed against origin — see
+           :meth:`_refresh_present_worktree` — then venv-linked + chowned.
+           Without this a worktree created once (a reviewer's first
+           ``claim_review``) stayed frozen at that commit across every
+           respawn even as new commits landed on origin (a QA/PR-gate round
+           2+ reviewing its own stale round-1 checkout).
         2. If the local ``refs/heads/{branch}`` ref is absent (a re-clone has
            none), fetch ``origin <branch>`` (token-aware) and create the local
            ref from ``refs/remotes/origin/{branch}`` when origin has it —
            recovering the pushed commits. (``create_branch`` always pushes at
-           claim time, so a claimed task's branch is on origin.)
-        3. ``ensure_worktree`` reuses the recovered ref, or — if origin doesn't
-           have it (a never-pushed branch) — re-creates it from ``origin/HEAD``;
-           no pushed work is lost because none existed.
+           claim time, so a claimed task's branch is on origin.) ``ensure_worktree``
+           then reuses the recovered ref, or — if origin doesn't have it (a
+           never-pushed branch) — re-creates it from ``origin/HEAD``; no pushed
+           work is lost because none existed.
+        3. If the local ref instead SURVIVED (the worktree itself was pruned —
+           disk pressure, manual cleanup, a reaper evict — while the branch ref
+           lived on), re-add via ``ensure_worktree`` and then run it through the
+           SAME :meth:`_refresh_present_worktree` classification as a present
+           worktree. Re-adding alone would resurrect whatever commit the local
+           ref happened to point at, however far origin moved since (a
+           reviewer's round-1 worktree evicted, then re-added by a round-2
+           respawn into its own stale round-1 checkout — the present-worktree
+           bug's absent-worktree twin).
 
         A transient fetch failure falls through to the ``origin/HEAD`` ``-b``
-        rather than fatal-looping; a diverged branch re-syncs on the agent's
-        first ``sync_branch``.
+        rather than fatal-looping; a diverged author-role branch re-syncs on
+        the agent's first ``sync_branch``.
+
+        ``can_author`` (see ``foundation.identity.WORKTREE_AUTHOR_ROLES``) is
+        the caller's role classification, not a task property — the SAME
+        branch is a developer's own worktree in the dev's clone and a pure
+        reader's worktree in QA's/the reviewer's own separate clone.
         """
         if worktree.exists() and (worktree / ".git").is_file():
+            await self._refresh_present_worktree(
+                clone_root, worktree, branch, project_slug, can_author=can_author
+            )
             self._link_shared_venv(worktree, clone_root)
             await asyncio.to_thread(_ensure_agent_owned, worktree)
             await asyncio.to_thread(_ensure_agent_owned, clone_root)
@@ -769,8 +793,150 @@ class WorkspaceService:
                     ["branch", branch, f"refs/remotes/origin/{branch}"],
                     check=False,
                 )
-        # Reuse refs/heads/{branch} if recovered; else -b from origin/HEAD.
+            # Reuse refs/heads/{branch} if recovered; else -b from origin/HEAD.
+            await self.ensure_worktree(clone_root, worktree, branch, "origin/HEAD")
+            return
+        # A surviving local ref means the WORKTREE was pruned, not that its
+        # would-be checkout is current — re-add, then classify it exactly like
+        # a present worktree instead of trusting the ref's stale commit.
         await self.ensure_worktree(clone_root, worktree, branch, "origin/HEAD")
+        await self._refresh_present_worktree(
+            clone_root, worktree, branch, project_slug, can_author=can_author
+        )
+        self._link_shared_venv(worktree, clone_root)
+        await asyncio.to_thread(_ensure_agent_owned, worktree)
+        await asyncio.to_thread(_ensure_agent_owned, clone_root)
+
+    @staticmethod
+    def _worktree_is_dirty(worktree: Path) -> bool:
+        """True iff the worktree has staged/unstaged/untracked changes.
+
+        A failed ``git status`` (nonzero returncode, empty stdout) fails
+        toward DIRTY, never clean — the caller uses this to decide whether an
+        author's uncommitted edits are safe to reset out from under, and a
+        false "clean" would discard them.
+        """
+        status = WorkspaceService._worktree_git(
+            worktree, ["status", "--porcelain"], check=False
+        )
+        return status.returncode != 0 or bool(status.stdout.strip())
+
+    @staticmethod
+    def _worktree_on_task_branch(worktree: Path, branch: str) -> bool:
+        """True iff the worktree's checked-out branch IS ``branch``.
+
+        Guards every ``reset --hard origin/<branch>`` below: a detached HEAD
+        (mid-rebase, a crashed checkout) or a worktree drifted onto a
+        different ref must never have ``branch``'s ref reset out from under
+        it.
+        """
+        current = WorkspaceService._worktree_git(
+            worktree, ["branch", "--show-current"], check=False
+        )
+        return current.returncode == 0 and current.stdout.strip() == branch
+
+    @classmethod
+    def _safe_to_reset(cls, worktree: Path, branch: str) -> bool:
+        if cls._worktree_on_task_branch(worktree, branch):
+            return True
+        logger.warning(
+            "ensure_worktree_self_heal: worktree not on its task branch, "
+            "skipping reset",
+            worktree=str(worktree),
+            branch=branch,
+        )
+        return False
+
+    async def _refresh_present_worktree(
+        self,
+        clone_root: Path,
+        worktree: Path,
+        branch: str,
+        project_slug: str,
+        *,
+        can_author: bool,
+    ) -> None:
+        """Re-sync an ALREADY-PRESENT worktree with origin before a respawn
+        reuses it.
+
+        A per-task worktree is created once (first claim, or a reviewer's
+        first ``claim_review``/``claim_gate_review``) and, before this, never
+        touched again — every later respawn saw whatever commit happened to
+        be checked out at creation time, however far origin moved since (a
+        QA/PR-gate reviewer bounced a task back for fixes, the dev pushed —
+        routinely a force-pushed rebase — then respawned the reviewer into
+        its OWN still-round-1 checkout: the root mechanism behind a live
+        multi-round QA bounce loop).
+
+        Best-effort throughout — an offline spawn must not break: a failed
+        fetch or an unresolvable ``origin/<branch>`` leaves the worktree
+        exactly as it was (the absent-worktree path already falls back to
+        ``origin/HEAD`` in that case; a present one has nothing better to do
+        than keep what it has).
+
+        Classifies local HEAD (normally ``branch`` — the worktree is meant to
+        be checked out on it) against ``origin/<branch>``:
+          - behind-or-equal (origin has every local commit): fast-forward to
+            origin. Safe content-wise for every role — nothing local is
+            unique — but an author-capable role's UNCOMMITTED edits are never
+            discarded to get there; a dirty author tree is left alone
+            entirely rather than resetting under it.
+          - strictly ahead (local has commits origin lacks): always leave
+            alone — real unpushed work, author or not.
+          - diverged: an author-capable role keeps its own history —
+            ``sync_branch``'s patch-equivalence probe is the self-heal path
+            for a diverged author. A pure reader's local history can only
+            ever be a stale prior-round checkout (readers never hold the
+            gateway's ``commit`` tool — see
+            ``foundation.identity.WORKTREE_AUTHOR_ROLES`` — and a reader's
+            durable outputs, e.g. QA's render previews, are written outside
+            the worktree entirely), so it hard-resets to origin, discarding
+            any uncommitted scratch along with it.
+
+        Every reset above is additionally guarded by ``_safe_to_reset``: a
+        detached HEAD or a worktree drifted onto a different branch (mid-rebase,
+        a crashed checkout) is left untouched with a warning rather than have
+        ``branch``'s ref moved under it.
+        """
+        await self._fetch_branch_ref(clone_root, branch, project_slug)
+        origin_ref = f"origin/{branch}"
+        if (
+            self._worktree_git(
+                clone_root,
+                ["rev-parse", "--verify", "--quiet", origin_ref],
+                check=False,
+            ).returncode
+            != 0
+        ):
+            return  # never pushed, or the fetch failed — nothing to compare to
+        ahead = self._worktree_git(
+            clone_root, ["rev-list", "--count", f"{origin_ref}..{branch}"], check=False
+        )
+        behind = self._worktree_git(
+            clone_root, ["rev-list", "--count", f"{branch}..{origin_ref}"], check=False
+        )
+        if ahead.returncode != 0 or behind.returncode != 0:
+            return
+        local_ahead = ahead.stdout.strip()
+        local_behind = behind.stdout.strip()
+        if local_ahead == "0":
+            if local_behind == "0":
+                return  # already in sync
+            if can_author and self._worktree_is_dirty(worktree):
+                return  # never discard an author's uncommitted edits
+            self._reset_worktree_if_safe(worktree, branch, origin_ref)
+            return
+        if local_behind == "0":
+            return  # strictly ahead — someone's real unpushed work
+        # Diverged: only a pure reader's copy is disposable.
+        if not can_author:
+            self._reset_worktree_if_safe(worktree, branch, origin_ref)
+
+    def _reset_worktree_if_safe(
+        self, worktree: Path, branch: str, origin_ref: str
+    ) -> None:
+        if self._safe_to_reset(worktree, branch):
+            self._worktree_git(worktree, ["reset", "--hard", origin_ref], check=False)
 
     async def remove_worktree(self, clone_root: Path, worktree: Path) -> None:
         """Remove a per-task worktree (cancel / terminal / reaper evict).

tests/unit/runtime/test_spawn_worktree_ensure.py

@@ -69,13 +69,40 @@ async def test_ensures_worktree_when_task_short_id_set() -> None:
     # Healthy clone -> no re-clone, just the worktree self-heal.
     ws.ensure_workspace.assert_not_awaited()
     ws.ensure_worktree_self_heal.assert_awaited_once()
-    args = ws.ensure_worktree_self_heal.call_args.args
-    assert args[0] == Path("/data/workspaces/roboco-api/backend/be-dev-1")
-    assert args[1] == Path(
+    call = ws.ensure_worktree_self_heal.call_args
+    assert call.args[0] == Path("/data/workspaces/roboco-api/backend/be-dev-1")
+    assert call.args[1] == Path(
         "/data/workspaces/roboco-api/backend/be-dev-1/.worktrees/a3c40fe7"
     )
-    assert args[2] == "feature/backend/abc12345"
-    assert args[3] == "roboco-api"
+    assert call.args[2] == "feature/backend/abc12345"
+    assert call.args[3] == "roboco-api"
+    # be-dev-1 is a developer — an author-capable role.
+    assert call.kwargs["can_author"] is True
+
+
+@pytest.mark.asyncio
+async def test_reader_role_classified_as_non_author_for_refresh() -> None:
+    # A QA/PR-reviewer/PM respawn onto the SAME task-branch worktree must be
+    # classified as a pure reader so ensure_worktree_self_heal knows a stale
+    # checkout there is always safe to reset to origin.
+    orch = _make_orchestrator()
+    ws = MagicMock()
+    ws.ensure_worktree_self_heal = AsyncMock()
+    ws.ensure_workspace = AsyncMock()
+
+    with (
+        patch("roboco.db.base.get_db_context", return_value=_fake_db_ctx(MagicMock())),
+        patch("roboco.services.workspace.WorkspaceService", return_value=ws),
+        patch(
+            "roboco.services.workspace.WorkspaceService._is_workspace_healthy",
+            return_value=True,
+        ),
+    ):
+        await orch._ensure_worktree_before_spawn(
+            _ctx(), "roboco-api", "backend", "be-qa", "task-1"
+        )
+
+    assert ws.ensure_worktree_self_heal.call_args.kwargs["can_author"] is False
 
 
 @pytest.mark.asyncio

tests/unit/services/test_workspace_worktree_lifecycle.py

@@ -263,7 +263,14 @@ def _ref_exists(repo: Path, ref: str) -> bool:
     )
 
 
-async def test_self_heal_noop_when_worktree_present(clone: Path) -> None:
+async def test_self_heal_present_worktree_fetches_but_noops_without_origin(
+    clone: Path,
+) -> None:
+    # A present worktree is no longer an unconditional no-op (the respawn
+    # bug): a fetch is now always attempted. This `clone` fixture carries no
+    # `origin` remote at all, so `origin/<branch>` can never resolve and the
+    # refresh has nothing to compare against — same observable outcome as
+    # the old no-op, but for a different reason (unresolvable, not skipped).
     svc = _service()
     wt = clone / ".worktrees" / "a3c40fe7"
     with patch("roboco.services.workspace._ensure_agent_owned"):
@@ -275,15 +282,21 @@ async def test_self_heal_noop_when_worktree_present(clone: Path) -> None:
         ) as fetch,
         patch("roboco.services.workspace._ensure_agent_owned"),
     ):
-        await svc.ensure_worktree_self_heal(clone, wt, "feature/a3c40fe7", "proj")
+        await svc.ensure_worktree_self_heal(
+            clone, wt, "feature/a3c40fe7", "proj", can_author=True
+        )
 
-    assert fetch.await_count == 0, "present worktree must not trigger a fetch"
+    assert fetch.await_count == 1, "present worktree must now attempt a fetch"
     assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == "feature/a3c40fe7"
 
 
 async def test_self_heal_readds_pruned_worktree_from_local_ref(clone: Path) -> None:
     # Common resume case: clone healthy, worktree pruned, local branch ref
-    # survives -> re-add with NO fetch (no origin round-trip on every spawn).
+    # survives -> re-add, THEN run it through the same fetch-and-classify
+    # refresh a present worktree gets (a stale local ref must not resurrect an
+    # untouched checkout). This `clone` fixture carries no `origin` remote, so
+    # there is nothing to classify against — the refresh's own fetch still
+    # runs, it just has no origin/<branch> to compare to.
     svc = _service()
     wt = clone / ".worktrees" / "a3c40fe7"
     with patch("roboco.services.workspace._ensure_agent_owned"):
@@ -297,13 +310,52 @@ async def test_self_heal_readds_pruned_worktree_from_local_ref(clone: Path) -> N
         ) as fetch,
         patch("roboco.services.workspace._ensure_agent_owned"),
     ):
-        await svc.ensure_worktree_self_heal(clone, wt, "feature/a3c40fe7", "proj")
+        # can_author is irrelevant on the absent-worktree path (pre-refresh).
+        await svc.ensure_worktree_self_heal(
+            clone, wt, "feature/a3c40fe7", "proj", can_author=True
+        )
 
-    assert fetch.await_count == 0, "local ref survives -> no fetch needed"
+    assert fetch.await_count == 1, (
+        "a re-add from a surviving local ref must now refresh"
+    )
     assert wt.exists()
     assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == "feature/a3c40fe7"
 
 
+async def test_self_heal_readd_from_stale_local_ref_lands_on_origin_tip_for_reader(
+    tmp_path: Path,
+) -> None:
+    # THE BUG SCENARIO: a reviewer's round-1 claim_review creates the worktree
+    # + local ref at origin's tip A; the worktree is evicted (disk pressure /
+    # manual cleanup) while the local ref survives; a dev then pushes tip B.
+    # A round-2 respawn's re-add must land on B, not resurrect the stale local
+    # ref's A.
+    branch = "feature/pruned-stale-ref"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "pruned-stale-ref"
+    _git(clone, "worktree", "remove", str(wt), "--force")  # evicted; local ref survives
+    assert not wt.exists()
+    assert _ref_exists(clone, f"refs/heads/{branch}"), (
+        "precondition: local ref survives"
+    )
+    _push_extra_commit(tmp_path, remote, branch, "other")  # origin advances to tip B
+    _git(clone, "fetch", "origin", branch)  # what the mocked _fetch_branch_ref would do
+
+    svc = _service()
+    await _run_self_heal(svc, clone, wt, branch, can_author=False)
+
+    assert wt.exists()
+    assert (wt / "origin_advance.txt").exists(), (
+        "a re-add from a stale local ref must land on origin's tip, not the "
+        "ref's stale commit"
+    )
+    assert (
+        _git(wt, "rev-parse", "HEAD").strip()
+        == _git(clone, "rev-parse", f"origin/{branch}").strip()
+    )
+
+
 def _bare_remote_with_branch(tmp_path: Path, branch: str, push_branch: bool) -> Path:
     """A bare remote carrying `main`; optionally also `branch` with a commit."""
     remote = tmp_path / "remote.git"
@@ -363,7 +415,8 @@ async def test_self_heal_recovers_branch_from_origin(tmp_path: Path) -> None:
         ) as fetch,
         patch("roboco.services.workspace._ensure_agent_owned"),
     ):
-        await svc.ensure_worktree_self_heal(clone, wt, branch, "proj")
+        # can_author is irrelevant on the absent-worktree path (pre-refresh).
+        await svc.ensure_worktree_self_heal(clone, wt, branch, "proj", can_author=True)
 
     assert fetch.await_count == 1, "missing local ref must trigger a fetch"
     assert wt.exists()
@@ -394,13 +447,204 @@ async def test_self_heal_falls_back_to_origin_head_when_branch_not_pushed(
         ) as fetch,
         patch("roboco.services.workspace._ensure_agent_owned"),
     ):
-        await svc.ensure_worktree_self_heal(clone, wt, branch, "proj")
+        # can_author is irrelevant on the absent-worktree path (pre-refresh).
+        await svc.ensure_worktree_self_heal(clone, wt, branch, "proj", can_author=True)
 
     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
 
 
+# ---------------------------------------------------------------------------
+# _refresh_present_worktree — role-aware respawn refresh of an ALREADY-PRESENT
+# worktree (the respawn bug). A worktree created once (first claim / first
+# claim_review) must not stay frozen at that commit across every later
+# respawn while origin moves on. `_fetch_branch_ref` is mocked (a spy, as
+# above) — the tests pre-seed `origin/<branch>`'s remote-tracking ref with a
+# real `git fetch` so the classification runs against real git state.
+# ---------------------------------------------------------------------------
+
+
+async def _synced_clone_and_worktree(tmp_path: Path, remote: Path, branch: str) -> Path:
+    """Clone `remote`, create+push a worktree on `branch` at origin's tip.
+
+    Mirrors `create_branch`'s real shape: the worktree branch is cut, then
+    pushed, so local and `origin/<branch>` start perfectly in sync.
+    """
+    clone = tmp_path / "clone"
+    subprocess.run(
+        ["git", "clone", str(remote), str(clone)], check=True, capture_output=True
+    )
+    svc = _service()
+    wt = clone / ".worktrees" / branch.rsplit("/", 1)[-1]
+    with patch("roboco.services.workspace._ensure_agent_owned"):
+        await svc.ensure_worktree(clone, wt, branch, "main")
+    _git(wt, "push", "origin", branch)
+    return clone
+
+
+def _push_extra_commit(tmp_path: Path, remote: Path, branch: str, name: str) -> None:
+    """A second clone pushes one more commit onto `branch` (simulates a dev's
+    force-pushed fix landing on origin between two reviewer respawns)."""
+    other = tmp_path / name
+    subprocess.run(
+        ["git", "clone", str(remote), str(other)], check=True, capture_output=True
+    )
+    _git(other, "checkout", branch)
+    (other / "origin_advance.txt").write_text(name)
+    _git(other, "add", "origin_advance.txt")
+    _git(other, "commit", "-m", "origin advances")
+    _git(other, "push", "origin", branch)
+
+
+def _commit_local_only(wt: Path) -> None:
+    """A commit in the worktree that never reaches origin (unpushed work)."""
+    (wt / "local_only.txt").write_text("mine")
+    _git(wt, "add", "local_only.txt")
+    _git(wt, "commit", "-m", "local unpushed work")
+
+
+async def _run_self_heal(
+    svc: WorkspaceService, clone: Path, wt: Path, branch: str, *, can_author: bool
+) -> None:
+    with (
+        patch.object(WorkspaceService, "_fetch_branch_ref", new_callable=AsyncMock),
+        patch("roboco.services.workspace._ensure_agent_owned"),
+    ):
+        await svc.ensure_worktree_self_heal(
+            clone, wt, branch, "proj", can_author=can_author
+        )
+
+
+async def test_refresh_reader_diverged_resets_to_origin(tmp_path: Path) -> None:
+    branch = "feature/reader-diverged"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "reader-diverged"
+    _commit_local_only(wt)  # local ref now ahead of origin/<branch>
+    _push_extra_commit(tmp_path, remote, branch, "other")  # ...and origin too
+    _git(clone, "fetch", "origin", branch)  # what the mocked _fetch_branch_ref would do
+    assert (wt / "local_only.txt").exists(), "precondition: local commit present"
+
+    svc = _service()
+    await _run_self_heal(svc, clone, wt, branch, can_author=False)
+
+    assert not (wt / "local_only.txt").exists(), (
+        "reader's diverged local history is disposable — must reset to origin"
+    )
+    assert (wt / "origin_advance.txt").exists(), "origin's tip must now be checked out"
+    assert (
+        _git(wt, "rev-parse", "HEAD").strip()
+        == _git(clone, "rev-parse", f"origin/{branch}").strip()
+    )
+
+
+async def test_refresh_author_ahead_untouched(tmp_path: Path) -> None:
+    branch = "feature/author-ahead"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "author-ahead"
+    _commit_local_only(wt)  # strictly ahead — origin never moved
+    _git(clone, "fetch", "origin", branch)
+
+    svc = _service()
+    await _run_self_heal(svc, clone, wt, branch, can_author=True)
+
+    assert (wt / "local_only.txt").exists(), "strictly-ahead work is never discarded"
+
+
+async def test_refresh_author_diverged_untouched(tmp_path: Path) -> None:
+    branch = "feature/author-diverged"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "author-diverged"
+    _commit_local_only(wt)
+    _push_extra_commit(tmp_path, remote, branch, "other")
+    _git(clone, "fetch", "origin", branch)
+
+    svc = _service()
+    await _run_self_heal(svc, clone, wt, branch, can_author=True)
+
+    assert (wt / "local_only.txt").exists(), (
+        "an author's diverged history is sync_branch's job, never a silent reset"
+    )
+    assert not (wt / "origin_advance.txt").exists(), "no reset must have run at all"
+
+
+async def test_refresh_behind_fast_forwards_for_any_role(tmp_path: Path) -> None:
+    branch = "feature/behind"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "behind"
+    _push_extra_commit(tmp_path, remote, branch, "other")  # local has nothing unique
+    _git(clone, "fetch", "origin", branch)
+
+    svc = _service()
+    await _run_self_heal(svc, clone, wt, branch, can_author=True)
+
+    assert (wt / "origin_advance.txt").exists(), (
+        "behind-or-equal is safe to fast-forward for every role"
+    )
+
+
+async def test_refresh_dirty_author_tree_preserved_even_when_behind(
+    tmp_path: Path,
+) -> None:
+    branch = "feature/dirty-author"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "dirty-author"
+    (wt / "pyproject.toml").write_text("[project]\nname = 'edited'\n")  # uncommitted
+    _push_extra_commit(tmp_path, remote, branch, "other")
+    _git(clone, "fetch", "origin", branch)
+
+    svc = _service()
+    await _run_self_heal(svc, clone, wt, branch, can_author=True)
+
+    assert (wt / "pyproject.toml").read_text() == "[project]\nname = 'edited'\n", (
+        "an author's uncommitted edit must never be discarded by a reset"
+    )
+    assert not (wt / "origin_advance.txt").exists(), "no reset must have run at all"
+
+
+def test_worktree_is_dirty_treats_failed_status_as_dirty(tmp_path: Path) -> None:
+    # A failed `git status` (nonzero returncode, empty stdout) must read as
+    # dirty, never clean — a false "clean" here lets an author+behind branch
+    # proceed straight to `reset --hard` and discard uncommitted edits.
+    failed = subprocess.CompletedProcess(
+        args=[], returncode=128, stdout="", stderr="fatal: not a git repository"
+    )
+    with patch.object(WorkspaceService, "_worktree_git", return_value=failed):
+        assert WorkspaceService._worktree_is_dirty(tmp_path) is True
+
+
+async def test_refresh_skips_reset_when_worktree_drifted_off_task_branch(
+    tmp_path: Path,
+) -> None:
+    # A worktree parked on some OTHER branch (a crashed mid-rebase, a drifted
+    # checkout) must never have the task branch's ref reset under it — a
+    # `reset --hard` runs in the worktree's own checked-out branch, not
+    # necessarily the task branch, so blindly resetting would move the wrong
+    # ref.
+    branch = "feature/drifted"
+    remote = _bare_remote_with_branch(tmp_path, branch, push_branch=False)
+    clone = await _synced_clone_and_worktree(tmp_path, remote, branch)
+    wt = clone / ".worktrees" / "drifted"
+    _push_extra_commit(tmp_path, remote, branch, "other")  # task branch now behind
+    _git(clone, "fetch", "origin", branch)
+    _git(wt, "checkout", "-b", "other-work")  # worktree drifts off the task branch
+
+    svc = _service()
+    with patch("roboco.services.workspace.logger.warning") as warn:
+        await _run_self_heal(svc, clone, wt, branch, can_author=True)
+
+    assert warn.called, "a drifted worktree must log a warning instead of resetting"
+    assert not (wt / "origin_advance.txt").exists(), (
+        "a worktree drifted off its task branch must be left alone"
+    )
+    assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == "other-work"
+
+
 # ---------------------------------------------------------------------------
 # 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
@@ -479,8 +723,16 @@ async def test_self_heal_recovers_clone_root_left_on_task_branch(clone: Path) ->
     _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")
+    with (
+        patch.object(WorkspaceService, "_fetch_branch_ref", new_callable=AsyncMock),
+        patch("roboco.services.workspace._ensure_agent_owned"),
+    ):
+        # The local ref here comes straight from `_clone_on_branch`, not a
+        # prior `ensure_worktree` — a surviving local ref now also re-adds
+        # through the present-worktree refresh; origin/<branch> is
+        # unresolvable (only origin/main was seeded), so the refresh's fetch
+        # is a no-op past the re-add.
+        await svc.ensure_worktree_self_heal(clone, wt, branch, "proj", can_author=True)
 
     assert (wt / ".git").is_file()
     assert _git(wt, "rev-parse", "--abbrev-ref", "HEAD").strip() == branch