fix(gateway): make i_will_work_on plan-precondition atomic; recover claimed-no-plan

Bug A from the 2026-05-09 smoke run. be-dev-1 called i_will_work_on
without `plan` on a pending task; claim() ran first (transitioned to
`claimed`), then the plan check failed → tracing_gap. The natural
retry path then dispatched to `_i_will_work_on_claimed`, which had no
plan-recovery logic and called start() against a still-plan-less task,
returning `start failed` forever. The dev kept looping; the parent
escalated up; the whole slice ended `blocked`.

Two changes (Task-5 atomicity pattern applied to i_will_work_on):

1. `_i_will_work_on_pending`: move the plan precondition BEFORE
   `claim()`. A missing-plan first call now returns tracing_gap with
   the task untouched in `pending`, so the agent's retry-with-plan
   succeeds cleanly.

2. `_i_will_work_on_claimed`: now accepts `plan` and calls set_plan
   before start() if the task has no plan yet. Recovery path for any
   already-stuck task (e.g. left over from the earlier image, or an
   orchestrator restart that left a partial claim).

Also wires `plan` through the dispatcher to the claimed branch.

Tests: 3137 passing (3135 + 2 regression tests pinning the atomic
invariant), 100% coverage, ruff clean.

roboco/services/gateway/choreographer/_impl.py

@@ -366,13 +366,28 @@ class Choreographer:
         plan: str | None,
         briefing: dict[str, Any],
     ) -> tuple[Envelope | None, Any]:
