Merge remote-tracking branch 'origin/slave' into feature/main_pm/5685ace0

.github/workflows/release.yml

@@ -122,6 +122,12 @@ jobs:
             # (unlike roboco-agent-grok above) it needs no special build-order
             # carve-out and is just another entry here.
             [roboco-agent-gemini]=docker/agent-gemini.Dockerfile
+            # Kimi (Moonshot, official kimi-code CLI) — same shape as gemini/
+            # codex: one-shot delivery roles only, plain FROM agent-base.
+            # Missing from this map on v0.28.0 shipped the release without the
+            # image; pull-smoke went red on exactly that pull. Every new
+            # docker/agent-*.Dockerfile MUST land here in the same PR.
+            [roboco-agent-kimi]=docker/agent-kimi.Dockerfile
           )
           for name in "${!IMAGES[@]}"; do
             echo "::group::build ${name}"

docs/rag/lifecycle/status-transitions.md

@@ -11,7 +11,6 @@
 | awaiting_documentation | claimed | claim | documenter |
 | awaiting_pm_review | awaiting_ceo_approval | escalate_to_ceo | head_marketing, main_pm, product_owner |
 | awaiting_pm_review | cancelled | cancel | cell_pm, ceo, main_pm |
-| awaiting_pm_review | claimed | claim | cell_pm, main_pm |
 | awaiting_pm_review | completed | complete | cell_pm, main_pm |
 | awaiting_pm_review | needs_revision | request_changes | cell_pm, main_pm |
 | awaiting_pr_review | awaiting_pm_review | pr_pass | pr_reviewer |

panel/lib/lifecycle.json

@@ -2,7 +2,6 @@
   "claim_rules": {
     "auditor": [],
     "cell_pm": [
-      "awaiting_pm_review",
       "needs_revision",
       "pending"
     ],
@@ -17,7 +16,6 @@
     ],
     "head_marketing": [],
     "main_pm": [
-      "awaiting_pm_review",
       "needs_revision",
       "pending"
     ],
@@ -531,15 +529,6 @@
       "source": "awaiting_pm_review",
       "target": "cancelled"
     },
