From c6c106e6d8e4b2f998fa946c4e2f13081ee7017f Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 04:29:50 +0800 Subject: [PATCH 1/4] fix: label every cooldown-fallback event with the leg it describes MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit `CooldownFallbackLLM`'s event payloads omitted `leg` on `retry`, `abandon_stream_retry` and `degrade`, and emitted no `request` event at all for the fallback leg. A tracing adapter therefore had to infer the leg, and the only available default ("fallback") mislabelled primary retries as fallback traffic; calls actually served by the fallback left no request record. Every event now carries `leg` — `retry` and `abandon_stream_retry` are primary-leg events, `degrade` is a fallback-leg event — and both the cooldown shortcut and the post-exhaustion path announce the fallback request they are about to make (`mode="cooldown"` / `"degraded"`). Stream-path degrade events gained the `degrade_from`/`degrade_to` pair and the `streaming` marker the chat path already had. The contract is now written down on the class. No behavioral change to the retry/cooldown state machine itself. Co-Authored-By: Claude Opus 5 (1M context) --- agent_core/providers/fallback.py | 58 +++++++++++++++- tests/test_cooldown_fallback.py | 111 +++++++++++++++++++++++++++++++ 2 files changed, 166 insertions(+), 3 deletions(-) diff --git a/agent_core/providers/fallback.py b/agent_core/providers/fallback.py index 6517458..d4839e0 100644 --- a/agent_core/providers/fallback.py +++ b/agent_core/providers/fallback.py @@ -206,6 +206,26 @@ class CooldownFallbackLLM: consumer, because a retry would emit those deltas twice. Set ``replay_partial_stream=True`` to restore the historical duplicating behavior; prefer :class:`LLMFallbackChain` for rewind-safe semantics. + + Event hook contract + ------------------- + Every event payload carries ``leg`` — ``"primary"`` or ``"fallback"`` — + naming the client the event is *about*, so a tracing adapter never has to + infer it. ``retry`` and ``abandon_stream_retry`` are primary-leg events: + they describe what happened to the primary, not the leg that serves next. + + Events emitted: + + - ``request`` — a leg is about to be called. ``attempt`` on the primary + leg; ``mode`` (``"cooldown"`` / ``"degraded"``) on the fallback leg. + - ``error`` — that leg's call raised. + - ``retry`` — the primary failed and will be retried after ``delay_s``. + - ``abandon_stream_retry`` — the primary stream failed after deltas had + already reached the consumer, so it degrades instead of replaying. + - ``degrade`` — traffic moves to the fallback leg (``leg="fallback"``); + always carries ``reason``, ``degrade_from`` and ``degrade_to``. + + Stream-path events additionally carry ``streaming=True``. """ def __init__( @@ -282,10 +302,12 @@ async def chat( if self._clock() < self._cooldown_until: await self._emit( "degrade", + leg="fallback", reason="primary_cooldown", degrade_from=_model_id(self.primary), degrade_to=_model_id(self.fallback), ) + await self._emit("request", leg="fallback", mode="cooldown") return await self.fallback.chat(messages, **kwargs) last_error: Exception | None = None @@ -308,17 +330,21 @@ async def chat( # emits a ``retry`` event that no retry follows. break delay = min(0.5 * (2**attempt), 8.0) + self._jitter() * 0.25 - await self._emit("retry", attempt=attempt + 1, delay_s=delay) + await self._emit( + "retry", leg="primary", attempt=attempt + 1, delay_s=delay, + ) await self._sleep(delay) self._cooldown_until = self._clock() + self.cooldown_seconds await self._emit( "degrade", + leg="fallback", reason="primary_exhausted", degrade_from=_model_id(self.primary), degrade_to=_model_id(self.fallback), cooldown_seconds=self.cooldown_seconds, ) + await self._emit("request", leg="fallback", mode="degraded") try: return await self.fallback.chat(messages, **kwargs) except Exception as fallback_error: @@ -345,7 +371,17 @@ async def stream( timeout=timeout, ) if self._clock() < self._cooldown_until: - await self._emit("degrade", reason="primary_stream_cooldown") + await self._emit( + "degrade", + leg="fallback", + reason="primary_stream_cooldown", + degrade_from=_model_id(self.primary), + degrade_to=_model_id(self.fallback), + streaming=True, + ) + await self._emit( + "request", leg="fallback", mode="cooldown", streaming=True, + ) async for delta in self.fallback.stream(messages, **kwargs): yield delta return @@ -380,6 +416,10 @@ async def stream( # would duplicate them. Degrade to the fallback leg instead. await self._emit( "abandon_stream_retry", + leg="primary", + reason="primary_stream_partial", + degrade_from=_model_id(self.primary), + degrade_to=_model_id(self.fallback), attempt=attempt + 1, streaming=True, yielded=True, @@ -390,6 +430,7 @@ async def stream( delay = min(0.5 * (2**attempt), 8.0) + self._jitter() * 0.25 await self._emit( "retry", + leg="primary", attempt=attempt + 1, delay_s=delay, streaming=True, @@ -398,7 +439,18 @@ async def stream( await self._sleep(delay) self._cooldown_until = self._clock() + self.cooldown_seconds - await self._emit("degrade", reason="primary_stream_exhausted") + await self._emit( + "degrade", + leg="fallback", + reason="primary_stream_exhausted", + degrade_from=_model_id(self.primary), + degrade_to=_model_id(self.fallback), + cooldown_seconds=self.cooldown_seconds, + streaming=True, + ) + await self._emit( + "request", leg="fallback", mode="degraded", streaming=True, + ) try: async for delta in self.fallback.stream(messages, **kwargs): yield delta diff --git a/tests/test_cooldown_fallback.py b/tests/test_cooldown_fallback.py index a254a0a..d016236 100644 --- a/tests/test_cooldown_fallback.py +++ b/tests/test_cooldown_fallback.py @@ -197,3 +197,114 @@ async def test_stream_replay_is_opt_in() -> None: seen = [delta.content async for delta in llm.stream([user_msg("x")])] assert seen == ["part1 ", "part2 ", "part1 ", "part2 ", "FALLBACK"] assert primary.starts == 2 + + +@pytest.mark.asyncio +async def test_every_event_names_the_leg_it_is_about() -> None: + """``leg`` is always present, and ``retry`` belongs to the primary. + + A tracing adapter labels each record by ``leg``; leaving it off any event + forced the adapter to guess, which mislabelled primary retries as fallback + traffic. + """ + primary = ScriptedLLM( + "primary-model", + [TimeoutError("timeout"), TimeoutError("timeout")], + ) + fallback = ScriptedLLM("fallback-model", [LLMResponse(content="ok")]) + events: list[tuple[str, dict[str, object]]] = [] + + async def hook(name: str, payload: dict[str, object]) -> None: + events.append((name, payload)) + + llm = CooldownFallbackLLM( + primary, + fallback, + max_retries=2, + cooldown_seconds=60, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + event_hook=hook, + ) + assert (await llm.chat([user_msg("x")])).content == "ok" + + assert all("leg" in payload for _name, payload in events), events + assert [(name, payload["leg"]) for name, payload in events] == [ + ("request", "primary"), + ("error", "primary"), + ("retry", "primary"), + ("request", "primary"), + ("error", "primary"), + ("degrade", "fallback"), + ("request", "fallback"), + ] + degrade = next(payload for name, payload in events if name == "degrade") + assert degrade["degrade_from"] == "primary-model" + assert degrade["degrade_to"] == "fallback-model" + + +@pytest.mark.asyncio +async def test_fallback_leg_call_is_traceable_in_cooldown() -> None: + """The cooldown shortcut still announces the fallback request it makes.""" + fallback = ScriptedLLM("fallback-model", [LLMResponse(content="ok")]) + events: list[tuple[str, dict[str, object]]] = [] + + async def hook(name: str, payload: dict[str, object]) -> None: + events.append((name, payload)) + + llm = CooldownFallbackLLM( + ScriptedLLM("primary-model", [TimeoutError("timeout")]), + fallback, + max_retries=1, + cooldown_seconds=60, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + event_hook=hook, + ) + await llm.chat([user_msg("one")]) # enters cooldown + events.clear() + await llm.chat([user_msg("two")]) # served from cooldown + + assert [(name, payload.get("mode")) for name, payload in events] == [ + ("degrade", None), + ("request", "cooldown"), + ] + assert fallback.calls == 2 + + +@pytest.mark.asyncio +async def test_stream_events_carry_leg_labels_and_degrade_pair() -> None: + """Stream-path events reach the same contract as the ``chat`` path.""" + events: list[tuple[str, dict[str, object]]] = [] + + async def hook(name: str, payload: dict[str, object]) -> None: + events.append((name, payload)) + + llm = CooldownFallbackLLM( + PartialStream(), + OkStream(), + max_retries=3, + cooldown_seconds=60, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + event_hook=hook, + ) + seen = [delta.content async for delta in llm.stream([user_msg("x")])] + assert seen == ["part1 ", "part2 ", "FALLBACK"] + + assert [(name, payload["leg"]) for name, payload in events] == [ + ("request", "primary"), + ("error", "primary"), + ("abandon_stream_retry", "primary"), + ("degrade", "fallback"), + ("request", "fallback"), + ] + assert all(payload.get("streaming") for _name, payload in events), events + for name in ("abandon_stream_retry", "degrade"): + payload = next(p for n, p in events if n == name) + assert payload["degrade_from"] == "primary" + assert payload["degrade_to"] == "fallback" + assert payload["reason"] From 1b886e75f21e93fa3cb17b76b1af10f6609b74a2 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 07:56:43 +0800 Subject: [PATCH 2/4] fix: log the fallback state machine's own transitions, unify its labels MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Four follow-ups found while two host products wrote tracing adapters over this class (drafted in ApodexHarness temp/2026-09-03_agentcore-cooldown-fallback-followups.md). - Degrades were invisible in logs. The class logged nothing about its own behaviour, so traffic moving to a weaker model for the whole cooldown window left no operational trace unless a host had wired a telemetry sink and re-implemented the log lines off `event_hook` — which both hosts had started to do, guaranteeing divergent lines for identical behaviour. Every event now also logs: retry / abandoned stream retry / degrade at warning, a failing fallback leg at error, a primary-leg error at debug (the line after it carries the signal). `event_hook` goes back to being purely structured telemetry. - `_model_id` was annotated `-> str` but returned whatever `model_name` / `model` held on a duck-typed client. It coerces now. - `model_name` is a property while `model` is a construction-time snapshot, so a swapped primary reports a stale `model`. That split is forced, not an oversight, and now says so: `LLMClient` declares `model` as a settable attribute and the concrete clients subclass the Protocol and assign it in `__init__`, so the slot is a real descriptor — making it a property here stops the class satisfying `LLMClient` under pyright, and making it a read-only member on the Protocol breaks every concrete client at runtime (both measured, hence the comment rather than a change). `model_name` carries no such constraint and is the label to read when the primary may be swapped. - `components.middleware.llm.base` kept a second copy of the retryable keyword table, with a comment pointing at a file in another repo that no longer has the thing it named. It re-binds to the one table now, which is public as `LEGACY_RETRYABLE_KEYWORDS`. `abandon_stream_retry` also drops the `degrade_from`/`degrade_to` pair added in c6c106e: it is a retry-shaped event, and the `degrade` that follows it is what describes the move — carrying the pair on both invited hosts to write two degrade records for one degradation. The exhausted `degrade` gained `error`, naming what exhausted the primary. Co-Authored-By: Claude Opus 5 (1M context) --- agent_core/components/middleware/llm/base.py | 40 ++------ agent_core/providers/fallback.py | 85 +++++++++++++++-- tests/test_cooldown_fallback.py | 99 +++++++++++++++++++- 3 files changed, 179 insertions(+), 45 deletions(-) diff --git a/agent_core/components/middleware/llm/base.py b/agent_core/components/middleware/llm/base.py index 9a40c63..e702fd1 100644 --- a/agent_core/components/middleware/llm/base.py +++ b/agent_core/components/middleware/llm/base.py @@ -10,6 +10,10 @@ from agent_core.llm import LLMResponse, StreamDelta from agent_core.messages import Message +from agent_core.providers.fallback import ( + LEGACY_RETRYABLE_KEYWORDS, + legacy_retryable, +) logger = logging.getLogger(__name__) @@ -31,38 +35,10 @@ def unwrap_runnable_binding( return model, kwargs -# Keywords that indicate a transient error worth retrying. -# Shared with FallbackLLM in llm_adapter.py — keep in sync. -_RETRYABLE_KEYWORDS = frozenset( - { - "timeout", - "timed out", - "429", - "500", - "502", - "503", - "504", - "529", - "overloaded", - "rate limit", - "rate_limit", - "server error", - "connection reset", - "connection error", - "econnreset", - "gateway timeout", - "model_dump", - "model_not_found", - } -) - - -def _is_retryable(error: Exception) -> bool: - """Return True if *error* looks transient and worth retrying.""" - if isinstance(error, AttributeError): - return True - msg = str(error).lower() - return any(kw in msg for kw in _RETRYABLE_KEYWORDS) +# One table, owned by the two-model fallback wrapper that also needs it. +# Both names stay bound here: host products re-export them from this module. +_RETRYABLE_KEYWORDS = LEGACY_RETRYABLE_KEYWORDS +_is_retryable = legacy_retryable # ── Context ────────────────────────────────────────────────────────────── diff --git a/agent_core/providers/fallback.py b/agent_core/providers/fallback.py index d4839e0..3065e90 100644 --- a/agent_core/providers/fallback.py +++ b/agent_core/providers/fallback.py @@ -66,6 +66,7 @@ logger = logging.getLogger(__name__) __all__ = [ + "LEGACY_RETRYABLE_KEYWORDS", "CooldownFallbackLLM", "FallbackEntry", "FallbackTrigger", @@ -76,7 +77,10 @@ type FallbackEventHook = Callable[[str, dict[str, Any]], Awaitable[None]] -_LEGACY_RETRYABLE_KEYWORDS = frozenset({ +# The single retryable-keyword table. ``components.middleware.llm.base`` +# re-binds its own ``_RETRYABLE_KEYWORDS`` to this rather than keeping a second +# copy — two tables for one policy drift. +LEGACY_RETRYABLE_KEYWORDS = frozenset({ "timeout", "timed out", "429", "500", "502", "503", "504", "529", "overloaded", "rate limit", "rate_limit", "server error", "connection reset", "connection error", "econnreset", "gateway timeout", @@ -87,7 +91,7 @@ def legacy_retryable(error: Exception) -> bool: """Match the historical two-model fallback wrapper's retry policy.""" return isinstance(error, AttributeError) or any( - keyword in str(error).lower() for keyword in _LEGACY_RETRYABLE_KEYWORDS + keyword in str(error).lower() for keyword in LEGACY_RETRYABLE_KEYWORDS ) @@ -138,8 +142,13 @@ def matches(self, exc: BaseException) -> bool: def _model_id(model: Any) -> str: - """Best-effort short label for a model. Used for telemetry.""" - return ( + """Best-effort short label for a model. Used for telemetry. + + ``model_name`` / ``model`` come off arbitrary duck-typed clients, so the + result is coerced: an enum, a pydantic field or a ``Mock`` must not flow + into ``self.model`` and every telemetry payload as a non-string. + """ + return str( getattr(model, "model_name", None) or getattr(model, "model", None) or type(model).__name__ @@ -221,11 +230,22 @@ class CooldownFallbackLLM: - ``error`` — that leg's call raised. - ``retry`` — the primary failed and will be retried after ``delay_s``. - ``abandon_stream_retry`` — the primary stream failed after deltas had - already reached the consumer, so it degrades instead of replaying. + already reached the consumer, so it degrades instead of replaying. A + retry-shaped event: the ``degrade`` that follows describes the move. - ``degrade`` — traffic moves to the fallback leg (``leg="fallback"``); always carries ``reason``, ``degrade_from`` and ``degrade_to``. Stream-path events additionally carry ``streaming=True``. + + Logging + ------- + Every event is also logged, independently of ``event_hook`` — a degrade + moves traffic to a different model for ``cooldown_seconds`` and an operator + has to be able to see that in logs whether or not a telemetry sink is + wired. Retries, abandoned stream retries and degrades log at ``warning``; a + failing fallback leg logs at ``error``; a primary-leg error logs at + ``debug``, since the ``retry`` or ``degrade`` line that follows it carries + the operational signal. Hosts should not re-log off the hook. """ def __init__( @@ -255,11 +275,56 @@ def __init__( self._replay_partial_stream = replay_partial_stream self._cooldown_until = 0.0 + # ``model_name`` is derived, ``model`` is a snapshot, and the split is + # forced: ``LLMClient`` declares ``model`` as a settable attribute (the + # concrete clients subclass the Protocol and assign it in ``__init__``, so + # it is a real descriptor slot), and a property there stops the class + # satisfying that. ``model_name`` carries no such constraint, so it reads + # through to the current primary — that is the label to use when a primary + # may be lazily initialised or swapped by middleware after construction. @property def model_name(self) -> str: return f"fallback({_model_id(self.primary)})" + def _log_event(self, name: str, payload: dict[str, Any]) -> None: + """Operational log line for one event. See the class docstring.""" + if name == "error": + if payload.get("leg") == "fallback": + logger.error( + "CooldownFallbackLLM: fallback leg also failed: %s", + payload.get("error"), + ) + else: + logger.debug( + "CooldownFallbackLLM: primary leg failed: %s", + payload.get("error"), + ) + elif name == "retry": + logger.warning( + "CooldownFallbackLLM: primary attempt %s/%s failed (%s), " + "retrying in %.1fs", + payload.get("attempt"), + self.max_retries, + payload.get("error_type"), + float(payload.get("delay_s") or 0.0), + ) + elif name == "abandon_stream_retry": + logger.warning( + "CooldownFallbackLLM: primary stream failed on attempt %s " + "after emitting deltas; degrading instead of replaying them", + payload.get("attempt"), + ) + elif name == "degrade": + logger.warning( + "CooldownFallbackLLM: degrading %s -> %s (%s), cooldown %ss", + payload.get("degrade_from"), + payload.get("degrade_to"), + payload.get("reason"), + payload.get("cooldown_seconds", self.cooldown_seconds), + ) + async def _emit(self, name: str, **payload: Any) -> None: + self._log_event(name, payload) try: await self._event_hook(name, payload) except Exception: @@ -331,7 +396,11 @@ async def chat( break delay = min(0.5 * (2**attempt), 8.0) + self._jitter() * 0.25 await self._emit( - "retry", leg="primary", attempt=attempt + 1, delay_s=delay, + "retry", + leg="primary", + attempt=attempt + 1, + delay_s=delay, + error_type=type(error).__name__, ) await self._sleep(delay) @@ -343,6 +412,7 @@ async def chat( degrade_from=_model_id(self.primary), degrade_to=_model_id(self.fallback), cooldown_seconds=self.cooldown_seconds, + error=str(last_error) if last_error else "", ) await self._emit("request", leg="fallback", mode="degraded") try: @@ -418,8 +488,6 @@ async def stream( "abandon_stream_retry", leg="primary", reason="primary_stream_partial", - degrade_from=_model_id(self.primary), - degrade_to=_model_id(self.fallback), attempt=attempt + 1, streaming=True, yielded=True, @@ -435,6 +503,7 @@ async def stream( delay_s=delay, streaming=True, yielded=yielded, + error_type=type(error).__name__, ) await self._sleep(delay) diff --git a/tests/test_cooldown_fallback.py b/tests/test_cooldown_fallback.py index d016236..b145186 100644 --- a/tests/test_cooldown_fallback.py +++ b/tests/test_cooldown_fallback.py @@ -303,8 +303,97 @@ async def hook(name: str, payload: dict[str, object]) -> None: ("request", "fallback"), ] assert all(payload.get("streaming") for _name, payload in events), events - for name in ("abandon_stream_retry", "degrade"): - payload = next(p for n, p in events if n == name) - assert payload["degrade_from"] == "primary" - assert payload["degrade_to"] == "fallback" - assert payload["reason"] + + # ``abandon_stream_retry`` is retry-shaped: it says why the primary was not + # retried. The ``degrade`` that follows is the one carrying the model pair, + # so a host mapping both to degrade records would write two per degradation. + abandon = next(p for n, p in events if n == "abandon_stream_retry") + assert abandon["reason"] == "primary_stream_partial" + assert "degrade_from" not in abandon + degrade = next(p for n, p in events if n == "degrade") + assert degrade["degrade_from"] == "primary" + assert degrade["degrade_to"] == "fallback" + + +@pytest.mark.asyncio +async def test_degrades_are_visible_in_logs_without_any_hook( + caplog: pytest.LogCaptureFixture, +) -> None: + """Logging is the always-wired channel: no ``event_hook``, still visible. + + An operator has to be able to see that traffic moved to another model for + the cooldown window, and a telemetry sink may not be registered at all. + """ + llm = CooldownFallbackLLM( + # The message has to match the retry policy for a retry to happen. + ScriptedLLM("primary-model", [TimeoutError("timed out")]), + ScriptedLLM("fallback-model", [LLMResponse(content="ok")]), + max_retries=2, + cooldown_seconds=30, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + ) + with caplog.at_level("DEBUG", logger="agent_core.providers.fallback"): + await llm.chat([user_msg("x")]) + + warnings = [r.getMessage() for r in caplog.records if r.levelname == "WARNING"] + assert any("retrying in 0.5s" in line and "TimeoutError" in line for line in warnings) + assert any( + "degrading primary-model -> fallback-model (primary_exhausted)" in line + and "cooldown 30s" in line + for line in warnings + ) + # The primary's own error stays at debug: the lines above carry the signal. + assert [r.levelname for r in caplog.records].count("ERROR") == 0 + assert any("primary leg failed: timed out" in r.getMessage() for r in caplog.records) + + +@pytest.mark.asyncio +async def test_failing_fallback_leg_logs_an_error( + caplog: pytest.LogCaptureFixture, +) -> None: + llm = CooldownFallbackLLM( + ScriptedLLM("primary-model", [TimeoutError("boom")]), + ScriptedLLM("fallback-model", [RuntimeError("fallback down")]), + max_retries=1, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + ) + with ( + caplog.at_level("ERROR", logger="agent_core.providers.fallback"), + pytest.raises(TimeoutError), + ): + await llm.chat([user_msg("x")]) + + assert any( + "fallback leg also failed: fallback down" in r.getMessage() + for r in caplog.records + ) + + +def test_model_name_follows_a_swapped_primary() -> None: + """``model_name`` reads through; ``model`` cannot (see its comment).""" + primary = ScriptedLLM("first", []) + llm = CooldownFallbackLLM(primary, ScriptedLLM("fallback", [])) + assert llm.model == "first" + assert llm.model_name == "fallback(first)" + + primary.model = "rotated" + assert llm.model_name == "fallback(rotated)" + assert llm.model == "first" # the settable protocol slot stays a snapshot + + +def test_model_label_is_always_a_string() -> None: + """A duck-typed client may hold a non-str model; telemetry needs a str.""" + class EnumishModel: + def __str__(self) -> str: + return "enum-model" + + class OddClient: + model = EnumishModel() + + llm = CooldownFallbackLLM(OddClient(), ScriptedLLM("fallback", [])) + assert llm.model == "enum-model" + assert isinstance(llm.model, str) From 0e2c5cfff622712502dad895f591b923bf9756f2 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 09:24:42 +0800 Subject: [PATCH 3/4] fix: close the cooldown leg's observability gaps in the fallback wrapper MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Review of the two preceding commits found the new event contract leaking in four places, all on the cooldown path — which carries the bulk of fallback traffic, since one primary failure routes every request for the next `cooldown_seconds` through it. - The cooldown shortcuts called the fallback unguarded, so a fallback that also fails during the window raised out of `CooldownFallbackLLM` with no `error` event, no error log, and a `request` record that never terminated. Both paths now wrap the call the way the exhausted paths do. A "fallback leg also failed" metric built on this contract was blind to precisely the case operators care about. - `degrade` fires once per call, not once per transition, so logging it at warning meant ~1200 WARNING lines per 60s window at 20 req/s, burying the one line marking the actual move. Only the two `exhausted` reasons — the transitions — log at warning now; the cooldown reasons log at debug. - The cooldown degrades carried no `cooldown_seconds`, so the log line fell back to the configured constant and claimed "cooldown 60s" 59 seconds in. They now carry both the configured window and `cooldown_remaining_s`, and the line reports what is left. - `error` was added to the chat exhausted degrade only. A host reading it off a degrade record would `KeyError` on any streaming or cooldown degrade. All four sites carry it now (`""` on the cooldown path), and the contract section documents the full payload. Also rename `abandon_stream_retry`'s `reason` to `abandon_reason`. The contract makes `reason` the discriminator of `degrade`, so a host dispatching on it wrote two degrade records for one stream degradation. Finally, move the retryable-keyword table into the leaf `agent_core.retry_policy`. De-duplicating it added a `components -> providers -> runtime` import edge for a frozenset, which any future `runtime -> components.middleware` import would have turned into a partially-initialised-module ImportError. Both sides import the leaf; both public names stay re-exported where they were. Co-Authored-By: Claude Opus 5 (1M context) --- agent_core/components/middleware/llm/base.py | 8 +- agent_core/providers/fallback.py | 109 +++++++---- agent_core/retry_policy.py | 28 +++ tests/test_cooldown_fallback.py | 179 ++++++++++++++++++- 4 files changed, 287 insertions(+), 37 deletions(-) create mode 100644 agent_core/retry_policy.py diff --git a/agent_core/components/middleware/llm/base.py b/agent_core/components/middleware/llm/base.py index e702fd1..776b139 100644 --- a/agent_core/components/middleware/llm/base.py +++ b/agent_core/components/middleware/llm/base.py @@ -10,10 +10,7 @@ from agent_core.llm import LLMResponse, StreamDelta from agent_core.messages import Message -from agent_core.providers.fallback import ( - LEGACY_RETRYABLE_KEYWORDS, - legacy_retryable, -) +from agent_core.retry_policy import LEGACY_RETRYABLE_KEYWORDS, legacy_retryable logger = logging.getLogger(__name__) @@ -35,7 +32,8 @@ def unwrap_runnable_binding( return model, kwargs -# One table, owned by the two-model fallback wrapper that also needs it. +# One table, defined in the leaf ``agent_core.retry_policy`` so this framework +# module does not import the whole ``providers`` package for a frozenset. # Both names stay bound here: host products re-export them from this module. _RETRYABLE_KEYWORDS = LEGACY_RETRYABLE_KEYWORDS _is_retryable = legacy_retryable diff --git a/agent_core/providers/fallback.py b/agent_core/providers/fallback.py index 3065e90..6c779b3 100644 --- a/agent_core/providers/fallback.py +++ b/agent_core/providers/fallback.py @@ -63,6 +63,11 @@ from agent_core.llm import LLMResponse, StreamDelta from agent_core.messages import Message +# Re-exported below: host products and ``components.middleware.llm.base`` +# import these from here. The definitions live in a leaf module so neither +# side has to import the other's package for them — see that module. +from agent_core.retry_policy import LEGACY_RETRYABLE_KEYWORDS, legacy_retryable + logger = logging.getLogger(__name__) __all__ = [ @@ -77,24 +82,13 @@ type FallbackEventHook = Callable[[str, dict[str, Any]], Awaitable[None]] -# The single retryable-keyword table. ``components.middleware.llm.base`` -# re-binds its own ``_RETRYABLE_KEYWORDS`` to this rather than keeping a second -# copy — two tables for one policy drift. -LEGACY_RETRYABLE_KEYWORDS = frozenset({ - "timeout", "timed out", "429", "500", "502", "503", "504", "529", - "overloaded", "rate limit", "rate_limit", "server error", - "connection reset", "connection error", "econnreset", "gateway timeout", - "model_dump", "model_not_found", +# The two ``degrade`` reasons that mark an actual transition to the fallback +# leg, as opposed to the cooldown window re-routing another call. +_DEGRADE_TRANSITIONS = frozenset({ + "primary_exhausted", "primary_stream_exhausted", }) -def legacy_retryable(error: Exception) -> bool: - """Match the historical two-model fallback wrapper's retry policy.""" - return isinstance(error, AttributeError) or any( - keyword in str(error).lower() for keyword in LEGACY_RETRYABLE_KEYWORDS - ) - - async def _noop_event(_name: str, _payload: dict[str, Any]) -> None: return None @@ -231,9 +225,20 @@ class CooldownFallbackLLM: - ``retry`` — the primary failed and will be retried after ``delay_s``. - ``abandon_stream_retry`` — the primary stream failed after deltas had already reached the consumer, so it degrades instead of replaying. A - retry-shaped event: the ``degrade`` that follows describes the move. - - ``degrade`` — traffic moves to the fallback leg (``leg="fallback"``); - always carries ``reason``, ``degrade_from`` and ``degrade_to``. + retry-shaped event: the ``degrade`` that follows describes the move. Its + "why" is ``abandon_reason``, deliberately not ``reason`` — ``reason`` + belongs to ``degrade``, and a host dispatching on it would otherwise + write two degrade records for one degradation. + - ``degrade`` — traffic moves to the fallback leg (``leg="fallback"``). + Always carries ``reason``, ``degrade_from``, ``degrade_to``, + ``cooldown_seconds`` (the configured window) and ``error`` (the primary + failure that forced it; ``""`` when the cooldown window, not a fresh + failure, is what routed this call). The ``primary_cooldown`` / + ``primary_stream_cooldown`` reasons additionally carry + ``cooldown_remaining_s``. + + ``degrade`` fires once per *call* served by the fallback, not once per + transition: expect one per request for the length of the cooldown window. Stream-path events additionally carry ``streaming=True``. @@ -242,10 +247,13 @@ class CooldownFallbackLLM: Every event is also logged, independently of ``event_hook`` — a degrade moves traffic to a different model for ``cooldown_seconds`` and an operator has to be able to see that in logs whether or not a telemetry sink is - wired. Retries, abandoned stream retries and degrades log at ``warning``; a - failing fallback leg logs at ``error``; a primary-leg error logs at - ``debug``, since the ``retry`` or ``degrade`` line that follows it carries - the operational signal. Hosts should not re-log off the hook. + wired. Retries, abandoned stream retries and the two *exhausted* degrades — + the transitions — log at ``warning``; a failing fallback leg logs at + ``error``. The cooldown degrades log at ``debug``: they repeat once per + request for the whole window, so at warning they would bury the one line + that marks the transition. A primary-leg error also logs at ``debug``, + since the ``retry`` or ``degrade`` line that follows it carries the + operational signal. Hosts should not re-log off the hook. """ def __init__( @@ -315,12 +323,23 @@ def _log_event(self, name: str, payload: dict[str, Any]) -> None: payload.get("attempt"), ) elif name == "degrade": - logger.warning( + reason = payload.get("reason") + # A transition is news; the cooldown window repeating itself once + # per request is not. See the class docstring. + log = ( + logger.warning + if reason in _DEGRADE_TRANSITIONS + else logger.debug + ) + remaining = payload.get("cooldown_remaining_s") + log( "CooldownFallbackLLM: degrading %s -> %s (%s), cooldown %ss", payload.get("degrade_from"), payload.get("degrade_to"), - payload.get("reason"), - payload.get("cooldown_seconds", self.cooldown_seconds), + reason, + payload.get("cooldown_seconds", self.cooldown_seconds) + if remaining is None + else f"{remaining:.0f}s remaining of {self.cooldown_seconds}", ) async def _emit(self, name: str, **payload: Any) -> None: @@ -364,16 +383,30 @@ async def chat( extra_headers=extra_headers, timeout=timeout, ) - if self._clock() < self._cooldown_until: + now = self._clock() + if now < self._cooldown_until: await self._emit( "degrade", leg="fallback", reason="primary_cooldown", degrade_from=_model_id(self.primary), degrade_to=_model_id(self.fallback), + cooldown_seconds=self.cooldown_seconds, + cooldown_remaining_s=self._cooldown_until - now, + error="", ) await self._emit("request", leg="fallback", mode="cooldown") - return await self.fallback.chat(messages, **kwargs) + # Guarded like the degraded path below: cooldown-window traffic is + # the bulk of fallback traffic, so "the fallback leg also failed" + # has to be visible here or the metric under-counts the case that + # matters most. + try: + return await self.fallback.chat(messages, **kwargs) + except Exception as fallback_error: + await self._emit( + "error", leg="fallback", error=str(fallback_error), + ) + raise last_error: Exception | None = None for attempt in range(self.max_retries): @@ -440,20 +473,33 @@ async def stream( extra_headers=extra_headers, timeout=timeout, ) - if self._clock() < self._cooldown_until: + now = self._clock() + if now < self._cooldown_until: await self._emit( "degrade", leg="fallback", reason="primary_stream_cooldown", degrade_from=_model_id(self.primary), degrade_to=_model_id(self.fallback), + cooldown_seconds=self.cooldown_seconds, + cooldown_remaining_s=self._cooldown_until - now, + error="", streaming=True, ) await self._emit( "request", leg="fallback", mode="cooldown", streaming=True, ) - async for delta in self.fallback.stream(messages, **kwargs): - yield delta + try: + async for delta in self.fallback.stream(messages, **kwargs): + yield delta + except Exception as fallback_error: + await self._emit( + "error", + leg="fallback", + streaming=True, + error=str(fallback_error), + ) + raise return last_error: Exception | None = None @@ -487,7 +533,7 @@ async def stream( await self._emit( "abandon_stream_retry", leg="primary", - reason="primary_stream_partial", + abandon_reason="primary_stream_partial", attempt=attempt + 1, streaming=True, yielded=True, @@ -515,6 +561,7 @@ async def stream( degrade_from=_model_id(self.primary), degrade_to=_model_id(self.fallback), cooldown_seconds=self.cooldown_seconds, + error=str(last_error) if last_error else "", streaming=True, ) await self._emit( diff --git a/agent_core/retry_policy.py b/agent_core/retry_policy.py new file mode 100644 index 0000000..46a59b1 --- /dev/null +++ b/agent_core/retry_policy.py @@ -0,0 +1,28 @@ +"""Retryable-error policy shared by provider wrappers and LLM middleware. + +A leaf module on purpose. Both :mod:`agent_core.providers.fallback` and +:mod:`agent_core.components.middleware.llm.base` need this one table, and +having either import the other's package to get it would add a +``components -> providers -> runtime`` import edge for the sake of a +frozenset — an edge that turns into a partially-initialised-module +``ImportError`` the day anything under ``runtime/`` imports the middleware. +Two copies of one policy drift, so the table lives here and both import it. +""" + +from __future__ import annotations + +__all__ = ["LEGACY_RETRYABLE_KEYWORDS", "legacy_retryable"] + +LEGACY_RETRYABLE_KEYWORDS = frozenset({ + "timeout", "timed out", "429", "500", "502", "503", "504", "529", + "overloaded", "rate limit", "rate_limit", "server error", + "connection reset", "connection error", "econnreset", "gateway timeout", + "model_dump", "model_not_found", +}) + + +def legacy_retryable(error: Exception) -> bool: + """Match the historical two-model fallback wrapper's retry policy.""" + return isinstance(error, AttributeError) or any( + keyword in str(error).lower() for keyword in LEGACY_RETRYABLE_KEYWORDS + ) diff --git a/tests/test_cooldown_fallback.py b/tests/test_cooldown_fallback.py index b145186..8f8164c 100644 --- a/tests/test_cooldown_fallback.py +++ b/tests/test_cooldown_fallback.py @@ -272,6 +272,14 @@ async def hook(name: str, payload: dict[str, object]) -> None: ("request", "cooldown"), ] assert fallback.calls == 2 + # Same keys as the transition degrade, so a host adapter reads one shape. + degrade = next(payload for name, payload in events if name == "degrade") + assert degrade["reason"] == "primary_cooldown" + assert degrade["degrade_from"] == "primary-model" + assert degrade["degrade_to"] == "fallback-model" + assert degrade["cooldown_seconds"] == 60 + assert degrade["cooldown_remaining_s"] == 60.0 + assert degrade["error"] == "" @pytest.mark.asyncio @@ -307,8 +315,10 @@ async def hook(name: str, payload: dict[str, object]) -> None: # ``abandon_stream_retry`` is retry-shaped: it says why the primary was not # retried. The ``degrade`` that follows is the one carrying the model pair, # so a host mapping both to degrade records would write two per degradation. + # Hence ``abandon_reason``: ``reason`` is degrade's, and nothing else. abandon = next(p for n, p in events if n == "abandon_stream_retry") - assert abandon["reason"] == "primary_stream_partial" + assert abandon["abandon_reason"] == "primary_stream_partial" + assert "reason" not in abandon assert "degrade_from" not in abandon degrade = next(p for n, p in events if n == "degrade") assert degrade["degrade_from"] == "primary" @@ -397,3 +407,170 @@ class OddClient: llm = CooldownFallbackLLM(OddClient(), ScriptedLLM("fallback", [])) assert llm.model == "enum-model" assert isinstance(llm.model, str) + + +@pytest.mark.asyncio +async def test_every_degrade_carries_the_documented_keys() -> None: + """One payload shape across all four degrade sites, ``chat`` and ``stream``. + + A host adapter reading ``payload["error"]`` off a degrade record must not + hit a ``KeyError`` because the degradation happened to be a streaming one. + """ + events: list[tuple[str, dict[str, object]]] = [] + + async def hook(name: str, payload: dict[str, object]) -> None: + events.append((name, payload)) + + def make(primary: object, fallback: object) -> CooldownFallbackLLM: + return CooldownFallbackLLM( + primary, + fallback, + max_retries=1, + cooldown_seconds=60, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + event_hook=hook, + ) + + chat = make( + ScriptedLLM("primary-model", [TimeoutError("timed out")]), + ScriptedLLM("fallback-model", [LLMResponse(content="ok")]), + ) + await chat.chat([user_msg("one")]) # primary_exhausted + await chat.chat([user_msg("two")]) # primary_cooldown + + stream = make(PartialStream(), OkStream()) + async for _delta in stream.stream([user_msg("x")]): # primary_stream_* + pass + async for _delta in stream.stream([user_msg("y")]): # ..._stream_cooldown + pass + + degrades = [payload for name, payload in events if name == "degrade"] + assert [d["reason"] for d in degrades] == [ + "primary_exhausted", + "primary_cooldown", + "primary_stream_exhausted", + "primary_stream_cooldown", + ] + for degrade in degrades: + assert degrade["leg"] == "fallback" + assert degrade["degrade_from"] and degrade["degrade_to"] + assert degrade["cooldown_seconds"] == 60 + assert isinstance(degrade["error"], str) + assert degrades[0]["error"] == "timed out" + assert degrades[1]["error"] == "" + + +@pytest.mark.asyncio +async def test_cooldown_degrade_logs_at_debug_not_warning( + caplog: pytest.LogCaptureFixture, +) -> None: + """The transition is a warning; the window repeating itself is not. + + At 20 req/s a 60 s cooldown would otherwise emit ~1200 WARNING lines and + bury the one line that marks the actual move to the other model. + """ + llm = CooldownFallbackLLM( + ScriptedLLM("primary-model", [TimeoutError("timed out")]), + ScriptedLLM("fallback-model", [LLMResponse(content="ok")]), + max_retries=1, + cooldown_seconds=60, + clock=iter([0.0, 0.0, 1.0]).__next__, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + ) + with caplog.at_level("DEBUG", logger="agent_core.providers.fallback"): + await llm.chat([user_msg("one")]) # the transition + caplog.clear() + await llm.chat([user_msg("two")]) # served from the cooldown window + + assert [r.levelname for r in caplog.records] == ["DEBUG"] + # ..and it reports what is left of the window, not the configured constant. + assert "59s remaining of 60" in caplog.records[0].getMessage() + + +@pytest.mark.asyncio +async def test_failing_fallback_leg_in_cooldown_is_reported( + caplog: pytest.LogCaptureFixture, +) -> None: + """Cooldown-window traffic is most fallback traffic; its failures count. + + Without this the whole cooldown window can fail silently: no ``error`` + event, no error log, and a ``request`` record that never terminates. + """ + events: list[tuple[str, dict[str, object]]] = [] + + async def hook(name: str, payload: dict[str, object]) -> None: + events.append((name, payload)) + + llm = CooldownFallbackLLM( + ScriptedLLM("primary-model", [TimeoutError("timed out")]), + ScriptedLLM( + "fallback-model", + [LLMResponse(content="ok"), RuntimeError("fallback down")], + ), + max_retries=1, + cooldown_seconds=60, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + event_hook=hook, + ) + await llm.chat([user_msg("one")]) # enters cooldown + events.clear() + with ( + caplog.at_level("ERROR", logger="agent_core.providers.fallback"), + pytest.raises(RuntimeError, match="fallback down"), + ): + await llm.chat([user_msg("two")]) + + assert [(name, payload["leg"]) for name, payload in events] == [ + ("degrade", "fallback"), + ("request", "fallback"), + ("error", "fallback"), + ] + assert any( + "fallback leg also failed: fallback down" in r.getMessage() + for r in caplog.records + ) + + +@pytest.mark.asyncio +async def test_failing_fallback_stream_in_cooldown_is_reported() -> None: + events: list[tuple[str, dict[str, object]]] = [] + + async def hook(name: str, payload: dict[str, object]) -> None: + events.append((name, payload)) + + class DownStream: + model = "fallback-model" + + async def stream( + self, + _messages: list[Message], + **_kwargs: object, + ) -> AsyncIterator[StreamDelta]: + raise RuntimeError("fallback down") + yield StreamDelta(content="") # pragma: no cover + + llm = CooldownFallbackLLM( + ScriptedLLM("primary-model", [TimeoutError("timed out")]), + DownStream(), + max_retries=1, + cooldown_seconds=60, + clock=lambda: 0.0, + sleep=lambda _d: _completed(), + jitter=lambda: 0.0, + event_hook=hook, + ) + llm._cooldown_until = 60.0 # already in the window + with pytest.raises(RuntimeError, match="fallback down"): + async for _delta in llm.stream([user_msg("x")]): + pass + + assert [(name, payload["leg"]) for name, payload in events] == [ + ("degrade", "fallback"), + ("request", "fallback"), + ("error", "fallback"), + ] From 214ced6b89a4bf4c10ee8950f7c15f0fda9f21c9 Mon Sep 17 00:00:00 2001 From: zhanghanduo Date: Thu, 3 Sep 2026 09:46:12 +0800 Subject: [PATCH 4/4] fix: clarify fallback logging contract --- agent_core/providers/fallback.py | 21 +++++++++++---------- agent_core/retry_policy.py | 3 ++- 2 files changed, 13 insertions(+), 11 deletions(-) diff --git a/agent_core/providers/fallback.py b/agent_core/providers/fallback.py index 6c779b3..183c387 100644 --- a/agent_core/providers/fallback.py +++ b/agent_core/providers/fallback.py @@ -244,16 +244,17 @@ class CooldownFallbackLLM: Logging ------- - Every event is also logged, independently of ``event_hook`` — a degrade - moves traffic to a different model for ``cooldown_seconds`` and an operator - has to be able to see that in logs whether or not a telemetry sink is - wired. Retries, abandoned stream retries and the two *exhausted* degrades — - the transitions — log at ``warning``; a failing fallback leg logs at - ``error``. The cooldown degrades log at ``debug``: they repeat once per - request for the whole window, so at warning they would bury the one line - that marks the transition. A primary-leg error also logs at ``debug``, - since the ``retry`` or ``degrade`` line that follows it carries the - operational signal. Hosts should not re-log off the hook. + Retry, abandonment, degradation and error events are also logged, + independently of ``event_hook`` — a degrade moves traffic to a different + model for ``cooldown_seconds`` and an operator has to be able to see that + in logs whether or not a telemetry sink is wired. ``request`` events remain + hook-only to avoid per-call log noise. Retries, abandoned stream retries + and the two *exhausted* degrades — the transitions — log at ``warning``; a + failing fallback leg logs at ``error``. The cooldown degrades log at + ``debug``: they repeat once per request for the whole window, so at warning + they would bury the one line that marks the transition. A primary-leg error + also logs at ``debug``, since the ``retry`` or ``degrade`` line that follows + it carries the operational signal. Hosts should not re-log off the hook. """ def __init__( diff --git a/agent_core/retry_policy.py b/agent_core/retry_policy.py index 46a59b1..17e71c6 100644 --- a/agent_core/retry_policy.py +++ b/agent_core/retry_policy.py @@ -23,6 +23,7 @@ def legacy_retryable(error: Exception) -> bool: """Match the historical two-model fallback wrapper's retry policy.""" + message = str(error).lower() return isinstance(error, AttributeError) or any( - keyword in str(error).lower() for keyword in LEGACY_RETRYABLE_KEYWORDS + keyword in message for keyword in LEGACY_RETRYABLE_KEYWORDS )