-        """Pending-branch dispatch for i_will_work_on. Extracted to keep
+        """Pending-branch dispatch for i_will_work_on. Atomic: validates
-        the parent's return count under PLR0911. Returns (rejection|None,
+        the plan precondition BEFORE calling claim() so a missing-plan
-        task). Caller emits rejection via _emit_rejection and falls
+        rejection doesn't leave the task in `claimed` with no plan.
-        through to the OK envelope when rejection is None.
+
+        Pre-fix (2026-05-09 smoke Bug A): claim() ran first, then the
+        plan check failed → task was stuck in `claimed` because the
+        `_i_will_work_on_claimed` branch (the natural retry path) had
+        no plan-recovery logic. Now ordering is: guards → plan check →
+        claim → set_plan → start.
         """
         if guard := await self._run_claim_guards(agent_id=agent_id, task=t):
             return self._with_briefing(guard, briefing), t
+        # Plan precondition BEFORE any state mutation. Atomic invariant.
+        if not t.plan and not plan:
+            return Envelope.tracing_gap(
+                missing=["plan"],
+                remediate=(
+                    f"call i_will_work_on(task_id='{task_id}',"
+                    f" plan='<one-paragraph plan describing what you will do>')"
+                ),
+                context_briefing=briefing,
+            ), t
         # claim() transitions pending → claimed; idempotent for same assignee.
         # Branch creation runs inside _finalize_claim and rolls back on
         # failure (audit P0-7 / S-01); we surface the failure as an envelope
@@ -395,15 +410,6 @@ class Choreographer:
                 remediate="task may already be claimed by another agent",
                 context_briefing=briefing,
             ), t
-        if not t.plan and not plan:
-            return Envelope.tracing_gap(
-                missing=["plan"],
-                remediate=(
-                    f"call i_will_work_on(task_id='{task_id}',"
-                    f" plan='<one-paragraph plan describing what you will do>')"
-                ),
-                context_briefing=briefing,
-            ), t
         if plan:
             t = await self.task.set_plan(task_id, plan)
         t = await self.task.start(task_id, agent_id)
@@ -454,14 +460,34 @@ class Choreographer:
         agent_id: UUID,
         task_id: UUID,
         t: Any,
+        plan: str | None,
         briefing: dict[str, Any],
     ) -> tuple[Envelope | None, Any]:
-        """claimed branch for i_will_work_on. Returns (rejection|None, task)."""
+        """claimed branch for i_will_work_on. Returns (rejection|None, task).
+
+        Recovery path (Bug A from 2026-05-09 smoke): if the task is in
+        `claimed` without a plan (e.g. orchestrator restart, prior
+        partial-claim race) and the caller now supplies one, set it
+        before start() instead of failing with "no plan recorded". If
+        the task still has no plan and none is supplied, surface the
+        same tracing_gap shape `_i_will_work_on_pending` uses.
+        """
         guard = await self._run_claim_guards(
             agent_id=agent_id, task=t, skip_sequence=True
         )
         if guard:
             return self._with_briefing(guard, briefing), None
+        if not t.plan and not plan:
+            return Envelope.tracing_gap(
+                missing=["plan"],
+                remediate=(
+                    f"call i_will_work_on(task_id='{task_id}',"
+                    f" plan='<one-paragraph plan describing what you will do>')"
+                ),
+                context_briefing=briefing,
+            ), None
+        if plan and not t.plan:
+            t = await self.task.set_plan(task_id, plan)
         t = await self.task.start(task_id, agent_id)
         if t is None:
             return self._start_failed_envelope(task_id, briefing), None
@@ -495,7 +521,7 @@ class Choreographer:
             )
         elif status == "claimed" and t.assigned_to == agent_id:
             rejection, t = await self._i_will_work_on_claimed(
-                agent_id, task_id, t, briefing
+                agent_id, task_id, t, plan, briefing
             )
         elif status == "in_progress" and t.assigned_to == agent_id:
             # Idempotent re-entry: respawned dev re-calling i_will_work_on

tests/unit/gateway/test_choreographer_impl_branches.py

@@ -1084,3 +1084,57 @@ async def test_open_pr_does_not_create_pr_if_no_commits() -> None:
     assert "no commits" in body["message"]
     git_svc.create_pr.assert_not_called()
     git_svc.push_branch.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_missing_plan_does_not_claim_pending_task() -> None:
+    """Atomic invariant (Task 5 pattern, Bug A from 2026-05-09 smoke):
+    if `plan` is missing on the FIRST i_will_work_on call against a
+    pending task, the task must NOT be claimed. Pre-fix the verb ran
+    claim() BEFORE checking plan, leaving the task in `claimed` with
+    no plan — and `_i_will_work_on_claimed` had no recovery path so
+    the agent looped forever on `start failed`.
+    """
+    agent_id = uuid4()
+    task_id = uuid4()
+    task_svc = _wire_dev_task_svc(task_id, status="pending")
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+    env = await c.i_will_work_on(agent_id, task_id, plan=None)
+    body = env.as_dict()
+    assert body["error"] == "tracing_gap"
+    assert "plan" in body["missing"]
+    task_svc.claim.assert_not_called(), (
+        "claim() ran before plan precondition was satisfied — atomicity broken"
+    )
+
+
+@pytest.mark.asyncio
+async def test_i_will_work_on_claimed_with_no_plan_accepts_recovery_plan() -> None:
+    """Recovery path (Bug A from 2026-05-09 smoke): if the task is in
+    `claimed` state without a plan (e.g. from a prior partial-claim race
+    or an orchestrator restart), a fresh i_will_work_on call WITH plan
+    must set the plan and then start, not just call start() against a
+    plan-less task.
+    """
+    agent_id = uuid4()
+    task_id = uuid4()
+    task_svc = _wire_dev_task_svc(
+        task_id, status="claimed", assigned_to=agent_id, plan=None
+    )
+    started = MagicMock(
+        status="in_progress",
+        assigned_to=agent_id,
+        plan="recovery plan",
+        id=task_id,
+        title="t",
+        task_type="code",
+    )
+    task_svc.set_plan.return_value = started
+    task_svc.start.return_value = started
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+    env = await c.i_will_work_on(agent_id, task_id, plan="recovery plan")
+    body = env.as_dict()
+    assert body["error"] is None, f"expected success, got {body}"
+    task_svc.set_plan.assert_awaited_once()