[F095] orchestrator: parked-provider spawn short-circuits before expensive prepare

spawn_agent ran the full _prepare_agent_spawn (writes blueprint/settings/
briefing/MCP files, ensures the image, registers a STARTING instance) every
dispatcher tick only to bail at the after-prepare parked-provider check —
wasting all that file I/O while the provider stayed parked and leaving a
STARTING instance registered then downgraded to OFFLINE.

Move the parked check before _prepare_agent_spawn: resolve the route cheaply
via _resolve_agent_route (only provider_type is needed) and bail with a
minimal unregistered OFFLINE instance. The existing-running check stays
first (inside the lock) so a live agent is never replaced; a TOCTOU
re-check guards the unlocked window before prepare; the after-prepare
check is kept as a rare-race defense (a park landing during prepare).

roboco/runtime/orchestrator.py

@@ -2040,16 +2040,58 @@ class AgentOrchestrator:
             existing = self._existing_running_instance(agent_id)
             if existing is not None:
                 return existing
+
+        # Provider-parking loop-breaker (cheap pre-check): while this agent's
+        # provider is parked (rate-limited or overloaded), do NOT run the full
+        # ``_prepare_agent_spawn`` — which writes the blueprint / settings /
+        # briefing / MCP-config files, ensures the agent image, and registers a
+        # STARTING instance — only to bail. The dispatcher re-ticks a parked
+        # agent every cycle, so running the full prepare each tick wasted all
+        # that file I/O and left a STARTING instance registered then downgraded
+        # to OFFLINE. The parked check only needs ``provider_type``, cheaply
+        # resolvable via ``_resolve_agent_route``. Bailing here returns a
+        # minimal UNREGISTERED OFFLINE instance (no stale ``_instances`` entry),
+        # so the next tick re-checks cheaply until the provider recovers. The
+        # existing-running check above stays first, so a live agent is never
+        # replaced by this bail. Fail-open: a tracker read error never blocks.
+        route = await self._resolve_agent_route(agent_id)
+        if await self._provider_spawn_parked(route.provider_type.value):
+            self._mark_task_handled(task_id)
+            logger.info(
+                "Spawn skipped: provider rate-limited (parked)",
+                agent_id=agent_id,
+                task_id=task_id,
+                provider=route.provider_type.value,
+            )
+            return AgentInstance(
+                agent_id=agent_id,
+                state=AgentState.OFFLINE,
+                config=AgentConfig(
+                    agent_id=agent_id,
+                    blueprint_path=Path(),  # not launching — no blueprint written
+                    model=route.model_name,
+                    provider_type=route.provider_type.value,
+                    provider_base_url=route.base_url,
+                    provider_auth_token=route.auth_token,
+                    git_context=git_context,
+                ),
+                current_task_id=task_id,
+            )
+
+        async with self._lock:
+            # TOCTOU re-check: another tick may have started this agent during
+            # the unlocked route resolve + parked check above. Re-check before
+            # the expensive prepare so two concurrent ticks don't double-spawn.
+            existing = self._existing_running_instance(agent_id)
+            if existing is not None:
+                return existing
             config, instance, agent_settings_path = await self._prepare_agent_spawn(
                 agent_id, task_id, model, git_context
             )
-        # Provider-parking loop-breaker: while this agent's provider is parked
-        # (rate-limited or overloaded), do NOT launch another container — the
-        # dispatcher would otherwise re-spawn the same task every tick, hit the
-        # limit again, and burn cost. The probe-resume loop clears the park when
-        # the provider recovers and the next tick spawns normally. Covers both
-        # the GROK 429 path and the Claude session/overload paths.
-        # Fail-open: a tracker read error must never block spawning.
+        # Rare-race defense: a park could land during prepare. The every-tick
+        # parked case is already handled above; this guards the window between
+        # the pre-check and the launch. Fail-open: a tracker read error never
+        # blocks spawning.
         if await self._provider_spawn_parked(config.provider_type):
             self._mark_task_handled(task_id)
             instance.state = AgentState.OFFLINE

tests/unit/runtime/test_parked_spawn_shortcut.py

