diff --git a/agent_core/components/middleware/llm/base.py b/agent_core/components/middleware/llm/base.py index 9a40c63..776b139 100644 --- a/agent_core/components/middleware/llm/base.py +++ b/agent_core/components/middleware/llm/base.py @@ -10,6 +10,7 @@ from agent_core.llm import LLMResponse, StreamDelta from agent_core.messages import Message +from agent_core.retry_policy import LEGACY_RETRYABLE_KEYWORDS, legacy_retryable logger = logging.getLogger(__name__) @@ -31,38 +32,11 @@ 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, 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 # ── Context ────────────────────────────────────────────────────────────── diff --git a/agent_core/providers/fallback.py b/agent_core/providers/fallback.py index 6517458..183c387 100644 --- a/agent_core/providers/fallback.py +++ b/agent_core/providers/fallback.py @@ -63,9 +63,15 @@ 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__ = [ + "LEGACY_RETRYABLE_KEYWORDS", "CooldownFallbackLLM", "FallbackEntry", "FallbackTrigger", @@ -76,21 +82,13 @@ type FallbackEventHook = Callable[[str, dict[str, Any]], Awaitable[None]] -_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 @@ -138,8 +136,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__ @@ -206,6 +209,52 @@ 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. A + 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``. + + Logging + ------- + 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__( @@ -235,11 +284,67 @@ 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": + 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"), + 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: + self._log_event(name, payload) try: await self._event_hook(name, payload) except Exception: @@ -279,14 +384,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="", ) - return await self.fallback.chat(messages, **kwargs) + await self._emit("request", leg="fallback", mode="cooldown") + # 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): @@ -308,17 +429,26 @@ 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, + error_type=type(error).__name__, + ) 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, + error=str(last_error) if last_error else "", ) + await self._emit("request", leg="fallback", mode="degraded") try: return await self.fallback.chat(messages, **kwargs) except Exception as fallback_error: @@ -344,10 +474,33 @@ async def stream( extra_headers=extra_headers, timeout=timeout, ) - if self._clock() < self._cooldown_until: - await self._emit("degrade", reason="primary_stream_cooldown") - async for delta in self.fallback.stream(messages, **kwargs): - yield delta + 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, + ) + 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 @@ -380,6 +533,8 @@ async def stream( # would duplicate them. Degrade to the fallback leg instead. await self._emit( "abandon_stream_retry", + leg="primary", + abandon_reason="primary_stream_partial", attempt=attempt + 1, streaming=True, yielded=True, @@ -390,15 +545,29 @@ 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, yielded=yielded, + error_type=type(error).__name__, ) 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, + error=str(last_error) if last_error else "", + 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/agent_core/retry_policy.py b/agent_core/retry_policy.py new file mode 100644 index 0000000..17e71c6 --- /dev/null +++ b/agent_core/retry_policy.py @@ -0,0 +1,29 @@ +"""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.""" + message = str(error).lower() + return isinstance(error, AttributeError) or any( + keyword in message for keyword in LEGACY_RETRYABLE_KEYWORDS + ) diff --git a/tests/test_cooldown_fallback.py b/tests/test_cooldown_fallback.py index a254a0a..8f8164c 100644 --- a/tests/test_cooldown_fallback.py +++ b/tests/test_cooldown_fallback.py @@ -197,3 +197,380 @@ 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 + # 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 +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 + + # ``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["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" + 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) + + +@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"), + ]