fix(gateway): let a dev idle past lane-held code-queue siblings

Per-dev sequenced queues (the prior commit) have a PM delegate a dev's whole
code queue up front, so a dev owns several pending, assigned-but-unclaimed code
leaves at once (seq0 + seq2). The orchestrator's lane barrier holds the seq2
SPAWN while seq0 is non-terminal — but _pending_assignment_guard rejected
i_am_idle for ANY pending assigned task, with no lane awareness. So a dev whose
current leaf just moved to QA (awaiting_qa) could neither idle (guard rejects)
nor proceed cleanly: it was steered to claim seq2 early (the claim path has no
lane/sequence check, since delegate sets `sequence` not `dependency_ids`),
jumping its own queue order, or it looped on the rejection. An adversarial
review of the queue work surfaced this; it is latent until PMs actually
delegate multi-item per-dev queues, so the green suite hid it.

Fix: TaskService.has_earlier_incomplete_code_sibling mirrors the orchestrator's
lane barrier in the service layer; _pending_assignment_guard now drops a dev's
lane-held pending code leaves (via _pending_blocking_idle / _pending_not_lane_held)
so the dev idles cleanly and the orchestrator spawns the next queue item when
the lane clears — preserving one-leaf-at-a-time, in order. `is not True` keeps
it inert under partial test mocks. Tests cover the service primitive (live /
terminal / higher-seq / non-code / missing-field) and the guard (dev idles when
lane-held; still blocks a non-lane-held pending leaf). Full mypy + xenon green.

tests/unit/gateway/test_choreographer_idle_guards.py

@@ -263,3 +263,43 @@ async def test_i_am_idle_decomposition_guard_is_pm_only() -> None:
     env = await c.i_am_idle(agent_id)
     assert env.status == "idle"
     task_svc.unclaimed_parent_acceptance_criteria.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_i_am_idle_lets_dev_idle_when_pending_is_lane_held() -> None:
+    """A dev's pending code leaf waiting behind its own earlier queue item must
+    NOT pin it to idle — the orchestrator spawns it when the lane clears."""
+    agent_id = uuid4()
+    pending = MagicMock(id=uuid4(), status="pending")
+    task_svc = AsyncMock()
+    task_svc.list_assigned_for_agent.return_value = [pending]
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.agent_for.return_value = MagicMock(role="developer")
+    task_svc.has_earlier_incomplete_code_sibling.return_value = True
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_am_idle(agent_id)
+    assert env.status == "idle"
+    task_svc.mark_agent_idle.assert_awaited_once()
+
+
+@pytest.mark.asyncio
+async def test_i_am_idle_still_blocks_dev_with_non_lane_held_pending() -> None:
+    """A pending leaf that is NOT lane-held (no earlier live sibling) still
+    blocks idle — the dev must claim or unclaim it."""
+    agent_id = uuid4()
+    pending = MagicMock(id=uuid4(), status="pending")
+    task_svc = AsyncMock()
+    task_svc.list_assigned_for_agent.return_value = [pending]
+    task_svc.list_in_progress_for_agent.return_value = []
+    task_svc.agent_for.return_value = MagicMock(role="developer")
+    task_svc.has_earlier_incomplete_code_sibling.return_value = False
+    deps = _make_deps(task=task_svc)
+    c = Choreographer(deps)
+
+    env = await c.i_am_idle(agent_id)
+    body = env.as_dict()
+    assert body["error"] == "invalid_state"
+    assert "i_will_work_on" in body["remediate"]
+    task_svc.mark_agent_idle.assert_not_awaited()

tests/unit/services/test_task.py

@@ -746,6 +746,81 @@ async def test_unclaimed_parent_acs_counts_live_children_not_just_completed() ->
     ]
 
 
+def _svc_with_sibling_status_seq(rows: list[tuple]) -> TaskService:
+    """TaskService whose execute() yields (status, sequence) sibling rows."""
+    res = MagicMock()
+    res.all.return_value = rows
+    return TaskService(MagicMock(execute=AsyncMock(return_value=res)))
+
+
+@pytest.mark.asyncio
+async def test_earlier_incomplete_code_sibling_true_for_live_lower_seq() -> None:
+    # A dev's queued code leaf (seq 2) is lane-held while its own seq-0 sibling
+    # is still in flight — so it must not pin the dev to idle.
+    task = _build_task(
+        task_type=TaskType.CODE.value,
+        parent_task_id=uuid4(),
+        assigned_to=uuid4(),
+        sequence=2,
+    )
+    svc = _svc_with_sibling_status_seq([(TaskStatus.IN_PROGRESS, 0)])
+    assert await svc.has_earlier_incomplete_code_sibling(task) is True
+
+
+@pytest.mark.asyncio
+async def test_earlier_incomplete_code_sibling_false_when_earlier_terminal() -> None:
+    task = _build_task(
+        task_type=TaskType.CODE.value,
+        parent_task_id=uuid4(),
+        assigned_to=uuid4(),
+        sequence=2,
+    )
+    svc = _svc_with_sibling_status_seq(
+        [(TaskStatus.COMPLETED, 0), (TaskStatus.CANCELLED, 1)]
+    )
+    assert await svc.has_earlier_incomplete_code_sibling(task) is False
+
+
+@pytest.mark.asyncio
+async def test_earlier_incomplete_code_sibling_false_for_higher_seq_only() -> None:
+    # A LATER sibling (seq 3) does not hold an earlier leaf (seq 2).
+    task = _build_task(
+        task_type=TaskType.CODE.value,
+        parent_task_id=uuid4(),
+        assigned_to=uuid4(),
+        sequence=2,
+    )
+    svc = _svc_with_sibling_status_seq([(TaskStatus.IN_PROGRESS, 3)])
+    assert await svc.has_earlier_incomplete_code_sibling(task) is False
+
+
+@pytest.mark.asyncio
+async def test_earlier_incomplete_code_sibling_non_code_short_circuits() -> None:
+    # Only code queues sequence this way; a planning/doc leaf never queries.
+    session = MagicMock(execute=AsyncMock())
+    svc = TaskService(session)
+    task = _build_task(
+        task_type="planning",
+        parent_task_id=uuid4(),
+        assigned_to=uuid4(),
+        sequence=2,
+    )
+    assert await svc.has_earlier_incomplete_code_sibling(task) is False
+    session.execute.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_earlier_incomplete_code_sibling_false_when_fields_missing() -> None:
+    task = _build_task(
+        task_type=TaskType.CODE.value,
+        parent_task_id=None,
+        assigned_to=uuid4(),
+        sequence=2,
+    )
+    svc = _svc_with_sibling_status_seq([(TaskStatus.IN_PROGRESS, 0)])
+    assert await svc.has_earlier_incomplete_code_sibling(task) is False
+
+
 @pytest.mark.asyncio
 async def test_unblock_with_branch_resumes_in_progress() -> None:
     # A task claimed (has a branch) before it blocked resumes in_progress.