From a4a5756f67291b48db5bc59a7b3416516b72003e Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 20:51:45 +0200 Subject: [PATCH] [F097] orchestrator: back off grok re-park retry_after within a rate-limit episode MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _probe_target returns (None, {}) for grok — the grok CLI's xAI endpoint is closed and the SuperGrok OIDC access token is not a valid bearer for the metered api.x.ai, so a real probe would either no-op or strand grok parked forever. _do_probe treats url-is-None as success (time-expiry optimism), so once the 60s retry_after passes the probe loop optimistically clears the grok park, a cleared park dispatches a fresh grok agent that hits the still-active xAI 429, exits 75, and re-parks — a flat ~90s crash-retry cycle for the whole xAI rate-limit window (each cycle costs container startup + a rejected grok call). Fix: track _grok_repark_count + _grok_last_park_at in _park_grok_rate_limited and back the re-park retry_after off exponentially within one episode (60 -> 120 -> 240 -> ... capped at 2**4 = ~16min cycle) so the churn dampens. A gap past _GROK_REPARK_EPISODE_GAP_S (25min, > the capped cycle) means no re-park for that long => the rate limit actually lifted => a fresh episode resets the count to the base 60s, so recovery latency isn't penalized across episodes. The first park in a fresh episode is unchanged at 60s. --- roboco/runtime/orchestrator.py | 49 +++++++++- tests/unit/runtime/test_grok_rate_limit.py | 106 +++++++++++++++++++++ 2 files changed, 153 insertions(+), 2 deletions(-) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 643ada57..c3fd6330 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -304,6 +304,18 @@ _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 +# F097: grok has no real recovery probe (the grok CLI's xAI endpoint is closed +# and the SuperGrok OIDC access token is not a valid bearer for the metered +# api.x.ai, so a probe would either no-op or strand grok parked forever). So +# the probe loop clears a grok park optimistically on a timer, a cleared park +# dispatches a fresh grok agent that immediately hits the still-active xAI 429, +# exits 75, and re-parks — a flat ~90s crash-retry cycle for the whole xAI +# window. Back the re-park retry_after off exponentially within one episode so +# the churn dampens (60 -> 120 -> 240 -> ... capped) instead of spinning flat. +# Cap bounds the cycle; the episode gap (> the max cycle) resets the count once +# the rate limit has actually lifted (no re-park for the gap => fresh episode). +_GROK_REPARK_BACKOFF_CAP = 4 # max 2**4 = 16x base (~16min cycle) +_GROK_REPARK_EPISODE_GAP_S = 1500.0 # 25min — > the capped ~16min cycle # 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 @@ -909,6 +921,16 @@ class AgentOrchestrator: # kill-switch parity (the grok CLI exposes no live usage hook). 0 disables. # See _enforce_grok_cost_budget. self._grok_max_cost_usd: float = settings.grok_max_cost_usd + # F097: grok re-park backoff state. Grok has no real recovery probe, so + # the probe loop clears a grok park optimistically on a timer; a cleared + # park respawns a grok agent that hits the still-active xAI 429 and + # re-parks. Track the re-park count within one episode so the retry_after + # can back off exponentially (dampening the ~90s crash-retry churn), and + # the last park time so a gap (the rate limit actually lifted) resets + # the count for the next episode. Per-provider state would be cleaner, + # but grok is a single provider key, so a scalar suffices. + self._grok_last_park_at: datetime | None = None + self._grok_repark_count: int = 0 # ========================================================================= # LIFECYCLE @@ -6865,14 +6887,37 @@ Start by: ) 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).""" + """Park a grok agent whose run hit an xAI 429 (entrypoint exit 75). + + F097: grok has no real recovery probe, so the probe loop clears a grok + park optimistically on a timer — a cleared park dispatches a fresh + grok agent that hits the still-active xAI 429, exits 75, and re-parks. + Without a backoff this is a flat ~90s crash-retry cycle for the whole + xAI rate-limit window. Back the re-park retry_after off exponentially + within one episode (60 -> 120 -> 240 -> ... capped) so the churn + dampens. A gap past ``_GROK_REPARK_EPISODE_GAP_S`` (no re-park for that + long => the rate limit actually lifted) starts a fresh episode at the + base retry_after, so recovery latency isn't penalized across episodes. + """ from roboco.models.base import ModelProvider + now = datetime.now(UTC) + last = self._grok_last_park_at + if ( + last is not None + and (now - last).total_seconds() < _GROK_REPARK_EPISODE_GAP_S + ): + self._grok_repark_count += 1 + else: + self._grok_repark_count = 0 + self._grok_last_park_at = now + backoff = 2 ** min(self._grok_repark_count, _GROK_REPARK_BACKOFF_CAP) + retry_after = _GROK_RATE_LIMIT_RETRY_AFTER_S * backoff await self._park_provider_unavailable( agent_id, instance, provider=ModelProvider.GROK.value, - retry_after=_GROK_RATE_LIMIT_RETRY_AFTER_S, + retry_after=retry_after, kind="rate_limited", ) diff --git a/tests/unit/runtime/test_grok_rate_limit.py b/tests/unit/runtime/test_grok_rate_limit.py index 91de5a61..e030064a 100644 --- a/tests/unit/runtime/test_grok_rate_limit.py +++ b/tests/unit/runtime/test_grok_rate_limit.py @@ -10,6 +10,7 @@ Claude session/overload paths get the same loop-break for free. from __future__ import annotations +from datetime import UTC, datetime, timedelta from unittest.mock import AsyncMock import pytest @@ -17,6 +18,8 @@ from roboco.models.runtime import AgentInstance from roboco.runtime.orchestrator import ( _GROK_AUTH_EXIT_CODE, _GROK_RATE_LIMIT_EXIT_CODE, + _GROK_RATE_LIMIT_RETRY_AFTER_S, + _GROK_REPARK_BACKOFF_CAP, AgentOrchestrator, AgentState, ) @@ -113,6 +116,9 @@ async def test_park_grok_rate_limited_activates_and_offlines( # needs the dict + persist stub to exercise that without AttributeError. orch._waiting_records = {} orch._rate_limit_ceo_notified = set() + # F097 backoff state — the constructor (skipped here) initializes these. + orch._grok_last_park_at = None + orch._grok_repark_count = 0 inst = _grok_instance() inst.error_count = 2 # pretend prior crashes — parking must NOT count one tracker = _FakeTracker() @@ -215,3 +221,103 @@ async def test_park_grok_auth_unavailable_activates_with_auth_missing_kind( "affected_agents": ["be-dev-1"], "kind": "auth_missing", } + + +# --------------------------------------------------------------------------- # +# F097 — grok has no real probe, so an optimistic clear respawns into a still- +# active xAI 429 every ~90s. Back off the re-park retry_after within one rate- +# limit episode so the churn dampens instead of spinning flat at 60s. +# --------------------------------------------------------------------------- # + + +class _RecordingTracker: + """Records every activate() retry_after across multiple re-parks.""" + + def __init__(self) -> None: + self.retry_afters: list[float] = [] + self.kinds: list[str] = [] + self.agents_lists: list[list[str]] = [] + + async def activate( + self, *, retry_after: float, affected_agents: list[str], kind: str + ) -> None: + self.retry_afters.append(retry_after) + self.kinds.append(kind) + self.agents_lists.append(affected_agents) + + +def _backoff_orchestrator() -> AgentOrchestrator: + orch = AgentOrchestrator.__new__(AgentOrchestrator) + orch._waiting_records = {} + orch._rate_limit_ceo_notified = set() + # Backoff state — the constructor (skipped here) initializes these. + orch._grok_last_park_at = None + orch._grok_repark_count = 0 + return orch + + +@pytest.mark.asyncio +async def test_grok_repark_backs_off_within_episode( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """Three re-parks within one episode (microseconds apart) must grow the + retry_after — 60 -> 120 -> 240 — not stay flat at 60s (the ~90s crash-retry + cycle the optimistic clear produces today).""" + orch = _backoff_orchestrator() + tracker = _RecordingTracker() + monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker) + monkeypatch.setattr(orch, "_finalize_spawn_session", AsyncMock()) + monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock()) + inst = _grok_instance() + + await orch._park_grok_rate_limited("be-dev-1", inst) + await orch._park_grok_rate_limited("be-dev-1", inst) + await orch._park_grok_rate_limited("be-dev-1", inst) + + assert tracker.retry_afters == [60.0, 120.0, 240.0] + assert tracker.kinds == ["rate_limited", "rate_limited", "rate_limited"] + + +@pytest.mark.asyncio +async def test_grok_repark_resets_after_episode_gap( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A re-park AFTER the episode gap is a fresh episode — the retry_after + resets to the base 60s even if the prior episode had backed off.""" + orch = _backoff_orchestrator() + orch._grok_repark_count = 3 # pretend a prior episode backed off hard + orch._grok_last_park_at = datetime.now(UTC) - timedelta(hours=2) + tracker = _RecordingTracker() + monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker) + monkeypatch.setattr(orch, "_finalize_spawn_session", AsyncMock()) + monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock()) + inst = _grok_instance() + + await orch._park_grok_rate_limited("be-dev-1", inst) + + # Fresh episode -> base retry_after, no carried-over backoff. + assert tracker.retry_afters == [60.0] + assert orch._grok_repark_count == 0 + + +@pytest.mark.asyncio +async def test_grok_repark_backoff_caps(monkeypatch: pytest.MonkeyPatch) -> None: + """The backoff is capped so a long xAI rate-limit window doesn't push the + retry_after toward infinity (bounded cycle, recovery still reachable).""" + orch = _backoff_orchestrator() + tracker = _RecordingTracker() + monkeypatch.setattr(orch, "_make_tracker", lambda _p: tracker) + monkeypatch.setattr(orch, "_finalize_spawn_session", AsyncMock()) + monkeypatch.setattr(orch, "_persist_waiting_record", AsyncMock()) + inst = _grok_instance() + + # Park once, then re-park well past the cap. + for _ in range(_GROK_REPARK_BACKOFF_CAP + 3): + await orch._park_grok_rate_limited("be-dev-1", inst) + + max_expected = _GROK_RATE_LIMIT_RETRY_AFTER_S * (2**_GROK_REPARK_BACKOFF_CAP) + # Every retry_after from the cap onward is the same capped value. + assert all( + r == max_expected for r in tracker.retry_afters[_GROK_REPARK_BACKOFF_CAP:] + ) + assert max(tracker.retry_afters) == max_expected