Warn On Cache Busts

Warn when a conversation’s prompt cache hit collapses between model requests, within a run or across the runs that continue it, so a moved cacheable prefix or an expired provider cache surfaces instead of quietly re-charging tokens it could have served from cache.

While Pydantic AI Harness is on 0.x releases, the API may change between minor releases; when it does, deprecation warnings and release-note migration guidance tell you (or your agent) exactly how to upgrade. See the version policy.

What it watches

Prompt caching pays off only while the cacheable prefix (tools, then system instructions, then message history) stays byte-stable across a run’s consecutive requests. When something moves that prefix — reordered tools, a timestamp injected into instructions, a serialization-level block hop — the provider re-charges tokens it could have served from cache.

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 conversation has 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.

When a request reads back less than collapse_ratio of the established prefix, the monitor emits a CacheBustWarning once and latches that key, staying quiet about the collapse until a healthy read-back re-stabilizes the cache. A sustained collapse — caching toggled off mid-run (read == 0, write == 0), or a prefix that moves every request so the provider keeps writing a cache nothing reads back — therefore warns once, not on every request.

from pydantic_ai import Agent
from pydantic_ai_harness import WarnOnCacheBusts

agent = Agent('anthropic:claude-sonnet-4-5', capabilities=[WarnOnCacheBusts()])
result = await agent.run('...')  # a CacheBustWarning fires if a cached prefix collapses mid-run
# ...and on the next turn, if the prefix the first turn cached no longer reads back:
await agent.run('...', message_history=result.all_messages())

The verdict is cross-provider for free — pyai normalizes every provider into the cache_read_tokens / cache_write_tokens fields on RequestUsage.

Model switches and expiry

Keying per provider and model means a mid-run model switch does not warn: a FallbackModel failover or a per-step model change uses a different cache key, so the monitor starts a fresh mark for it instead of comparing against the previous model’s. 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 same model’s 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 gap is timed per model, so switching away and back measures the returning model’s own idle time, not whatever ran in between.

Conversations

Marks are kept per conversation (RunContext.conversation_id), not per run. A run that continues an earlier one via message_history — including history that was serialized and loaded back, which carries the conversation id with it — is judged against the prefix the earlier run established, so the first request of the next turn is checked against what the previous turn cached. That is where a moved prefix most often hides: history rewritten between turns, or a tool or instruction that differs from one turn to the next. A run that starts a new conversation (no history, or conversation_id='new') starts from a clean mark.

A conversation idle for longer than cache_ttl_seconds is forgotten: the provider cache has expired by then, so a low read-back at the start of its next run is an expiry, not a bust, and would only be noise. The warning says whether the mark it compared against came from this run or from an earlier run of the conversation.

Options

• collapse_ratio (default 0.5): warn when a request reads back less than this fraction of the established prefix. Conservative by default so ordinary rounding or a partial miss does not fire; raise toward 1.0 to warn on smaller regressions. It must be greater than 0.0 — a ratio of 0.0 could never warn, so it is rejected rather than treated as a silent disable switch.
• 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. Within a run it is message-only — when the gap since the same model’s previous request exceeds it, the warning notes the collapse may be a cache expiry rather than a moved prefix, without changing whether it fires. Between runs it bounds memory: a conversation idle for longer than this is forgotten, so its next run starts from a clean mark instead of warning about a cache the provider has already dropped. Lower it for providers with a shorter cache lifetime; raise it when the model is configured for a longer one (e.g. Anthropic’s 1-hour cache).

Silencing and escalation

There is no bespoke suppression API. Use the stdlib warnings machinery, exactly as you would manage any other UserWarning:

import warnings
from pydantic_ai_harness.warn_on_cache_busts import CacheBustWarning

# Silence the whole category:
warnings.filterwarnings('ignore', category=CacheBustWarning)

# Silence one intentional bust, scoped to the operation that causes it:
with warnings.catch_warnings():
    warnings.simplefilter('ignore', CacheBustWarning)
    result = agent.run_sync('...')  # e.g. a step that switches models or adds a file

# Treat every bust as an error (dev/CI enforcement):
warnings.filterwarnings('error', category=CacheBustWarning)

