github-featured/chopratejas--headroom
2
0
Fork 0
Code Issues Pull Requests Actions 16 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(cache): stabilize Anthropic block-growing lineages (#2917)

Browse Source 
## Description

Fixes the remaining Anthropic prompt-cache failure in #2671 and the
newly reported parallel-tool-profile variant.

The production failure has three connected parts:

1. `SessionTrackerStore.resolve_tracker` only recognized whole-message
prefixes. A caller that grows or regenerates blocks inside one message
therefore received a fresh tracker every turn, so previous forwarded
state was always empty and breakpoint relocation could never run.
2. `normalize_message_cache_control` always moved the message breakpoint
to the newest block. That is correct for a pure block append, but a
message that rewrites its tail can never match the prior newest-block
write and repeatedly rewrites the full message prefix.
3. Parallel Anthropic sub-calls can carry identical messages but
different tools. Because tools precede messages in the provider cache
key, sharing one frozen-prefix tracker across those calls
cross-contaminates cache state even when message lineage is identical.

This PR deliberately combines the valid parts of #2699 and #2702, fixes
the discriminator between their two shapes, and adds cache-key affinity
for the second pattern reported on #2671. In particular, a pure append
is identified by `stable_prefix_blocks == previous_block_count`;
rewritten-tail relocation is only possible when `stable_prefix_blocks <
previous_block_count`. This prevents a pure append from being pinned to
an old boundary.

Closes #2671.

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [x] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- Added one canonical history classifier with distinct exact,
whole-message append, pure block append, rewritten-tail, and diverged
outcomes.
- Kept pure block appends on newest-block breakpoint placement so each
request reads the old prefix and writes only its appended blocks.
- Added block-level replay of the prior forwarded bytes for pure
appends; the whole-message delta path explicitly refuses this shape so
it cannot silently discard appended blocks.
- Kept a rewritten-tail request on its existing tracker and anchored its
breakpoint to the end of the byte-stable leading run.
- Made rewritten-tail matching conservative: one changed message,
unchanged message count, no shrink, at least 8 stable leading blocks
covering at least half of old and new content, and a fixed suffix of at
least 2 blocks.
- Required a unique best rewritten-tail lineage match. Ambiguity creates
a fresh lineage instead of making sibling sub-calls ping-pong one
tracker.
- Added a stable affinity fingerprint over model, deterministically
forwarded tools, tool choice, thinking, and output configuration.
Different provider cache-key profiles cannot share frozen-prefix state.
- Snapshotted previous original/forwarded messages once in the Anthropic
handler and reused that exact state for delta extraction, replay, and
breakpoint placement.
- Added `HEADROOM_STABLE_BOUNDARY_BREAKPOINT=0` as a rollback switch for
rewritten-tail relocation.

The canonical projection is used only for comparison. Replayed content
always comes from the exact previously forwarded bytes or the current
raw/optimized tail; canonicalized data is never reconstructed into an
upstream request.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check` on changed files)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ python -m pytest tests/test_issue_2671_block_growth_cache.py -q
12 passed in 0.10s

$ python -m pytest tests/test_cache -q
253 passed, 3 skipped in 1.94s

$ python -m pytest <Anthropic handler/proxy regression set> -q
134 passed, 1 warning in 9.42s

$ ruff format --check <changed files>
4 files already formatted

$ ruff check <changed files>
All checks passed!

$ git diff --check
# clean
```

The Anthropic regression set covers beta stickiness, CCR injection,
compaction transforms, pre-upstream backpressure, streaming
reconstruction, upstream headers, model sanitization, diagnostics, and
cache stability.

## Real Behavior Proof

- Environment: macOS, Python 3.12.13, real FastAPI Anthropic handler
with a local upstream stub plus a deterministic provider-cache oracle.
- Exact steps: send a cold 35-block aggregate message, then three
requests that preserve a 30-block prefix and fixed two-block suffix
while regenerating a growing middle tail. Resolve the real session
tracker, normalize the real handler body, record the response, and
repeat.
- Observed result: handler breakpoint indices are `34 -> 29 -> 29`; the
cache oracle transitions from a cold 35-block write to establishing the
30-block stable boundary, then produces `(read=30, write=0)` on
subsequent rewritten-tail turns. A separate pure-append sequence
produces `(0,30) -> (30,4) -> (34,4) -> (38,5)`, proving its breakpoint
continues to advance.
- Also observed: identical message histories with different tool schemas
resolve to distinct trackers in the real handler path.
- Not tested: a live Anthropic billing soak, the complete repository
test suite, or mypy. #2702 contains earlier live production measurements
for the rewritten-tail mechanism; this PR adds the pure-append
correction, affinity isolation, and broader regression model.

## 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
- [x] I have made corresponding documentation updates where applicable
(internal behavior is documented in code; no user-facing surface
changed)
- [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 relevant unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md` — it is generated by
release-please from the Conventional Commit PR title

## Additional Notes

- Consolidates the complementary approaches in #2699 and #2702. Credit
to @axisrow and @nangsontay for the traces, root-cause work, and live
validation that made the two production shapes distinguishable.
- The 20-block minimum for relocation mirrors the provider lookup-window
risk boundary and keeps short ordinary messages on the established
newest-block behavior.
- Disabling stable-boundary relocation does not disable improved lineage
resolution or tool-profile isolation; it restores only the previous
breakpoint placement.

Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local>
This commit is contained in:
Tejas Chopra
2026-08-10 22:38:51 -07:00
committed by GitHub
parent 4925bf6a82
commit 1a04c957f5
4 changed files with 833 additions and 35 deletions
Download Patch File Download Diff File Expand all files Collapse all files
headroom/cache/prefix_tracker.py
Vendored
+359 -27
View File
@@ -21,6 +21,7 @@ import hashlib
import itertools
import json
import logging
import os
import time
from collections import OrderedDict
from dataclasses import dataclass
@@ -229,6 +230,184 @@ def _canonicalize_for_prefix_compare(obj: Any) -> Any:
return obj
# Canonical relationships between consecutive histories. These constants are
# strings (rather than an Enum) so they remain cheap to log on the request hot
# path and easy to assert in tests.
RELATION_EXACT = "exact"
RELATION_MESSAGE_APPEND = "message_append"
RELATION_BLOCK_APPEND = "block_append"
RELATION_BLOCK_REWRITE_TAIL = "block_rewrite_tail"
RELATION_DIVERGED = "diverged"
@dataclass(frozen=True)
class HistoryRelation:
"""How a current message history relates to one recorded last turn.
``block_append`` and ``block_rewrite_tail`` are deliberately distinct:
Anthropic should keep the breakpoint on the newest block for a pure append,
but anchor it to ``stable_prefix_blocks - 1`` when the previous tail was
rewritten and therefore can never match a prior cache write (#2671).
"""
kind: str
message_index: int | None = None
stable_prefix_blocks: int = 0
stable_suffix_blocks: int = 0
previous_block_count: int = 0
current_block_count: int = 0
# A rewritten-tail match is intentionally conservative. The production shape
# behind #2671 has a hundred-plus-block stable prefix and a fixed two-block
# suffix. Requiring both avoids merging sibling sub-calls which merely share a
# short injected preamble or a single generic reminder at the end.
_MIN_REWRITE_PREFIX_BLOCKS = 8
_MIN_REWRITE_SUFFIX_BLOCKS = 2
def _message_fields_outside_content(message: dict[str, Any]) -> dict[str, Any]:
"""Return message identity fields, excluding the block list itself."""
return {key: value for key, value in message.items() if key != "content"}
def _stable_leading_block_run(current: list[Any], previous: list[Any]) -> int:
"""Number of canonical-equal blocks at the start of both lists."""
limit = min(len(current), len(previous))
run = 0
while run < limit and current[run] == previous[run]:
run += 1
return run
def _stable_trailing_block_run(current: list[Any], previous: list[Any], *, leading_run: int) -> int:
"""Non-overlapping canonical-equal suffix length."""
limit = min(len(current), len(previous)) - leading_run
run = 0
while run < limit and current[-(run + 1)] == previous[-(run + 1)]:
run += 1
return run
def _classify_history_canonical(
current_messages: list[Any], previous_messages: list[Any]
) -> HistoryRelation:
"""Classify two already-canonical, structurally snapshotted histories."""
if not previous_messages or len(current_messages) < len(previous_messages):
return HistoryRelation(RELATION_DIVERGED)
changed: HistoryRelation | None = None
for index, previous_message in enumerate(previous_messages):
current_message = current_messages[index]
if current_message == previous_message:
continue
if changed is not None:
return HistoryRelation(RELATION_DIVERGED)
if not isinstance(previous_message, dict) or not isinstance(current_message, dict):
return HistoryRelation(RELATION_DIVERGED)
if _message_fields_outside_content(previous_message) != _message_fields_outside_content(
current_message
):
return HistoryRelation(RELATION_DIVERGED)
previous_blocks = previous_message.get("content")
current_blocks = current_message.get("content")
if not isinstance(previous_blocks, list) or not isinstance(current_blocks, list):
return HistoryRelation(RELATION_DIVERGED)
previous_count = len(previous_blocks)
current_count = len(current_blocks)
leading = _stable_leading_block_run(current_blocks, previous_blocks)
# Pure block append. The previous write remains intact and Anthropic's
# lookback can find it, so the breakpoint must advance to the newest
# block and cover the newly appended tail.
if current_count > previous_count and leading == previous_count:
changed = HistoryRelation(
RELATION_BLOCK_APPEND,
message_index=index,
stable_prefix_blocks=leading,
previous_block_count=previous_count,
current_block_count=current_count,
)
continue
# Rewritten-tail growth. This is narrower than a fuzzy prefix match:
# message count may not change, content may not shrink, most of the old
# prefix must survive, and a substantial fixed suffix must identify the
# sub-call. Crucially ``leading < previous_count`` proves this is NOT a
# pure append (the bug in the original #2702 discriminator).
trailing = _stable_trailing_block_run(current_blocks, previous_blocks, leading_run=leading)
if (
len(current_messages) == len(previous_messages)
and current_count >= previous_count
and _MIN_REWRITE_PREFIX_BLOCKS <= leading < previous_count
and leading * 2 >= previous_count
and leading * 2 >= current_count
and trailing >= _MIN_REWRITE_SUFFIX_BLOCKS
):
changed = HistoryRelation(
RELATION_BLOCK_REWRITE_TAIL,
message_index=index,
stable_prefix_blocks=leading,
stable_suffix_blocks=trailing,
previous_block_count=previous_count,
current_block_count=current_count,
)
continue
return HistoryRelation(RELATION_DIVERGED)
if changed is not None:
return changed
return HistoryRelation(
RELATION_EXACT
if len(current_messages) == len(previous_messages)
else RELATION_MESSAGE_APPEND
)
def classify_history_relation(
current_messages: list[dict[str, Any]],
previous_messages: list[dict[str, Any]],
) -> HistoryRelation:
"""Return the canonical cross-turn relationship for two raw histories.
The canonical projection may drop a whole directive-only message. Refuse
classification when that would shift raw message indices: block replay
always slices the raw lists and must never consume a canonical index as a
raw one.
"""
if not current_messages or not previous_messages:
return HistoryRelation(RELATION_DIVERGED)
current = _lineage_snapshot(_canonicalize_for_prefix_compare(current_messages))
previous = _lineage_snapshot(_canonicalize_for_prefix_compare(previous_messages))
prefix_len = len(previous_messages)
if len(previous) != prefix_len:
return HistoryRelation(RELATION_DIVERGED)
if len(_canonicalize_for_prefix_compare(current_messages[:prefix_len])) != prefix_len:
return HistoryRelation(RELATION_DIVERGED)
return _classify_history_canonical(current, previous)
def segment_fingerprint(value: Any) -> str:
"""Stable hash for non-message provider cache-key segments.
Cache-control placement and transport annotations are deliberately ignored;
semantic tool/model/thinking changes remain visible. The hash is affinity
metadata only and is never used to reconstruct or forward request content.
"""
canonical = _lineage_snapshot(_canonicalize_for_prefix_compare(value))
encoded = json.dumps(
canonical,
sort_keys=True,
ensure_ascii=False,
separators=(",", ":"),
default=str,
)
return hashlib.sha256(encoded.encode()).hexdigest()[:24]
def extract_cache_stable_delta(
current_messages: list[dict[str, Any]],
previous_original_messages: list[dict[str, Any]] | None,
@@ -251,13 +430,13 @@ def extract_cache_stable_delta(
"""
if not previous_original_messages or previous_forwarded_messages is None:
return None
relation = classify_history_relation(current_messages, previous_original_messages)
if relation.kind not in (RELATION_EXACT, RELATION_MESSAGE_APPEND):
# A same-message block append needs a block-level splice in
# ``overlay_cached_prefix``; slicing only whole messages would silently
# discard its new blocks. Rewritten tails are not append-only deltas.
return None
prefix_len = len(previous_original_messages)
if len(current_messages) < prefix_len:
return None
if _canonicalize_for_prefix_compare(
current_messages[:prefix_len]
) != _canonicalize_for_prefix_compare(previous_original_messages):
return None
return (
copy.deepcopy(previous_forwarded_messages),
copy.deepcopy(current_messages[prefix_len:]),
@@ -281,12 +460,12 @@ def overlay_cached_prefix(
the corresponding leading messages so the forwarded prefix stays byte-for-byte
what the provider hashed for its cache key.
Safe only when this turn append-only-extends the previous turn (the standard
growing-conversation shape): the previous ORIGINAL messages must be an exact
prefix of the current ORIGINAL messages, and there is exactly one forwarded
message per original. Otherwise the previous forwarded bytes may not
correspond to the same positions, so we return ``optimized_messages``
unchanged (accept a possible bust rather than forward wrong content).
Safe only when this turn extends the previous turn in a proven positional
shape: either whole-message append or pure block append inside one message.
There must be exactly one previous forwarded message per original. Otherwise
the previous bytes may not correspond to the same positions, so we return
``optimized_messages`` unchanged (accept a possible bust rather than forward
wrong content).
This makes freezing byte-identical in BOTH proxy modes, so the only remaining
difference between them is how large a mutable (still-compressible) tail each
@@ -311,6 +490,55 @@ def overlay_cached_prefix(
n,
)
return optimized_messages
relation = classify_history_relation(current_original_messages, prev_orig)
if relation.kind == RELATION_BLOCK_APPEND and relation.message_index is not None:
message_index = relation.message_index
if message_index < len(optimized_messages):
previous_message = prev_fwd[message_index]
previous_original_message = prev_orig[message_index]
current_message = optimized_messages[message_index]
previous_content = (
previous_message.get("content") if isinstance(previous_message, dict) else None
)
previous_original_content = (
previous_original_message.get("content")
if isinstance(previous_original_message, dict)
else None
)
current_content = (
current_message.get("content") if isinstance(current_message, dict) else None
)
split = (
len(previous_original_content)
if isinstance(previous_original_content, list)
else -1
)
if (
isinstance(previous_content, list)
and isinstance(previous_original_content, list)
and isinstance(current_content, list)
and len(previous_content) == split
and len(current_content) >= split
and _canonicalize_for_prefix_compare(current_content[:split])
== _canonicalize_for_prefix_compare(previous_original_content)
):
merged = copy.deepcopy(previous_message)
merged["content"] = copy.deepcopy(previous_content) + copy.deepcopy(
current_content[split:]
)
logger.debug(
"overlay: replayed %d forwarded blocks and appended %d new blocks "
"inside message %d",
split,
len(current_content) - split,
message_index,
)
return (
list(prev_fwd[:message_index])
+ [merged]
+ list(optimized_messages[message_index + 1 :])
)
# Append-only guard on CONTENT ONLY, message-by-message. Replay the
# previously-forwarded (cached, compressed) bytes for the longest LEADING
# run of messages that is byte-for-byte (content-canonical) identical to
@@ -359,8 +587,55 @@ def overlay_cached_prefix(
return list(prev_fwd[:k]) + list(optimized_messages[k:])
_STABLE_BOUNDARY_ENV = "HEADROOM_STABLE_BOUNDARY_BREAKPOINT"
_MIN_BLOCKS_FOR_RELOCATION = 20
def _stable_boundary_enabled() -> bool:
return os.environ.get(_STABLE_BOUNDARY_ENV, "").strip().lower() not in (
"0",
"false",
"no",
"off",
)
def _breakpoint_index(
content: list[Any],
message: dict[str, Any],
message_index: int,
previous_forwarded_messages: list[dict[str, Any]] | None,
) -> int:
"""Choose newest for appends, stable-prefix end for rewritten tails."""
newest = len(content) - 1
if (
not previous_forwarded_messages
or not _stable_boundary_enabled()
or len(content) < _MIN_BLOCKS_FOR_RELOCATION
or message_index >= len(previous_forwarded_messages)
):
return newest
previous = previous_forwarded_messages[message_index]
if not isinstance(previous, dict):
return newest
relation = classify_history_relation([message], [previous])
if relation.kind != RELATION_BLOCK_REWRITE_TAIL:
return newest
logger.debug(
"cache breakpoint anchored to stable run %d/%d blocks in message %d "
"(previous=%d, stable_suffix=%d)",
relation.stable_prefix_blocks,
relation.current_block_count,
message_index,
relation.previous_block_count,
relation.stable_suffix_blocks,
)
return relation.stable_prefix_blocks - 1
def normalize_message_cache_control(
messages: list[dict[str, Any]],
previous_forwarded_messages: list[dict[str, Any]] | None = None,
) -> list[dict[str, Any]]:
"""Own message-level cache_control placement so breakpoints stay bounded.
@@ -371,9 +646,12 @@ def normalize_message_cache_control(
so on a long conversation the accumulation eventually 400s.
Fix: strip EVERY message-level cache_control and re-place a **single**
ephemeral breakpoint on the last block of the last block-style message. One
breakpoint caches the whole message prefix up to it, and — because the
provider's cache key is message CONTENT, not marker presence (moving the
ephemeral breakpoint. Pure append-only growth keeps it on the newest block,
which both reads the prior write and writes the appended tail. If last turn's
counterpart proves that the tail was rewritten, it is placed at the end of
the byte-stable leading run instead. One breakpoint caches the prefix up to
it, and — because the provider's cache key is message CONTENT, not marker
presence (moving the
breakpoint forward is the documented client pattern and it hits) — stripping
and re-placing markers never busts. system/tools breakpoints live outside
``messages`` and are left untouched (they still count toward the 4 limit, so
@@ -417,7 +695,16 @@ def normalize_message_cache_control(
msg = out[last_block_idx]
content = list(msg["content"])
marker = dict(last_marker) if last_marker else {"type": "ephemeral"}
content[-1] = {**content[-1], "cache_control": marker}
breakpoint_index = _breakpoint_index(
content, msg, last_block_idx, previous_forwarded_messages
)
# Anthropic content blocks are dictionaries, but callers can still
# supply mixed list content. The newest block is known to be a dict
# from the scan above; fall back to it rather than attempting ``**``
# on a scalar stable-boundary element.
if not isinstance(content[breakpoint_index], dict):
breakpoint_index = len(content) - 1
content[breakpoint_index] = {**content[breakpoint_index], "cache_control": marker}
out[last_block_idx] = {**msg, "content": content}
changed = True
return out if changed else messages
@@ -829,6 +1116,10 @@ class SessionTrackerStore:
# value, so a synthetic key can never collide with a client-supplied
# x-headroom-session-id.
self._lineages: dict[str, OrderedDict[str, list[Any]]] = {}
# Exact non-message cache-key affinity per tracker. Anthropic renders
# tools before system/messages, so two sub-calls with identical history
# but different tool profiles must never share frozen-prefix state.
self._lineage_affinities: dict[str, str | None] = {}
self._lineage_counter = itertools.count(1)
def get_or_create(self, session_id: str, provider: str) -> PrefixCacheTracker:
@@ -854,6 +1145,7 @@ class SessionTrackerStore:
session_id: str,
provider: str,
messages: list[dict[str, Any]] | None = None,
cache_affinity: str | None = None,
) -> PrefixCacheTracker:
"""Resolve the tracker for THIS conversation within a session id (#2085).
@@ -866,10 +1158,11 @@ class SessionTrackerStore:
Lineage resolution keys trackers by conversation content instead:
reuse the tracker whose previous request messages are a prefix of the
incoming history (client histories are append-only, so a
conversation's next request always extends its previous one); start a
fresh lineage when the history diverges or was rewritten (client-side
compaction — the provider cache line is gone then anyway).
incoming history. It also recognizes a conservative block-level shape
where a large leading run and two-block identity suffix survive while
the middle tail is regenerated; all other rewrites start a fresh
lineage. This keeps #2671's stable cache boundary attached without
merging unrelated parallel sub-calls.
Byte-identical histories (templated fan-outs before they diverge)
intentionally share a tracker: their provider cache line is identical
too, so sharing is harmless.
@@ -888,6 +1181,9 @@ class SessionTrackerStore:
compares like against like across turns. ``None``/empty
(legacy callers, stub stores in tests) falls back to plain
:meth:`get_or_create`.
cache_affinity: Stable fingerprint of the provider's non-message
cache-key segments (model/tools/tool choice/thinking). Lineages
with different affinity never share a tracker.
Returns:
The ``PrefixCacheTracker`` for this conversation's lineage.
@@ -918,14 +1214,48 @@ class SessionTrackerStore:
family = self._lineages.setdefault(session_id, OrderedDict())
# Longest recorded chain that prefixes the incoming history wins.
# Strict whole-message continuations win first, then pure block appends.
# Rewritten-tail matches are deliberately last and require a unique best
# structural score; ambiguity starts a fresh lineage instead of making
# sibling sub-calls ping-pong one tracker.
by_length = sorted(family.items(), key=lambda item: len(item[1]), reverse=True)
best_key: str | None = None
best_len = -1
for key, chain in family.items():
if len(chain) > len(snap) or len(chain) <= best_len:
continue
if snap[: len(chain)] == chain:
best_key, best_len = key, len(chain)
for accepted in (
(RELATION_EXACT, RELATION_MESSAGE_APPEND),
(RELATION_BLOCK_APPEND,),
):
for key, chain in by_length:
if self._lineage_affinities.get(key) != cache_affinity:
continue
relation = _classify_history_canonical(snap, chain)
if relation.kind in accepted:
best_key = key
break
if best_key is not None:
break
if best_key is None:
rewrite_candidates: list[tuple[tuple[int, int, int], str]] = []
for key, chain in by_length:
if self._lineage_affinities.get(key) != cache_affinity:
continue
relation = _classify_history_canonical(snap, chain)
if relation.kind == RELATION_BLOCK_REWRITE_TAIL:
rewrite_candidates.append(
(
(
relation.stable_prefix_blocks,
relation.stable_suffix_blocks,
relation.previous_block_count,
),
key,
)
)
rewrite_candidates.sort(reverse=True)
if rewrite_candidates and (
len(rewrite_candidates) == 1 or rewrite_candidates[0][0] != rewrite_candidates[1][0]
):
best_key = rewrite_candidates[0][1]
if best_key is None:
cap = self._default_config.max_lineages_per_session
@@ -963,6 +1293,7 @@ class SessionTrackerStore:
# the family before the stamp below.
tracker = self.get_or_create(best_key, provider)
family[best_key] = snap
self._lineage_affinities[best_key] = cache_affinity
return tracker
def compute_session_id(
@@ -1031,6 +1362,7 @@ class SessionTrackerStore:
family = self._lineages[base]
for key in [k for k in family if k not in self._trackers]:
del family[key]
self._lineage_affinities.pop(key, None)
if not family:
del self._lineages[base]
logger.debug("SessionTrackerStore: cleaned up %d expired sessions", len(expired))
headroom/proxy/handlers/anthropic.py
+35 -8
View File
@@ -1091,6 +1091,25 @@ class AnthropicHandlerMixin:
session_id = self.session_tracker_store.compute_session_id(
request, model, session_messages
)
# Prefix trackers must follow the provider's cache key, not just
# message history. Anthropic renders tools before system/messages;
# parallel sub-calls commonly share model+system+history while
# carrying different tool sets. Sharing one tracker across those
# requests pins cache reads at the early tools segment (#2671).
from headroom.cache.prefix_tracker import segment_fingerprint
affinity_tools = self._tools_for_forwarding(
body.get("tools"), preserve_order=preserve_tool_order
)
cache_affinity = segment_fingerprint(
{
"model": model,
"tools": affinity_tools,
"tool_choice": body.get("tool_choice"),
"thinking": body.get("thinking"),
"output_config": body.get("output_config"),
}
)
# Resolve the tracker by conversation lineage within the session id
# (#2085): one model + system prompt spans a Claude Code session and
# all its parallel subagents, so concurrent conversations share this
@@ -1098,8 +1117,17 @@ class AnthropicHandlerMixin:
# thrash the frozen-prefix state and the provider prompt cache is
# re-written on nearly every call.
prefix_tracker = self.session_tracker_store.resolve_tracker(
session_id, "anthropic", messages=session_messages
session_id,
"anthropic",
messages=session_messages,
cache_affinity=cache_affinity,
)
# Snapshot lineage state once. Reusing the same pair for delta
# extraction, byte-stable replay, and breakpoint placement keeps
# all three decisions tied to one previous request (and avoids
# repeatedly deep-copying a multi-megabyte agent transcript).
previous_original_messages = prefix_tracker.get_last_original_messages()
previous_forwarded_messages = prefix_tracker.get_last_forwarded_messages()
frozen_message_count = prefix_tracker.get_frozen_message_count()
# Idle gap since the previous turn's response, snapshotted at fetch
# (before get_or_create bumped the access clock). Forwarded to the
@@ -1505,8 +1533,6 @@ class AnthropicHandlerMixin:
optimized_tokens = tokenizer.count_messages(optimized_messages)
transforms_applied = _cold_transforms
else:
previous_original_messages = prefix_tracker.get_last_original_messages()
previous_forwarded_messages = prefix_tracker.get_last_forwarded_messages()
delta = self._extract_cache_stable_delta(
original_client_messages,
previous_original_messages,
@@ -1625,8 +1651,8 @@ class AnthropicHandlerMixin:
_ov = overlay_cached_prefix(
optimized_messages,
original_client_messages,
prefix_tracker.get_last_original_messages(),
prefix_tracker.get_last_forwarded_messages(),
previous_original_messages,
previous_forwarded_messages,
)
_overlay_replayed = _ov != optimized_messages
if _overlay_replayed:
@@ -1636,10 +1662,11 @@ class AnthropicHandlerMixin:
# Own cache_control placement: the client moves the breakpoint each
# turn and the overlay replays past markers, so they accumulate ~1/turn
# and Anthropic hard-errors at >4. Strip message-level markers and keep
# a single breakpoint on the last block (caches the whole prefix;
# content-keyed cache so re-placing never busts). Applied last so the
# one breakpoint. Pure appends advance it to the newest block; a
# proven rewritten tail anchors it at the byte-stable boundary so
# that the same prefix is readable next turn. Applied last so the
# forwarded AND recorded (next_forwarded) messages stay bounded.
_norm = normalize_message_cache_control(optimized_messages)
_norm = normalize_message_cache_control(optimized_messages, previous_forwarded_messages)
if _norm is not optimized_messages:
optimized_messages = _norm
tests/test_issue_2671_block_growth_cache.py
+288
View File
@@ -0,0 +1,288 @@
"""Comprehensive regression for #2671's block-growing Anthropic histories.
The provider writes cache entries only at explicit breakpoints and searches at
most 20 block boundaries backwards on the next request. Consequently:
* a pure append must advance the breakpoint to the newest block;
* a rewritten tail must anchor at the last byte-stable leading block;
* both shapes must retain one conversation lineage across turns;
* different tools/thinking profiles must never share that lineage, because
Anthropic renders those segments before messages in its cache key.
The small cache oracle below models those write/lookback rules. It catches a
green-but-inert implementation: merely moving a marker in a unit-built message
is insufficient unless the real resolve -> normalize -> record sequence carries
the previous turn's state forward.
"""
from __future__ import annotations
import json
from dataclasses import dataclass, field
from typing import Any
from headroom.cache.prefix_tracker import (
RELATION_BLOCK_APPEND,
RELATION_BLOCK_REWRITE_TAIL,
RELATION_DIVERGED,
PrefixFreezeConfig,
SessionTrackerStore,
_strip_cache_control,
classify_history_relation,
extract_cache_stable_delta,
normalize_message_cache_control,
overlay_cached_prefix,
segment_fingerprint,
)
def _text(text: str, *, cache: bool = False) -> dict[str, Any]:
block: dict[str, Any] = {"type": "text", "text": text}
if cache:
block["cache_control"] = {"type": "ephemeral"}
return block
def _message(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]:
return [{"role": "user", "content": blocks}]
def _pure_append(total: int) -> list[dict[str, Any]]:
return _message([_text(f"block-{index}") for index in range(total)])
def _rewritten_tail(
turn: int,
churn_blocks: int,
*,
stable_blocks: int = 30,
instruction: str = "instruction: summarize",
) -> list[dict[str, Any]]:
blocks = [_text(f"stable-{index}") for index in range(stable_blocks)]
blocks += [_text(f"turn-{turn}-changing-{index}") for index in range(churn_blocks)]
# The captured production shape keeps a two-block identity suffix pinned at
# the end while the blocks immediately before it are rewritten.
blocks += [_text(instruction), _text("fixed end-of-transcript reminder")]
return _message(blocks)
def _breakpoint(messages: list[dict[str, Any]]) -> tuple[int, int]:
found = [
(message_index, block_index)
for message_index, message in enumerate(messages)
if isinstance(message.get("content"), list)
for block_index, block in enumerate(message["content"])
if isinstance(block, dict) and "cache_control" in block
]
assert len(found) == 1
return found[0]
@dataclass
class _AnthropicBreakpointCache:
"""Deterministic model of Anthropic's explicit-breakpoint cache lookup."""
entries: dict[str, int] = field(default_factory=dict)
lookback_blocks: int = 20
@staticmethod
def _blocks(messages: list[dict[str, Any]]) -> list[Any]:
blocks: list[Any] = []
for message in messages:
content = message.get("content")
if isinstance(content, list):
blocks.extend(_strip_cache_control(content))
return blocks
@staticmethod
def _key(blocks: list[Any], end: int) -> str:
return json.dumps(blocks[: end + 1], sort_keys=True, separators=(",", ":"))
def request(self, messages: list[dict[str, Any]]) -> tuple[int, int]:
"""Return simulated ``(cache_read_blocks, cache_write_blocks)``."""
_, breakpoint = _breakpoint(messages)
blocks = self._blocks(messages)
read = 0
first = max(0, breakpoint - self.lookback_blocks + 1)
for candidate in range(breakpoint, first - 1, -1):
key = self._key(blocks, candidate)
if key in self.entries:
read = self.entries[key]
break
written_prefix = breakpoint + 1
write = max(0, written_prefix - read)
self.entries[self._key(blocks, breakpoint)] = written_prefix
return read, write
def _record(tracker, original, forwarded, *, read=0, write=10_000): # noqa: ANN001
tracker.update_from_response(
cache_read_tokens=read,
cache_write_tokens=write,
messages=forwarded,
original_messages=original,
)
def test_classifier_separates_pure_append_from_rewritten_tail() -> None:
append = classify_history_relation(_pure_append(35), _pure_append(30))
rewrite = classify_history_relation(_rewritten_tail(2, 5), _rewritten_tail(1, 3))
assert append.kind == RELATION_BLOCK_APPEND
assert append.stable_prefix_blocks == 30
assert rewrite.kind == RELATION_BLOCK_REWRITE_TAIL
assert rewrite.stable_prefix_blocks == 30
assert rewrite.stable_suffix_blocks == 2
def test_rewritten_tail_requires_a_real_previous_divergence() -> None:
"""The #2702 bug classified a pure append as a rewritten tail."""
previous = _pure_append(30)
current = _pure_append(31)
relation = classify_history_relation(current, previous)
assert relation.kind == RELATION_BLOCK_APPEND
assert relation.stable_prefix_blocks == relation.previous_block_count
def test_rewritten_tail_requires_a_two_block_identity_suffix() -> None:
"""Sibling sub-calls sharing a transcript and generic reminder must split."""
previous = _rewritten_tail(1, 3, instruction="instruction: summarize")
sibling = _rewritten_tail(2, 5, instruction="instruction: title")
assert classify_history_relation(sibling, previous).kind == RELATION_DIVERGED
def test_lineage_survives_rewritten_tail_growth_and_delivers_previous_state() -> None:
store = SessionTrackerStore(PrefixFreezeConfig(min_cached_tokens=0))
first_tracker = None
for turn, churn in enumerate((3, 5, 8, 11), start=1):
original = _rewritten_tail(turn, churn)
tracker = store.resolve_tracker("shared", "anthropic", messages=original)
first_tracker = first_tracker or tracker
assert tracker is first_tracker
previous = tracker.get_last_forwarded_messages()
if turn > 1:
assert previous, "lineage match must deliver the previous forwarded request"
forwarded = normalize_message_cache_control(original, previous)
_record(tracker, original, forwarded)
assert store.active_sessions == 1
assert first_tracker._turn_number == 4
def test_sibling_rewritten_tail_streams_do_not_ping_pong() -> None:
store = SessionTrackerStore()
seen = {}
for turn, churn in enumerate((3, 5, 8), start=1):
for instruction in ("instruction: summarize", "instruction: title"):
original = _rewritten_tail(turn, churn, instruction=instruction)
tracker = store.resolve_tracker("shared", "anthropic", messages=original)
seen.setdefault(instruction, tracker)
assert tracker is seen[instruction]
forwarded = normalize_message_cache_control(
original, tracker.get_last_forwarded_messages()
)
_record(tracker, original, forwarded)
assert seen["instruction: summarize"] is not seen["instruction: title"]
def test_cache_affinity_splits_identical_histories_with_different_tools() -> None:
store = SessionTrackerStore()
history = _pure_append(30)
shell = segment_fingerprint({"model": "claude", "tools": [{"name": "shell"}]})
search = segment_fingerprint({"model": "claude", "tools": [{"name": "search"}]})
shell_tracker = store.resolve_tracker(
"shared", "anthropic", messages=history, cache_affinity=shell
)
search_tracker = store.resolve_tracker(
"shared", "anthropic", messages=history, cache_affinity=search
)
assert search_tracker is not shell_tracker
assert (
store.resolve_tracker("shared", "anthropic", messages=history, cache_affinity=shell)
is shell_tracker
)
def test_cache_affinity_ignores_only_cache_directive_movement() -> None:
base = {
"model": "claude",
"tools": [{"name": "shell", "cache_control": {"type": "ephemeral"}}],
}
moved = {"model": "claude", "tools": [{"name": "shell"}]}
changed = {"model": "claude", "tools": [{"name": "search"}]}
assert segment_fingerprint(base) == segment_fingerprint(moved)
assert segment_fingerprint(base) != segment_fingerprint(changed)
def test_pure_append_replays_forwarded_blocks_and_advances_breakpoint() -> None:
previous_original = _pure_append(30)
previous_forwarded = _message([_text(f"COMPRESSED-{index}") for index in range(30)])
current = _pure_append(34)
overlaid = overlay_cached_prefix(current, current, previous_original, previous_forwarded)
normalized = normalize_message_cache_control(overlaid, previous_forwarded)
assert [block["text"] for block in normalized[0]["content"][:30]] == [
f"COMPRESSED-{index}" for index in range(30)
]
assert [block["text"] for block in normalized[0]["content"][30:]] == [
f"block-{index}" for index in range(30, 34)
]
assert _breakpoint(normalized) == (0, 33)
def test_whole_message_delta_path_cannot_discard_appended_blocks() -> None:
"""Block appends require a splice, never an empty whole-message delta."""
previous = _pure_append(30)
assert extract_cache_stable_delta(_pure_append(34), previous, previous) is None
def test_cache_oracle_proves_pure_append_chains_without_rewrites() -> None:
oracle = _AnthropicBreakpointCache()
previous = None
outcomes = []
for total in (30, 34, 38, 43):
current = _pure_append(total)
forwarded = normalize_message_cache_control(current, previous)
outcomes.append(oracle.request(forwarded))
previous = forwarded
assert outcomes == [(0, 30), (30, 4), (34, 4), (38, 5)]
def test_cache_oracle_proves_rewritten_tail_stops_perpetual_full_writes() -> None:
oracle = _AnthropicBreakpointCache()
previous = None
outcomes = []
breakpoints = []
for turn, churn in enumerate((3, 5, 8, 11), start=1):
current = _rewritten_tail(turn, churn)
forwarded = normalize_message_cache_control(current, previous)
breakpoints.append(_breakpoint(forwarded)[1])
outcomes.append(oracle.request(forwarded))
previous = forwarded
# Cold turn writes its varying tail. Turn two establishes the new stable
# boundary; subsequent turns read it and perform no repeated full write.
assert breakpoints == [34, 29, 29, 29]
assert outcomes[0] == (0, 35)
assert outcomes[1] == (0, 30)
assert outcomes[2:] == [(30, 0), (30, 0)]
def test_relocation_kill_switch_restores_newest_block(monkeypatch) -> None: # noqa: ANN001
previous = normalize_message_cache_control(_rewritten_tail(1, 3))
monkeypatch.setenv("HEADROOM_STABLE_BOUNDARY_BREAKPOINT", "0")
current = _rewritten_tail(2, 5)
forwarded = normalize_message_cache_control(current, previous)
assert _breakpoint(forwarded) == (0, len(current[0]["content"]) - 1)
tests/test_proxy_anthropic_cache_stability.py
+151
View File
@@ -2,6 +2,7 @@
from __future__ import annotations
import copy
from types import SimpleNamespace
from unittest.mock import AsyncMock
@@ -1243,6 +1244,156 @@ def test_cache_mode_reuses_prior_forwarded_prefix_and_compresses_only_new_suffix
]
def test_anthropic_handler_splits_prefix_trackers_when_tool_profiles_differ() -> None:
"""The handler must pass its non-message cache affinity into resolution.
Anthropic's cache key begins with tools. Identical messages on two parallel
sub-calls therefore cannot safely share frozen-prefix state when their tool
arrays differ (#2671 Pattern B).
"""
resolved = []
with _make_proxy_client() as client:
proxy = client.app.state.proxy
proxy.config.optimize = True
proxy.config.mode = "cache"
proxy.config.image_optimize = False
real_resolve = proxy.session_tracker_store.resolve_tracker
def _spy_resolve(session_id, provider, messages=None, cache_affinity=None): # noqa: ANN001
tracker = real_resolve(
session_id,
provider,
messages=messages,
cache_affinity=cache_affinity,
)
resolved.append((cache_affinity, tracker))
return tracker
proxy.session_tracker_store.resolve_tracker = _spy_resolve
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
return httpx.Response(
200,
json={
"id": "msg_affinity",
"type": "message",
"role": "assistant",
"content": [{"type": "text", "text": "ok"}],
"usage": {
"input_tokens": 10,
"output_tokens": 1,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 0,
},
},
)
proxy._retry_request = _fake_retry
headers = {"x-api-key": "test-key", "anthropic-version": "2023-06-01"}
messages = [{"role": "user", "content": "same parent transcript"}]
for tool_name in ("shell", "search"):
response = client.post(
"/v1/messages",
headers=headers,
json={
"model": "claude-sonnet-4-6",
"max_tokens": 32,
"messages": messages,
"tools": [
{
"name": tool_name,
"description": tool_name,
"input_schema": {"type": "object", "properties": {}},
}
],
},
)
assert response.status_code == 200
assert len(resolved) == 2
assert resolved[0][0] != resolved[1][0]
assert resolved[0][1] is not resolved[1][1]
def test_anthropic_handler_anchors_a_proven_rewritten_tail_to_stable_blocks() -> None:
"""The real handler must feed last turn's bytes into normalization."""
bodies = []
def _history(turn: int, churn: int) -> list[dict]:
content = [{"type": "text", "text": f"stable-{index}"} for index in range(30)]
content.extend(
{"type": "text", "text": f"turn-{turn}-changing-{index}"} for index in range(churn)
)
content.extend(
[
{"type": "text", "text": "instruction: summarize"},
{"type": "text", "text": "fixed end-of-transcript reminder"},
]
)
return [{"role": "user", "content": content}]
with _make_proxy_client() as client:
proxy = client.app.state.proxy
# Breakpoint ownership and lineage tracking apply even in passthrough
# mode. Keeping optimization off isolates that real handler wiring from
# the compression pipeline.
proxy.config.optimize = False
proxy.config.image_optimize = False
# This regression models a client whose next request replaces the same
# aggregate message. Do not synthesize an assistant history entry in
# the tracker, because that is a separate response-reconstruction path.
proxy._assistant_message_from_response_json = lambda _body: None
async def _fake_retry(method, url, headers, body, stream=False, **kwargs): # noqa: ANN001
bodies.append(copy.deepcopy(body))
return httpx.Response(
200,
json={
"id": "msg_rewrite",
"type": "message",
"role": "assistant",
"content": [],
"usage": {
"input_tokens": 10_000,
"output_tokens": 1,
"cache_read_input_tokens": 0,
"cache_creation_input_tokens": 10_000,
},
},
)
proxy._retry_request = _fake_retry
headers = {"x-api-key": "test-key", "anthropic-version": "2023-06-01"}
for turn, churn in enumerate((3, 5, 8), start=1):
response = client.post(
"/v1/messages",
headers=headers,
json={
"model": "claude-sonnet-4-6",
"max_tokens": 32,
"messages": _history(turn, churn),
},
)
assert response.status_code == 200
breakpoint_indices = []
for body in bodies:
marked = [
index
for index, block in enumerate(body["messages"][0]["content"])
if "cache_control" in block
]
assert len(marked) == 1
breakpoint_indices.append(marked[0])
# Cold request caches through its newest block. Once rewrite is proven,
# all subsequent calls pin the same 30-block boundary instead of creating
# an ever-growing full write on each turn.
assert breakpoint_indices == [34, 29, 29]
def test_cache_mode_skips_same_message_append_rewrite_to_preserve_stability() -> None:
captured = {"calls": []}
with _make_proxy_client() as client: