fix(cache_stability): key marks per model, hedge cache-expiry cause

Address Douwe's review on #327:

- A run-global high-water mark warned on every legitimate mid-run model
  switch, because the new provider's cache is empty. Key the mark per
  (provider_name, model_name) -- the truthful source, since FallbackModel
  resolves the inner model inside request() -- so a failover / per-step
  switch starts a fresh mark. Keep marks per key (not reset) so a switch
  back within the old model's TTL still compares to its prefix.
- "The cacheable prefix moved" overclaimed: a cache_read collapse is also a
  provider-side cache expiry under a byte-identical prefix (Anthropic's 5m
  default TTL, refreshed on each hit). Hedge the message and name the
  inter-request gap when it exceeds cache_ttl_seconds (new message-only
  field, default 300s) to avoid mis-attributing a long pause to a moved prefix.
- collapse_ratio doc said "lower toward 1.0"; raising is what sharpens it.
- README: warnings don't reach Logfire on their own (it bridges stdlib
  logging, not warnings) -- point users at logging.captureWarnings(True).

pydantic_ai_harness/experimental/cache_stability/README.md

@@ -12,10 +12,24 @@ that collapse visible.
 This is the **observe** signal: it reads the provider's own verdict rather than
 guessing from the structured request. On each response it reads
 `usage.cache_read_tokens` and tracks the largest cacheable prefix the run has
-established (`cache_read_tokens + cache_write_tokens`, a high-water mark). Because
-message history is append-only, a stable prefix means each request reads back at
-least what the previous one cached; a large drop is the observable signature of a
-bust, whatever the cause.
+established (`cache_read_tokens + cache_write_tokens`, a high-water mark), keyed by
+the response's `(provider_name, model_name)`. Because message history is
+append-only, a stable prefix means each request for that model reads back at least
+what the previous one cached; a large drop is the observable signature of a
+collapse.
+
+Keying per provider and model means a mid-run model switch does not warn: a
+`FallbackModel` failover or a per-step model change hits a different provider's
+cache, which is empty, so it starts its own high-water mark. Marks are kept per key
+rather than reset, so switching back to an earlier model within its cache TTL still
+compares against that model's prefix.
+
+A collapse has two shapes the monitor cannot tell apart, so the warning names both:
+the cacheable prefix moved, or the provider's cache expired under an unchanged prefix
+(a gap between requests longer than the cache TTL -- Anthropic's default is 5 minutes,
+refreshed on each hit). When the gap since the previous request exceeds
+`cache_ttl_seconds`, the message reports the gap so a long tool or approval pause
+isn't mistaken for a moved prefix.
 
 The verdict is cross-provider for free -- pyai normalizes every provider into the
 `cache_read_tokens` / `cache_write_tokens` fields on `RequestUsage`.
