mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
## Description A low prompt-cache hit rate is hard to act on without knowing *why* turns miss. Two very different causes need very different responses: - **TTL lapse** — the session went idle longer than the provider's cache lifetime, so the entry expired. The fix is a longer TTL (e.g. Anthropic's 1h breakpoint instead of the 5m default). - **Prefix change** — the cacheable message prefix shifted, so the new request couldn't match the cached key. A longer TTL won't help here at all. Right now those look identical from the dashboard (just "cache_read was 0"). This adds the attribution so a user can actually decide 5m vs 1h. Closes #1313 ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made `PrefixCacheTracker` already kept the previous turn's forwarded messages and a per-turn activity timestamp, so the signal was already there — it just wasn't being read. - **`prefix_tracker.py`** — `classify_cache_miss()`: when a turn expected a cached prefix (non-zero cached tokens last turn) but read 0 this turn, returns `ttl_expiry` if the idle gap exceeded the provider cache TTL, else `prefix_change` if the forwarded prefix differs from last turn's, else `unknown`. **TTL wins ties** — once the entry lapsed, a coincident content change is moot, and the 5m-vs-1h decision is exactly what the TTL signal answers. A 1h-breakpoint session can widen the window via `PrefixFreezeConfig.cache_ttl_seconds`. Cold starts and hits return `is_miss=False`. - **Anthropic handlers (streaming + non-streaming)** — classify BEFORE `update_from_response` overwrites the last-turn state the classifier reads, then record the reason. - **`prometheus_metrics.py`** — a per-provider/per-reason counter, `record_cache_miss_attribution()`, reset handling, and a `headroom_cache_miss_attribution_total{provider,reason}` export series. - **`cost.py`** — `build_prefix_cache_stats()` aggregates a `miss_attribution` block (per-provider + totals, with the ttl/prefix split as a % of *attributed* misses, so `unknown` doesn't dilute the headline). - **dashboard** — a "Cache Miss Attribution" panel (TTL expiry / prefix change / unknown / total) with a "mostly TTL lapse" vs "mostly prefix change" headline. Scoped to Anthropic for this first cut (where the tracker is fully wired); OpenAI/Gemini can follow once the shape is proven. ## Testing - [x] Unit tests pass (`pytest`) - [ ] Linting passes (`ruff check .`) - [ ] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [ ] Manual testing performed ### Test Output ```text $ python -m pytest tests/test_cache/test_prefix_tracker.py -q 38 passed # 29 existing + 9 new classifier tests (TestClassifyCacheMiss). $ python -m pytest tests/test_proxy_cache_ttl_metrics.py -k "miss_attribution or reset_runtime_clears" -q 5 passed, 8 deselected # new: counter bucketing, stats aggregation, empty case, /metrics export, reset. ``` The full `test_proxy_cache_ttl_metrics.py` / `test_proxy_dashboard_stats_cache.py` files have some failures in this sandbox (`test_stats_endpoint_*`, streaming-parser, reset-counters) — those spin up the proxy server / Rust `_core` extension, which isn't built here. I confirmed via `git stash` that they fail identically on `main` without my changes, so they're pre-existing and unrelated. My additions to the stats dict are purely additive and don't break any passing assertion. ## Real Behavior Proof - Environment: Windows 11, Python 3.10. The Rust `_core` extension and a live proxy aren't available in this checkout. - Exact command / steps: drove `classify_cache_miss()` through every branch with a faithful warm-then-miss sequence; drove `record_cache_miss_attribution()` → `build_prefix_cache_stats()` → `export()` end to end. - Observed result: classifier returns `cold_start`/`hit`/`ttl_expiry`/`prefix_change`/`unknown` correctly, TTL wins the tie when both signals fire, a growing (append-only) prefix is treated as stable, and the 1h override widens the window. The stats builder produces `miss_attribution.totals` (`ttl_expiry`/`prefix_change`/`unknown`/`total` + `ttl_expiry_pct`/`prefix_change_pct` over attributed misses) and `by_provider`; `/metrics` emits `headroom_cache_miss_attribution_total{provider="anthropic",reason="ttl_expiry"}`. - Not tested: a live Anthropic session through the running proxy with a real idle-then-resume to confirm the handler wiring fires end-to-end. I verified the handler integration by reading scope/order (classify before `update_from_response`, `provider_name`/`self.metrics` in scope) and unit-tested every layer it calls, but didn't exercise the actual server loop. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [ ] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Additional Notes - The classifier is intentionally pure (takes the cache-read result + current forwarded messages + an optional idle override) so it's order-independent and unit-testable without a live tracker clock. - No README/docs change yet — this surfaces in the dashboard and `/metrics`, which are self-describing; happy to add a docs page if you'd like one. - CHANGELOG.md isn't touched — release-please generates it from the `feat(cache):` commit subject. - Follow-ups if useful: extend to OpenAI/Gemini handlers, and add a per-provider breakdown row in the dashboard panel (the stats already carry `by_provider`).
This commit is contained in:
Lakshya Sharma
GitHub
parent
530318b425
commit
4658721ea0
8 changed files with 505 additions and 0 deletions
156
headroom/cache/prefix_tracker.py
vendored
156
headroom/cache/prefix_tracker.py
vendored
|
|
@ -41,6 +41,21 @@ _PROVIDER_WRITE_PENALTY = {
|
|||
"bedrock": 0.25,
|
||||
}
|
||||
|
||||
# Default prompt-cache lifetime per provider, in seconds. Used by
|
||||
# `classify_cache_miss` to decide whether a miss is most likely a TTL
|
||||
# lapse (idle longer than this) versus a prefix-content change. Anthropic's
|
||||
# default ephemeral cache is 5 minutes (matches
|
||||
# headroom.cache.anthropic.ANTHROPIC_CACHE_TTL_SECONDS); the others are best-
|
||||
# effort defaults and only matter once those providers are wired in. A
|
||||
# session that opts into Anthropic's 1h cache breakpoint can override this
|
||||
# via the tracker config (see PrefixFreezeConfig.cache_ttl_seconds).
|
||||
_PROVIDER_CACHE_TTL_SECONDS = {
|
||||
"anthropic": 300, # 5 minutes (default ephemeral cache)
|
||||
"openai": 300, # automatic prefix cache, ~5-10 min; conservative floor
|
||||
"gemini": 300,
|
||||
"bedrock": 300,
|
||||
}
|
||||
|
||||
|
||||
@dataclass
|
||||
class PrefixFreezeConfig:
|
||||
|
|
@ -50,6 +65,11 @@ class PrefixFreezeConfig:
|
|||
min_cached_tokens: int = 1024 # Min cached tokens to activate freeze
|
||||
session_ttl_seconds: int = 600 # Session tracker cleanup TTL
|
||||
force_compress_threshold: float = 0.5 # Bust cache if compression saves > this fraction
|
||||
# Provider prompt-cache lifetime used by `classify_cache_miss` to tell a
|
||||
# TTL lapse from a prefix change. `None` falls back to the per-provider
|
||||
# default in `_PROVIDER_CACHE_TTL_SECONDS`. Set to 3600 for a session that
|
||||
# uses Anthropic's 1h cache breakpoint so idle-gap attribution stays honest.
|
||||
cache_ttl_seconds: int | None = None
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -64,6 +84,34 @@ class FreezeStats:
|
|||
turn_number: int = 0
|
||||
|
||||
|
||||
# Cache-miss attribution verdicts. `reason` is one of these literals so
|
||||
# metrics/dashboard can bucket without re-deriving the logic. See
|
||||
# PrefixCacheTracker.classify_cache_miss.
|
||||
MISS_TTL_EXPIRY = "ttl_expiry"
|
||||
MISS_PREFIX_CHANGE = "prefix_change"
|
||||
MISS_COLD_START = "cold_start" # no prior cached prefix to miss against
|
||||
MISS_UNKNOWN = "unknown" # expected a hit, content stable, idle within TTL
|
||||
|
||||
|
||||
@dataclass
|
||||
class CacheMissAttribution:
|
||||
"""Why a turn that expected a prompt-cache hit missed instead.
|
||||
|
||||
Produced by :meth:`PrefixCacheTracker.classify_cache_miss`. ``is_miss``
|
||||
is False when the turn actually hit cache (or there was nothing to hit),
|
||||
in which case ``reason`` is informational only.
|
||||
"""
|
||||
|
||||
is_miss: bool
|
||||
reason: str # one of the MISS_* literals
|
||||
idle_seconds: float = 0.0
|
||||
cache_ttl_seconds: int = 0
|
||||
expected_cached_tokens: int = 0
|
||||
cache_read_tokens: int = 0
|
||||
prefix_changed: bool = False
|
||||
ttl_exceeded: bool = False
|
||||
|
||||
|
||||
class PrefixCacheTracker:
|
||||
"""Tracks provider prefix cache state across turns in a session.
|
||||
|
||||
|
|
@ -185,6 +233,114 @@ class PrefixCacheTracker:
|
|||
def get_last_forwarded_messages(self) -> list[dict[str, Any]]:
|
||||
return copy.deepcopy(self._last_forwarded_messages)
|
||||
|
||||
def resolved_cache_ttl_seconds(self) -> int:
|
||||
"""Effective prompt-cache lifetime for this session's provider."""
|
||||
if self.config.cache_ttl_seconds is not None:
|
||||
return self.config.cache_ttl_seconds
|
||||
return _PROVIDER_CACHE_TTL_SECONDS.get(self.provider, 300)
|
||||
|
||||
def classify_cache_miss(
|
||||
self,
|
||||
cache_read_tokens: int,
|
||||
current_forwarded_messages: list[dict[str, Any]],
|
||||
idle_seconds: float | None = None,
|
||||
) -> CacheMissAttribution:
|
||||
"""Attribute *this turn's* cache outcome: hit, TTL lapse, or prefix change.
|
||||
|
||||
Call this BEFORE :meth:`update_from_response` — it reads the state
|
||||
captured from the *previous* turn (``_cached_token_count``,
|
||||
``_last_forwarded_messages``, ``_last_activity``), all of which
|
||||
``update_from_response`` overwrites.
|
||||
|
||||
Attribution only fires when the previous turn left a cacheable prefix
|
||||
(``_cached_token_count > 0``); the very first warm turn has nothing to
|
||||
miss against, so it is reported as ``cold_start`` with ``is_miss=False``.
|
||||
|
||||
When a hit was expected but ``cache_read_tokens == 0``:
|
||||
|
||||
* If the idle gap since the last turn exceeded the provider cache TTL,
|
||||
the cache entry had already lapsed — ``ttl_expiry``. **TTL wins ties:**
|
||||
once the entry expired, a coincident prefix change is moot (the issue
|
||||
asks "should I move 5m → 1h?", which only the TTL signal answers).
|
||||
* Otherwise, if the forwarded prefix changed versus last turn, the new
|
||||
bytes couldn't match the cached prefix — ``prefix_change``.
|
||||
* If neither signal fires (stable prefix, within TTL) we can't explain
|
||||
it from local state — ``unknown`` (e.g. provider-side eviction).
|
||||
|
||||
A partial read (``0 < cache_read_tokens``) counts as a hit here; the
|
||||
existing model-aware bust detection in PrometheusMetrics already covers
|
||||
partial-invalidation accounting, and double-counting it as a "miss"
|
||||
would muddy the 5m-vs-1h signal this method exists to provide.
|
||||
|
||||
Returns a :class:`CacheMissAttribution`; ``is_miss`` is False for hits
|
||||
and cold starts.
|
||||
"""
|
||||
if idle_seconds is None:
|
||||
idle_seconds = self.seconds_since_activity()
|
||||
ttl = self.resolved_cache_ttl_seconds()
|
||||
expected = self._cached_token_count
|
||||
|
||||
# Nothing was cached last turn → cold start, not a miss.
|
||||
if expected <= 0:
|
||||
return CacheMissAttribution(
|
||||
is_miss=False,
|
||||
reason=MISS_COLD_START,
|
||||
idle_seconds=idle_seconds,
|
||||
cache_ttl_seconds=ttl,
|
||||
expected_cached_tokens=expected,
|
||||
cache_read_tokens=cache_read_tokens,
|
||||
)
|
||||
|
||||
# We expected a hit. A non-zero read means the prefix cache worked.
|
||||
if cache_read_tokens > 0:
|
||||
return CacheMissAttribution(
|
||||
is_miss=False,
|
||||
reason="hit",
|
||||
idle_seconds=idle_seconds,
|
||||
cache_ttl_seconds=ttl,
|
||||
expected_cached_tokens=expected,
|
||||
cache_read_tokens=cache_read_tokens,
|
||||
)
|
||||
|
||||
# Full miss on a prefix we expected cached. Attribute it.
|
||||
ttl_exceeded = idle_seconds > ttl
|
||||
prefix_changed = not self._forwarded_prefix_stable(current_forwarded_messages)
|
||||
|
||||
if ttl_exceeded:
|
||||
reason = MISS_TTL_EXPIRY # TTL wins ties (see docstring)
|
||||
elif prefix_changed:
|
||||
reason = MISS_PREFIX_CHANGE
|
||||
else:
|
||||
reason = MISS_UNKNOWN
|
||||
|
||||
return CacheMissAttribution(
|
||||
is_miss=True,
|
||||
reason=reason,
|
||||
idle_seconds=idle_seconds,
|
||||
cache_ttl_seconds=ttl,
|
||||
expected_cached_tokens=expected,
|
||||
cache_read_tokens=cache_read_tokens,
|
||||
prefix_changed=prefix_changed,
|
||||
ttl_exceeded=ttl_exceeded,
|
||||
)
|
||||
|
||||
def _forwarded_prefix_stable(self, current_forwarded_messages: list[dict[str, Any]]) -> bool:
|
||||
"""True if last turn's forwarded prefix is still an exact prefix of this turn's.
|
||||
|
||||
The cached prefix is whatever we forwarded last turn. If those exact
|
||||
messages still lead the current forwarded list, the bytes the provider
|
||||
hashed for its cache key are unchanged, so a miss can't be blamed on
|
||||
content. Anything else (a frozen message rewritten, the prefix
|
||||
reordered, the list now shorter) counts as a prefix change.
|
||||
"""
|
||||
prev = self._last_forwarded_messages
|
||||
if not prev:
|
||||
# No recorded prefix to compare — can't claim it changed.
|
||||
return True
|
||||
if len(current_forwarded_messages) < len(prev):
|
||||
return False
|
||||
return current_forwarded_messages[: len(prev)] == prev
|
||||
|
||||
def record_bust_avoided(self, tokens_preserved: int, compression_foregone: int) -> None:
|
||||
"""Record when we chose to preserve cache over compressing."""
|
||||
self._busts_avoided += 1
|
||||
|
|
|
|||
|
|
@ -889,6 +889,42 @@
|
|||
</div>
|
||||
</div>
|
||||
</template>
|
||||
<!-- Cache miss attribution: why expected-cache turns missed (#1313) -->
|
||||
<template x-if="hasMissAttribution">
|
||||
<div class="mt-5 border-t border-border pt-4">
|
||||
<div class="flex items-center justify-between mb-3">
|
||||
<div>
|
||||
<div class="text-xs text-gray-500 uppercase tracking-wide">Cache Miss Attribution</div>
|
||||
<div class="text-[11px] text-gray-600 mt-0.5">Why turns that expected a prompt-cache hit missed — TTL lapse (consider a longer TTL) vs the cacheable prefix changing</div>
|
||||
</div>
|
||||
<div data-testid="miss-attr-headline" class="rounded-full border bg-black/20 px-3 py-1 text-[11px] uppercase tracking-[0.18em]"
|
||||
:class="(missAttribution.ttl_expiry_pct || 0) >= (missAttribution.prefix_change_pct || 0) ? 'border-violet-400/25 text-violet-300/90' : 'border-amber-400/25 text-amber-300/90'"
|
||||
x-text="(missAttribution.ttl_expiry_pct || 0) >= (missAttribution.prefix_change_pct || 0) ? 'Mostly TTL lapse' : 'Mostly prefix change'"></div>
|
||||
</div>
|
||||
<div class="grid grid-cols-2 md:grid-cols-4 gap-3">
|
||||
<div class="rounded-2xl border border-white/8 bg-black/20 p-4">
|
||||
<div class="text-[11px] uppercase tracking-[0.18em] text-violet-300/75">TTL Expiry</div>
|
||||
<div data-testid="miss-attr-ttl-value" class="mt-2 text-3xl font-light text-violet-50" x-text="formatNumber(missAttribution.ttl_expiry || 0)"></div>
|
||||
<div class="mt-1 text-xs text-gray-500" x-text="(missAttribution.ttl_expiry_pct || 0).toFixed(1) + '% of attributed — idle past cache TTL'"></div>
|
||||
</div>
|
||||
<div class="rounded-2xl border border-white/8 bg-black/20 p-4">
|
||||
<div class="text-[11px] uppercase tracking-[0.18em] text-amber-300/75">Prefix Change</div>
|
||||
<div data-testid="miss-attr-prefix-value" class="mt-2 text-3xl font-light text-amber-50" x-text="formatNumber(missAttribution.prefix_change || 0)"></div>
|
||||
<div class="mt-1 text-xs text-gray-500" x-text="(missAttribution.prefix_change_pct || 0).toFixed(1) + '% of attributed — cached prefix shifted'"></div>
|
||||
</div>
|
||||
<div class="rounded-2xl border border-white/8 bg-black/20 p-4">
|
||||
<div class="text-[11px] uppercase tracking-[0.18em] text-gray-400">Unknown</div>
|
||||
<div data-testid="miss-attr-unknown-value" class="mt-2 text-3xl font-light text-gray-300" x-text="formatNumber(missAttribution.unknown || 0)"></div>
|
||||
<div class="mt-1 text-xs text-gray-500">stable prefix, within TTL</div>
|
||||
</div>
|
||||
<div class="rounded-2xl border border-white/8 bg-black/20 p-4">
|
||||
<div class="text-[11px] uppercase tracking-[0.18em] text-gray-400">Total Misses</div>
|
||||
<div data-testid="miss-attr-total-value" class="mt-2 text-3xl font-light text-gray-100" x-text="formatNumber(missAttribution.total || 0)"></div>
|
||||
<div class="mt-1 text-xs text-gray-500">expected a cache hit, got none</div>
|
||||
</div>
|
||||
</div>
|
||||
</div>
|
||||
</template>
|
||||
<!-- Per-provider breakdown -->
|
||||
<template x-if="Object.keys(stats.prefix_cache?.by_provider || {}).length > 0">
|
||||
<div class="mt-4 border-t border-border pt-3">
|
||||
|
|
@ -2396,6 +2432,16 @@
|
|||
|| (pf.compression_foregone_tokens || 0) > 0;
|
||||
},
|
||||
|
||||
// --- Cache miss attribution (#1313) ---
|
||||
|
||||
get missAttribution() {
|
||||
return this.stats.prefix_cache?.miss_attribution?.totals || {};
|
||||
},
|
||||
|
||||
get hasMissAttribution() {
|
||||
return (this.missAttribution.total || 0) > 0;
|
||||
},
|
||||
|
||||
// --- Waste Signals ---
|
||||
|
||||
wasteSignalLabel(signal) {
|
||||
|
|
|
|||
|
|
@ -287,9 +287,50 @@ def build_prefix_cache_stats(
|
|||
],
|
||||
}
|
||||
|
||||
# Cache-miss attribution (#1313): why turns that expected a prompt-cache
|
||||
# hit missed instead. Per-provider reason buckets plus an aggregate total,
|
||||
# so the dashboard can show "of N expected-cache misses, X were TTL lapses
|
||||
# vs Y prefix changes" — the signal a user needs to decide 5m vs 1h TTL.
|
||||
_miss_by_provider: dict[str, dict[str, int]] = {}
|
||||
# Holds integer counts AND float percentages (ttl_expiry_pct etc.), so the
|
||||
# value type is float — ints coerce cleanly and the counts stay whole.
|
||||
_miss_totals: dict[str, float] = {
|
||||
"ttl_expiry": 0,
|
||||
"prefix_change": 0,
|
||||
"unknown": 0,
|
||||
"total": 0,
|
||||
}
|
||||
for _provider, _reasons in metrics.cache_miss_attribution_by_provider.items():
|
||||
provider_reasons = {reason: int(count) for reason, count in _reasons.items()}
|
||||
provider_total = sum(provider_reasons.values())
|
||||
if provider_total == 0:
|
||||
continue
|
||||
provider_reasons["total"] = provider_total
|
||||
_miss_by_provider[_provider] = provider_reasons
|
||||
for reason, count in provider_reasons.items():
|
||||
if reason == "total":
|
||||
continue
|
||||
_miss_totals[reason] = _miss_totals.get(reason, 0) + count
|
||||
_miss_totals["total"] += provider_total
|
||||
|
||||
# Share of misses attributable to TTL lapse vs prefix change — the headline
|
||||
# the dashboard renders. Computed against attributed (non-unknown) misses
|
||||
# so an "unknown" bucket doesn't dilute the actionable split.
|
||||
_attributed = _miss_totals["ttl_expiry"] + _miss_totals["prefix_change"]
|
||||
_miss_totals["ttl_expiry_pct"] = (
|
||||
round(_miss_totals["ttl_expiry"] / _attributed * 100, 1) if _attributed > 0 else 0.0
|
||||
)
|
||||
_miss_totals["prefix_change_pct"] = (
|
||||
round(_miss_totals["prefix_change"] / _attributed * 100, 1) if _attributed > 0 else 0.0
|
||||
)
|
||||
|
||||
return {
|
||||
"by_provider": by_provider,
|
||||
"totals": totals,
|
||||
"miss_attribution": {
|
||||
"totals": _miss_totals,
|
||||
"by_provider": _miss_by_provider,
|
||||
},
|
||||
"prefix_freeze": {
|
||||
"busts_avoided": metrics.prefix_freeze_busts_avoided,
|
||||
"tokens_preserved": metrics.prefix_freeze_tokens_preserved,
|
||||
|
|
|
|||
|
|
@ -2516,6 +2516,35 @@ class AnthropicHandlerMixin:
|
|||
if assistant_message is not None:
|
||||
next_original_messages.append(copy.deepcopy(assistant_message))
|
||||
next_forwarded_messages.append(copy.deepcopy(assistant_message))
|
||||
|
||||
# Cache-miss attribution (#1313): when this turn expected a
|
||||
# prompt-cache hit but got cr_tokens == 0, decide whether the
|
||||
# cache most likely lapsed (idle > provider TTL → suggest a
|
||||
# longer TTL) or the cacheable prefix changed (content shifted).
|
||||
# Classify BEFORE update_from_response, which overwrites the
|
||||
# last-turn state the classifier reads (idle clock, prefix,
|
||||
# cached-token count). `optimized_messages` is the prefix we
|
||||
# forwarded this turn; compare it against last turn's.
|
||||
# `hasattr` guard: some tests inject a SimpleNamespace stub
|
||||
# tracker that only implements the freeze API, not the full
|
||||
# PrefixCacheTracker surface.
|
||||
if hasattr(prefix_tracker, "classify_cache_miss"):
|
||||
miss = prefix_tracker.classify_cache_miss(
|
||||
cache_read_tokens=cr_tokens,
|
||||
current_forwarded_messages=optimized_messages,
|
||||
)
|
||||
if miss.is_miss:
|
||||
logger.info(
|
||||
f"[{request_id}] CACHE-MISS-ATTRIBUTION: reason={miss.reason} "
|
||||
f"idle={miss.idle_seconds:.0f}s ttl={miss.cache_ttl_seconds}s "
|
||||
f"expected_cached={miss.expected_cached_tokens:,} "
|
||||
f"prefix_changed={miss.prefix_changed} "
|
||||
f"ttl_exceeded={miss.ttl_exceeded}"
|
||||
)
|
||||
await self.metrics.record_cache_miss_attribution(
|
||||
provider_name, miss.reason
|
||||
)
|
||||
|
||||
prefix_tracker.update_from_response(
|
||||
cache_read_tokens=cr_tokens,
|
||||
cache_write_tokens=cw_tokens,
|
||||
|
|
|
|||
|
|
@ -750,6 +750,27 @@ class StreamingMixin:
|
|||
next_forwarded.append(_copy.deepcopy(asst_msg))
|
||||
next_original.append(_copy.deepcopy(asst_msg))
|
||||
|
||||
# Cache-miss attribution (#1313), streaming Anthropic path. Mirror
|
||||
# the non-streaming handler: classify BEFORE update_from_response
|
||||
# overwrites the last-turn state the classifier reads. Compare the
|
||||
# prefix we forwarded this turn (`forwarded_messages`, pre-assistant
|
||||
# append) against last turn's.
|
||||
# `hasattr` guard: stub trackers in tests may implement only the
|
||||
# freeze API, not the full PrefixCacheTracker surface.
|
||||
if provider == "anthropic" and hasattr(prefix_tracker, "classify_cache_miss"):
|
||||
miss = prefix_tracker.classify_cache_miss(
|
||||
cache_read_tokens=cache_read_tokens,
|
||||
current_forwarded_messages=forwarded_messages,
|
||||
)
|
||||
if miss.is_miss:
|
||||
logger.info(
|
||||
f"[{request_id}] CACHE-MISS-ATTRIBUTION: reason={miss.reason} "
|
||||
f"idle={miss.idle_seconds:.0f}s ttl={miss.cache_ttl_seconds}s "
|
||||
f"expected_cached={miss.expected_cached_tokens:,} "
|
||||
f"prefix_changed={miss.prefix_changed} ttl_exceeded={miss.ttl_exceeded}"
|
||||
)
|
||||
await self.metrics.record_cache_miss_attribution(provider, miss.reason)
|
||||
|
||||
prefix_tracker.update_from_response(
|
||||
cache_read_tokens=cache_read_tokens,
|
||||
cache_write_tokens=cache_write_tokens,
|
||||
|
|
|
|||
|
|
@ -224,6 +224,16 @@ class PrometheusMetrics:
|
|||
self.cache_bust_tokens_lost: int = 0
|
||||
self.cache_bust_count: int = 0
|
||||
|
||||
# Cache-miss attribution (#1313): when a turn expected a prompt-cache
|
||||
# hit but got none, why? Bucketed by reason so operators can tell a
|
||||
# TTL lapse (idle longer than the cache lifetime → consider a longer
|
||||
# TTL) from a prefix-content change (the cacheable prefix shifted).
|
||||
# Reasons are the MISS_* literals from prefix_tracker. Per provider so
|
||||
# the dashboard can scope; populated by `record_cache_miss_attribution`.
|
||||
self.cache_miss_attribution_by_provider: dict[str, dict[str, int]] = defaultdict(
|
||||
lambda: defaultdict(int)
|
||||
)
|
||||
|
||||
# Cumulative savings history (timestamp → cumulative tokens saved)
|
||||
self.savings_history: list[tuple[str, int]] = []
|
||||
self.savings_tracker = savings_tracker or SavingsTracker()
|
||||
|
|
@ -328,6 +338,7 @@ class PrometheusMetrics:
|
|||
self.prefix_freeze_compression_foregone = 0
|
||||
self.cache_bust_tokens_lost = 0
|
||||
self.cache_bust_count = 0
|
||||
self.cache_miss_attribution_by_provider.clear()
|
||||
self.savings_history = []
|
||||
|
||||
with self._stage_timing_lock:
|
||||
|
|
@ -735,6 +746,18 @@ class PrometheusMetrics:
|
|||
self.cache_bust_count += 1
|
||||
self._get_otel_metrics().record_proxy_cache_bust(tokens_lost=tokens_lost)
|
||||
|
||||
async def record_cache_miss_attribution(self, provider: str, reason: str) -> None:
|
||||
"""Record why a turn that expected a prompt-cache hit missed instead.
|
||||
|
||||
``reason`` is one of the MISS_* literals produced by
|
||||
``PrefixCacheTracker.classify_cache_miss`` (``ttl_expiry``,
|
||||
``prefix_change``, ``unknown``). Cold starts and hits are not recorded
|
||||
— only actual misses against a previously-cached prefix reach here.
|
||||
Bucketed per provider so the dashboard can scope or aggregate.
|
||||
"""
|
||||
async with self._lock:
|
||||
self.cache_miss_attribution_by_provider[provider][reason] += 1
|
||||
|
||||
# ------------------------------------------------------------------
|
||||
# Unit 3: WS session lifecycle gauges / histogram
|
||||
# ------------------------------------------------------------------
|
||||
|
|
@ -977,6 +1000,22 @@ class PrometheusMetrics:
|
|||
value=self.cache_bust_tokens_lost,
|
||||
)
|
||||
|
||||
if self.cache_miss_attribution_by_provider:
|
||||
lines.extend(
|
||||
[
|
||||
"# HELP headroom_cache_miss_attribution_total Cache misses on an "
|
||||
"expected-cached prefix, bucketed by reason (ttl_expiry|prefix_change|unknown)",
|
||||
"# TYPE headroom_cache_miss_attribution_total counter",
|
||||
]
|
||||
)
|
||||
for _provider, _reasons in self.cache_miss_attribution_by_provider.items():
|
||||
for _reason, _count in _reasons.items():
|
||||
lines.append(
|
||||
f'headroom_cache_miss_attribution_total{{provider="{_provider}",'
|
||||
f'reason="{_reason}"}} {_count}'
|
||||
)
|
||||
lines.append("")
|
||||
|
||||
lines.extend(
|
||||
[
|
||||
"# HELP headroom_requests_by_provider Requests by provider",
|
||||
|
|
|
|||
|
|
@ -5,6 +5,10 @@ import time
|
|||
import pytest
|
||||
|
||||
from headroom.cache.prefix_tracker import (
|
||||
MISS_COLD_START,
|
||||
MISS_PREFIX_CHANGE,
|
||||
MISS_TTL_EXPIRY,
|
||||
MISS_UNKNOWN,
|
||||
FreezeStats,
|
||||
PrefixCacheTracker,
|
||||
PrefixFreezeConfig,
|
||||
|
|
@ -466,3 +470,103 @@ class TestMultiTurnScenario:
|
|||
|
||||
# After a bust with 0 total, freeze should reset
|
||||
assert tracker.get_frozen_message_count() == 0
|
||||
|
||||
|
||||
class TestClassifyCacheMiss:
|
||||
"""Cache-miss attribution (#1313): TTL lapse vs prefix change vs unknown."""
|
||||
|
||||
BASE = [
|
||||
{"role": "system", "content": "x" * 4000},
|
||||
{"role": "user", "content": "hello"},
|
||||
]
|
||||
CHANGED = [
|
||||
{"role": "system", "content": "DIFFERENT" * 400},
|
||||
{"role": "user", "content": "hello"},
|
||||
]
|
||||
|
||||
def _warm(self, tracker, messages, read=500, write=500):
|
||||
"""Simulate a turn that left `messages` cached."""
|
||||
tracker.update_from_response(
|
||||
cache_read_tokens=read, cache_write_tokens=write, messages=messages
|
||||
)
|
||||
|
||||
def test_cold_start_is_not_a_miss(self):
|
||||
"""No prior cached prefix → cold start, is_miss False."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
result = tracker.classify_cache_miss(0, self.BASE)
|
||||
assert result.is_miss is False
|
||||
assert result.reason == MISS_COLD_START
|
||||
|
||||
def test_cache_read_is_a_hit(self):
|
||||
"""A non-zero read on an expected-cached prefix is a hit, not a miss."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
self._warm(tracker, self.BASE)
|
||||
result = tracker.classify_cache_miss(800, self.BASE)
|
||||
assert result.is_miss is False
|
||||
assert result.reason == "hit"
|
||||
|
||||
def test_ttl_expiry_when_idle_exceeds_ttl(self):
|
||||
"""Idle longer than the cache TTL → ttl_expiry."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
self._warm(tracker, self.BASE)
|
||||
result = tracker.classify_cache_miss(0, self.BASE, idle_seconds=400)
|
||||
assert result.is_miss is True
|
||||
assert result.reason == MISS_TTL_EXPIRY
|
||||
assert result.ttl_exceeded is True
|
||||
assert result.cache_ttl_seconds == 300
|
||||
|
||||
def test_ttl_wins_tie_when_prefix_also_changed(self):
|
||||
"""When idle past TTL AND prefix changed, TTL expiry wins (docstring)."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
self._warm(tracker, self.BASE)
|
||||
result = tracker.classify_cache_miss(0, self.CHANGED, idle_seconds=400)
|
||||
assert result.reason == MISS_TTL_EXPIRY
|
||||
assert result.ttl_exceeded is True
|
||||
assert result.prefix_changed is True
|
||||
|
||||
def test_prefix_change_within_ttl(self):
|
||||
"""Within TTL but the forwarded prefix differs → prefix_change."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
self._warm(tracker, self.BASE)
|
||||
result = tracker.classify_cache_miss(0, self.CHANGED, idle_seconds=10)
|
||||
assert result.is_miss is True
|
||||
assert result.reason == MISS_PREFIX_CHANGE
|
||||
assert result.prefix_changed is True
|
||||
assert result.ttl_exceeded is False
|
||||
|
||||
def test_unknown_when_stable_prefix_within_ttl(self):
|
||||
"""Within TTL, prefix unchanged, but still no read → unknown."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
self._warm(tracker, self.BASE)
|
||||
result = tracker.classify_cache_miss(0, self.BASE, idle_seconds=10)
|
||||
assert result.is_miss is True
|
||||
assert result.reason == MISS_UNKNOWN
|
||||
|
||||
def test_growing_prefix_is_stable(self):
|
||||
"""A turn that appends to last turn's forwarded prefix is not a change."""
|
||||
tracker = PrefixCacheTracker("anthropic")
|
||||
self._warm(tracker, self.BASE)
|
||||
grown = self.BASE + [{"role": "assistant", "content": "hi back"}]
|
||||
result = tracker.classify_cache_miss(0, grown, idle_seconds=10)
|
||||
# Prefix preserved (only appended) → not a prefix_change.
|
||||
assert result.prefix_changed is False
|
||||
assert result.reason == MISS_UNKNOWN
|
||||
|
||||
def test_one_hour_ttl_override(self):
|
||||
"""cache_ttl_seconds override widens the TTL window (1h breakpoint)."""
|
||||
tracker = PrefixCacheTracker("anthropic", PrefixFreezeConfig(cache_ttl_seconds=3600))
|
||||
self._warm(tracker, self.BASE)
|
||||
# 400s idle is past the 300s default but within 3600s → not TTL expiry.
|
||||
result = tracker.classify_cache_miss(0, self.BASE, idle_seconds=400)
|
||||
assert result.cache_ttl_seconds == 3600
|
||||
assert result.ttl_exceeded is False
|
||||
assert result.reason == MISS_UNKNOWN
|
||||
|
||||
def test_resolved_ttl_falls_back_to_provider_default(self):
|
||||
assert PrefixCacheTracker("anthropic").resolved_cache_ttl_seconds() == 300
|
||||
assert (
|
||||
PrefixCacheTracker(
|
||||
"anthropic", PrefixFreezeConfig(cache_ttl_seconds=3600)
|
||||
).resolved_cache_ttl_seconds()
|
||||
== 3600
|
||||
)
|
||||
|
|
|
|||
|
|
@ -250,3 +250,72 @@ def test_stats_endpoint_reports_langfuse_configuration(monkeypatch: pytest.Monke
|
|||
assert langfuse["enabled"] is True
|
||||
assert langfuse["service_name"] == "headroom-proxy"
|
||||
assert langfuse["endpoint"] == "https://cloud.langfuse.com/api/public/otel/v1/traces"
|
||||
|
||||
|
||||
# --- Cache-miss attribution (#1313) ---
|
||||
|
||||
|
||||
def test_record_cache_miss_attribution_buckets_by_provider_and_reason() -> None:
|
||||
metrics = PrometheusMetrics()
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "ttl_expiry"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "ttl_expiry"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "prefix_change"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "unknown"))
|
||||
|
||||
buckets = metrics.cache_miss_attribution_by_provider["anthropic"]
|
||||
assert buckets["ttl_expiry"] == 2
|
||||
assert buckets["prefix_change"] == 1
|
||||
assert buckets["unknown"] == 1
|
||||
|
||||
|
||||
def test_prefix_cache_stats_include_miss_attribution() -> None:
|
||||
metrics = PrometheusMetrics()
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "ttl_expiry"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "ttl_expiry"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "prefix_change"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "unknown"))
|
||||
|
||||
stats = build_prefix_cache_stats(metrics, None)
|
||||
ma = stats["miss_attribution"]
|
||||
|
||||
assert ma["totals"]["ttl_expiry"] == 2
|
||||
assert ma["totals"]["prefix_change"] == 1
|
||||
assert ma["totals"]["unknown"] == 1
|
||||
assert ma["totals"]["total"] == 4
|
||||
# Percentages are over attributed (non-unknown) misses: 2 / 3, 1 / 3.
|
||||
assert ma["totals"]["ttl_expiry_pct"] == 66.7
|
||||
assert ma["totals"]["prefix_change_pct"] == 33.3
|
||||
assert ma["by_provider"]["anthropic"]["total"] == 4
|
||||
|
||||
|
||||
def test_prefix_cache_stats_miss_attribution_empty_when_no_misses() -> None:
|
||||
metrics = PrometheusMetrics()
|
||||
stats = build_prefix_cache_stats(metrics, None)
|
||||
ma = stats["miss_attribution"]
|
||||
assert ma["totals"]["total"] == 0
|
||||
assert ma["totals"]["ttl_expiry_pct"] == 0.0
|
||||
assert ma["by_provider"] == {}
|
||||
|
||||
|
||||
def test_prometheus_export_includes_miss_attribution() -> None:
|
||||
metrics = PrometheusMetrics()
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "ttl_expiry"))
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "prefix_change"))
|
||||
|
||||
exported = asyncio.run(metrics.export())
|
||||
|
||||
assert (
|
||||
'headroom_cache_miss_attribution_total{provider="anthropic",reason="ttl_expiry"} 1'
|
||||
in exported
|
||||
)
|
||||
assert (
|
||||
'headroom_cache_miss_attribution_total{provider="anthropic",reason="prefix_change"} 1'
|
||||
in exported
|
||||
)
|
||||
|
||||
|
||||
def test_reset_runtime_clears_miss_attribution() -> None:
|
||||
metrics = PrometheusMetrics()
|
||||
asyncio.run(metrics.record_cache_miss_attribution("anthropic", "ttl_expiry"))
|
||||
asyncio.run(metrics.reset_runtime())
|
||||
assert dict(metrics.cache_miss_attribution_by_provider) == {}
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue