[F041] park grok exit-78 (auth missing/expired) instead of crash-retrying

A one-shot grok container whose entrypoint ran grok_auth --check and found
the token missing/expired exits 78 (EX_CONFIG). Crash-retrying 3x burns
tokens for zero progress — the agent cannot start without a valid token.
Park the provider with kind=auth_missing (same shape as the 429 exit-75
path) so the probe-resume loop revives the task once grok_auth.refresh_if_stale
mints a fresh token; if still expired, the next exit 78 re-parks (no burn).

Also fixes a latent F035 regression: _park_provider_unavailable now registers
a WaitingRecord, so the bare-__new__ rate-limit park test had to set
_waiting_records + stub _persist_waiting_record (mirrors the overload-test
fixture).

roboco/runtime/orchestrator.py

@@ -126,6 +126,8 @@ def _system_api_headers() -> dict[str, str]:
             _SYSTEM_API_HEADERS["X-Agent-ID"], "system", ""
         ),
     }
+
+
 # Consecutive failed recovery probes before the CEO is notified once per episode.
 _CEO_NOTIFY_THRESHOLD = 10
 
@@ -270,6 +272,17 @@ _GROK_INTERACTIVE_DOCKERFILES = {
 # the retry window (unknown-provider time-expiry fallback in _probe_target).
 _GROK_RATE_LIMIT_EXIT_CODE = 75
 _GROK_RATE_LIMIT_RETRY_AFTER_S = 60.0
+# A one-shot Grok container exits with this code (EX_CONFIG) when the
+# entrypoint's `grok_auth --check` backstop found the access token missing or
+# expired (it can't be refreshed headlessly, so the CLI would hang at an
+# interactive login prompt). Park the provider instead of crash-retrying 3x —
+# the agent cannot start without a valid token, so respawning burns tokens for
+# zero progress. The probe-resume loop revives the task once
+# grok_auth.refresh_if_stale (run once per dispatch tick) mints a fresh token
+# from the offline-access refresh token; if still expired, the next exit 78
+# re-parks (no token burn). Same shape as the 429 exit-75 path (F041).
+_GROK_AUTH_EXIT_CODE = 78
+_GROK_AUTH_RETRY_AFTER_S = 60.0
 
 
 # =============================================================================
@@ -5599,6 +5612,14 @@ Start by:
         if self._is_grok_rate_limit_exit(instance, exit_code):
             await self._park_grok_rate_limited(agent_id, instance)
             return
+        # Grok auth-missing parking (F041): a one-shot grok run whose entrypoint
+        # found the token missing/expired exits 78 (EX_CONFIG). Park the provider
+        # instead of crash-retrying — the agent can't start without a valid token,
+        # so respawning burns tokens for zero progress. The probe-resume loop
+        # revives the task once grok_auth.refresh_if_stale mints a fresh token.
+        if self._is_grok_auth_exit(instance, exit_code):
+            await self._park_grok_auth_unavailable(agent_id, instance)
+            return
         graceful = exit_code == 0
         # Session/usage-limit parking: the Claude session ("5-hour") limit is a
         # 429 the SDK does not retry — the container exits non-zero with a
@@ -6376,6 +6397,23 @@ Start by:
             and instance.config.provider_type == ModelProvider.GROK.value
         )
 
+    @staticmethod
+    def _is_grok_auth_exit(instance: Any, exit_code: int | None) -> bool:
+        """True for a one-shot grok container that exited 78 (auth missing/expired).
+
+        The entrypoint runs ``grok_auth --check`` as a backstop and exits 78
+        (EX_CONFIG) when the access token is missing or expired — the CLI cannot
+        refresh it headlessly and would otherwise hang at an interactive login
+        prompt. See ``_GROK_AUTH_EXIT_CODE`` for the full rationale (F041).
+        """
+        from roboco.models.base import ModelProvider
+
+        return (
+            exit_code == _GROK_AUTH_EXIT_CODE
+            and instance.config is not None
+            and instance.config.provider_type == ModelProvider.GROK.value
+        )
+
     @staticmethod
     async def _tail_container_logs(container_name: str, lines: int = 80) -> str:
         """Return the last ``lines`` of a container's combined output, '' on error.
@@ -6530,9 +6568,7 @@ Start by:
         # call ``mark_waiting_long`` itself — the container is already dead (it
         # exited), so there is nothing to stop, and parking keeps OFFLINE (not
         # WAITING_LONG) so the reaper's live-skip / health loop ignore it.
-        task_id = (
-            str(instance.current_task_id) if instance.current_task_id else None
-        )
+        task_id = str(instance.current_task_id) if instance.current_task_id else None
         record = WaitingRecord(
             agent_id=agent_id,
             task_id=task_id,
@@ -6563,6 +6599,26 @@ Start by:
             kind="rate_limited",
         )
 
+    async def _park_grok_auth_unavailable(self, agent_id: str, instance: Any) -> None:
+        """Park a grok agent whose token was missing/expired (entrypoint exit 78).
+
+        Same park-and-probe shape as the 429 exit-75 path, but with
+        ``kind="auth_missing"``: the agent cannot start without a valid token, so
+        crash-retrying burns tokens for zero progress. The probe-resume loop
+        revives the task once ``grok_auth.refresh_if_stale`` mints a fresh token
+        (run once per dispatch tick); if still expired, the next exit 78 re-parks
+        (no token burn). See ``_GROK_AUTH_EXIT_CODE`` (F041).
+        """
+        from roboco.models.base import ModelProvider
+
+        await self._park_provider_unavailable(
+            agent_id,
+            instance,
+            provider=ModelProvider.GROK.value,
+            retry_after=_GROK_AUTH_RETRY_AFTER_S,
+            kind="auth_missing",
+        )
+
     @staticmethod
     def _too_early_to_probe(state: dict[str, Any]) -> bool:
         """True while the estimated lift time (activated_at + retry_after) is future.