@@ -48,6 +62,10 @@ belongs at the wire level in tests, not here.
 - `min_prefix_tokens` (default `1024`): only judge collapse once the established
   prefix reaches this many tokens. Below a provider's minimum cacheable size
   (Anthropic's is 1024) `cache_read_tokens` is noisy or zero.
+- `cache_ttl_seconds` (default `300`): the assumed provider cache TTL. Message-only
+  -- when the gap since the previous request exceeds it, the warning notes the
+  collapse may be a cache expiry rather than a moved prefix. It does not change
+  whether a warning fires. Lower it for providers with a shorter cache lifetime.
 
 ## Silencing and escalation
 
@@ -74,6 +92,21 @@ In tests, assert an intentional bust with `pytest.warns(CacheBustWarning)`, or
 silence a legitimately-busting test with
 `@pytest.mark.filterwarnings('ignore::pydantic_ai_harness.experimental.cache_stability.CacheBustWarning')`.
 
+## Logfire
+
+Logfire bridges the stdlib `logging` module, not the `warnings` module, so a
+`CacheBustWarning` does not reach your traces on its own. To route busts into
+Logfire, redirect Python warnings to the `logging` system once at startup:
+
+```python
+import logging
+
+logging.captureWarnings(True)  # warnings.warn(...) -> the 'py.warnings' logger -> Logfire
+```
+
+This experimental version emits a warning only; a dedicated Logfire event may be
+added when the capability graduates.
+
 ## Composition
 
 - The monitor only implements `for_run` and `after_model_request`; it adds no
@@ -85,7 +118,12 @@ silence a legitimately-busting test with
 
 ## Scope
 
-- **Observational only.** It reports that a cached prefix collapsed, not why. The
-  structural explanation ("what moved the prefix this turn") is a separate job.
+- **Observational only.** It reports that a cached prefix collapsed, not why -- a
+  moved prefix and a provider-side cache expiry look the same from the token counts,
+  so the warning names both. The structural explanation ("what moved the prefix this
+  turn") is a separate job.
 - **Fires only when caching is enabled and reported.** A run that never establishes
   a cache never warns.
+- **A mid-run model switch does not warn.** Marks are per `(provider_name,
+  model_name)`, so a `FallbackModel` failover starts a fresh mark rather than
+  collapsing the previous model's.

pydantic_ai_harness/experimental/cache_stability/_capability.py

@@ -17,12 +17,16 @@ from __future__ import annotations
 
 import warnings
 from dataclasses import dataclass, field, replace
+from datetime import datetime
 
 from pydantic_ai.capabilities import AbstractCapability
 from pydantic_ai.messages import ModelResponse
 from pydantic_ai.models import ModelRequestContext
 from pydantic_ai.tools import AgentDepsT, RunContext
 
+# A response's (provider_name, model_name) -- identifies which provider cache the tokens came from.
+_CacheKey = tuple[str | None, str | None]
+
 _SILENCE_HINT = (
     '    import warnings\n'
     '    from pydantic_ai_harness.experimental.cache_stability import CacheBustWarning\n'
@@ -34,9 +38,11 @@ _SILENCE_HINT = (
 class CacheBustWarning(UserWarning):
     """Warned when a previously-established prompt cache hit collapses on a later request.
 
-    Emitted by `CacheStabilityMonitor` when this run read back far fewer cached tokens
-    than a prior request in the same run established, i.e. the cacheable prefix moved and
-    the provider re-charged tokens it could have served from cache.
+    Emitted by `CacheStabilityMonitor` when this run read back far fewer cached tokens for the
+    same provider and model than a prior request established. The likely causes are a moved
+    cacheable prefix (reordered tools, injected timestamps, a serialization-level block hop) or a
+    provider-side cache expiry under an unchanged prefix (a gap between requests longer than the
+    cache TTL). The monitor observes the collapse; it does not attribute the cause.
 
     Silence it, or escalate it to an error in dev/CI, with the stdlib `warnings` machinery
     (no bespoke API):
@@ -67,13 +73,21 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
 
     Attach it to any agent whose model uses prompt caching. On each response the monitor
     reads `usage.cache_read_tokens` and tracks the largest cacheable prefix the run has
-    established so far (`cache_read_tokens + cache_write_tokens`, a high-water mark). When a
-    later request reads back fewer than `collapse_ratio` of that established prefix, it emits
-    a `CacheBustWarning` -- the prefix moved and the provider re-charged those tokens.
+    established so far (`cache_read_tokens + cache_write_tokens`, a high-water mark), keyed by
+    the response's `(provider_name, model_name)`. When a later request for the same key reads
+    back fewer than `collapse_ratio` of that established prefix, it emits a `CacheBustWarning`.
+
+    Keying per provider and model means a mid-run model switch does not warn: a `FallbackModel`
+    failover or a per-step model change hits a different provider's cache, which is empty, so it
+    starts a fresh high-water mark instead of collapsing an existing one. Marks are kept per key
+    rather than reset, so switching back to an earlier model within its cache TTL still compares
+    against that model's established prefix.
 
     Because message history is append-only, a stable prefix means each request reads back at
-    least what the previous one cached. A large drop is the observable signature of a bust,
-    whatever the cause (reordered tools, injected timestamps, a serialization-level block hop).
+    least what the previous one cached. A large drop is the observable signature of a collapse,
+    whether the cause is a moved prefix (reordered tools, injected timestamps, a
+    serialization-level block hop) or a provider-side cache expiry when the gap between requests
+    exceeds the cache TTL. The monitor surfaces the collapse; it does not attribute the cause.
 
     ```python
     from pydantic_ai import Agent
@@ -92,7 +106,7 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
     """Warn when a request reads back less than this fraction of the established prefix.
 
     Conservative by default (0.5): only a drop below half the previously-cached prefix counts
-    as a collapse, so ordinary provider rounding or a partial cache miss does not fire. Lower
+    as a collapse, so ordinary provider rounding or a partial cache miss does not fire. Raise
     it toward 1.0 to warn on smaller regressions.
     """
 
@@ -103,12 +117,21 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
     noisy or zero, so small prefixes are ignored to avoid false positives.
     """
 
-    _established: int = field(default=0, compare=False)
+    cache_ttl_seconds: float = 300.0
+    """Assumed provider cache TTL, in seconds (Anthropic's default is 300, refreshed on each hit).
+
+    Message-only: when the gap since the previous request in the run exceeds this, the warning
+    notes that the collapse may be a provider-side cache expiry rather than a moved prefix. It
+    does not change whether a warning fires. Lower it for providers with a shorter cache lifetime.
+    """
+
+    _marks: dict[_CacheKey, int] = field(default_factory=dict[_CacheKey, int], compare=False)
     _step: int = field(default=0, compare=False)
+    _last_time: datetime | None = field(default=None, compare=False)
 
     async def for_run(self, ctx: RunContext[AgentDepsT]) -> AbstractCapability[AgentDepsT]:
-        """Reset the per-run high-water mark so each `Agent.run` is judged independently."""
-        return replace(self, _established=0, _step=0)
+        """Reset per-run state (per-key high-water marks, step, timing) so each run is judged alone."""
+        return replace(self, _marks={}, _step=0, _last_time=None)
 
     async def after_model_request(
         self,
@@ -117,20 +140,31 @@ class CacheStabilityMonitor(AbstractCapability[AgentDepsT]):
         request_context: ModelRequestContext,
         response: ModelResponse,
     ) -> ModelResponse:
-        """Compare this response's cache read against the run's established prefix, then update it."""
+        """Compare this response's cache read against the established prefix for its model, then update it."""
         self._step += 1
         usage = response.usage
         read = usage.cache_read_tokens
-        established = self._established
+        key = (response.provider_name, response.model_name)
+        established = self._marks.get(key, 0)
+        # 0.0 on the run's first request, which has no predecessor to measure a gap against.
+        gap = 0.0 if self._last_time is None else (response.timestamp - self._last_time).total_seconds()
+        self._last_time = response.timestamp
         if established >= self.min_prefix_tokens and read < established * self.collapse_ratio:
             wasted = established - read
+            if gap > self.cache_ttl_seconds:
+                expiry = (
+                    f' -- the previous request was ~{gap:.0f}s earlier, '
+                    f'past the assumed ~{self.cache_ttl_seconds:.0f}s cache TTL'
+                )
+            else:
+                expiry = ' (e.g. a gap longer than the cache TTL)'
             warnings.warn(
                 f'Cache hit collapsed at model request {self._step}: read {read} cached tokens but '
                 f'a prior request established ~{established} (~{wasted} tokens re-sent uncached). '
-                f'The cacheable prefix moved between requests.\n\n'
+                f"The cacheable prefix moved between requests, or the provider's cache expired{expiry}.\n\n"
                 f'To silence or escalate:\n\n{_SILENCE_HINT}\n',
                 CacheBustWarning,
                 stacklevel=2,
             )
-        self._established = max(established, read + usage.cache_write_tokens)
+        self._marks[key] = max(established, read + usage.cache_write_tokens)
         return response

tests/experimental/cache_stability/test_cache_stability.py

@@ -11,10 +11,13 @@ that explicitly.
 from __future__ import annotations
 
 import warnings
+from datetime import datetime, timedelta, timezone
 
 import pytest
 from pydantic_ai import Agent
+from pydantic_ai.exceptions import ModelAPIError
 from pydantic_ai.messages import ModelMessage, ModelResponse, TextPart, ToolCallPart
+from pydantic_ai.models.fallback import FallbackModel
 from pydantic_ai.models.function import AgentInfo, FunctionModel
 from pydantic_ai.usage import RequestUsage
 
@@ -58,6 +61,26 @@ def _agent(usages: list[RequestUsage], monitor: CacheStabilityMonitor[None]) ->
     return Agent(FunctionModel(fn), deps_type=type(None), capabilities=[monitor], tools=[noop])
 
 
+def _agent_from_responses(responses: list[ModelResponse], monitor: CacheStabilityMonitor[None]) -> Agent[None, str]:
+    """Agent whose model replays preset `ModelResponse`s, one per step.
+
+    Lets a test control `provider_name` and `timestamp` per response (a model switch, a wall-clock
+    gap) -- fields `FunctionModel` leaves untouched -- which the simpler `_agent` helper can't.
+    Every response but the last must carry a tool call so the run keeps stepping.
+    """
+    state = {'i': 0}
+
+    def fn(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        i = state['i']
+        state['i'] += 1
+        return responses[i]
+
+    def noop() -> str:
+        return 'ok'
+
+    return Agent(FunctionModel(fn), deps_type=type(None), capabilities=[monitor], tools=[noop])
+
+
 async def test_collapse_warns() -> None:
     """A large drop in cache_read below the established prefix warns."""
     usages = [_usage(read=0, write=8000), _usage(read=8000, write=200), _usage(read=500)]
@@ -119,3 +142,77 @@ async def test_for_run_resets_between_runs() -> None:
     with warnings.catch_warnings():
         warnings.simplefilter('error', CacheBustWarning)
         await silent.run('second')
+
+
+async def test_model_failover_does_not_warn() -> None:
+    """A mid-run `FallbackModel` failover reads an empty cache on the new model, which must not warn."""
+    a_calls = {'n': 0}
+
+    def model_a(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        a_calls['n'] += 1
+        if a_calls['n'] == 1:
+            # Establish a large cached prefix on model A, then keep the run stepping.
+            return ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000))
+        raise ModelAPIError('model-a', 'model A is down')
+
+    def model_b(messages: list[ModelMessage], info: AgentInfo) -> ModelResponse:
+        # B's cache is empty: it reads back nothing of A's prefix.
+        return ModelResponse(parts=[TextPart('done')], usage=_usage(read=0))
+
+    def noop() -> str:
+        return 'ok'
+
+    fallback = FallbackModel(
+        FunctionModel(model_a, model_name='model-a'),
+        FunctionModel(model_b, model_name='model-b'),
+    )
+    agent = Agent(fallback, deps_type=type(None), capabilities=[CacheStabilityMonitor()], tools=[noop])
+    with warnings.catch_warnings():
+        warnings.simplefilter('error', CacheBustWarning)
+        result = await agent.run('hi')
+    assert result.output == 'done'
+
+
+async def test_switch_back_within_ttl_uses_preserved_mark() -> None:
+    """Marks are kept per model, so a collapse after switching back to an earlier model still warns.
+
+    A reset-on-switch design would have discarded model A's mark at the switch to B, so the return
+    to A would compare against nothing and stay silent. The warning proves the mark survived.
+    """
+    responses = [
+        ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), provider_name='anthropic'),
+        ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), provider_name='openai'),
+        ModelResponse(parts=[TextPart('done')], usage=_usage(read=100), provider_name='anthropic'),
+    ]
+    agent = _agent_from_responses(responses, CacheStabilityMonitor())
+    with pytest.warns(CacheBustWarning, match='request 3'):
+        result = await agent.run('hi')
+    assert result.output == 'done'
+
+
+async def test_expiry_gap_named_when_beyond_ttl() -> None:
+    """A collapse after a gap longer than the assumed TTL names the gap, avoiding mis-attribution."""
+    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    responses = [
+        ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), timestamp=base),
+        ModelResponse(parts=[TextPart('done')], usage=_usage(read=100), timestamp=base + timedelta(seconds=400)),
+    ]
+    agent = _agent_from_responses(responses, CacheStabilityMonitor())
+    with pytest.warns(CacheBustWarning, match='past the assumed') as record:
+        await agent.run('hi')
+    assert '400s earlier' in str(record[0].message)
+
+
+async def test_small_gap_keeps_generic_expiry_hedge() -> None:
+    """A collapse with a short inter-request gap keeps the generic TTL hedge, not a concrete gap."""
+    base = datetime(2026, 1, 1, tzinfo=timezone.utc)
+    responses = [
+        ModelResponse(parts=[ToolCallPart('noop', {})], usage=_usage(read=0, write=8000), timestamp=base),
+        ModelResponse(parts=[TextPart('done')], usage=_usage(read=100), timestamp=base + timedelta(seconds=5)),
+    ]
+    agent = _agent_from_responses(responses, CacheStabilityMonitor())
+    with pytest.warns(CacheBustWarning) as record:
+        await agent.run('hi')
+    message = str(record[0].message)
+    assert 'e.g. a gap longer than the cache TTL' in message
+    assert 'past the assumed' not in message