In tests, assert an intentional bust with pytest.warns(CacheBustWarning), or silence a legitimately-busting test with @pytest.mark.filterwarnings('ignore::pydantic_ai_harness.warn_on_cache_busts.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:

import logging

logging.captureWarnings(True)  # warnings.warn(...) -> the 'py.warnings' logger -> Logfire

The monitor’s signal is the CacheBustWarning; routing it through logging is how it reaches Logfire.

Composition

• The monitor only implements for_run and after_model_request; it adds no tools, instructions, or model settings, so it composes with any other capability, toolset, or ToolSearch setup without interference.
• The marks live on the WarnOnCacheBusts instance the agent was built with, keyed by conversation; for_run binds each run to its conversation’s marks. Reuse one instance across many Agent.run calls: runs of the same conversation share a mark, runs of different conversations are judged apart.

Scope

• 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.
• Marks are in-process memory. They are held on the capability instance, so a conversation continued through a different instance, a different process, or a different worker starts from a clean mark; nothing is persisted or shared.

API reference

• pydantic_ai_harness.warn_on_cache_busts source
• Pydantic AI capabilities
• Pydantic AI hooks

The public module exports WarnOnCacheBusts and CacheBustWarning. Import them from pydantic_ai_harness.warn_on_cache_busts.

WarnOnCacheBusts

Bases: AbstractCapability[AgentDepsT]

Warn when a conversation’s prompt cache hit collapses between requests.

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 conversation has established (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 once and then stays quiet about that collapse until a healthy read-back re-stabilizes the cache, so a sustained collapse warns once rather than on every subsequent request.

Marks are kept per conversation (RunContext.conversation_id), not per run, so a run that continues an earlier one via message_history — including history that was serialized and loaded back, which carries the conversation id with it — is judged against the prefix the earlier run established. That is where a moved prefix most often hides: the first request of the next turn re-sends what the previous turn cached. A run that starts a new conversation (no history, or conversation_id='new') starts from a clean mark. Marks are forgotten once a conversation has been idle for longer than cache_ttl_seconds: by then the provider cache has expired too, so a low read-back at the start of the next run is an expiry, not a bust, and remembering the conversation would only cost memory and a false warning.

Keying per provider and model means a mid-run model switch does not warn: a FallbackModel failover or a per-step model change uses a different cache key, so it starts a fresh mark for that key instead of comparing against the previous model’s. 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 — and the expiry hedge measures the gap against that same model’s previous request, not whatever ran in between.

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 collapse, whether the cause is a moved prefix (reordered tools, injected timestamps, a serialization-level block hop, history rewritten between turns) 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.

from pydantic_ai import Agent
from pydantic_ai_harness.warn_on_cache_busts import WarnOnCacheBusts

agent = Agent('anthropic:claude-sonnet-4-5', capabilities=[WarnOnCacheBusts()])
result = await agent.run('...')  # a CacheBustWarning fires if a cached prefix collapses mid-run
# ...and on the next turn, if the prefix the first turn cached no longer reads back:
await agent.run('...', message_history=result.all_messages())

The monitor is silent when caching is off or unreported (cache_read_tokens stays 0), so it never fires spuriously in tests that don’t exercise caching. Silencing and dev/CI escalation both go through the stdlib warnings filters — see CacheBustWarning.

Attributes

collapse_ratio

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. Raise it toward 1.0 to warn on smaller regressions. Must be greater than 0.0 (a ratio of 0.0 could never warn, so it is rejected rather than treated as a silent disable switch).

Type: float Default: 0.5

min_prefix_tokens

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, so small prefixes are ignored to avoid false positives.

Type: int Default: 1024

cache_ttl_seconds

Assumed provider cache TTL, in seconds (Anthropic’s default is 300, refreshed on each hit).

Two uses. Within a run it is message-only: when the gap since the previous request for the same model exceeds this, the warning notes that the collapse may be a provider-side cache expiry rather than a moved prefix, without changing whether it fires. Between runs it bounds memory: a conversation idle for longer than this is forgotten, so its next run starts from a clean mark instead of warning about a cache the provider has already dropped. Lower it for providers with a shorter cache lifetime; raise it when the model is configured for a longer one (e.g. Anthropic’s 1-hour cache).

Type: float Default: 300.0

Methods

for_run

@async

def for_run(ctx: RunContext[AgentDepsT]) -> AbstractCapability[AgentDepsT]

Bind this run to its conversation’s marks, forgetting conversations whose cache has expired.

The marks live on the instance the agent was built with, so every run of a conversation that goes through it — in this process — shares them. A run without a conversation id gets private marks and is judged alone.

Returns

AbstractCapability[AgentDepsT]

after_model_request

@async

def after_model_request(
    ctx: RunContext[AgentDepsT],
    *,
    request_context: ModelRequestContext,
    response: ModelResponse,
) -> ModelResponse

Compare this response’s cache read against the established prefix for its model, then update it.

Returns

ModelResponse

CacheBustWarning

Bases: UserWarning

Warned when a previously-established prompt cache hit collapses on a later request.

Emitted by WarnOnCacheBusts when a request read back far fewer cached tokens for the same provider and model than a prior request in the same conversation established — whether that prior request was earlier in this run or in an earlier run continued via message_history. The likely causes are a moved cacheable prefix (reordered tools, injected timestamps, a serialization-level block hop, history rewritten between turns) 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):

import warnings from pydantic_ai_harness.warn_on_cache_busts import CacheBustWarning

Silence the whole category:

warnings.filterwarnings(‘ignore’, category=CacheBustWarning)

Silence one intentional bust, scoped to the operation that causes it:

with warnings.catch_warnings(): warnings.simplefilter(‘ignore’, CacheBustWarning) result = agent.run_sync(’…’) # e.g. a step that switches models or adds a file

Treat every bust as an error (dev/CI enforcement):

warnings.filterwarnings(‘error’, category=CacheBustWarning)

In tests, assert an intentional bust with pytest.warns(CacheBustWarning), or silence a legitimately-busting test with @pytest.mark.filterwarnings('ignore::pydantic_ai_harness.warn_on_cache_busts.CacheBustWarning').