@@ -0,0 +1,163 @@
+"""Parked-provider spawn must short-circuit BEFORE the expensive prepare.
+
+``spawn_agent`` runs the parked-provider check (``_provider_spawn_parked``) to
+avoid re-spawning a container into a rate-limited / overloaded provider every
+dispatcher tick. The check only needs ``provider_type``, which is cheaply
+resolvable via ``_resolve_agent_route``. Running the full
+``_prepare_agent_spawn`` first — which writes the blueprint / settings /
+briefing / MCP-config files, ensures the agent image, and registers a STARTING
+``AgentInstance`` in ``_instances`` — only to bail at the parked check wastes
+all that file I/O every tick the provider stays parked, and leaves a STARTING
+instance registered then downgraded to OFFLINE.
+
+The fix: resolve the route + run the parked check BEFORE ``_prepare_agent_spawn``,
+bailing with a minimal unregistered OFFLINE instance. The existing-running
+check stays first (inside the lock) so a running agent is never bailed.
+"""
+
+from __future__ import annotations
+
+import asyncio
+from types import SimpleNamespace
+from typing import Any
+from unittest.mock import AsyncMock
+
+import pytest
+from roboco.models.runtime import AgentInstance
+from roboco.runtime.orchestrator import AgentOrchestrator, AgentState
+
+
+def _make_orchestrator() -> AgentOrchestrator:
+    # __new__ + skip __init__: avoid all constructor I/O.
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._instances = {}
+    orch._lock = asyncio.Lock()
+    orch._tick_handled_tasks = set()
+    orch._bg_tasks = set()
+    orch._running = True
+    return orch
+
+
+def _wire(monitor: dict[str, Any]) -> Any:
+    """Build the mock wiring closure capturing call counts in ``monitor``."""
+
+    async def _readiness_gate(_aid: str, _tid: str | None) -> None:
+        return None
+
+    async def _git_context(_gc: Any, _tid: str | None) -> None:
+        return None
+
+    async def _route(_aid: str) -> Any:
+        monitor["route_calls"] += 1
+        return SimpleNamespace(
+            provider_type=SimpleNamespace(value="anthropic"),
+            model_name="opus",
+            base_url=None,
+            auth_token=None,
+        )
+
+    async def _prepare(*_a: Any, **_k: Any) -> Any:
+        monitor["prepare_calls"] += 1
+        # Mirrors the real prepare's registration side-effect so the RED test
+        # observes the STARTING instance the current code leaks.
+        cfg = SimpleNamespace(provider_type="anthropic", model="opus")
+        inst = AgentInstance(agent_id="be-dev-1", state=AgentState.STARTING, config=cfg)
+        return cfg, inst, None
+
+    return _readiness_gate, _git_context, _route, _prepare
+
+
+@pytest.mark.asyncio
+async def test_parked_spawn_skips_prepare_and_does_not_register(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """When the provider is parked, ``_prepare_agent_spawn`` (all the file
+    writes + image ensure + STARTING registration) must NOT run, and no
+    instance may be left registered in ``_instances``."""
+    orch = _make_orchestrator()
+    monitor = {"route_calls": 0, "prepare_calls": 0}
+    _rg, _gc, _route, _prepare = _wire(monitor)
+
+    monkeypatch.setattr(orch, "_readiness_gate", _rg)
+    monkeypatch.setattr(orch, "_resolve_spawn_git_context", _gc)
+    monkeypatch.setattr(orch, "_resolve_agent_route", _route)
+    monkeypatch.setattr(orch, "_prepare_agent_spawn", _prepare)
+    monkeypatch.setattr(orch, "_provider_spawn_parked", AsyncMock(return_value=True))
+
+    result = await orch.spawn_agent(agent_id="be-dev-1", task_id="task-9")
+
+    # The parked check ran (route resolved for the cheap provider_type lookup).
+    assert monitor["route_calls"] >= 1
+    # The expensive prepare was NOT called — the whole point of the fix.
+    assert monitor["prepare_calls"] == 0
+    # Bailed with an OFFLINE instance, no container launched.
+    assert isinstance(result, AgentInstance)
+    assert result.state is AgentState.OFFLINE
+    # No STARTING/OFFLINE instance left lingering in _instances.
+    assert "be-dev-1" not in orch._instances
+    # Task marked handled so later dispatchers in this tick skip it.
+    assert "task-9" in orch._tick_handled_tasks
+
+
+@pytest.mark.asyncio
+async def test_not_parked_spawn_still_runs_prepare_and_launches(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """Sanity: when the provider is NOT parked, the normal prepare + launch
+    path runs unchanged — the shortcut only short-circuits the parked case."""
+    orch = _make_orchestrator()
+    monitor = {"route_calls": 0, "prepare_calls": 0}
+    _rg, _gc, _route, _prepare = _wire(monitor)
+
+    launched: list[bool] = []
+
+    async def _launch(*_a: Any, **_k: Any) -> AgentInstance:
+        launched.append(True)
+        return AgentInstance(
+            agent_id="be-dev-1",
+            state=AgentState.ACTIVE,
+            config=SimpleNamespace(provider_type="anthropic", model="opus"),
+        )
+
+    monkeypatch.setattr(orch, "_readiness_gate", _rg)
+    monkeypatch.setattr(orch, "_resolve_spawn_git_context", _gc)
+    monkeypatch.setattr(orch, "_resolve_agent_route", _route)
+    monkeypatch.setattr(orch, "_prepare_agent_spawn", _prepare)
+    monkeypatch.setattr(orch, "_provider_spawn_parked", AsyncMock(return_value=False))
+    monkeypatch.setattr(orch, "_launch_spawn", _launch)
+
+    await orch.spawn_agent(agent_id="be-dev-1", task_id="task-9")
+
+    assert monitor["prepare_calls"] == 1
+    assert launched == [True]
+
+
+@pytest.mark.asyncio
+async def test_running_agent_not_bailed_by_parked_check(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    """A running (ACTIVE) agent whose provider gets parked mid-flight must be
+    returned as-is — the parked shortcut must NOT replace a live instance with
+    a fresh OFFLINE one. The existing-running check stays first."""
+    orch = _make_orchestrator()
+    monitor = {"route_calls": 0, "prepare_calls": 0}
+    _rg, _gc, _route, _prepare = _wire(monitor)
+
+    existing = AgentInstance(
+        agent_id="be-dev-1",
+        state=AgentState.ACTIVE,
+        config=SimpleNamespace(provider_type="anthropic", model="opus"),
+    )
+    orch._instances["be-dev-1"] = existing
+
+    monkeypatch.setattr(orch, "_readiness_gate", _rg)
+    monkeypatch.setattr(orch, "_resolve_spawn_git_context", _gc)
+    monkeypatch.setattr(orch, "_resolve_agent_route", _route)
+    monkeypatch.setattr(orch, "_prepare_agent_spawn", _prepare)
+    monkeypatch.setattr(orch, "_provider_spawn_parked", AsyncMock(return_value=True))
+
+    result = await orch.spawn_agent(agent_id="be-dev-1", task_id="task-9")
+
+    # The running instance is returned untouched — parked check never reached.
+    assert result is existing
+    assert monitor["prepare_calls"] == 0