From aaf6b19303e464c22b0f6d28479cbd9bf64f7cc2 Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 20:24:56 +0200 Subject: [PATCH] [F094] add a persistent-probe-failure escape hatch to provider parking MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit _on_probe_failure only incremented the failure counter and, at 10 failures, sent a one-shot CEO notification. It never cleared the tracker, never gave up, never fell back to time-expiry. _do_probe returns False for any non-2xx AND any httpx error, so a permanently unreachable probe endpoint (removed API key, network partition to the probe host, misconfigured base URL) kept the provider parked forever — every agent on it gated by _provider_spawn_parked, their tasks reaped to pending but the spawn gate queuing every spawn, sitting pending forever. The only recovery was the operator manually clearing the Redis key. Past _PROBE_GIVE_UP_THRESHOLD (30) persistent failures, fall back to the same time-expiry optimism the unprobeable-provider path uses (_do_probe returns True when there is no probe URL): clear the park and resume parked agents. If the provider is genuinely still down the real workload attempts re-park via the 429/5xx path, so this is bounded burn — strictly better than a silent forever-strand. Kept above the CEO-notify threshold (10) so the operator still gets the notification first. --- roboco/runtime/orchestrator.py | 38 ++++- tests/unit/runtime/test_rate_limit_sweep.py | 152 +++++++++++++++++++- 2 files changed, 188 insertions(+), 2 deletions(-) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 13836ec2..2b178fbc 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -150,6 +150,18 @@ def _system_api_headers() -> dict[str, str]: # Consecutive failed recovery probes before the CEO is notified once per episode. _CEO_NOTIFY_THRESHOLD = 10 +# Persistent-probe-failure escape hatch (F094): if the recovery probe keeps +# failing past this threshold, the probe endpoint itself is the problem (a +# misconfigured URL, a removed API key, a network partition to the probe host) +# while the provider may well be fine for real workloads. Hold the park any +# longer and every agent on the provider strands forever with only a one-shot +# CEO notification. Past this threshold, fall back to the same time-expiry +# optimism the unprobeable-provider path uses (``_do_probe`` returns True when +# there is no probe URL): clear the park and resume. If the provider is +# genuinely still down the real workload attempts re-park via the 429/5xx path, +# so this is bounded burn — strictly better than a silent forever-strand. Kept +# above the CEO-notify threshold so the operator gets the notification first. +_PROBE_GIVE_UP_THRESHOLD = 30 # Persistent server-overload parking (HTTP 529 / 500 / 503). The model API's # SDK already retries transient overloads in-process; only a persistent one @@ -6887,7 +6899,21 @@ Start by: async def _on_probe_failure( self, provider: str, tracker: Any, activated_at_raw: str | None ) -> None: - """Count a failed probe; notify the CEO once at the failure threshold.""" + """Count a failed probe; notify the CEO once at the failure threshold. + + F094 escape hatch: past ``_PROBE_GIVE_UP_THRESHOLD`` persistent failures + the probe endpoint itself is the problem (misconfigured URL / removed API + key / network partition to the probe host) while the provider may be fine + for real workloads. Holding the park any longer strands every agent on + the provider forever with only a one-shot CEO notification. Fall back to + the same time-expiry optimism the unprobeable-provider path uses: clear + the park and resume. If the provider is genuinely still down the real + workload attempts re-park via the 429/5xx path, so this is bounded burn — + strictly better than a silent forever-strand. ``_on_probe_success`` + clears the tracker (so the loop won't probe this provider again until a + real 429 re-parks) and discards the CEO-notified flag (a fresh episode + later gets a fresh notification). + """ failure_count = await tracker.increment_probe_failures() logger.debug( "Rate-limit probe failed", provider=provider, probe_failures=failure_count @@ -6902,6 +6928,16 @@ Start by: activated_at_str=activated_at_raw or "unknown", paused_agent_count=len(self._parked_agents_for(provider)), ) + if failure_count >= _PROBE_GIVE_UP_THRESHOLD: + logger.warning( + "Rate-limit probe persistently failing; giving up on the probe " + "and falling back to time-expiry optimism (clearing the park + " + "resuming parked agents). If the provider is genuinely still down " + "they will re-park via the real 429/5xx path.", + provider=provider, + probe_failures=failure_count, + ) + await self._on_probe_success(provider, tracker) async def _probe_one_provider(self, provider: str, state: dict[str, Any]) -> None: """Probe a single rate-limited provider and handle the outcome.""" diff --git a/tests/unit/runtime/test_rate_limit_sweep.py b/tests/unit/runtime/test_rate_limit_sweep.py index 558ee8c0..82c67e12 100644 --- a/tests/unit/runtime/test_rate_limit_sweep.py +++ b/tests/unit/runtime/test_rate_limit_sweep.py @@ -20,7 +20,11 @@ from httpx import ASGITransport, AsyncClient from roboco.api.app import create_app from roboco.models.events import EventType from roboco.models.runtime import WaitingRecord -from roboco.runtime.orchestrator import AgentOrchestrator +from roboco.runtime.orchestrator import ( + _CEO_NOTIFY_THRESHOLD, + _PROBE_GIVE_UP_THRESHOLD, + AgentOrchestrator, +) from roboco.services.gateway.rate_limit_tracker import RateLimitStateTracker _HTTP_OK = 200 @@ -307,6 +311,152 @@ class TestProbeFailurePath: tracker_mock.clear.assert_not_awaited() +# --------------------------------------------------------------------------- +# Tests: persistent-probe-failure escape hatch (F094) +# --------------------------------------------------------------------------- + + +class TestProbeGiveUpEscapeHatch: + """When the probe has failed persistently past the give-up threshold, the + provider is no longer held parked forever (the probe endpoint may be + misconfigured / unreachable while the provider is actually fine for real + workloads). Fall back to the same time-expiry optimism the unprobeable + provider path uses (``_do_probe`` returns True when there is no probe URL): + clear the park and resume parked agents. If the provider is genuinely still + down the real workload attempts re-park via the 429/5xx path — strictly + better than a silent forever-strand with only a one-shot CEO notification. + """ + + async def test_persistent_failure_clears_park_and_resumes_at_threshold( + self, + ) -> None: + """At the give-up threshold the park is cleared and parked agents + resumed (the escape hatch), not held forever.""" + orch = _make_orchestrator() + provider = "anthropic" + state = _make_active_state(provider, retry_after=None) + orch._waiting_records = { + "be-dev-1": _waiting_record("be-dev-1", provider), + } + + tracker_mock = _make_tracker_mock(failure_return=_PROBE_GIVE_UP_THRESHOLD) + resolve_mock = AsyncMock(return_value=None) + + with ( + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "resolve_wait", new=resolve_mock), + patch.object(orch, "_do_probe", new=AsyncMock(return_value=False)), + patch.object(orch, "_notify_rate_limit_ceo", new=AsyncMock()), + patch("roboco.events.get_event_bus") as mock_bus_fn, + ): + bus_mock = AsyncMock() + bus_mock.publish = AsyncMock() + mock_bus_fn.return_value = bus_mock + await orch._probe_one_provider(provider, state) + + tracker_mock.clear.assert_awaited_once() # the park was cleared + # the parked agent was resumed + assert resolve_mock.await_count == 1 + assert resolve_mock.call_args_list[0].args[0] == "be-dev-1" + + async def test_give_up_does_not_fire_below_threshold(self) -> None: + """One failure below the give-up threshold: hold the park (no clear).""" + orch = _make_orchestrator() + provider = "anthropic" + state = _make_active_state(provider, retry_after=None) + + tracker_mock = _make_tracker_mock(failure_return=_PROBE_GIVE_UP_THRESHOLD - 1) + + with ( + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "resolve_wait", new=AsyncMock(return_value=None)), + patch.object(orch, "_do_probe", new=AsyncMock(return_value=False)), + patch.object(orch, "_notify_rate_limit_ceo", new=AsyncMock()), + patch("roboco.events.get_event_bus") as mock_bus_fn, + ): + bus_mock = AsyncMock() + bus_mock.publish = AsyncMock() + mock_bus_fn.return_value = bus_mock + await orch._probe_one_provider(provider, state) + + tracker_mock.clear.assert_not_awaited() # still parked + + async def test_give_up_publishes_rate_limit_lifted(self) -> None: + """Resuming via the escape hatch publishes RATE_LIMIT_LIFTED (the panel + + parked agents see the lift, not a silent resume).""" + orch = _make_orchestrator() + provider = "anthropic" + state = _make_active_state(provider, retry_after=None) + orch._waiting_records = {"be-dev-1": _waiting_record("be-dev-1", provider)} + + tracker_mock = _make_tracker_mock(failure_return=_PROBE_GIVE_UP_THRESHOLD) + published: list[Any] = [] + + with ( + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "resolve_wait", new=AsyncMock(return_value=None)), + patch.object(orch, "_do_probe", new=AsyncMock(return_value=False)), + patch.object(orch, "_notify_rate_limit_ceo", new=AsyncMock()), + patch("roboco.events.get_event_bus") as mock_bus_fn, + ): + bus_mock = AsyncMock() + bus_mock.publish = AsyncMock(side_effect=published.append) + mock_bus_fn.return_value = bus_mock + await orch._probe_one_provider(provider, state) + + assert any(e.type == EventType.RATE_LIMIT_LIFTED for e in published) + + async def test_give_up_clears_ceo_notified_flag_for_next_episode(self) -> None: + """The give-up ends the episode (park cleared), so the CEO-notified flag + is cleared — a fresh rate-limit episode later gets a fresh notification.""" + orch = _make_orchestrator() + provider = "anthropic" + state = _make_active_state(provider, retry_after=None) + orch._rate_limit_ceo_notified.add(provider) # prior episode notified + + tracker_mock = _make_tracker_mock(failure_return=_PROBE_GIVE_UP_THRESHOLD) + + with ( + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "resolve_wait", new=AsyncMock(return_value=None)), + patch.object(orch, "_do_probe", new=AsyncMock(return_value=False)), + patch.object(orch, "_notify_rate_limit_ceo", new=AsyncMock()), + patch("roboco.events.get_event_bus") as mock_bus_fn, + ): + bus_mock = AsyncMock() + bus_mock.publish = AsyncMock() + mock_bus_fn.return_value = bus_mock + await orch._probe_one_provider(provider, state) + + assert provider not in orch._rate_limit_ceo_notified + + async def test_ceo_notification_still_fires_on_the_give_up_sweep(self) -> None: + """The CEO is still notified on the give-up sweep (failure_count >= the + CEO threshold AND >= the give-up threshold on the same sweep) — the + escape hatch does not silence the operator signal.""" + assert _PROBE_GIVE_UP_THRESHOLD >= _CEO_NOTIFY_THRESHOLD # invariant + orch = _make_orchestrator() + provider = "anthropic" + state = _make_active_state(provider, retry_after=None) + + tracker_mock = _make_tracker_mock(failure_return=_PROBE_GIVE_UP_THRESHOLD) + notify_mock = AsyncMock() + + with ( + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "resolve_wait", new=AsyncMock(return_value=None)), + patch.object(orch, "_do_probe", new=AsyncMock(return_value=False)), + patch.object(orch, "_notify_rate_limit_ceo", new=notify_mock), + patch("roboco.events.get_event_bus") as mock_bus_fn, + ): + bus_mock = AsyncMock() + bus_mock.publish = AsyncMock() + mock_bus_fn.return_value = bus_mock + await orch._probe_one_provider(provider, state) + + notify_mock.assert_awaited_once() + + # --------------------------------------------------------------------------- # Tests: CEO notification threshold # ---------------------------------------------------------------------------