@@ -7835,9 +7891,7 @@ Start now: evidence(task_id="{task_id}")
             # ACTIVE; the reaper's Docker-liveness fallback covers it).
             container_id: str | None = None
             try:
-                container_id = await self._resolve_container_id(
-                    f"roboco-agent-{slug}"
-                )
+                container_id = await self._resolve_container_id(f"roboco-agent-{slug}")
             except Exception:
                 container_id = None
             self._instances[slug] = AgentInstance(

tests/unit/runtime/test_grok_rate_limit.py

@@ -15,6 +15,7 @@ from unittest.mock import AsyncMock
 import pytest
 from roboco.models.runtime import AgentInstance
 from roboco.runtime.orchestrator import (
+    _GROK_AUTH_EXIT_CODE,
     _GROK_RATE_LIMIT_EXIT_CODE,
     AgentOrchestrator,
     AgentState,
@@ -107,12 +108,18 @@ async def test_park_grok_rate_limited_activates_and_offlines(
     monkeypatch: pytest.MonkeyPatch,
 ) -> None:
     orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    # _park_provider_unavailable registers a WaitingRecord (F035) so the
+    # probe-resume loop can revive the task; the bare __new__ orchestrator
+    # needs the dict + persist stub to exercise that without AttributeError.
+    orch._waiting_records = {}
+    orch._rate_limit_ceo_notified = set()
     inst = _grok_instance()
     inst.error_count = 2  # pretend prior crashes — parking must NOT count one
     tracker = _FakeTracker()
     monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker)
     finalize = AsyncMock()
     monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+    monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock())
 
     await orch._park_grok_rate_limited("be-dev-1", inst)
 
@@ -143,3 +150,68 @@ async def test_handle_stopped_container_parks_on_grok_429(
     park.assert_awaited_once_with("be-dev-1", inst)
     # Early-return: the normal crash/graceful finalize path never runs.
     finalize.assert_not_awaited()
+
+
+# ---------------------------------------------------------------------------
+# F041: exit 78 (auth missing/expired) parks instead of crash-retrying
+# ---------------------------------------------------------------------------
+
+
+def test_is_grok_auth_exit() -> None:
+    inst = _grok_instance()
+    assert AgentOrchestrator._is_grok_auth_exit(inst, _GROK_AUTH_EXIT_CODE)
+    # Wrong exit code, or a non-grok provider, is not a grok-auth exit.
+    assert not AgentOrchestrator._is_grok_auth_exit(inst, 0)
+    assert not AgentOrchestrator._is_grok_auth_exit(inst, 1)
+    assert not AgentOrchestrator._is_grok_auth_exit(
+        _grok_instance(provider_type="anthropic"), _GROK_AUTH_EXIT_CODE
+    )
+
+
+@pytest.mark.asyncio
+async def test_handle_stopped_container_parks_on_grok_auth_exit(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # F041: a grok container whose entrypoint ran `grok_auth --check` and found
+    # the token missing/expired exits 78 (EX_CONFIG). Crash-retrying 3x burns
+    # tokens for zero progress (the agent can't start without a valid token);
+    # park it like the 429 exit-75 path so the probe-resume loop revives the
+    # task once grok_auth.refresh_if_stale mints a fresh token.
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    inst = _grok_instance()
+    park = AsyncMock()
+    finalize = AsyncMock()
+    monkeypatch.setattr(orch, "_park_grok_auth_unavailable", park)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", finalize)
+
+    await orch._handle_stopped_container("be-dev-1", inst, _GROK_AUTH_EXIT_CODE)
+
+    park.assert_awaited_once_with("be-dev-1", inst)
+    # Early-return: the crash-retry path never runs (no token burn).
+    finalize.assert_not_awaited()
+
+
+@pytest.mark.asyncio
+async def test_park_grok_auth_unavailable_activates_with_auth_missing_kind(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    orch = AgentOrchestrator.__new__(AgentOrchestrator)
+    orch._waiting_records = {}
+    orch._rate_limit_ceo_notified = set()
+    inst = _grok_instance()
+    inst.error_count = 2  # prior crashes — parking must NOT count one
+    tracker = _FakeTracker()
+    monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker)
+    monkeypatch.setattr(orch, "_finalize_spawn_session", AsyncMock())
+    monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock())
+
+    await orch._park_grok_auth_unavailable("be-dev-1", inst)
+
+    assert inst.state == AgentState.OFFLINE
+    assert inst.container_id is None
+    assert inst.error_count == 0  # an auth-missing exit is not a crash
+    assert tracker.activated_with == {
+        "retry_after": pytest.approx(60.0),
+        "affected_agents": ["be-dev-1"],
+        "kind": "auth_missing",
+    }