fix(orchestrator): don't auto-restart on graceful exit; tighten role-status

Smoke-8 surfaced a tight respawn loop: QA failed a PR cleanly, container
exited 0, then _check_health bumped error_count and respawned QA with
the same task_id. But by then the task was in needs_revision (dev's
state), so QA's claim_review was rejected — and the cycle repeated on
the next health tick. Token-burning loop.

Two layers:

1. _check_health now reads docker's exit code. exit_code == 0 →
   graceful (intentional handoff via i_am_idle / clean shutdown) →
   reset error_count, do NOT auto-restart. Non-zero → keep the
   existing crash-retry behavior. Refactored into
   _inspect_container_state + _handle_stopped_container to keep
   xenon's complexity check happy.

2. _readiness_check_role_for_status now includes the dev-owned
   states (needs_revision, verifying) so a misrouted spawn for QA /
   PM / board on these statuses fails the readiness gate before the
   gateway has to reject it. Defense in depth — the right path is
   #1 (don't respawn on clean exit at all), but if some other code
   path tries to spawn QA on needs_revision the gate now catches it.

Tests: 12 new (5 for _check_health graceful/crash matrix + 7 for the
expanded role-status table). Pre-gateway names (none of which were
needed here) untouched.

tests/unit/runtime/test_check_health_graceful_exit.py

@@ -0,0 +1,153 @@
+"""Smoke-8: _check_health distinguishes graceful (exit 0) from crash exits.
+
+Original bug: every container stop bumped error_count and triggered
+spawn_agent(agent_id, task_id=instance.current_task_id). After QA failed a
+PR and cleanly idled, the health check respawned QA on the (now
+needs_revision) task — the gateway rejected claim_review every time, and
+QA respawned again on the next health tick. Token-burning tight loop.
+
+Fix: read exit code via `docker inspect`. exit_code == 0 → graceful;
+reset error_count and DO NOT auto-restart. Non-zero → crash; keep
+existing retry behavior.
+"""
+
+from __future__ import annotations
+
+from unittest.mock import AsyncMock, MagicMock, patch
+from uuid import uuid4
+
+import pytest
+from roboco.runtime.orchestrator import AgentOrchestrator, AgentState
+
+
+def _make_orchestrator() -> AgentOrchestrator:
+    with patch.object(AgentOrchestrator, "__init__", return_value=None):
+        orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._instances = {}
+    orch._lock = MagicMock()
+    return orch
+
+
+def _instance(task_id: str | None) -> MagicMock:
+    inst = MagicMock()
+    inst.state = AgentState.ACTIVE
+    inst.container_id = "deadbeef1234"
+    inst.current_task_id = task_id
+    inst.error_count = 0
+    inst.config = MagicMock(git_context=None)
+    return inst
+
+
+async def _docker_inspect_returning(*, running: bool, exit_code: int) -> bytes:
+    return f"{'true' if running else 'false'} {exit_code}\n".encode()
+
+
+@pytest.mark.asyncio
+async def test_graceful_exit_does_not_respawn() -> None:
+    """Container exit_code=0 means clean shutdown. No auto-restart."""
+    orch = _make_orchestrator()
+    inst = _instance(task_id=str(uuid4()))
+    orch._instances["be-qa"] = inst
+
+    proc = MagicMock()
+    proc.communicate = AsyncMock(return_value=(b"false 0\n", b""))
+    spawn = AsyncMock()
+    orch.spawn_agent = spawn
+
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+        await orch._check_health()
+
+    spawn.assert_not_awaited()
+    assert inst.state == AgentState.OFFLINE
+    assert inst.error_count == 0, (
+        "Graceful exit must reset error_count, not bump it. Otherwise a "
+        "long-running agent that idles clean every time eventually trips "
+        "max_retries and gets flagged as stranded."
+    )
+
+
+@pytest.mark.asyncio
+async def test_crash_exit_triggers_restart() -> None:
+    """Container exit_code != 0 means crash. Auto-restart (existing behavior)."""
+    orch = _make_orchestrator()
+    task_id = str(uuid4())
+    inst = _instance(task_id=task_id)
+    orch._instances["be-dev-1"] = inst
+
+    proc = MagicMock()
+    proc.communicate = AsyncMock(return_value=(b"false 137\n", b""))
+    spawn = AsyncMock()
+    orch.spawn_agent = spawn
+
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+        await orch._check_health()
+
+    spawn.assert_awaited_once()
+    args = spawn.await_args.kwargs
+    assert args["agent_id"] == "be-dev-1"
+    assert args["task_id"] == task_id
+    assert inst.error_count == 1
+
+
+@pytest.mark.asyncio
+async def test_still_running_no_action() -> None:
+    """If the container is still running, no state change."""
+    orch = _make_orchestrator()
+    inst = _instance(task_id=str(uuid4()))
+    orch._instances["be-dev-1"] = inst
+
+    proc = MagicMock()
+    proc.communicate = AsyncMock(return_value=(b"true 0\n", b""))
+    spawn = AsyncMock()
+    orch.spawn_agent = spawn
+
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+        await orch._check_health()
+
+    spawn.assert_not_awaited()
+    assert inst.state == AgentState.ACTIVE
+    assert inst.error_count == 0
+    assert inst.container_id == "deadbeef1234"
+
+
+@pytest.mark.asyncio
+async def test_crash_max_retries_does_not_restart() -> None:
+    """Hit max_retries → don't restart (existing behavior preserved)."""
+    orch = _make_orchestrator()
+    inst = _instance(task_id=str(uuid4()))
+    starting_error_count = 3
+    inst.error_count = starting_error_count
+    orch._instances["be-dev-1"] = inst
+
+    proc = MagicMock()
+    proc.communicate = AsyncMock(return_value=(b"false 1\n", b""))
+    spawn = AsyncMock()
+    orch.spawn_agent = spawn
+    orch._notify_agent_stranded = AsyncMock()
+
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+        await orch._check_health()
+
+    spawn.assert_not_awaited()
+    assert inst.error_count == starting_error_count + 1
+
+
+@pytest.mark.asyncio
+async def test_malformed_inspect_treated_as_crash() -> None:
+    """If `docker inspect` returns malformed output, default to crash path."""
+    orch = _make_orchestrator()
+    inst = _instance(task_id=str(uuid4()))
+    orch._instances["be-dev-1"] = inst
+
+    proc = MagicMock()
+    # No exit code field at all.
+    proc.communicate = AsyncMock(return_value=(b"false\n", b""))
+    spawn = AsyncMock()
+    orch.spawn_agent = spawn
+
+    with patch("asyncio.create_subprocess_exec", AsyncMock(return_value=proc)):
+        await orch._check_health()
+
+    # exit_code is None → not graceful → counts as crash.
+    spawn.assert_awaited_once()
+    assert inst.error_count == 1