-    {
-      "action": "claim",
-      "roles": [
-        "cell_pm",
-        "main_pm"
-      ],
-      "source": "awaiting_pm_review",
-      "target": "claimed"
-    },
     {
       "action": "complete",
       "roles": [

roboco/enforcement/task_lifecycle.py

@@ -92,8 +92,14 @@ _LEGACY_OPERATIONAL_EDGES: dict[Status, frozenset[Status]] = {
     Status.VERIFYING: frozenset({Status.NEEDS_REVISION, Status.PENDING}),
     # QA can park a task as blocked while waiting on dev clarification.
     Status.AWAITING_QA: frozenset({Status.BLOCKED}),
-    # PM claim + PM reject path on review queue.
+    # PM reject path on review queue. No CLAIMED edge here (removed): a PM
-    Status.AWAITING_PM_REVIEW: frozenset({Status.CLAIMED, Status.NEEDS_REVISION}),
+    # claiming its own awaiting_pm_review task used to legally reset it to
+    # in_progress via i_will_plan's composed claim, looping submit_up ->
+    # pr_pass -> awaiting_pm_review forever. lifecycle.CLAIM_RULES /
+    # _ROLE_CLAIM_STATUSES close the edge upstream of this legacy view; the
+    # choreographer's _handle_pm_reentry now steers a re-entering PM to
+    # complete/request_changes instead.
+    Status.AWAITING_PM_REVIEW: frozenset({Status.NEEDS_REVISION}),
     # Re-entry from revision back into active dev work (without re-claim), or
     # voluntary unclaim back to the pool (TaskService.unclaim_for_agent) — a
     # dev sent back for revision otherwise had no legal exit but in_progress.
@@ -113,13 +119,6 @@ _LEGACY_ROLE_GATES: dict[tuple[Status, Status], tuple[str, ...]] = {
         "main_pm",
         "product_owner",
     ),
-    # PM claim of review queue.
-    (Status.AWAITING_PM_REVIEW, Status.CLAIMED): (
-        "cell_pm",
-        "head_marketing",
-        "main_pm",
-        "product_owner",
-    ),
     # PM reject back to dev (needs_revision).
     (Status.AWAITING_PM_REVIEW, Status.NEEDS_REVISION): (
         "cell_pm",

roboco/foundation/policy/lifecycle.py

@@ -245,14 +245,12 @@ _STATUS_TRANSITIONS: tuple[StatusTransition, ...] = (
         frozenset({Role.DOCUMENTER}),
     ),
     StatusTransition(Status.NEEDS_REVISION, Status.CLAIMED, "claim", None),
-    # A PM re-claims an awaiting_pm_review task it already owns (CLAIM_RULES
+    # No AWAITING_PM_REVIEW -> CLAIMED edge: a PM re-entering its own
-    # grants this to CELL_PM/MAIN_PM) — see the "claim" ActionSpec comment.
+    # review-queue task is steered by the choreographer's i_will_plan
-    StatusTransition(
+    # re-entry contract straight to complete/request_changes, never via a
-        Status.AWAITING_PM_REVIEW,
+    # claim (a claim edge here let i_will_plan legally reset the task and
-        Status.CLAIMED,
+    # loop the submit_up -> pr_pass -> awaiting_pm_review cycle forever —
-        "claim",
+    # see the CLAIM_RULES comment below).
-        frozenset({Role.CELL_PM, Role.MAIN_PM}),
-    ),
     # Start
     StatusTransition(Status.CLAIMED, Status.IN_PROGRESS, "start", None),
     # Block / pause / resume
@@ -454,14 +452,6 @@ _ATOMIC_ACTIONS: dict[str, ActionSpec] = {
                 Status.AWAITING_QA,
                 Status.AWAITING_DOCUMENTATION,
                 Status.AWAITING_PR_REVIEW,
-                # A PM re-claims a review/queue task it already owns (e.g. after
-                # a respawn) via i_will_plan — CLAIM_RULES[CELL_PM/MAIN_PM]
-                # grants AWAITING_PM_REVIEW. Without it here, can_invoke_intent
-                # rejected i_will_plan on an awaiting_pm_review task with
-                # invalid_state even though CLAIM_RULES said it was allowed.
-                # test_claim_rules_match_pre_gateway_table keeps this source set
-                # and CLAIM_RULES in sync.
-                Status.AWAITING_PM_REVIEW,
             }
         ),
         target_status=Status.CLAIMED,
@@ -749,21 +739,22 @@ CLAIM_RULES: dict[Role, frozenset[Status]] = {
     # agent its own assigned tasks. (A per-instance ownership gate at the gateway
     # would diverge from this spec — the parity invariant forbids that.)
     #
-    # AWAITING_PM_REVIEW: a PM re-claims its own review-queue task (e.g. after
+    # AWAITING_PM_REVIEW is deliberately ABSENT here — do not re-add it. A task
-    # a respawn) via i_will_plan. roboco/services/task.py's
+    # in this status already passed the in-path PR gate and is waiting on the
-    # _ROLE_CLAIM_STATUSES already granted this to "cell_pm"/"main_pm" on the
+    # owning PM's merge decision (complete / request_changes), not on
-    # stated belief that "the spec (lifecycle.CLAIM_RULES) grants it" — but
+    # re-planning. A prior revision granted CELL_PM/MAIN_PM a claim from
-    # CLAIM_RULES never actually did, so can_invoke_intent silently rejected
+    # AWAITING_PM_REVIEW so a respawned PM could "re-claim its own review-queue
-    # every i_will_plan attempt on an awaiting_pm_review task with
+    # task", but claim composes into i_will_plan's (claim, set_plan, start)
-    # invalid_state regardless of what the service layer allowed. Added here
+    # sequence: every respawn legally re-claimed the task, reset it to
-    # to match; test_claim_rules_match_pre_gateway_table asserts CLAIM_RULES
+    # in_progress, and re-ran the full submit_up -> pr_pass ->
-    # and the service table stay in sync so they can't diverge again.
+    # awaiting_pm_review cycle — looping forever with no progress (one
-    Role.CELL_PM: frozenset(
+    # production task cycled 11 times across 37 spawns in 4h before this was
-        {Status.PENDING, Status.NEEDS_REVISION, Status.AWAITING_PM_REVIEW}
+    # caught). A PM re-entering its own AWAITING_PM_REVIEW task is now steered
-    ),
+    # by the choreographer's i_will_plan re-entry contract (_handle_pm_reentry)
-    Role.MAIN_PM: frozenset(
+    # straight to complete/request_changes, with no claim and no status change
-        {Status.PENDING, Status.NEEDS_REVISION, Status.AWAITING_PM_REVIEW}
+    # — closing the edge that made the reset possible in the first place.
-    ),
+    Role.CELL_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
+    Role.MAIN_PM: frozenset({Status.PENDING, Status.NEEDS_REVISION}),
     Role.PRODUCT_OWNER: frozenset(),
     Role.HEAD_MARKETING: frozenset(),
     Role.AUDITOR: frozenset(),

roboco/services/gateway/choreographer/_impl.py

@@ -519,7 +519,7 @@ class Choreographer:
         role_str: str,
         briefing: dict[str, Any],
     ) -> Envelope | None:
-        """Handle two distinct re-entry contracts for i_will_plan.
+        """Handle three distinct re-entry contracts for i_will_plan.
 
         Idempotent heartbeat: the PM already owns the task in in_progress —
         touch the heartbeat and return OK without re-running the spec gate.
@@ -531,10 +531,20 @@ class Choreographer:
         re-claim (claimed is not a valid source for the claim transition) and
         run set_plan+start to complete the interrupted sequence.
 
-        Returns None when neither condition applies, signalling the caller to
+        Review-queue steering: the task already passed the in-path PR gate and
-        continue to the normal claim-plan-start path. PLR0911 budget is the
+        is sitting in awaiting_pm_review awaiting this PM's merge decision. A
-        secondary reason this lives in a helper; the domain contract above is
+        respawned PM re-offered this task by give_me_work used to call
-        the primary one.
+        i_will_plan on it, and CLAIM_RULES used to let that legally re-claim
+        it — running the full composed (claim, set_plan, start) sequence and
+        resetting the task to in_progress, which re-ran submit_up -> pr_pass
+        -> awaiting_pm_review forever (one production task looped 11 cycles /
+        37 spawns in 4h before this was caught). Steer to complete /
+        request_changes instead, with NO claim and NO status change.
+
+        Returns None when none of the three conditions applies, signalling the
+        caller to continue to the normal claim-plan-start path. PLR0911 budget
+        is the secondary reason this lives in a helper; the domain contract
+        above is the primary one.
         """
         status = str(t.status)
         if status == "in_progress" and t.assigned_to == pm_agent_id:
@@ -550,6 +560,20 @@ class Choreographer:
             return await self._post_claim_journal_gate(
                 "i_will_plan", pm_agent_id, task_id, envelope
             )
+        if status == "awaiting_pm_review" and t.assigned_to == pm_agent_id:
+            return Envelope.ok(
+                status=status,
+                task_id=str(task_id),
+                next=(
+                    "this task already passed the PR-review gate and is"
+                    " awaiting your merge decision — do NOT re-plan or"
+                    " re-submit it. Call complete(task_id) to merge the"
+                    " assembled PR, or request_changes(task_id,"
+                    " findings=[...]) to bounce it back with concrete"
+                    " findings."
+                ),
+                context_briefing=briefing,
+            ).with_introspection(task=t, role=role_str)
         return None
 
     async def _pm_sub_tasks_gate(
@@ -830,6 +854,12 @@ class Choreographer:
         handed an awaiting_documentation task (or QA an awaiting_qa
         task) was told to call a dev verb it doesn't have — it looped.
         Map to the verb that actually claims the task for this role.
+
+        awaiting_pm_review is a review-queue state, not a re-plan state: a
+        PM offered its own already-gated task here must be steered to
+        complete/request_changes, never i_will_plan (i_will_plan legally
+        re-claiming from this status used to reset the task and loop the
+        submit_up -> pr_pass -> awaiting_pm_review cycle forever).
         """
         tid = str(getattr(task, "id", ""))
         status = str(getattr(task, "status", ""))
@@ -837,6 +867,13 @@ class Choreographer:
             return f"call claim_doc_task(task_id='{tid}') to start"
         if status == "awaiting_qa":
             return f"call claim_review(task_id='{tid}') to start"
+        if status == "awaiting_pm_review" and role in ("cell_pm", "main_pm"):
+            return (
+                f"this task already passed the PR-review gate — call"
+                f" complete(task_id='{tid}') to merge, or"
+                f" request_changes(task_id='{tid}', findings=[...]) to bounce"
+                " it back; do NOT call i_will_plan"
+            )
         if role in ("cell_pm", "main_pm", "product_owner", "head_marketing"):
             return f"call i_will_plan(task_id='{tid}', plan='<plan>') to start"
         return f"call i_will_work_on(task_id='{tid}', plan='<plan>') to start"
@@ -3363,6 +3400,19 @@ class Choreographer:
         satisfies the gate, no duplicate is written. Best-effort — a journal
         write failure is logged and swallowed so the verb falls through to
         the normal gate (which rejects as before), never crashing the verb.
+
+        Savepoint-guarded: the journal INSERT can lock-timeout on a
+        concurrent claim holding the task row's FK share lock (live
+        incident: an escalate_up request's write hit
+        ``LockNotAvailableError`` mid-flush). Swallowing that without a
+        rollback left the session's transaction poisoned — the very next
+        attribute touch anywhere in the request (``_escalate_up_preflight``
+        reading ``t.id``) raised an unhandled ``PendingRollbackError``
+        instead of the clean gate rejection this docstring promises.
+        ``begin_nested()`` scopes the failure to a SAVEPOINT the except
+        below rolls back to, leaving the outer transaction — and every
+        object this call didn't itself touch, e.g. the caller's ``t`` —
+        exactly as usable as if the write had never been attempted.
         """
         from roboco.config import settings as _settings
 
@@ -3370,16 +3420,17 @@ class Choreographer:
         if not text:
             return
         try:
-            latest = await self.journal.latest_decision_at(agent_id, task_id)
+            async with self.task.session.begin_nested():
-            window = _settings.pm_decision_window_seconds
+                latest = await self.journal.latest_decision_at(agent_id, task_id)
-            if (
+                window = _settings.pm_decision_window_seconds
-                latest is not None
+                if (
-                and (datetime.now(UTC) - latest).total_seconds() <= window
+                    latest is not None
-            ):
+                    and (datetime.now(UTC) - latest).total_seconds() <= window
-                return
+                ):
-            await self.journal.write_decision(
+                    return
-                agent_id=agent_id, task_id=task_id, content=text
+                await self.journal.write_decision(
-            )
+                    agent_id=agent_id, task_id=task_id, content=text
+                )
         except Exception as exc:  # best-effort; gate rejects normally on failure
             logger.warning(
                 "auto-record pm decision failed",

roboco/services/git.py

@@ -448,6 +448,13 @@ class GitService(BaseService):
         with "unable to append to .git/logs/refs/heads/...". A read-only
         op (status, log, diff, ...) never writes, so it skips the repair
         entirely.
+
+        This is also the ONE chokepoint every root-side git write routes
+        through, so it's where the ownership-repair root-sentinel marker
+        (``_ensure_agent_owned``'s ``.git/roboco-owned``, workspace.py) gets
+        invalidated — BEFORE the op runs, since the op is what's about to
+        create new root-owned files a live marker would otherwise let a
+        later ``_ensure_agent_owned`` call wrongly skip.
         """
         effective_timeout = timeout if timeout is not None else _default_git_timeout()
 
@@ -477,6 +484,12 @@ class GitService(BaseService):
 
         loop = asyncio.get_running_loop()
         op = " ".join(args[:2])
+        if _git_ownership_scope(args) != "none":
+            from roboco.services.workspace import invalidate_owned_marker
+
+            await loop.run_in_executor(
+                _GIT_EXECUTOR, invalidate_owned_marker, workspace
+            )
         t0 = time.monotonic()
         try:
             result = await loop.run_in_executor(_GIT_EXECUTOR, _run)

roboco/services/task.py

@@ -107,15 +107,26 @@ _ROLE_CLAIM_STATUSES: dict[str, set[TaskStatus]] = {
     # claim() returns None -> INVALID_STATE -> the PM respawn-loops on its own
     # rejected root, unable to plan or idle it. Parity is locked by
     # tests/unit/services/test_pm_claim_needs_revision.py.
+    #
+    # AWAITING_PM_REVIEW is deliberately absent: granting it here once let a
+    # respawned PM's i_will_plan legally re-claim its own review-queue task,
+    # resetting it to in_progress and re-running submit_up -> pr_pass ->
+    # awaiting_pm_review forever. lifecycle.CLAIM_RULES agrees, and unlike the
+    # NEEDS_REVISION parity above (spec ⊆ runtime only, via
+    # test_runtime_pm_claim_mapping_covers_spec_claim_rules), this exact
+    # per-PM-role SET is pinned bidirectionally against spec.CLAIM_RULES by
+    # test_claim_rules_and_role_statuses_are_identical — re-adding
+    # AWAITING_PM_REVIEW here alone (without touching the spec) fails that
+    # test instead of silently reopening the loop. The choreographer's
+    # _handle_pm_reentry now steers a re-entering PM straight to
+    # complete/request_changes instead, with no claim involved.
     "cell_pm": {
         TaskStatus.PENDING,
         TaskStatus.NEEDS_REVISION,
-        TaskStatus.AWAITING_PM_REVIEW,
     },
     "main_pm": {
         TaskStatus.PENDING,
         TaskStatus.NEEDS_REVISION,
-        TaskStatus.AWAITING_PM_REVIEW,
     },
 }
 
@@ -3741,11 +3752,18 @@ class TaskService(BaseService):
         if task.assigned_to and str(task.assigned_to) != str(agent.id):
             markers.set_original_developer(task, task.assigned_to)
 
+    # Consulted only inside _finalize_claim (both its target-status write and
+    # its branch-failure rollback's reversal-audit check), reached only via
+    # claim() -> _validate_claim_preconditions -> _get_valid_claim_statuses,
+    # which already screens the pre-claim status against _ROLE_CLAIM_STATUSES
+    # per role. AWAITING_PM_REVIEW is deliberately absent: no role's
+    # _ROLE_CLAIM_STATUSES grants it anymore (the i_will_plan re-claim loop
+    # fix), so _finalize_claim can never run with that pre-claim status —
+    # listing it here would be dead weight, not an independent gate.
     _CLAIMABLE_STATUSES: ClassVar[set[TaskStatus]] = {
         TaskStatus.PENDING,
         TaskStatus.AWAITING_QA,
         TaskStatus.AWAITING_DOCUMENTATION,
-        TaskStatus.AWAITING_PM_REVIEW,
     }
 
     async def _claim_blocked_by_dependencies(self, task: TaskTable) -> bool:

roboco/services/workspace.py

@@ -84,19 +84,35 @@ _PRUNE_DIRS = frozenset(
 )
 
 
-def _chown_entry(entry: str) -> bool:
+def _chown_entry(entry: str, st: os.stat_result) -> bool:
-    """Chown a single entry; return True on success (or already correct)."""
+    """Chown a single entry to the agent uid/gid per the given (already-known)
+    stat; return True on success (or already correct)."""
+    if st.st_uid == _AGENT_UID and st.st_gid == _AGENT_GID:
+        return True
     try:
-        st = Path(entry).stat()
+        os.chown(entry, _AGENT_UID, _AGENT_GID)
-        if st.st_uid != _AGENT_UID or st.st_gid != _AGENT_GID:
-            os.chown(entry, _AGENT_UID, _AGENT_GID)
     except OSError:
         return False
     return True
 
 
-def _make_owner_and_group_rw(entry: str) -> None:
+def _rw_mode_for(st_mode: int) -> int:
-    """Best-effort chmod ensuring owner+group have rw (+x for dirs).
+    """The owner+group rw (+x for dirs) bits ``_make_owner_and_group_rw``
+    ensures, given an entry's current mode. Shared with the root-sentinel
+    check in ``_root_already_owned`` so both agree on what "already has the
+    required bits" means.
+    """
+    import stat as _stat
+
+    mode = st_mode | _stat.S_IRUSR | _stat.S_IWUSR | _stat.S_IRGRP | _stat.S_IWGRP
+    if _stat.S_ISDIR(st_mode):
+        mode |= _stat.S_IXUSR | _stat.S_IXGRP
+    return mode
+
+
+def _make_owner_and_group_rw(entry: str, st: os.stat_result) -> None:
+    """Best-effort chmod ensuring owner+group have rw (+x for dirs), given the
+    entry's already-known stat.
 
     NAS volumes with POSIX ACL inheritance can land cloned files with
     owner=0 (e.g. `.git/config` arriving as `----rw----`). POSIX permission
@@ -107,25 +123,36 @@ def _make_owner_and_group_rw(entry: str) -> None:
     capabilities; if chown failed earlier (we're not root), we still
     can't chmod files we don't own, so this is best-effort by design.
     """
-    import stat as _stat
+    new_mode = _rw_mode_for(st.st_mode)
-
+    if new_mode == st.st_mode:
-    try:
+        return
-        st = Path(entry).stat()
+    with contextlib.suppress(OSError):
-        new_mode = (
+        Path(entry).chmod(new_mode)
-            st.st_mode | _stat.S_IRUSR | _stat.S_IWUSR | _stat.S_IRGRP | _stat.S_IWGRP
-        )
-        if _stat.S_ISDIR(st.st_mode):
-            new_mode |= _stat.S_IXUSR | _stat.S_IXGRP
-        if new_mode != st.st_mode:
-            Path(entry).chmod(new_mode)
-    except OSError:
-        pass
 
 
 def _own_and_grant_rw(entry: str) -> int:
-    """Chown + grant owner/group rw on one entry; return 1 if the chown failed."""
+    """Chown + grant owner/group rw on one entry; return 1 if the chown failed.
-    failed = 0 if _chown_entry(entry) else 1
+
-    _make_owner_and_group_rw(entry)
+    One ``stat`` (follows symlinks, matching os.chown/Path.chmod's own
+    default follow behavior) now backs BOTH the chown-needed and
+    chmod-needed checks below, replacing what used to be two separate
+    ``Path.stat()`` calls (one inside each helper). The common case — an
+    agent re-claiming its own already-correctly-owned clone — costs one
+    read syscall and zero metadata-write syscalls per entry instead of two
+    stats plus, on some hosts, always re-testing each write independently.
+    On a NAS volume with tens of thousands of files, where every
+    chown/chmod is a copy-on-write metadata write, that is the difference
+    between a sub-second ownership pass and one that stacks tens of
+    seconds per claim verb. A stat failure (e.g. a broken symlink) is
+    treated as a chown failure, matching the prior behavior where the same
+    OSError surfaced from inside ``_chown_entry``'s own stat call.
+    """
+    try:
+        st = Path(entry).stat()
+    except OSError:
+        return 1
+    failed = 0 if _chown_entry(entry, st) else 1
+    _make_owner_and_group_rw(entry, st)
     return failed
 
 
@@ -166,10 +193,27 @@ def _ensure_agent_owned(workspace: Path) -> None:
        userns hosts) we log the failure instead of swallowing it, so a
        still-failing agent write is diagnosable rather than silent.
     2. chmod owner+group rw. Belt + suspenders for ACL-inheriting NAS volumes.
+
+    Root-sentinel short-circuit: if the workspace root itself is already
+    agent-owned with the right bits AND ``_root_already_owned`` finds the
+    marker from the last zero-failure pass, the ENTIRE walk is skipped — one
+    stat instead of walking tens of thousands of files, the remaining cost
+    of a re-claim on an already-correctly-owned NAS clone (this repo's
+    per-entry stat-collapse already halved the walk itself; this removes it
+    outright in the common case). The marker is deleted at the single
+    root-side git-write chokepoint (``GitService._run_git`` →
+    ``invalidate_owned_marker``) before any op that could create new
+    root-owned files, so a live marker is trustworthy: it can only ever be
+    stale-and-wrongly-trusted if some OTHER path creates root-owned files
+    without going through that chokepoint, which is the class this repo's
+    ownership repair exists to fix in the first place.
     """
     if not workspace.exists():
         return
 
+    if _root_already_owned(workspace):
+        return
+
     failed_chowns = sum(
         _own_and_grant_rw(entry) for entry in _iter_ownable_entries(workspace)
     )
@@ -182,6 +226,8 @@ def _ensure_agent_owned(workspace: Path) -> None:
             workspace=str(workspace),
             failures=failed_chowns,
         )
+    else:
+        _write_owned_marker(workspace)
 
 
 def _resolve_clone_root(workspace: Path) -> Path:
@@ -198,6 +244,111 @@ def _resolve_clone_root(workspace: Path) -> Path:
     return workspace
 
 
+# Sentinel filename recording "the last full _ensure_agent_owned pass over
+# this clone found zero wrong-owned entries". Lives under `.git/` — git
+# never tracks its own metadata dir, so this needs no .gitignore entry and
+# sits outside every `_PRUNE_DIRS` exemption — instead of the working tree,
+# so writing/deleting it never touches a tracked file.
+_OWNED_MARKER_NAME = "roboco-owned"
+
+
+def _owned_marker_path(workspace: Path) -> Path:
+    """The sentinel's path for a workspace or one of its worktrees.
+
+    Worktree-aware via ``_resolve_clone_root``: a worktree checkout and its
+    clone root share the ONE ``.git`` they both ultimately read/write, so
+    they share one marker too.
+    """
+    return _resolve_clone_root(workspace) / ".git" / _OWNED_MARKER_NAME
+
+
+def _root_already_owned(workspace: Path) -> bool:
+    """True iff the workspace root is already agent-owned with the required
+    rw(+x) bits AND the marker attests the last full walk found zero
+    wrong-owned entries anywhere under the tree.
+
+    Only the root gets stat'd here — the marker stands in for "every other
+    entry was already correct as of the last zero-failure pass", so the
+    common re-claim case costs one stat instead of walking the whole clone.
+    A missing/unreadable root or marker just falls through to the real walk
+    (safe default — this is a pure perf short-circuit, never a correctness
+    one).
+    """
+    try:
+        st = workspace.stat()
+    except OSError:
+        return False
+    if st.st_uid != _AGENT_UID or st.st_gid != _AGENT_GID:
+        return False
+    if _rw_mode_for(st.st_mode) != st.st_mode:
+        return False
+    return _owned_marker_path(workspace).is_file()
+
+
+def _write_owned_marker(workspace: Path) -> None:
+    """Record a zero-failure ``_ensure_agent_owned`` pass so the next call
+    can trust ``_root_already_owned`` and skip the walk entirely.
+
+    Best-effort: a write failure just means the next call re-walks — it
+    only costs the perf win, never correctness.
+    """
+    with contextlib.suppress(OSError):
+        marker = _owned_marker_path(workspace)
+        marker.parent.mkdir(parents=True, exist_ok=True)
+        marker.touch()
+
+
+def invalidate_owned_marker(workspace: Path) -> None:
+    """Delete the ownership sentinel so the next ``_ensure_agent_owned``
+    call re-walks instead of trusting stale state.
+
+    Called from every root-side git-write chokepoint before the git
+    invocation that could create new root-owned files runs:
+    ``GitService._run_git`` (orchestrator-driven git ops) and, in this
+    module, ``WorkspaceService._worktree_git`` (mutating verbs only),
+    ``_fetch_branch_ref``, and ``_fetch_origin_best_effort`` — the
+    raw-subprocess worktree/fetch paths the spawn-time self-heal flow
+    (``ensure_worktree_self_heal`` -> ``_refresh_present_worktree``) hits on
+    (nearly) every spawn. A marker written BEFORE those files land would let
+    the very next ``_ensure_agent_owned`` call (concurrent or later) wrongly
+    skip them, stranding root-owned files the agent can't write. Best-effort:
+    a missing marker/workspace is a silent no-op, never an error.
+    """
+    with contextlib.suppress(OSError):
+        _owned_marker_path(workspace).unlink()
+
+
+# Verbs `_worktree_git` receives that never write anything, hand-enumerated
+# against every real call site in this module (rev-parse --verify, rev-list
+# --count, status --porcelain, symbolic-ref --short — the SET form of
+# symbolic-ref is never used here). `branch` is the one ambiguous verb this
+# helper is called with: `--show-current` only reads, while a create
+# (`branch <name> <ref>`) or delete (`branch -d/-D <name>`) writes — handled
+# separately in `_worktree_git_is_mutating` below rather than folded into
+# this set. `git.py`'s `_git_ownership_scope` can't be imported here (git.py
+# imports FROM workspace.py; the reverse would cycle).
+_WORKTREE_GIT_ALWAYS_READ_ONLY = frozenset(
+    {"rev-parse", "rev-list", "status", "symbolic-ref"}
+)
+
+
+def _worktree_git_is_mutating(args: list[str]) -> bool:
+    """True iff a ``_worktree_git`` invocation could write anything.
+
+    Everything this helper is ever called with besides the always-read-only
+    set and `branch` is mutating (checkout, reset, worktree add/remove/
+    prune) — the safe default for an unrecognized/empty verb too.
+    """
+    if not args:
+        return True
+    verb = args[0]
+    if verb in _WORKTREE_GIT_ALWAYS_READ_ONLY:
+        return False
+    if verb == "branch":
+        return any(not a.startswith("-") for a in args[1:])
+    return True
+
+
 def _iter_git_dir_entries(clone_root: Path) -> Iterator[str]:
     """Yield ``clone_root/.git`` and every entry beneath it.
 
@@ -517,6 +668,19 @@ class WorkspaceService:
     def _worktree_git(
         clone_root: Path, args: list[str], check: bool = True
     ) -> subprocess.CompletedProcess[str]:
+        """Run a raw ``git -C <clone_root> <args>`` invocation (sync, no
+        token injection — internal worktree plumbing only).
+
+        A mutating verb (anything but rev-parse/rev-list/status/
+        symbolic-ref, or a `branch` create/delete) invalidates the
+        ownership-sentinel marker BEFORE the subprocess runs — this is one
+        of the root-side git-write chokepoints ``invalidate_owned_marker``
+        documents; without it a stale marker lets ``_ensure_agent_owned``
+        skip the walk that would repair the root-owned files this call is
+        about to create.
+        """
+        if _worktree_git_is_mutating(args):
+            invalidate_owned_marker(clone_root)
         return subprocess.run(
             ["git", "-C", str(clone_root), *args],
             capture_output=True,
@@ -684,6 +848,10 @@ class WorkspaceService:
             prefix = ["-c", f"http.extraheader=Authorization: Basic {basic}"]
 
         def _do_fetch() -> subprocess.CompletedProcess[str]:
+            # fetch always writes .git/objects + refs — a root-side git-write
+            # chokepoint (see invalidate_owned_marker's docstring). Invalidate
+            # BEFORE the subprocess runs so a marker can never straddle it.
+            invalidate_owned_marker(clone_root)
             return subprocess.run(
                 [
                     "git",
@@ -1133,6 +1301,10 @@ class WorkspaceService:
             return refs or ["master"]
 
         def _do_fetch() -> subprocess.CompletedProcess[str]:
+            # fetch always writes .git/objects + refs — a root-side git-write
+            # chokepoint (see invalidate_owned_marker's docstring). Invalidate
+            # BEFORE the subprocess runs so a marker can never straddle it.
+            invalidate_owned_marker(workspace)
             return subprocess.run(
                 ["git", "fetch", "--no-tags", "--prune", "origin", *_scoped_refs()],
                 cwd=str(workspace),

tests/foundation/test_lifecycle_consumer_parity.py

@@ -869,6 +869,26 @@ async def test_complete_matches_spec(role: str, status: str) -> None:
         id=task_id, status="awaiting_ceo_approval", assigned_to=None, team="backend"
     )
     task_svc.all_subtasks_terminal.return_value = True
+    # complete's merge path runs _stamp_pm_findings_verified_or_rejection,
+    # which opens a session.begin_nested() savepoint and reads the findings
+    # ledger via ReviewFindingsRepository.list_for_task (session.execute ->
+    # .scalars().all()). Unconfigured, both calls resolve to auto-generated
+    # AsyncMock children: `async with <AsyncMock>():` fails the async
+    # context-manager protocol and `<AsyncMock-result>.scalars()` returns an
+    # unawaited coroutine — caught by the verb's own except Exception, but
+    # the never-awaited coroutine leaks a RuntimeWarning at GC time.
+    task_svc.session = MagicMock()
+    task_svc.session.execute = AsyncMock(
+        return_value=MagicMock(
+            scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
+        )
+    )
+    task_svc.session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     git_svc = AsyncMock()
     git_svc.pr_merge.return_value = {"merged": True, "merge_commit_sha": "x"}
     git_svc.create_pr.return_value = {"pr_number": 99, "pr_url": "x"}
@@ -1268,6 +1288,18 @@ async def test_claim_review_matches_spec(role: str, status: str) -> None:
     task_svc.qa_claim.return_value = after
     task_svc.list_in_progress_for_agent.return_value = []
     task_svc.list_paused_for_agent.return_value = []
+    # claim_review's full=True briefing reads the findings ledger
+    # (findings.open_findings_for_task -> session.execute -> .scalars().all()).
+    # Unconfigured, session.execute auto-generates as an AsyncMock child whose
+    # call returns another AsyncMock; .scalars() on that is itself an
+    # unawaited coroutine — caught by open_findings_for_task's own fail-open
+    # except, but the dangling coroutine leaks a RuntimeWarning at GC time.
+    task_svc.session = MagicMock()
+    task_svc.session.execute = AsyncMock(
+        return_value=MagicMock(
+            scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
+        )
+    )
     deps = _make_deps(task_svc=task_svc)
     c = Choreographer(deps)
 
@@ -1546,6 +1578,14 @@ async def test_claim_doc_task_matches_spec(role: str, status: str) -> None:
     task_svc.doc_claim.return_value = after
     task_svc.list_in_progress_for_agent.return_value = []
     task_svc.list_paused_for_agent.return_value = []
+    # claim_doc_task's full=True briefing reads the findings ledger the same
+    # way claim_review's does (see that test's comment) — same fix.
+    task_svc.session = MagicMock()
+    task_svc.session.execute = AsyncMock(
+        return_value=MagicMock(
+            scalars=MagicMock(return_value=MagicMock(all=MagicMock(return_value=[])))
+        )
+    )
     deps = _make_deps(task_svc=task_svc)
     c = Choreographer(deps)
 

tests/foundation/test_lifecycle_spec.py

@@ -491,6 +491,11 @@ def test_claim_rules_match_pre_gateway_table() -> None:
     re-delegating fixes (scoped by give_me_work routing, which offers only the
     caller's own assigned tasks). BACKLOG → PENDING is a separate `activate`
     action (strict transitions; no implicit activate-on-claim).
+
+    AWAITING_PM_REVIEW is deliberately absent from both PM roles — a claim
+    edge there let a respawned PM's i_will_plan legally re-claim its own
+    review-queue task and loop the submit_up -> pr_pass -> awaiting_pm_review
+    cycle forever. See test_awaiting_pm_review_not_claimable_by_any_role.
     """
     assert spec.CLAIM_RULES[spec.Role.DEVELOPER] == frozenset(
         {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
@@ -500,21 +505,26 @@ def test_claim_rules_match_pre_gateway_table() -> None:
         {spec.Status.PENDING, spec.Status.AWAITING_DOCUMENTATION}
     )
     assert spec.CLAIM_RULES[spec.Role.CELL_PM] == frozenset(
-        {
+        {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
-            spec.Status.PENDING,
-            spec.Status.NEEDS_REVISION,
-            spec.Status.AWAITING_PM_REVIEW,
-        }
     )
     assert spec.CLAIM_RULES[spec.Role.MAIN_PM] == frozenset(
-        {
+        {spec.Status.PENDING, spec.Status.NEEDS_REVISION}
-            spec.Status.PENDING,
-            spec.Status.NEEDS_REVISION,
-            spec.Status.AWAITING_PM_REVIEW,
-        }
     )
 
 
+def test_awaiting_pm_review_not_claimable_by_any_role() -> None:
+    """The claim edge that let a respawned PM's i_will_plan reset an
+    awaiting_pm_review task (looping submit_up -> pr_pass -> awaiting_pm_review
+    forever) is permanently closed: no role may claim from this status. A PM
+    re-entering its own review task is steered by the choreographer directly
+    to complete/request_changes, never through claim.
+    """
+    for role, statuses in spec.CLAIM_RULES.items():
+        assert spec.Status.AWAITING_PM_REVIEW not in statuses, role
+    # PR_REVIEWER's own review-gate claim (a different status) is untouched.
+    assert spec.Status.AWAITING_PR_REVIEW in spec.CLAIM_RULES[spec.Role.PR_REVIEWER]
+
+
 def test_team_rules_pin_team_for_seeded_agents() -> None:
     assert spec.ROLE_TEAM_RULES["be-dev-1"] == "backend"
     assert spec.ROLE_TEAM_RULES["be-pm"] == "backend"

tests/unit/enforcement/test_task_lifecycle.py

@@ -277,6 +277,40 @@ def test_status_classification_is_mutually_disjoint() -> None:
     assert waiting & terminal == set()
 
 
+# ---------------------------------------------------------------------------
+# awaiting_pm_review -> claimed is closed (the i_will_plan re-claim loop):
+# a PM re-entering its own review-queue task is steered by the choreographer
+# straight to complete/request_changes, never via a claim that resets the
+# task and re-runs submit_up -> pr_pass -> awaiting_pm_review forever. This
+# legacy shim (_LEGACY_OPERATIONAL_EDGES / _LEGACY_ROLE_GATES) used to grant
+# the same edge lifecycle.CLAIM_RULES had already closed — the identical
+# two-tables-drift shape that caused the incident.
+# ---------------------------------------------------------------------------
+
+
+def test_awaiting_pm_review_claim_no_longer_allowed_for_pm_roles() -> None:
+    assert can_agent_transition("awaiting_pm_review", "claimed", "cell_pm") is False
+    assert can_agent_transition("awaiting_pm_review", "claimed", "main_pm") is False
+    with pytest.raises(TaskLifecycleError):
+        validate_task_transition("awaiting_pm_review", "claimed", "cell_pm")
+    with pytest.raises(TaskLifecycleError):
+        validate_task_transition("awaiting_pm_review", "claimed", "main_pm")
+
+
+def test_awaiting_pm_review_needs_revision_still_allowed() -> None:
+    """The PM reject-back-to-dev path (request_changes) is untouched."""
+    assert (
+        can_agent_transition("awaiting_pm_review", "needs_revision", "cell_pm") is True
+    )
+    assert (
+        can_agent_transition("awaiting_pm_review", "needs_revision", "main_pm") is True
+    )
+    assert (
+        validate_task_transition("awaiting_pm_review", "needs_revision", "cell_pm")
+        is True
+    )
+
+
 def test_status_classification_covers_every_enum_member() -> None:
     """Every Status enum member must be classified by EXACTLY one of
     is_terminal_state / is_active_state / is_waiting_state — the coverage

tests/unit/gateway/test_budget_unblock_guard.py

@@ -34,6 +34,15 @@ def _make_deps(task_svc: AsyncMock) -> ChoreographerDeps:
     }
     base["journal"].has_decision_for_task.return_value = True
     base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_choreographer_delegate_guards.py

@@ -53,6 +53,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
     _ldef = base["journal"].latest_decision_at.return_value
     if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
         base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_choreographer_pm.py

@@ -12,6 +12,7 @@ from uuid import uuid4
 
 import pytest
 from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+from sqlalchemy.exc import OperationalError
 
 
 def _make_deps(**overrides: Any) -> ChoreographerDeps:
@@ -969,6 +970,52 @@ async def test_escalate_up_blocks_without_journal_decision() -> None:
     assert "journal:decision" in body["missing"]
 
 
+@pytest.mark.asyncio
+async def test_escalate_up_survives_journal_write_lock_timeout() -> None:
+    """Regression: a journal:decision INSERT that lock-times out (a
+    concurrent claim transaction holding the task row's FK share lock —
+    live production 500) used to be swallowed by ``_ensure_pm_decision``
+    with no rollback/savepoint, poisoning the session so the very next
+    attribute touch (``_escalate_up_preflight`` reading ``t.id``) raised an
+    unhandled ``PendingRollbackError``. The write is now savepoint-guarded
+    (``begin_nested()``): the failure is contained, the verb falls through
+    cleanly to the normal tracing_gap rejection (no decision was actually
+    persisted), and the task stays fully readable — no unhandled exception
+    escapes ``escalate_up``."""
+    pm_id = uuid4()
+    task_id = uuid4()
+    t = MagicMock(id=task_id, status="blocked", assigned_to=pm_id, team="backend")
+    task_svc = AsyncMock()
+    task_svc.get.return_value = t
+    task_svc.agent_for.return_value = MagicMock(
+        role="cell_pm", escalation_target="main-pm"
+    )
+    journal_svc = AsyncMock()
+    journal_svc.has_decision_for_task.return_value = False
+    journal_svc.latest_decision_at.return_value = None
+    journal_svc.write_decision.side_effect = OperationalError(
+        "INSERT INTO journal_entries (id, ...) VALUES (...)",
+        {},
+        Exception("canceling statement due to lock timeout"),
+    )
+    deps = _make_deps(task=task_svc, journal=journal_svc)
+    c = Choreographer(deps)
+
+    env = await c.escalate_up(pm_id, task_id, reason="needs cross-cell coordination")
+
+    # The savepoint was actually engaged — proves the fix is wired in, not
+    # merely that AsyncMock happened to swallow the raise on its own.
+    task_svc.session.begin_nested.assert_called()
+    # No unhandled exception escaped escalate_up: the gate falls through to
+    # its normal clean rejection since the decision write never landed.
+    body = env.as_dict()
+    assert body["error"] == "tracing_gap"
+    assert "journal:decision" in body["missing"]
+    # The task is still fully readable afterward — this is exactly where
+    # the production trace crashed with PendingRollbackError on t.id.
+    assert t.id == task_id
+
+
 @pytest.mark.asyncio
 async def test_escalate_up_no_target_returns_invalid_state() -> None:
     """Verb-specific preflight: PM whose escalation_target is unconfigured.

tests/unit/gateway/test_claim_doc_task_checkout.py

@@ -86,6 +86,16 @@ def test_claim_verb_hint_pm_for_planning() -> None:
     assert "i_will_plan" in hint
 
 
+def test_claim_verb_hint_pm_for_awaiting_pm_review_steers_to_complete() -> None:
+    """A PM's own review-queue task must never hint i_will_plan — that verb
+    used to legally re-claim and reset it, looping submit_up -> pr_pass ->
+    awaiting_pm_review forever."""
+    for role in ("cell_pm", "main_pm"):
+        hint = Choreographer._claim_verb_hint(role, _task("awaiting_pm_review"))
+        assert "complete" in hint
+        assert "call i_will_plan(" not in hint
+
+
 def test_claim_verb_hint_dev_default() -> None:
     hint = Choreographer._claim_verb_hint("developer", _task("pending"))
     assert "i_will_work_on" in hint

tests/unit/gateway/test_delegate_ac_coverage_gate.py

@@ -64,6 +64,18 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
     _ldef = base["journal"].latest_decision_at.return_value
     if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
         base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use. Only stub it for a mocked
+    # task service: one test below passes a REAL TaskService (get_task_service)
+    # over a live db_session, whose genuine begin_nested must stay intact.
+    if isinstance(base["task"], AsyncMock | MagicMock):
+        base["task"].session.begin_nested = MagicMock(
+            return_value=MagicMock(
+                __aenter__=AsyncMock(return_value=None),
+                __aexit__=AsyncMock(return_value=False),
+            )
+        )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_delegate_incomplete_input.py

@@ -52,6 +52,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
     _ldef = base["journal"].latest_decision_at.return_value
     if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
         base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_delegate_parent_lock.py

@@ -50,6 +50,15 @@ def _make_deps(task: AsyncMock) -> ChoreographerDeps:
     # A fresh decision within the recency window so the delegate tracing gate
     # (journal:decision required) passes without a separate write.
     base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_delegate_project_routing.py

@@ -38,6 +38,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
     _ldef = base["journal"].latest_decision_at.return_value
     if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
         base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_oscillation_breaker.py

@@ -41,6 +41,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
     base.update(overrides)
     base["journal"].has_decision_for_task.return_value = True
     base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/gateway/test_pm_review_reentry_loop_fix.py

@@ -0,0 +1,148 @@
+"""PM re-entry on an awaiting_pm_review task must steer, never re-claim.
+
+Live incident: an awaiting_pm_review task (already past the in-path PR gate)
+kept getting re-offered to its owning PM by give_me_work. The respawned PM
+called i_will_plan, and CLAIM_RULES used to grant CELL_PM/MAIN_PM a claim from
+AWAITING_PM_REVIEW — so the composed (claim, set_plan, start) sequence legally
+reset the task to in_progress and re-ran submit_up -> pr_pass ->
+awaiting_pm_review forever (one production task looped 11 cycles across 37
+spawns in 4h). ``_handle_pm_reentry`` now recognizes this status for the
+owning PM and returns a steering-only OK envelope (complete / request_changes)
+with no claim and no state change; CLAIM_RULES no longer permits the claim at
+all, so a non-owner (or any other caller) falls through to a normal spec
+rejection.
+"""
+
+from __future__ import annotations
+
+from datetime import UTC, datetime
+from typing import Any
+from unittest.mock import AsyncMock, MagicMock
+from uuid import uuid4
+
+import pytest
+from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps
+
+# ---------------------------------------------------------------------------
+# Shared fixture helpers — same pattern as test_i_will_plan_sub_tasks_gate.py
+# ---------------------------------------------------------------------------
+
+
+def _make_deps(**overrides: Any) -> ChoreographerDeps:
+    base: dict[str, Any] = {
+        "task": AsyncMock(),
+        "work_session": AsyncMock(),
+        "git": AsyncMock(),
+        "a2a": AsyncMock(),
+        "journal": AsyncMock(),
+        "audit": AsyncMock(),
+        "evidence_repo": AsyncMock(),
+    }
+    base.update(overrides)
+    task = base["task"]
+    task.session = MagicMock()
+    task.session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
+    repo = base["evidence_repo"]
+    for method in (
+        "list_unread_a2a",
+        "list_unread_mentions",
+        "list_pending_notifications",
+        "task_metadata_gaps",
+        "recent_team_activity",
+        "blockers_in_lane",
+        "journal_highlights_for_task",
+    ):
+        getattr(repo, method).return_value = []
+    _ldef = base["journal"].latest_decision_at.return_value
+    if type(_ldef).__name__ in ("MagicMock", "AsyncMock"):
+        base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    return ChoreographerDeps(**base)
+
+
+def _review_task_svc(task_id: object, pm_id: object, *, role: str) -> AsyncMock:
+    """TaskService mock for a PM re-entering its own awaiting_pm_review task."""
+    task_svc = AsyncMock()
+    task_svc.get.return_value = MagicMock(
+        id=task_id,
+        status="awaiting_pm_review",
+        plan={"text": "already planned"},
+        assigned_to=pm_id,
+        task_type="planning",
+        parent_task_id=None,
+        sequence=0,
+        team="backend",
+        commits=["abc123"],
+        pr_number=42,
+        branch_name="feature/backend/abc",
+        quick_context=None,
+    )
+    task_svc.agent_for.return_value = MagicMock(
+        id=pm_id, role=role, team="backend", slug=None
+    )
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.list_paused_for_agent.return_value = []
+    task_svc.get_subtasks.return_value = []
+    task_svc.session = MagicMock()
+    task_svc.session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
+    return task_svc
+
+
+async def _assert_steers_without_reclaiming(role: str) -> None:
+    pm_id = uuid4()
+    task_id = uuid4()
+    task_svc = _review_task_svc(task_id, pm_id, role=role)
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_plan(pm_id, task_id, plan="resume")
+    body = env.as_dict()
+
+    assert body.get("error") is None, body
+    assert body["status"] == "awaiting_pm_review", body
+    assert "complete" in body["next"], body
+
+    task_svc.claim.assert_not_awaited()
+    task_svc.set_plan.assert_not_awaited()
+    task_svc.start.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_cell_pm_reentry_awaiting_pm_review_steers_to_complete() -> None:
+    await _assert_steers_without_reclaiming("cell_pm")
+
+
+@pytest.mark.asyncio
+async def test_main_pm_reentry_awaiting_pm_review_steers_to_complete() -> None:
+    await _assert_steers_without_reclaiming("main_pm")
+
+
+@pytest.mark.asyncio
+async def test_non_owner_awaiting_pm_review_is_rejected_not_reclaimed() -> None:
+    """A PM that does NOT own the review task gets the normal spec rejection —
+    CLAIM_RULES no longer grants a claim from awaiting_pm_review to anyone, so
+    this falls straight through to invalid_state instead of resetting the task.
+    """
+    pm_id = uuid4()
+    other_pm_id = uuid4()
+    task_id = uuid4()
+    task_svc = _review_task_svc(task_id, other_pm_id, role="cell_pm")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_will_plan(pm_id, task_id, plan="resume")
+    body = env.as_dict()
+
+    assert body.get("error") == "invalid_state", body
+    task_svc.claim.assert_not_awaited()
+    task_svc.set_plan.assert_not_awaited()
+    task_svc.start.assert_not_awaited()

tests/unit/gateway/test_unblock_flip_breaker.py

@@ -38,6 +38,15 @@ def _make_deps(**overrides: Any) -> ChoreographerDeps:
     base.update(overrides)
     base["journal"].has_decision_for_task.return_value = True
     base["journal"].latest_decision_at.return_value = datetime.now(UTC)
+    # _ensure_pm_decision's journal write is savepoint-guarded — an
+    # unconfigured AsyncMock's begin_nested() call returns a raw unawaited
+    # coroutine, which `async with` cannot use.
+    base["task"].session.begin_nested = MagicMock(
+        return_value=MagicMock(
+            __aenter__=AsyncMock(return_value=None),
+            __aexit__=AsyncMock(return_value=False),
+        )
+    )
     return ChoreographerDeps(**base)
 
 

tests/unit/services/test_git_ownership_scope.py

@@ -194,6 +194,72 @@ async def test_full_scope_op_calls_full_repair_not_git_repair(
     git_repair.assert_not_called()
 
 
+@pytest.mark.asyncio
+async def test_read_only_op_does_not_invalidate_owned_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A read-only op writes nothing, so the ownership-sentinel marker
+    (`_ensure_agent_owned`'s root short-circuit) stays valid — invalidating
+    it here would force a needless full walk on the very next call."""
+    (tmp_path / ".git").mkdir()
+    monkeypatch.setattr(
+        "roboco.services.git.subprocess.run", lambda *_a, **_k: _ok(["status"])
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+
+    await _svc()._run_git(tmp_path, ["status", "--porcelain"])
+
+    invalidate.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_git_scoped_op_invalidates_owned_marker_before_running(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """A `.git`-only-writing op (commit) can still create new root-owned
+    files, so it must invalidate the marker too, not just checkout/reset/
+    etc. — and it must do so BEFORE the subprocess runs, so a marker still
+    trusted by a concurrent `_ensure_agent_owned` call can never straddle
+    the write."""
+    (tmp_path / ".git").mkdir()
+    order: list[str] = []
+
+    def _run_subprocess(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]:
+        order.append("subprocess.run")
+        return _ok(["commit"])
+
+    monkeypatch.setattr("roboco.services.git.subprocess.run", _run_subprocess)
+    monkeypatch.setattr(
+        "roboco.services.workspace.invalidate_owned_marker",
+        lambda _ws: order.append("invalidate_owned_marker"),
+    )
+    monkeypatch.setattr("roboco.services.workspace._ensure_git_dir_owned", MagicMock())
+
+    await _svc()._run_git(tmp_path, ["commit", "-m", "msg"])
+
+    assert order == ["invalidate_owned_marker", "subprocess.run"]
+
+
+@pytest.mark.asyncio
+async def test_full_scope_op_invalidates_owned_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """checkout/reset/rebase/pull can create root-owned working-tree files
+    too — the marker invalidation isn't scoped to `.git`-only writes."""
+    (tmp_path / ".git").mkdir()
+    monkeypatch.setattr(
+        "roboco.services.git.subprocess.run", lambda *_a, **_k: _ok(["checkout"])
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+    monkeypatch.setattr("roboco.services.workspace._ensure_agent_owned", MagicMock())
+
+    await _svc()._run_git(tmp_path, ["checkout", "some-branch"])
+
+    invalidate.assert_called_once_with(tmp_path)
+
+
 @pytest.mark.asyncio
 async def test_reown_after_git_op_returns_zero_ms_when_skipped() -> None:
     """The instrumentation must see a true near-zero cost for a skipped repair,

tests/unit/services/test_pm_claim_needs_revision.py

@@ -22,7 +22,11 @@ from typing import TYPE_CHECKING, cast
 import pytest
 from roboco.foundation.policy import lifecycle as spec
 from roboco.models.base import TaskStatus
-from roboco.services.task import _default_claim_statuses, _get_valid_claim_statuses
+from roboco.services.task import (
+    _ROLE_CLAIM_STATUSES,
+    _default_claim_statuses,
+    _get_valid_claim_statuses,
+)
 
 if TYPE_CHECKING:
     from roboco.db.tables import AgentTable
@@ -52,3 +56,26 @@ def test_runtime_pm_claim_mapping_covers_spec_claim_rules(role: spec.Role) -> No
             f"runtime claim mapping for {role.value} is missing spec-allowed "
             f"status '{status.value}'"
         )
+
+
+@pytest.mark.parametrize("role", [spec.Role.CELL_PM, spec.Role.MAIN_PM])
+def test_claim_rules_and_role_statuses_are_identical(role: spec.Role) -> None:
+    """Genuine bidirectional cross-check between the two claim tables.
+
+    ``test_runtime_pm_claim_mapping_covers_spec_claim_rules`` above only
+    checks spec ⊆ runtime — it would still pass if ``_ROLE_CLAIM_STATUSES``
+    carried an EXTRA status the spec doesn't grant (e.g. AWAITING_PM_REVIEW
+    re-added to the runtime table alone, with lifecycle.CLAIM_RULES left
+    untouched). That silent one-sided drift — "the service table already
+    granted this on the belief that the spec granted it too" — is exactly the
+    shape that caused the awaiting_pm_review re-claim loop
+    (``lifecycle.py``'s ``CLAIM_RULES`` comment covers the incident). This
+    test imports both tables directly and asserts the per-role sets are
+    IDENTICAL, not just one-way-covering.
+    """
+    spec_values = {s.value for s in spec.CLAIM_RULES[role]}
+    runtime_values = {s.value for s in _ROLE_CLAIM_STATUSES[role.value]}
+    assert spec_values == runtime_values, (
+        f"spec.CLAIM_RULES[{role.value}]={spec_values} != "
+        f"task._ROLE_CLAIM_STATUSES[{role.value!r}]={runtime_values}"
+    )

tests/unit/services/test_workspace_ensure_agent_owned_scope.py

@@ -10,14 +10,19 @@ approach) left the working tree root-owned and broke every agent file write.
 
 from __future__ import annotations
 
-from typing import TYPE_CHECKING
+import os
+import stat as stat_module
+from pathlib import Path
+from unittest.mock import MagicMock
 
 import pytest
 from roboco.services import workspace as workspace_module
-from roboco.services.workspace import _ensure_agent_owned
+from roboco.services.workspace import (
-
+    _AGENT_GID,
-if TYPE_CHECKING:
+    _AGENT_UID,
-    from pathlib import Path
+    _ensure_agent_owned,
+    _own_and_grant_rw,
+)
 
 
 def _build_workspace(root: Path) -> None:
@@ -48,11 +53,11 @@ def _record_touched(monkeypatch: pytest.MonkeyPatch) -> list[str]:
     """Record every path _ensure_agent_owned tries to chown/chmod."""
     touched: list[str] = []
 
-    def fake_chown_entry(entry: str) -> bool:
+    def fake_chown_entry(entry: str, _st: os.stat_result) -> bool:
         touched.append(entry)
         return True
 
-    def fake_make_rw(entry: str) -> None:
+    def fake_make_rw(entry: str, _st: os.stat_result) -> None:
         touched.append(entry)
 
     monkeypatch.setattr(workspace_module, "_chown_entry", fake_chown_entry)
@@ -105,9 +110,11 @@ def test_chown_failure_falls_back_to_chmod_and_warns(
     (tmp_path / "file.py").write_text("x = 1\n")
 
     chmod_calls: list[str] = []
-    monkeypatch.setattr(workspace_module, "_chown_entry", lambda _entry: False)
+    monkeypatch.setattr(workspace_module, "_chown_entry", lambda _entry, _st: False)
     monkeypatch.setattr(
-        workspace_module, "_make_owner_and_group_rw", chmod_calls.append
+        workspace_module,
+        "_make_owner_and_group_rw",
+        lambda entry, _st: chmod_calls.append(entry),
     )
     warning_calls: list[tuple[str, dict[str, object]]] = []
     monkeypatch.setattr(
@@ -123,3 +130,229 @@ def test_chown_failure_falls_back_to_chmod_and_warns(
     # The failure is surfaced, not swallowed.
     assert warning_calls
     assert warning_calls[0][1]["failures"]
+
+
+# ---------------------------------------------------------------------------
+# `_own_and_grant_rw`: one shared stat now backs both the chown-needed and
+# chmod-needed checks (previously two separate `Path.stat()` calls, one per
+# helper) — the fix for chown_ms: 39502 on the production NAS, where every
+# chown/chmod is a copy-on-write metadata write and the tree is almost always
+# ALREADY correctly owned on a re-claim. These exercise the real (unmocked)
+# `_own_and_grant_rw` / `_chown_entry` / `_make_owner_and_group_rw` at the
+# os-syscall boundary.
+# ---------------------------------------------------------------------------
+
+
+def _stat_result(mode: int, uid: int = 0, gid: int = 0) -> os.stat_result:
+    """A real ``os.stat_result`` exposing only the fields the ownership
+    helpers read (st_mode/st_uid/st_gid) — no filesystem entry needed."""
+    return os.stat_result((mode, 0, 0, 0, uid, gid, 0, 0, 0, 0))
+
+
+def _fake_stat(result: os.stat_result) -> object:
+    """A stand-in for ``os.stat`` accepting the ``(path, *, follow_symlinks)``
+    signature ``Path(entry).stat()`` actually calls it with underneath."""
+
+    def _stat(_path: object, **_kwargs: object) -> os.stat_result:
+        return result
+
+    return _stat
+
+
+_ALREADY_RW_MODE = (
+    stat_module.S_IFREG
+    | stat_module.S_IRUSR
+    | stat_module.S_IWUSR
+    | stat_module.S_IRGRP
+    | stat_module.S_IWGRP
+)
+_ROOT_NARROW_MODE = stat_module.S_IFREG | stat_module.S_IRUSR | stat_module.S_IWUSR
+
+
+def test_own_and_grant_rw_skips_syscalls_when_already_correct(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """An already agent-owned, already rw entry costs one stat and ZERO
+    metadata-write syscalls — the no-op-write cost that stacked to
+    chown_ms: 39502 across tens of thousands of files when the tree was
+    already correctly owned, as it almost always is on a re-claim."""
+    chown_mock = MagicMock()
+    chmod_mock = MagicMock()
+    monkeypatch.setattr(
+        workspace_module.os,
+        "stat",
+        _fake_stat(_stat_result(_ALREADY_RW_MODE, _AGENT_UID, _AGENT_GID)),
+    )
+    monkeypatch.setattr(workspace_module.os, "chown", chown_mock)
+    monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock)
+
+    failed = _own_and_grant_rw("/fake/already-owned")
+
+    assert failed == 0
+    chown_mock.assert_not_called()
+    chmod_mock.assert_not_called()
+
+
+def test_own_and_grant_rw_still_repairs_a_wrong_owned_entry(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A genuinely wrong-owned entry (fresh root-side clone/fetch/checkout
+    output — root uid/gid, no group-write bit) still gets chowned AND
+    chmodded exactly as before the single-stat merge."""
+    chown_mock = MagicMock()
+    chmod_mock = MagicMock()
+    monkeypatch.setattr(
+        workspace_module.os, "stat", _fake_stat(_stat_result(_ROOT_NARROW_MODE))
+    )
+    monkeypatch.setattr(workspace_module.os, "chown", chown_mock)
+    monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock)
+
+    failed = _own_and_grant_rw("/fake/root-owned")
+
+    assert failed == 0
+    chown_mock.assert_called_once_with("/fake/root-owned", _AGENT_UID, _AGENT_GID)
+    expected_mode = _ROOT_NARROW_MODE | stat_module.S_IRGRP | stat_module.S_IWGRP
+    # chmod runs via Path(entry).chmod(...), which calls os.chmod with a
+    # Path-wrapped first arg + follow_symlinks=True — not the bare string.
+    chmod_mock.assert_called_once_with(
+        Path("/fake/root-owned"), expected_mode, follow_symlinks=True
+    )
+
+
+def test_own_and_grant_rw_chown_failure_still_counted_and_chmod_still_attempted(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Unchanged failure-counting contract: a rejected chown (rootless /
+    userns host) still counts as one failure AND the chmod best-effort
+    fallback still runs (belt + suspenders for ACL-inheriting NAS volumes)."""
+    monkeypatch.setattr(
+        workspace_module.os, "stat", _fake_stat(_stat_result(_ROOT_NARROW_MODE))
+    )
+
+    def _raise_chown(*_args: object, **_kwargs: object) -> None:
+        raise OSError("Operation not permitted")
+
+    chmod_mock = MagicMock()
+    monkeypatch.setattr(workspace_module.os, "chown", _raise_chown)
+    monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock)
+
+    failed = _own_and_grant_rw("/fake/rootless-host")
+
+    assert failed == 1
+    chmod_mock.assert_called_once()
+
+
+def test_own_and_grant_rw_broken_symlink_counts_as_failure_without_crashing(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """Symlink decision: the shared stat call follows symlinks — matching
+    os.chown/os.chmod's own default follow behavior, unchanged from before
+    the merge (the old code's two separate `Path(entry).stat()` calls also
+    followed). A dangling symlink's stat raises OSError exactly as it did
+    from inside the old `_chown_entry`'s own stat call: counted as one
+    chown failure, and chmod is never attempted — the old chmod path hit
+    the identical OSError from its own separate stat call and silently
+    swallowed it, so the net effect (one counted failure, no chmod) is
+    unchanged."""
+    broken_link = tmp_path / "dangling"
+    broken_link.symlink_to(tmp_path / "does-not-exist")
+    chown_mock = MagicMock()
+    chmod_mock = MagicMock()
+    monkeypatch.setattr(workspace_module.os, "chown", chown_mock)
+    monkeypatch.setattr(workspace_module.os, "chmod", chmod_mock)
+
+    failed = _own_and_grant_rw(str(broken_link))
+
+    assert failed == 1
+    chown_mock.assert_not_called()
+    chmod_mock.assert_not_called()
+
+
+# ---------------------------------------------------------------------------
+# Root-sentinel short-circuit: `_ensure_agent_owned` skips the ENTIRE walk
+# when the workspace root is already agent-owned with the right bits AND a
+# `.git/roboco-owned` marker from the last zero-failure pass exists. The
+# marker is invalidated by `GitService._run_git`
+# (tests/unit/services/test_git_ownership_scope.py) before any root-side git
+# write, so a live marker is trustworthy — this only removes the remaining
+# per-entry-stat cost the earlier collapse (above) couldn't, the walk itself.
+# ---------------------------------------------------------------------------
+
+_OWNED_DIR_MODE = (
+    stat_module.S_IFDIR
+    | stat_module.S_IRUSR
+    | stat_module.S_IWUSR
+    | stat_module.S_IXUSR
+    | stat_module.S_IRGRP
+    | stat_module.S_IWGRP
+    | stat_module.S_IXGRP
+)
+
+
+def _fake_root_owned_stat(root: Path) -> object:
+    """Real ``os.stat`` for every path except ``root``, which reports as
+    agent-owned with the required rw+x bits. Lets the marker file's own
+    existence check (``Path.is_file()``, which also routes through
+    ``os.stat``) reflect the real filesystem instead of a blanket fake."""
+    real_stat = os.stat
+
+    def _stat(path: Path, *, follow_symlinks: bool = True) -> os.stat_result:
+        if path == root:
+            return _stat_result(_OWNED_DIR_MODE, _AGENT_UID, _AGENT_GID)
+        return real_stat(path, follow_symlinks=follow_symlinks)
+
+    return _stat
+
+
+def test_skips_walk_when_root_owned_and_marker_present(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _record_touched: list[str]
+) -> None:
+    _build_workspace(tmp_path)
+    (tmp_path / ".git" / "roboco-owned").touch()
+    monkeypatch.setattr(workspace_module.os, "stat", _fake_root_owned_stat(tmp_path))
+    walk_mock = MagicMock(return_value=iter(()))
+    monkeypatch.setattr(workspace_module.os, "walk", walk_mock)
+
+    _ensure_agent_owned(tmp_path)
+
+    walk_mock.assert_not_called()
+    assert _record_touched == []
+
+
+def test_full_walk_when_marker_absent(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch, _record_touched: list[str]
+) -> None:
+    """Root already agent-owned, but no marker: the last pass over this
+    clone is unattested, so the real walk still runs."""
+    _build_workspace(tmp_path)
+    monkeypatch.setattr(workspace_module.os, "stat", _fake_root_owned_stat(tmp_path))
+
+    _ensure_agent_owned(tmp_path)
+
+    assert str(tmp_path) in _record_touched
+    assert str(tmp_path / "README.md") in _record_touched
+
+
+def test_marker_written_only_on_zero_failure_pass(tmp_path: Path) -> None:
+    _build_workspace(tmp_path)
+    marker = tmp_path / ".git" / "roboco-owned"
+    assert not marker.exists()
+
+    # A real pass: chown to uid 1000 fails under the test's real (non-root)
+    # uid, exactly like a rootless/userns host — so no marker should land.
+    _ensure_agent_owned(tmp_path)
+    assert not marker.exists()
+
+
+def test_marker_written_after_successful_pass(
+    tmp_path: Path, _record_touched: list[str]
+) -> None:
+    """`_record_touched`'s fakes report every chown/chmod as succeeding, so
+    this exercises the zero-failure branch without needing real root."""
+    _build_workspace(tmp_path)
+    marker = tmp_path / ".git" / "roboco-owned"
+
+    _ensure_agent_owned(tmp_path)
+
+    assert marker.is_file()
+    assert _record_touched  # the walk actually ran (no marker existed yet)

tests/unit/services/test_workspace_owned_marker_bypass.py

@@ -0,0 +1,212 @@
+"""The ownership-sentinel marker (`_ensure_agent_owned`'s root short-circuit,
+see test_workspace_ensure_agent_owned_scope.py) must be invalidated by EVERY
+root-side git-write path, not just `GitService._run_git`
+(test_git_ownership_scope.py covers that one).
+
+Adversarial review found a deterministic hole: `WorkspaceService`'s raw-
+subprocess git helpers — `_worktree_git`, `_fetch_branch_ref`,
+`_fetch_origin_best_effort` — never invalidated the marker, and the most
+common spawn path hits them on (nearly) every respawn
+(`ensure_worktree_self_heal` -> `_refresh_present_worktree` ->
+`_fetch_branch_ref` + `_worktree_git(["reset", "--hard", ...])`). A stale
+marker then let `_ensure_agent_owned` skip the walk that would have repaired
+the root-owned files those calls had just created — a live Permission
+denied for the agent. These tests cover the bypass paths directly.
+"""
+
+from __future__ import annotations
+
+import subprocess
+from typing import TYPE_CHECKING
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+from roboco.services.workspace import WorkspaceService, _ensure_agent_owned
+from tests.unit.services.test_workspace_ensure_agent_owned_scope import (
+    _build_workspace,
+)
+
+if TYPE_CHECKING:
+    from pathlib import Path
+
+
+def _svc() -> WorkspaceService:
+    return WorkspaceService(MagicMock())
+
+
+def _ok(args: list[str]) -> subprocess.CompletedProcess[str]:
+    return subprocess.CompletedProcess(
+        args=["git", *args], returncode=0, stdout="", stderr=""
+    )
+
+
+# ---------------------------------------------------------------------------
+# `_worktree_git`: mutating verbs invalidate, read-only verbs don't.
+# ---------------------------------------------------------------------------
+
+
+def test_worktree_git_reset_hard_invalidates_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(
+        "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["reset"])
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+
+    WorkspaceService._worktree_git(tmp_path, ["reset", "--hard", "origin/x"])
+
+    invalidate.assert_called_once_with(tmp_path)
+
+
+def test_worktree_git_rev_parse_does_not_invalidate_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(
+        "roboco.services.workspace.subprocess.run",
+        lambda *_a, **_k: _ok(["rev-parse"]),
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+
+    WorkspaceService._worktree_git(
+        tmp_path, ["rev-parse", "--verify", "--quiet", "refs/heads/x"], check=False
+    )
+
+    invalidate.assert_not_called()
+
+
+def test_worktree_git_branch_show_current_does_not_invalidate_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The ambiguous verb's query form: `branch --show-current` only reads."""
+    monkeypatch.setattr(
+        "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["branch"])
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+
+    WorkspaceService._worktree_git(tmp_path, ["branch", "--show-current"], check=False)
+
+    invalidate.assert_not_called()
+
+
+def test_worktree_git_branch_delete_invalidates_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    """The ambiguous verb's write forms: `branch -d/-D <name>` writes."""
+    monkeypatch.setattr(
+        "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["branch"])
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+
+    WorkspaceService._worktree_git(
+        tmp_path, ["branch", "-D", "task-branch"], check=False
+    )
+
+    invalidate.assert_called_once_with(tmp_path)
+
+
+# ---------------------------------------------------------------------------
+# `_fetch_branch_ref` — always mutating (fetch writes .git/objects + refs).
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_fetch_branch_ref_invalidates_marker_before_subprocess(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    order: list[str] = []
+
+    def _run_subprocess(*_a: object, **_k: object) -> subprocess.CompletedProcess[str]:
+        order.append("subprocess.run")
+        return _ok(["fetch"])
+
+    monkeypatch.setattr("roboco.services.workspace.subprocess.run", _run_subprocess)
+    monkeypatch.setattr(
+        "roboco.services.workspace.invalidate_owned_marker",
+        lambda _ws: order.append("invalidate_owned_marker"),
+    )
+    mock_project_service = MagicMock()
+    mock_project_service.get_by_slug = AsyncMock(return_value=None)
+
+    with patch(
+        "roboco.services.project.get_project_service",
+        return_value=mock_project_service,
+    ):
+        await _svc()._fetch_branch_ref(tmp_path, "task-branch", "roboco-api")
+
+    assert order == ["invalidate_owned_marker", "subprocess.run"]
+
+
+# ---------------------------------------------------------------------------
+# `_fetch_origin_best_effort` — same shape, scoped multi-ref fetch.
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.asyncio
+async def test_fetch_origin_best_effort_invalidates_marker(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    monkeypatch.setattr(
+        "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["fetch"])
+    )
+    invalidate = MagicMock()
+    monkeypatch.setattr("roboco.services.workspace.invalidate_owned_marker", invalidate)
+
+    await WorkspaceService._fetch_origin_best_effort(tmp_path, "roboco-api")
+
+    invalidate.assert_called_once_with(tmp_path)
+
+
+# ---------------------------------------------------------------------------
+# End-to-end-shaped repro: a full zero-failure pass writes the marker; a
+# bypass-path root write (through the now-fixed helper) invalidates it; the
+# NEXT _ensure_agent_owned call walks again instead of trusting stale state.
+# ---------------------------------------------------------------------------
+
+
+def test_bypass_path_write_forces_next_ensure_agent_owned_to_walk(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    _build_workspace(tmp_path)
+    touched_pass_1: list[str] = []
+    touched_pass_2: list[str] = []
+
+    def make_fakes(sink: list[str]) -> tuple[object, object]:
+        def fake_chown_entry(entry: str, _st: object) -> bool:
+            sink.append(entry)
+            return True
+
+        def fake_make_rw(entry: str, _st: object) -> None:
+            sink.append(entry)
+
+        return fake_chown_entry, fake_make_rw
+
+    # Pass 1: a normal zero-failure ensure_agent_owned — writes the marker.
+    fake_chown, fake_rw = make_fakes(touched_pass_1)
+    monkeypatch.setattr("roboco.services.workspace._chown_entry", fake_chown)
+    monkeypatch.setattr("roboco.services.workspace._make_owner_and_group_rw", fake_rw)
+    _ensure_agent_owned(tmp_path)
+    marker = tmp_path / ".git" / "roboco-owned"
+    assert marker.is_file()
+    assert touched_pass_1  # the walk actually ran
+
+    # Bypass-path root write: a mutating _worktree_git call (the exact class
+    # of call ensure_worktree_self_heal's self-heal makes on nearly every
+    # respawn) — with subprocess mocked so no real git repo is needed, but
+    # invalidate_owned_marker running for real.
+    monkeypatch.setattr(
+        "roboco.services.workspace.subprocess.run", lambda *_a, **_k: _ok(["reset"])
+    )
+    WorkspaceService._worktree_git(tmp_path, ["reset", "--hard", "origin/x"])
+    assert not marker.exists()  # the fix: the bypass path invalidated it
+
+    # Pass 2: _ensure_agent_owned must walk again (marker gone), not trust
+    # the stale "fully owned" state from before the bypass-path write.
+    fake_chown_2, fake_rw_2 = make_fakes(touched_pass_2)
+    monkeypatch.setattr("roboco.services.workspace._chown_entry", fake_chown_2)
+    monkeypatch.setattr("roboco.services.workspace._make_owner_and_group_rw", fake_rw_2)
+    _ensure_agent_owned(tmp_path)
+    assert touched_pass_2  # the walk ran again — nothing was silently skipped