From a6de1330aa6a13aca0c1f25cafec35fe1751d59b Mon Sep 17 00:00:00 2001 From: Renn F Date: Sun, 28 Jun 2026 11:36:47 +0200 Subject: [PATCH] [F045] rate-limit: loud activate-failure log + in-memory orphan-probe fallback MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The in-verb i_am_blocked(rate_limited) path wrapped RateLimitStateTracker.activate in a bare contextlib.suppress. A silent activate failure stranded the fleet: agents were parked in _waiting_records but the provider never entered the tracker, so the tracker-driven _sweep_rate_limit_probes never probed it and no _on_probe_success ever resumed them — parked agents stuck in WAITING_LONG. Fix: (1) replace the bare suppress with a try/except that logs an error event naming the provider + affected agents; (2) in _sweep_rate_limit_probes, after probing the tracker-listed set, scan _waiting_records for any rate_limit_lifted provider the loop did NOT cover and probe it via the time-expiry fallback (empty state -> probe now) so _on_probe_success resumes the parked agents. The fallback reads only local memory, so it still resumes when Redis was down at park time (list_rate_limited_providers failure now falls through to the orphan scan instead of returning early). --- roboco/runtime/orchestrator.py | 37 +++++++- .../services/gateway/choreographer/_impl.py | 21 ++++- .../gateway/test_i_am_blocked_rate_limited.py | 29 +++++++ tests/unit/runtime/test_rate_limit_sweep.py | 87 +++++++++++++++++++ 4 files changed, 172 insertions(+), 2 deletions(-) diff --git a/roboco/runtime/orchestrator.py b/roboco/runtime/orchestrator.py index 35bb167d..3bdcec94 100644 --- a/roboco/runtime/orchestrator.py +++ b/roboco/runtime/orchestrator.py @@ -6337,6 +6337,17 @@ Start by: - **Failure**: increment probe_failures; if the count reaches 10 and we haven't already sent a CEO notification for this episode, send one now. + + F045: the loop is tracker-driven, but an ``activate()`` failure in the + in-verb ``i_am_blocked(rate_limited)`` path (or a Redis hiccup) can + leave agents parked in ``_waiting_records`` for a provider the tracker + never learned about — so the tracker-listed loop above never probes it + and the parked agents strand in WAITING_LONG forever. After probing the + tracker-listed set, scan the in-memory records for any + ``rate_limit_lifted`` provider the loop did NOT cover and probe it via + the time-expiry fallback (empty state → ``_too_early_to_probe`` returns + False → probe now) so ``_on_probe_success`` can resume them. The + fallback reads only local memory, so it works even when Redis is down. """ from roboco.services.gateway.rate_limit_tracker import RateLimitStateTracker @@ -6344,9 +6355,11 @@ Start by: providers = await RateLimitStateTracker.list_rate_limited_providers() except Exception as e: logger.warning("Failed to list rate-limited providers", error=str(e)) - return + providers = [] + probed_providers: set[str] = set() for provider, state in providers: + probed_providers.add(provider) try: await self._probe_one_provider(provider, state) except Exception as e: @@ -6356,6 +6369,28 @@ Start by: error=str(e), ) + # F045 orphan fallback: resume agents parked for a provider the + # tracker-listed loop above did not cover (activate failed silently or + # Redis was down at park time). Empty state => probe immediately; on + # success ``_on_probe_success`` clears the tracker (self-healing) and + # resumes the parked agents. + orphan_providers: set[str] = set() + for record in self._waiting_records.values(): + if record.waiting_for != "rate_limit_lifted": + continue + prov = record.context.get("provider") + if prov and prov not in probed_providers: + orphan_providers.add(prov) + for provider in orphan_providers: + try: + await self._probe_one_provider(provider, {}) + except Exception as e: + logger.error( + "Unhandled error probing orphaned rate-limited provider", + provider=provider, + error=str(e), + ) + def _make_tracker(self, provider: str) -> Any: """Return a RateLimitStateTracker for *provider*. diff --git a/roboco/services/gateway/choreographer/_impl.py b/roboco/services/gateway/choreographer/_impl.py index d5db00b1..679b9ac6 100644 --- a/roboco/services/gateway/choreographer/_impl.py +++ b/roboco/services/gateway/choreographer/_impl.py @@ -2941,8 +2941,19 @@ class Choreographer: # calls can gate new spawns for this provider. Skipped when the # provider is "unknown" (orchestrator not wired or not tracking the # agent) to avoid polluting the tracker with meaningless keys. + # + # F045: an activate() failure is logged loudly, NOT bare-suppressed. + # The probe-resume loop is tracker-driven — it iterates + # ``list_rate_limited_providers()`` — so a silent activate failure here + # leaves every parked agent in ``_waiting_records`` with a provider the + # tracker never learned about, and no probe ever runs to resume them + # (the stranded-fleet blind spot). The orchestrator's + # ``_sweep_rate_limit_probes`` has an in-memory ``_waiting_records`` + # fallback that resumes them when the provider recovers even without + # tracker state; this error log makes the condition visible to + # operators either way. if provider != "unknown": - with contextlib.suppress(Exception): + try: from roboco.services.gateway.rate_limit_tracker import ( RateLimitStateTracker, ) @@ -2951,6 +2962,14 @@ class Choreographer: retry_after=retry_after_seconds, affected_agents=affected_agents, ) + except Exception as exc: + logger.error( + "rate_limit_tracker.activate failed; parked agents rely on" + " the in-memory probe fallback to resume", + provider=provider, + affected_agents=affected_agents, + error=str(exc), + ) bus = self.stream_bus if bus is not None: diff --git a/tests/unit/gateway/test_i_am_blocked_rate_limited.py b/tests/unit/gateway/test_i_am_blocked_rate_limited.py index e77e873d..0e7b9360 100644 --- a/tests/unit/gateway/test_i_am_blocked_rate_limited.py +++ b/tests/unit/gateway/test_i_am_blocked_rate_limited.py @@ -20,6 +20,7 @@ from uuid import uuid4 from roboco.models.events import EventType from roboco.services.gateway.choreographer import Choreographer, ChoreographerDeps +from structlog.testing import capture_logs # --------------------------------------------------------------------------- # Helpers @@ -505,3 +506,31 @@ class TestRateLimitTrackerActivateOnParking: assert env.error is None assert env.status == "in_progress" + + async def test_activate_failure_is_logged_not_silent(self) -> None: + """F045: an activate() failure must be logged loudly, not bare-suppressed. + + The probe-resume loop is tracker-driven, so a silent activate failure + strands every parked agent in WAITING_LONG with no probe ever running. + A loud error log makes the stranded-fleet condition visible to + operators (and pairs with the orchestrator's in-memory fallback sweep). + """ + agent_id = uuid4() + task_id = uuid4() + orch = _make_orchestrator(active_agents=["be-dev-1"], provider=_PROVIDER) + deps = _make_deps(agent_id, task_id, orchestrator=orch) + c = Choreographer(deps) + + mock_tracker = AsyncMock() + mock_tracker.activate = AsyncMock(side_effect=RuntimeError("redis down")) + mock_tracker_cls = MagicMock(return_value=mock_tracker) + + with patch(_TRACKER_PATCH, mock_tracker_cls), capture_logs() as logs: + env = await c.i_am_blocked(agent_id, task_id, "rate_limited") + + assert env.error is None + assert any( + "activate" in str(e.get("event", "")).lower() + and e.get("log_level") == "error" + for e in logs + ), f"expected an error log about activate failure; got {logs!r}" diff --git a/tests/unit/runtime/test_rate_limit_sweep.py b/tests/unit/runtime/test_rate_limit_sweep.py index d86eb52f..558ee8c0 100644 --- a/tests/unit/runtime/test_rate_limit_sweep.py +++ b/tests/unit/runtime/test_rate_limit_sweep.py @@ -414,6 +414,93 @@ class TestCEONotificationThreshold: notify_mock.assert_awaited_once() +# --------------------------------------------------------------------------- +# Tests: orphan-provider fallback (F045) +# --------------------------------------------------------------------------- + + +class TestOrphanProviderFallback: + """F045: an activate() failure in the in-verb ``i_am_blocked(rate_limited)`` + path leaves agents parked in ``_waiting_records`` but the provider never + makes it into the tracker — so the tracker-driven loop never probes it and + the parked agents strand in WAITING_LONG forever. The sweep must scan the + in-memory records for any ``rate_limit_lifted`` provider the tracker-listed + set did NOT cover and probe it via the time-expiry fallback so + ``_on_probe_success`` can resume them. + """ + + async def test_orphan_parked_agent_resumed_when_tracker_lacks_provider( + self, + ) -> None: + orch = _make_orchestrator() + provider = "anthropic" + agent = "be-dev-1" + orch._waiting_records = {agent: _waiting_record(agent, provider)} + + tracker_mock = _make_tracker_mock() + resolve_mock = AsyncMock(return_value=None) + + with ( + patch.object(orch, "resolve_wait", new=resolve_mock), + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "_do_probe", new=AsyncMock(return_value=True)), + patch.object( + RateLimitStateTracker, + "list_rate_limited_providers", + new=AsyncMock(return_value=[]), + ), + 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._sweep_rate_limit_probes() + + # The orphan provider was probed and the parked agent resumed. + assert resolve_mock.await_count == 1 + assert resolve_mock.call_args.args[0] == agent + + async def test_orphan_skipped_when_tracker_already_covers_provider( + self, + ) -> None: + """A provider the tracker lists must NOT be double-probed via the fallback.""" + orch = _make_orchestrator() + provider = "anthropic" + agent = "be-dev-1" + orch._waiting_records = {agent: _waiting_record(agent, provider)} + + tracker_mock = _make_tracker_mock() + resolve_mock = AsyncMock(return_value=None) + state = _make_active_state(provider, retry_after=None) + + probe_calls: list[str] = [] + + async def fake_do_probe(p: str) -> bool: + probe_calls.append(p) + return True + + with ( + patch.object(orch, "resolve_wait", new=resolve_mock), + patch.object(orch, "_make_tracker", return_value=tracker_mock), + patch.object(orch, "_do_probe", new=fake_do_probe), + patch.object( + RateLimitStateTracker, + "list_rate_limited_providers", + new=AsyncMock(return_value=[(provider, state)]), + ), + 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._sweep_rate_limit_probes() + + # Probed exactly once (via the tracker-listed path), not twice. + assert probe_calls == [provider] + + # --------------------------------------------------------------------------- # Tests: list_rate_limited_providers # ---------------------------------------------------------------------------