tests/unit/runtime/test_readiness_role_status_match.py

@@ -0,0 +1,75 @@
+"""Smoke-8: _readiness_check_role_for_status covers dev-owned states.
+
+Original gap: the role-mismatch table only mapped handoff states
+(awaiting_qa, awaiting_documentation, awaiting_pm_review,
+awaiting_ceo_approval) to required roles. needs_revision/verifying had
+no entry, so a QA spawn for a needs_revision task passed the readiness
+check and the gateway rejected claim_review afterwards. Defense in depth
+layered behind the _check_health fix.
+"""
+
+from __future__ import annotations
+
+from roboco.runtime.orchestrator import AgentOrchestrator
+
+
+def test_qa_on_needs_revision_blocked() -> None:
+    """QA cannot be spawned for a needs_revision task."""
+    reason = AgentOrchestrator._readiness_check_role_for_status(
+        agent_id="be-qa", role="qa", status="needs_revision"
+    )
+    assert reason is not None
+    assert "needs_revision" in reason
+    assert "qa" in reason
+
+
+def test_pm_on_needs_revision_blocked() -> None:
+    """PM cannot be spawned for a needs_revision task either."""
+    reason = AgentOrchestrator._readiness_check_role_for_status(
+        agent_id="be-pm", role="cell_pm", status="needs_revision"
+    )
+    assert reason is not None
+
+
+def test_developer_on_needs_revision_allowed() -> None:
+    """Developer (and documenter) ARE the right roles for needs_revision."""
+    reason = AgentOrchestrator._readiness_check_role_for_status(
+        agent_id="be-dev-1", role="developer", status="needs_revision"
+    )
+    assert reason is None
+
+
+def test_documenter_on_needs_revision_allowed() -> None:
+    """Documenter can rework — same dev/doc-owned set."""
+    reason = AgentOrchestrator._readiness_check_role_for_status(
+        agent_id="be-doc", role="documenter", status="needs_revision"
+    )
+    assert reason is None
+
+
+def test_qa_on_verifying_blocked() -> None:
+    """Verifying belongs to the dev/doc roles, not QA."""
+    reason = AgentOrchestrator._readiness_check_role_for_status(
+        agent_id="be-qa", role="qa", status="verifying"
+    )
+    assert reason is not None
+
+
+def test_qa_on_awaiting_qa_allowed() -> None:
+    """The original handoff case still works — QA on awaiting_qa is fine."""
+    reason = AgentOrchestrator._readiness_check_role_for_status(
+        agent_id="be-qa", role="qa", status="awaiting_qa"
+    )
+    assert reason is None
+
+
+def test_unmapped_status_allows_any_role() -> None:
+    """Statuses with no role lock (pending, paused, blocked, etc.) pass."""
+    for status in ("pending", "claimed", "in_progress", "paused", "blocked"):
+        for role in ("developer", "qa", "documenter", "cell_pm"):
+            assert (
+                AgentOrchestrator._readiness_check_role_for_status(
+                    agent_id="x", role=role, status=status
+                )
+                is None
+            ), f"role={role} status={status} should not be rejected by this gate"