diff --git a/docker/scripts/grok-agent-entrypoint.sh b/docker/scripts/grok-agent-entrypoint.sh index e75ac877..6ac6e18e 100755 --- a/docker/scripts/grok-agent-entrypoint.sh +++ b/docker/scripts/grok-agent-entrypoint.sh @@ -58,21 +58,30 @@ fi # the post-mortem + silent-exit substitute below (the Claude SessionEnd / Stop # hooks have no opencode equivalent, so the boundary handles them). `set +e` # around the run so a non-zero opencode exit doesn't abort before the post-run -# hooks; PIPESTATUS preserves opencode's real code through the `tee`. +# hooks; tee captures the output for rate-limit detection and PIPESTATUS +# preserves opencode's real exit code through the pipe. +RUN_LOG="/tmp/opencode-run.log" set +e opencode run \ --model "xai/${ROBOCO_AGENT_MODEL:-grok-build-0.1}" \ "${variant_arg[@]}" \ - -- "${ROBOCO_INITIAL_PROMPT:-}" < /dev/null -run_rc=$? + -- "${ROBOCO_INITIAL_PROMPT:-}" < /dev/null 2>&1 | tee "$RUN_LOG" +run_rc=${PIPESTATUS[0]} set -e -# --- Post-run hooks (Claude SessionEnd + Stop parity) ---------------------- -# Read terminal state once, then (a) write a post-mortem journal entry and -# (b) if the agent exited WITHOUT a terminal verb (i_am_idle / i_am_done / -# pass / fail / ...), auto-substitute the task so it is not left stuck in -# claimed/in_progress for a human to hand-unstick. Best-effort; never change -# the exit code the orchestrator observes. +# --- Rate-limit detection (B4) --------------------------------------------- +# A 429 from xAI ends the one-shot run without the agent ever calling a terminal +# verb. Detect it from the run output and exit 75 (EX_TEMPFAIL) so the +# orchestrator PARKS the grok provider instead of the dispatcher re-spawning the +# same task every tick (429 -> exit -> respawn -> 429, a cost/token loop). A +# rate-limited task is NOT substituted — it must be retried once the limit lifts. +RATE_LIMITED=0 +if grep -qiE '(\b429\b|too many requests|rate.?limit|quota exceeded|rate_limit_exceeded)' \ + "$RUN_LOG" 2>/dev/null; then + RATE_LIMITED=1 +fi + +# --- Post-mortem (Claude SessionEnd parity) — always ----------------------- terminal=$(curl -sf -m 2 "${SDK_URL}/terminal/status" 2>/dev/null || echo "") last_tool="null" had_terminal="false" @@ -86,6 +95,16 @@ curl -sf -m 3 -X POST "${SDK_URL}/journal/post_mortem" \ -d "{\"terminal_tool\":\"${last_tool}\",\"reason\":\"session_end\"}" \ >/dev/null 2>&1 || true +if [ "$RATE_LIMITED" = "1" ]; then + echo "[grok] xAI rate-limited — exiting 75 so the orchestrator parks the" \ + "provider; the task is retried when the limit lifts (not substituted)." >&2 + exit 75 +fi + +# --- Silent-exit substitute (Claude Stop parity) --------------------------- +# Only when NOT rate-limited: if the agent exited WITHOUT a terminal verb +# (i_am_idle / i_am_done / pass / fail / ...), auto-substitute the task so it is +# not left stuck in claimed/in_progress for a human to hand-unstick. if [ "$had_terminal" != "true" ]; then curl -sf -m 3 -X POST "${SDK_URL}/terminal/force_substitute" >/dev/null 2>&1 || true echo "[grok] exited without a terminal verb (last tool: ${last_tool}) — auto-substituted." >&2 diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 14f37878..6f74bd9c 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -179,6 +179,14 @@ _GROK_INTERACTIVE_DOCKERFILES = { GROK_SECRETARY_IMAGE: "agent-grok-secretary.Dockerfile", } +# A one-shot Grok container exits with this code (EX_TEMPFAIL) when the run hit +# an xAI 429 (grok-agent-entrypoint.sh detects it). The orchestrator parks the +# grok provider rate-limited instead of crash-retrying, breaking the +# 429 -> exit -> respawn cost loop. The probe-resume loop clears the park after +# 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 + # ============================================================================= # ORCHESTRATOR @@ -1744,6 +1752,21 @@ class AgentOrchestrator: config, instance, agent_settings_path = await self._prepare_agent_spawn( agent_id, task_id, model, git_context ) + # Grok 429 loop-breaker (B4): while the xAI provider is parked + # rate-limited, do NOT launch another grok container — the dispatcher + # would otherwise re-spawn the same task every tick, 429, and burn + # cost. The probe-resume loop clears the park after the retry window and + # the next tick spawns normally. Grok-only; the Claude path is untouched. + # Fail-open: a tracker read error must never block spawning. + if await self._grok_spawn_parked(config.provider_type): + self._mark_task_handled(task_id) + instance.state = AgentState.OFFLINE + logger.info( + "Grok spawn skipped: provider rate-limited (parked)", + agent_id=agent_id, + task_id=task_id, + ) + return instance # Record the task as handled so later dispatchers in the same # tick don't act on it again. Safe even if _launch_spawn fails # — the next tick starts fresh. @@ -4798,6 +4821,13 @@ Start by: do nothing; non-zero exits keep the existing crash-retry behaviour. """ cid = instance.container_id[:12] if instance.container_id else None + # Grok 429 parking (B4): a one-shot grok run that hit an xAI 429 exits + # 75 (set by grok-agent-entrypoint.sh). Park the provider instead of + # crash-retrying so the spawn guard suppresses the respawn loop; the + # probe-resume loop revives the task when the limit lifts. + if self._is_grok_rate_limit_exit(instance, exit_code): + await self._park_grok_rate_limited(agent_id, instance) + return graceful = exit_code == 0 if graceful: logger.info( @@ -5340,6 +5370,65 @@ Start by: return RateLimitStateTracker(provider) + async def _grok_spawn_parked(self, provider_type: str | None) -> bool: + """True when *provider_type* is GROK and the provider is parked rate-limited. + + The grok 429 loop-breaker consults this before launching a grok + container. Grok-only and fail-open: any error reading the tracker + returns False so a Redis hiccup can never block spawning. + """ + from roboco.models.base import ModelProvider + + if provider_type != ModelProvider.GROK.value: + return False + try: + tracker = self._make_tracker(ModelProvider.GROK.value) + return bool(await tracker.is_rate_limited()) + except Exception as exc: + logger.warning( + "grok rate-limit check failed; allowing spawn", error=str(exc) + ) + return False + + @staticmethod + def _is_grok_rate_limit_exit(instance: Any, exit_code: int | None) -> bool: + """True for a one-shot grok container that exited 75 (xAI 429).""" + from roboco.models.base import ModelProvider + + return ( + exit_code == _GROK_RATE_LIMIT_EXIT_CODE + and instance.config is not None + and instance.config.provider_type == ModelProvider.GROK.value + ) + + async def _park_grok_rate_limited(self, agent_id: str, instance: Any) -> None: + """Park a grok agent whose run hit an xAI 429 (entrypoint exit 75). + + Finalize the session for usage capture, mark the instance OFFLINE + WITHOUT counting a crash (so it isn't escalated as stranded), and + activate the grok rate-limit tracker so the spawn guard suppresses + re-spawns until the probe-resume loop clears it after the retry window. + The task stays claimed/in_progress and is retried when the limit lifts. + """ + from roboco.models.base import ModelProvider + + await self._finalize_spawn_session(agent_id, exit_reason="rate_limited") + instance.state = AgentState.OFFLINE + instance.container_id = None + instance.error_count = 0 # a 429 is not a crash — don't escalate as stranded + try: + await self._make_tracker(ModelProvider.GROK.value).activate( + retry_after=_GROK_RATE_LIMIT_RETRY_AFTER_S, + affected_agents=[agent_id], + ) + except Exception as exc: + logger.warning("failed to park grok rate-limit state", error=str(exc)) + logger.warning( + "Grok provider rate-limited; parked (task retried when the limit lifts)", + agent_id=agent_id, + task_id=instance.current_task_id, + ) + @staticmethod def _too_early_to_probe(state: dict[str, Any]) -> bool: """True while the estimated lift time (activated_at + retry_after) is future. diff --git a/tests/unit/runtime/test_grok_rate_limit.py b/tests/unit/runtime/test_grok_rate_limit.py new file mode 100644 index 00000000..663fd184 --- /dev/null +++ b/tests/unit/runtime/test_grok_rate_limit.py @@ -0,0 +1,136 @@ +"""GROK 429 parking: break the 429 -> exit -> respawn cost loop (B4). + +A one-shot grok run that hits an xAI 429 exits 75; the orchestrator parks the +grok provider rate-limited instead of crash-retrying, and the spawn guard +suppresses re-spawns until the probe-resume loop clears the park. These tests +exercise the decision points deterministically (tracker + finalize stubbed). +""" + +from __future__ import annotations + +from unittest.mock import AsyncMock + +import pytest +from roboco.models.runtime import AgentInstance +from roboco.runtime.orchestrator import ( + _GROK_RATE_LIMIT_EXIT_CODE, + AgentOrchestrator, + AgentState, +) + + +def _grok_instance(provider_type: str = "grok") -> AgentInstance: + cfg = type("C", (), {"provider_type": provider_type, "model": "grok-build-0.1"})() + inst = AgentInstance(agent_id="be-dev-1", state=AgentState.ACTIVE, config=cfg) + inst.current_task_id = "task-1" + inst.container_id = "cid" + return inst + + +class _FakeTracker: + def __init__(self, *, limited: bool = False) -> None: + self._limited = limited + self.activated_with: dict[str, object] | None = None + + async def is_rate_limited(self) -> bool: + return self._limited + + async def activate(self, *, retry_after: float, affected_agents: list[str]) -> None: + self.activated_with = { + "retry_after": retry_after, + "affected_agents": affected_agents, + } + + +def test_is_grok_rate_limit_exit() -> None: + inst = _grok_instance() + assert AgentOrchestrator._is_grok_rate_limit_exit(inst, _GROK_RATE_LIMIT_EXIT_CODE) + # Wrong exit code, or a non-grok provider, is not a grok-429 exit. + assert not AgentOrchestrator._is_grok_rate_limit_exit(inst, 0) + assert not AgentOrchestrator._is_grok_rate_limit_exit(inst, 1) + assert not AgentOrchestrator._is_grok_rate_limit_exit( + _grok_instance(provider_type="anthropic"), _GROK_RATE_LIMIT_EXIT_CODE + ) + + +@pytest.mark.asyncio +async def test_grok_spawn_parked_true_when_limited( + monkeypatch: pytest.MonkeyPatch, +) -> None: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + monkeypatch.setattr(orch, "_make_tracker", lambda _p: _FakeTracker(limited=True)) + assert await orch._grok_spawn_parked("grok") is True + + +@pytest.mark.asyncio +async def test_grok_spawn_parked_false_when_not_limited( + monkeypatch: pytest.MonkeyPatch, +) -> None: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + monkeypatch.setattr(orch, "_make_tracker", lambda _p: _FakeTracker(limited=False)) + assert await orch._grok_spawn_parked("grok") is False + + +@pytest.mark.asyncio +async def test_grok_spawn_parked_false_for_non_grok( + monkeypatch: pytest.MonkeyPatch, +) -> None: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + # Non-grok never consults the tracker (a tracker call would error here). + monkeypatch.setattr( + orch, "_make_tracker", lambda _p: (_ for _ in ()).throw(AssertionError) + ) + assert await orch._grok_spawn_parked("anthropic") is False + + +@pytest.mark.asyncio +async def test_grok_spawn_parked_fails_open(monkeypatch: pytest.MonkeyPatch) -> None: + # A tracker error must never block spawning (fail-open -> False). + def _boom(_p: str) -> object: + raise RuntimeError("redis down") + + orch = AgentOrchestrator.__new__(AgentOrchestrator) + monkeypatch.setattr(orch, "_make_tracker", _boom) + assert await orch._grok_spawn_parked("grok") is False + + +@pytest.mark.asyncio +async def test_park_grok_rate_limited_activates_and_offlines( + monkeypatch: pytest.MonkeyPatch, +) -> None: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + 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) + + await orch._park_grok_rate_limited("be-dev-1", inst) + + finalize.assert_awaited_once() + assert inst.state == AgentState.OFFLINE + assert inst.container_id is None + assert inst.error_count == 0 # a 429 is not a crash + assert tracker.activated_with == { + "retry_after": pytest.approx(60.0), + "affected_agents": ["be-dev-1"], + } + + +@pytest.mark.asyncio +async def test_handle_stopped_container_parks_on_grok_429( + monkeypatch: pytest.MonkeyPatch, +) -> None: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + inst = _grok_instance() + park = AsyncMock() + finalize = AsyncMock() + monkeypatch.setattr(orch, "_park_grok_rate_limited", park) + monkeypatch.setattr(orch, "_finalize_spawn_session", finalize) + + await orch._handle_stopped_container("be-dev-1", inst, _GROK_RATE_LIMIT_EXIT_CODE) + + park.assert_awaited_once_with("be-dev-1", inst) + # Early-return: the normal crash/graceful finalize path never runs. + finalize.assert_not_awaited()