akadmin/headroom
1
0
Fork 0
mirror of https://github.com/headroomlabs-ai/headroom.git synced 2026-08-27 14:17:10 -04:00
Code Issues Projects Releases Packages Wiki Activity Actions

fix(cache): enforce Anthropic's 1h-before-5m cache_control ordering before forwarding (#2941)

Browse source 
## Description

Anthropic evaluates prompt-cache breakpoints in **one pass over the
whole request** — `tools`, then `system`, then `messages` — and rejects
the request outright when a `ttl='1h'` breakpoint appears after a
5-minute one. A bare `{"type": "ephemeral"}` marker counts as 5 minutes,
so this is easy to trip without any `ttl` field being visibly wrong:

```
API Error: 400 messages.15.content.1.cache_control.ttl: a ttl='1h' cache_control block must not
come after a ttl='5m' cache_control block. Note that blocks are processed in the following order:
tools, system, messages.
```

Headroom rewrites `cache_control` markers in several independent places,
each looking at one section, and nothing checked the invariant that
spans them. The failure mode is a dead turn, not a silent cost
regression.

Two paths can leave the forwarded body illegal today:

1. **Replayed 1h marker in a 5m request.** Claude Code picks its TTL
lane per request, not per session: the main loop asks for 1h and sends
the `extended-cache-ttl` beta header, while a side question (`/btw` in
the report) goes out in the 5m lane with bare markers and no beta
header. Headroom replays part of the previous turn's forwarded bytes
into `messages` to keep the prefix stable, and those bytes still carry
`ttl: "1h"`. `tools`/`system` at 5m, `messages` at 1h — 400.
2. **Tools breakpoint downgraded.** `inject_tool_search_deferral`
re-places the *last* marker it stripped, so a bare marker on a later
deferred tool overwrites a `ttl='1h'` one. The tools prefix goes
upstream at 5m while message breakpoints are still 1h — 400. Reported
separately as #2767.

The rule spans `tools`/`system`/`messages`, so no individual transform
is in a position to check it. The fix is a guard at the last seam before
the body goes on the wire, plus the one Python/Rust divergence that
manufactures the violation upstream of it.

Closes #2939

## 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
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `headroom/proxy/helpers.py`: new `enforce_cache_control_ttl_order`,
plus `cache_control_ttl_lane` / `cache_control_ttl_lanes` /
`walk_cache_control`. The walk visits markers in Anthropic's documented
order and matches the traversal `count_cache_breakpoints` already
performs, so the two cannot disagree about what counts as a breakpoint.
TTL ranking is ported verbatim from the Rust `TtlOrderingWalk::observe`:
absent or `"5m"` is short, `"1h"` is long, any other value is left alone
rather than guessed at. Two repairs:
- **Lane containment** — when the client sent no 1h marker of its own,
strip `ttl` from any 1h marker that leaked in. That request never sent
the `extended-cache-ttl` beta header, so it could not have written a 1h
entry anyway; nothing is lost. Other marker fields (`scope`, …) are
preserved.
- **Ordering** — when the client did ask for 1h, promote every 5m marker
preceding the last 1h one. Demoting would also make the request legal
but would discard 1h caching the client is explicitly paying for, which
is the regression #2375 / #2382 / #2651 were filed to stop. A violation
seen on the way out means headroom downgraded or introduced a marker, so
promoting restores what the client's own (legal) request asked for at
that position.
- Copy-on-write: a legal body is returned by identity, so the hot path
pays only a walk over at most a handful of markers. Kill switch
`HEADROOM_CACHE_CONTROL_TTL_GUARD=0`, matching the
`HEADROOM_TOOL_SEARCH=0` convention.
- `headroom/proxy/handlers/anthropic.py`:
- Capture the client's TTL lane from the inbound snapshot, before any
transform runs. This cannot be inferred from the session or from config
— the lane is a per-request property of Claude Code, which is the whole
reason the `/btw` case exists.
- Call the guard immediately before `log_cache_breakpoints`, i.e. after
every transform, the tool sort, the deferral, CCR injection and the
pipeline extensions. Mark the body mutated when a repair fires, and log
a WARNING carrying the repair kind, the counts and the offending
sections — the diagnostic the next report of this class will need.
- `_sort_tools_deterministically` now skips the sort when any tool
carries `cache_control`, logging `event=tool_sort_skipped
reason=marker_present`. A breakpoint on a tool means "cache through
here", so reordering changes what is inside the cached prefix and can
move a 1h-marked tool behind a 5m-marked one. The Rust proxy already
refuses for exactly this reason (`any_tool_has_cache_control` in
`crates/headroom-proxy/src/compression/live_zone_anthropic.rs:651`); the
Python path never got the same guard. Putting the check in
`_sort_tools_deterministically` rather than `_tools_for_forwarding`
covers all call sites including the batch path.
- `tests/test_cache_control_ttl_order.py`: new, 24 cases.

## Testing

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

The new tests validate against an independent reimplementation of
Anthropic's rule rather than against the guard's own walk, so a bug in
the walk cannot make the assertions pass. Coverage: lane classification
(bare marker is 5m, unknown TTLs are `other`); containment of a replayed
1h marker including preservation of non-`ttl` fields; promotion across
`tools`→`messages`, `system`→`messages` and within `messages`; markers
nested in `tool_result` sub-blocks; only markers before the *last* 1h
one are rewritten, so a legal 1h-then-5m ordering is left alone; legal
bodies returned by `is` identity; unknown `ttl` untouched; kill switch;
the tool sort skipping on marked tools and still sorting unmarked ones,
with one test pinning that the sort *would* have created a violation
without the guard; and an end-to-end regression running
`inject_tool_search_deferral` then the guard on the #2767 shape.

### Test Output

```text
$ pytest tests/test_cache_control_ttl_order.py -q
24 passed in 0.77s

$ pytest tests/test_cache_ttl_preserved.py tests/test_cache_control_move_bust.py \
         tests/test_cache_breakpoint_diagnostics.py tests/test_issue_746_tool_search.py -q
86 passed in 1.36s

$ pytest tests/test_cache/ tests/test_proxy/ -q
16 failed, 489 passed, 2 skipped in 91.41s

$ ruff check headroom/ tests/test_cache_control_ttl_order.py
All checks passed!

$ ruff format --check headroom/ tests/test_cache_control_ttl_order.py
1 file would be reformatted, 520 files already formatted

$ mypy --python-version 3.13 headroom --ignore-missing-imports
Found 12 errors in 3 files (checked 517 source files)
```

The three non-green results above are all pre-existing on a clean
`upstream/main` in this environment, verified by stashing the changes
and re-running:

- The 16 failures are all in
`tests/test_cache/test_client_integration.py` and are a Windows
temp-path problem in this sandbox (`OSError: [WinError 123] ...
'\\C:\\Users\\...\\Temp'`), not a code failure. They fail identically
with the branch stashed.
- `ruff format --check` flags `headroom/testing/README.md`, a docs code
block untouched by this PR.
- The mypy errors are in `headroom/memory/mcp_server.py` and
`headroom/release_version.py`; none are in the files this PR changes.
`--python-version 3.13` is needed locally because the pinned
`python_version = "3.10"` makes mypy reject the installed numpy stubs
before it checks anything.

## Real Behavior Proof

- Environment: Windows 11, Python 3.13.11, headroom at this branch's
head, run against the real `headroom.proxy` forwarding helpers. No
Anthropic API key is available in this environment, so Anthropic's
validator is reimplemented locally from its documented rule and its own
error string; the request bodies are produced by the real code path
(`_sort_tools_deterministically` then `inject_tool_search_deferral` then
the guard), not hand-written.
- Exact command / steps: build two request shapes — (A) a 5m-lane
request whose `messages` carries a replayed `ttl:"1h"` marker, the
`/btw` case; (B) 13 tools with markers on two deferred tools plus 1h
message breakpoints, the #2767 case — push each through the forwarding
helpers twice, once with `HEADROOM_CACHE_CONTROL_TTL_GUARD=0` and once
with the guard at its default, and validate the resulting body.
- Observed result: both scenarios are rejected with the issue's exact
400 when the guard is off, and both are legal with it on. Scenario A is
repaired by lane containment (the leaked 1h ttl is stripped), scenario B
by promotion (the downgraded tools breakpoint goes back to 1h). Full
output:

```text
########## Scenario A: /btw side question replays a 1h marker into a 5m request ##########

===== BEFORE (HEADROOM_CACHE_CONTROL_TTL_GUARD=0) =====
  tools.0                      5m
  system.0                     5m
  messages.1.content.0         1h
  RESULT: API Error: 400 messages.1.content.0.cache_control.ttl: a ttl='1h' cache_control block
  must not come after a ttl='5m' cache_control block. Note that blocks are processed in the
  following order: tools, system, messages.

===== AFTER (default) =====
WARNING event=cache_control_ttl_order request_id=repro-2939 repair=lane_containment demoted=1
leaked_from_section=messages; the client sent no 1h marker, so a replayed 1h breakpoint would
have been rejected upstream
  tools.0                      5m
  system.0                     5m
  messages.1.content.0         5m
  RESULT: 200 OK (request satisfies the ordering rule)

########## Scenario B: tool-search deferral downgrades the tools breakpoint ##########

===== BEFORE (HEADROOM_CACHE_CONTROL_TTL_GUARD=0) =====
INFO event=tool_sort_skipped reason=marker_present tool_count=13 marked=2
  tools.1                      5m
  messages.1.content.0         1h
  RESULT: API Error: 400 messages.1.content.0.cache_control.ttl: a ttl='1h' cache_control block
  must not come after a ttl='5m' cache_control block. Note that blocks are processed in the
  following order: tools, system, messages.

===== AFTER (default) =====
INFO event=tool_sort_skipped reason=marker_present tool_count=13 marked=2
WARNING event=cache_control_ttl_order request_id=repro-2939 repair=promote_to_1h promoted=1
first_short_section=tools first_long_section=messages; a 5m breakpoint preceded a 1h one, which
Anthropic rejects outright
  tools.1                      1h
  messages.1.content.0         1h
  RESULT: 200 OK (request satisfies the ordering rule)
```

- Not tested: no live call to `api.anthropic.com` — no credentials in
this environment — so the 400/200 above come from a local
reimplementation of the rule, not from the API itself. The reporter's
original `/btw` flow was not reproduced end to end through `headroom
wrap claude`. Cache-hit-rate impact of promoting a 5m marker to 1h was
not measured against real traffic; the reasoning for promoting over
demoting is argued above, not benchmarked. Nothing on the Rust side was
exercised.

## 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
- [x] I did **not** edit `CHANGELOG.md` — it is generated by
release-please from my Conventional Commit PR title (a CI guard enforces
this)

## Additional Notes

Documentation was not updated: the new env var is a kill switch for an
internal correctness guard with no user-facing behaviour when things are
working, matching how `HEADROOM_KOMPRESS_BACKGROUND_WARM` is handled.
Happy to add a line to the env-var reference if maintainers prefer.

**One trade-off worth a maintainer's eye.** `affinity_tools` feeds a
`segment_fingerprint` used for prefix-tracker affinity. Skipping the
sort makes that fingerprint depend on the client's tool order. Clients
that mark tools must already keep a stable order for their own prefix
cache to work, so this should be safe, but it is stated rather than
assumed.

**Deliberately out of scope, flagged rather than dropped:**

- The sibling hole in `inject_tool_search_deferral`: when
`resident_has_cache_control` is already true at 5m, a dropped 1h marker
is discarded outright. That is a cost regression rather than a 400, and
it sits in the same handful of lines that the open PR #2771 rewrites, so
touching it here would conflict. Better raised on #2767.
- `TtlOrderingWalk` in `crates/headroom-core/src/cache_control.rs` is
instantiated separately per field list, so it only ever sees violations
*within* `messages`, `system` or `tools` — never across them — and it
only warns. Its module doc justifies warn-only with "Anthropic itself
accepts both orderings (just with potentially-suboptimal cache
eviction)". #2939 and #2767 both show that premise is now stale.
Changing Rust behaviour is a separate blast radius.
- `cold_prefix._cache_control_ttls` never scans `tools[]`, so a client
whose only 1h marker rides on `tools` is read as 300s. Separate bug,
separate PR.

Co-authored-by: Claude Opus 5 <noreply@anthropic.com>
Co-authored-by: JD Davis <mxjerrett@gmail.com>
This commit is contained in:
Parideboy 2026-08-12 23:32:12 +02:00 • committed by GitHub
parent e540d64feb
commit 3752458022
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 705 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

70
headroom/proxy/handlers/anthropic.py
View file

@ -199,9 +199,30 @@ class AnthropicHandlerMixin:
def _sort_tools_deterministically(
cls, tools: list[dict[str, Any]] | None
) -> list[dict[str, Any]] | None:
"""Return tools in deterministic order to preserve prompt-cache stability."""
"""Return tools in deterministic order to preserve prompt-cache stability.
Skipped entirely when any tool carries ``cache_control``. A breakpoint on
a tool means "cache everything up to and including this one", so
reordering the array changes which tools are inside that prefix -- and
with two markers of different TTLs it can put the 1h one behind the 5m
one, which Anthropic rejects outright (#2939). The Rust proxy already
refuses for the same reason (``any_tool_has_cache_control`` in
``crates/headroom-proxy/src/compression/live_zone_anthropic.rs``); this
is the Python side of that guard.
Clients that mark no tools -- the common case, and the one the sort was
written for -- are unaffected, so this costs nobody a cache bust.
"""
if not tools:
return tools
marked = sum(1 for t in tools if isinstance(t, dict) and t.get("cache_control"))
if marked:
logger.info(
"event=tool_sort_skipped reason=marker_present tool_count=%d marked=%d",
len(tools),
marked,
)
return tools
return sorted(tools, key=cls._tool_sort_key)
@classmethod
@ -754,11 +775,24 @@ class AnthropicHandlerMixin:
# transform runs; paired with the outbound count right before
# forwarding (event=cache_breakpoints) so a dropped or moved
# final breakpoint is self-diagnosing from proxy.log alone.
from headroom.proxy.helpers import count_cache_breakpoints
from headroom.proxy.helpers import (
CACHE_TTL_1H,
cache_control_ttl_lanes,
count_cache_breakpoints,
)
inbound_breakpoints = count_cache_breakpoints(
body.get("system"), messages, body.get("tools")
)
# Which TTL lane did the CLIENT ask for on THIS turn? Claude Code
# picks 5m or 1h per request (a `/btw` side question drops to 5m and
# omits the extended-cache-ttl beta header even mid-1h-session), so
# this cannot be inferred from the session or from config. The
# pre-forward guard needs it to tell a breakpoint the client asked
# for from one an earlier turn's replayed bytes dragged in (#2939).
client_uses_1h = CACHE_TTL_1H in cache_control_ttl_lanes(
body.get("system"), messages, body.get("tools")
)
# Validate message array size
if len(messages) > MAX_MESSAGE_ARRAY_LENGTH:
@ -3082,7 +3116,37 @@ class AnthropicHandlerMixin:
"upstream request for server-side retrieval handling"
)
from headroom.proxy.helpers import log_cache_breakpoints
# Last stop before the wire. Every transform, the tool sort, the
# tool-search deferral, CCR injection and the pipeline
# extensions have run, so this is the only place that can see
# the cache_control markers Anthropic will actually evaluate --
# and the ordering rule spans tools/system/messages, which no
# single transform is in a position to check (#2939).
from headroom.proxy.helpers import (
enforce_cache_control_ttl_order,
log_cache_breakpoints,
)
(
_ttl_system,
_ttl_messages,
_ttl_tools,
_ttl_stats,
) = enforce_cache_control_ttl_order(
body.get("system"),
body.get("messages"),
body.get("tools"),
client_uses_1h=client_uses_1h,
request_id=request_id,
)
if _ttl_stats["violation"]:
if body.get("system") is not None:
body["system"] = _ttl_system
body["messages"] = _ttl_messages
if body.get("tools") is not None:
body["tools"] = _ttl_tools
tools = _ttl_tools
body_mutation_tracker.mark_mutated("cache_control_ttl_order")
log_cache_breakpoints(
request_id=request_id,

284
headroom/proxy/helpers.py
View file

@ -18,6 +18,7 @@ import re
import threading
import time
from collections import OrderedDict
from collections.abc import Callable
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, Literal, cast
@ -406,6 +407,289 @@ def count_cache_breakpoints(
}
# --------------------------------------------------------------------------
# cache_control TTL lanes (issues #2939, #2767).
#
# Anthropic reads cache breakpoints in ONE global walk -- `tools`, then
# `system`, then `messages` -- and requires every ``ttl="1h"`` marker to appear
# before every 5-minute one. A bare ``{"type": "ephemeral"}`` marker IS 5m. Get
# it wrong and the whole turn dies with
#
# messages.15.content.1.cache_control.ttl: a ttl='1h' cache_control block must
# not come after a ttl='5m' cache_control block.
#
# The rule is already modelled in ``crates/headroom-core/src/cache_control.rs``
# (``TtlOrderingWalk``), but that walker is instantiated once per field list, so
# it only sees violations *within* `tools`, `system` or `messages` -- never
# across them -- and it only warns. The helpers below are the cross-section
# Python counterpart, and they repair rather than warn, because by the time the
# body reaches the forwarder any violation in it is one Headroom introduced.
#
# No Headroom code ever invents a ``ttl`` value: every marker we re-place is
# copied from some client marker. So an outbound 1h marker on a request whose
# client sent none can only have leaked in from an EARLIER turn (via
# ``overlay_cached_prefix`` replaying the previous turn's forwarded bytes), and
# the fix is to drop the leaked ttl rather than to spread it.
# --------------------------------------------------------------------------
CACHE_TTL_1H = "1h"
CACHE_TTL_5M = "5m"
#: Any other ``ttl`` value is a lane we don't model; markers carrying one are
#: reported but never rewritten, so a future Anthropic TTL can't be mangled by
#: guesswork here. Mirrors ``TtlOrderingWalk::observe`` in headroom-core.
CACHE_TTL_OTHER = "other"
_TTL_GUARD_ENV = "HEADROOM_CACHE_CONTROL_TTL_GUARD"
def cache_control_ttl_lane(marker: Any) -> str:
"""Return ``"1h"``, ``"5m"`` or ``"other"`` for one ``cache_control`` marker.
A marker with no ``ttl`` key is 5m -- that is Anthropic's default lane, and
treating it as "unknown" instead would make every ordinary Claude Code
request look like a violation.
"""
if not isinstance(marker, dict):
return CACHE_TTL_OTHER
ttl = marker.get("ttl")
if ttl is None:
return CACHE_TTL_5M
ttl = str(ttl)
return ttl if ttl in (CACHE_TTL_1H, CACHE_TTL_5M) else CACHE_TTL_OTHER
def _revisit_holder(
holder: Any,
section: str,
visit: Callable[[str, dict[str, Any]], dict[str, Any] | None],
) -> tuple[Any, bool]:
"""Offer ``holder``'s marker to ``visit``; return a copy only if replaced."""
if not isinstance(holder, dict):
return holder, False
marker = holder.get("cache_control")
if not isinstance(marker, dict):
return holder, False
replacement = visit(section, marker)
if replacement is None:
return holder, False
return {**holder, "cache_control": replacement}, True
def walk_cache_control(
system: Any,
messages: Any,
tools: Any,
visit: Callable[[str, dict[str, Any]], dict[str, Any] | None],
) -> tuple[Any, Any, Any, bool]:
"""Visit every ``cache_control`` marker in Anthropic's evaluation order.
``visit(section, marker)`` returns a replacement marker, or ``None`` to
leave it alone -- so the same traversal serves both a read-only survey and a
rewrite. Sections are rebuilt copy-on-write and the untouched originals are
returned by identity: the forwarded body shares structure with the prefix
tracker's snapshot of what we sent, so mutating a marker in place would
rewrite history the next turn compares against.
Traversal matches :func:`count_cache_breakpoints`, nested ``tool_result``
sub-blocks included, so the guard and the diagnostic can never disagree
about what counts as a breakpoint.
"""
changed = False
new_tools = tools
if isinstance(tools, list):
rebuilt_tools: list[Any] = []
hit = False
for tool in tools:
out, did = _revisit_holder(tool, "tools", visit)
hit = hit or did
rebuilt_tools.append(out)
if hit:
new_tools = rebuilt_tools
changed = True
new_system = system
if isinstance(system, list):
rebuilt_system: list[Any] = []
hit = False
for block in system:
out, did = _revisit_holder(block, "system", visit)
hit = hit or did
rebuilt_system.append(out)
if hit:
new_system = rebuilt_system
changed = True
new_messages = messages
if isinstance(messages, list):
rebuilt_messages: list[Any] = []
any_message_hit = False
for msg in messages:
if not isinstance(msg, dict):
rebuilt_messages.append(msg)
continue
# Message-level markers are non-standard but Headroom's own
# diagnostics count them, so keep the two traversals in step.
new_msg, message_hit = _revisit_holder(msg, "messages", visit)
content = new_msg.get("content")
if isinstance(content, list):
rebuilt_blocks: list[Any] = []
block_hit = False
for block in content:
new_block, did = _revisit_holder(block, "messages", visit)
inner = new_block.get("content") if isinstance(new_block, dict) else None
if isinstance(inner, list):
rebuilt_inner: list[Any] = []
inner_hit = False
for sub in inner:
new_sub, sub_did = _revisit_holder(sub, "messages", visit)
inner_hit = inner_hit or sub_did
rebuilt_inner.append(new_sub)
if inner_hit:
new_block = {**new_block, "content": rebuilt_inner}
did = True
block_hit = block_hit or did
rebuilt_blocks.append(new_block)
if block_hit:
new_msg = {**new_msg, "content": rebuilt_blocks}
message_hit = True
any_message_hit = any_message_hit or message_hit
rebuilt_messages.append(new_msg)
if any_message_hit:
new_messages = rebuilt_messages
changed = True
return new_system, new_messages, new_tools, changed
def cache_control_ttl_lanes(system: Any, messages: Any, tools: Any) -> set[str]:
"""Return the distinct TTL lanes the request's markers ask for."""
lanes: set[str] = set()
def _survey(_section: str, marker: dict[str, Any]) -> None:
lanes.add(cache_control_ttl_lane(marker))
return None
walk_cache_control(system, messages, tools, _survey)
return lanes
def enforce_cache_control_ttl_order(
system: Any,
messages: Any,
tools: Any,
*,
client_uses_1h: bool,
request_id: str = "",
) -> tuple[Any, Any, Any, dict[str, Any]]:
"""Make the outbound body satisfy Anthropic's cache_control TTL ordering.
Two repairs, in this order:
1. **Lane containment.** When the client's own request carried no 1h marker
anywhere (``client_uses_1h`` is False), strip ``ttl`` from every outbound
1h marker. Headroom never authors a ttl, so such a marker is a previous
turn's value replayed into this one -- and a client that did not ask for
the 1h lane has not sent the ``extended-cache-ttl`` beta header either,
so promoting the rest of the request to match it is not an option. This
is the ``/btw`` case in #2939: Claude Code forks a side question into the
5m lane, and the replayed prefix drags a 1h marker in behind the fork's
own 5m ``tools``/``system`` breakpoints.
2. **Ordering.** Any 5m marker still sitting before the last 1h marker is
promoted to 1h. Here the client *is* in the 1h lane, so the beta header
is present and the promotion is safe. This covers the mirror-image bug
where a transform downgrades an early breakpoint -- e.g.
``inject_tool_search_deferral`` losing a 1h marker off a deferred tool
(#2767) -- leaving the client's later 1h message breakpoints illegal.
Demoting the later 1h instead would also make the request legal, but it
throws away 1h caching the client asked and paid for, which is the exact
regression #2375 / #2382 / #2651 were filed to stop.
Returns ``(system, messages, tools, stats)``. When nothing needed repairing
the three sections are the objects that were passed in.
"""
stats: dict[str, Any] = {
"violation": False,
"demoted": 0,
"promoted": 0,
"first_short_section": "",
"first_long_section": "",
}
if os.environ.get(_TTL_GUARD_ENV, "1").strip().lower() in ("0", "false", "no", "off"):
return system, messages, tools, stats
if not client_uses_1h:
def _contain(section: str, marker: dict[str, Any]) -> dict[str, Any] | None:
if cache_control_ttl_lane(marker) != CACHE_TTL_1H:
return None
stats["demoted"] += 1
if not stats["first_long_section"]:
stats["first_long_section"] = section
return {k: v for k, v in marker.items() if k != "ttl"}
system, messages, tools, contained = walk_cache_control(system, messages, tools, _contain)
if contained:
stats["violation"] = True
logger.warning(
"event=cache_control_ttl_order request_id=%s repair=lane_containment "
"demoted=%d leaked_from_section=%s; the client sent no 1h marker, so a "
"replayed 1h breakpoint would have been rejected upstream",
request_id,
stats["demoted"],
stats["first_long_section"],
)
return system, messages, tools, stats
# Ordering pass. Survey first so we know how far the violation reaches, then
# rewrite only the markers ahead of the last 1h one.
lanes: list[tuple[int, str, str]] = []
index = 0
def _survey(section: str, marker: dict[str, Any]) -> None:
nonlocal index
lanes.append((index, section, cache_control_ttl_lane(marker)))
index += 1
return None
walk_cache_control(system, messages, tools, _survey)
last_long = max((i for i, _s, lane in lanes if lane == CACHE_TTL_1H), default=-1)
offenders = [
(i, section) for i, section, lane in lanes if lane == CACHE_TTL_5M and i < last_long
]
if not offenders:
return system, messages, tools, stats
stats["violation"] = True
stats["first_short_section"] = offenders[0][1]
stats["first_long_section"] = next(s for i, s, lane in lanes if lane == CACHE_TTL_1H)
offending_indices = {i for i, _section in offenders}
cursor = 0
def _promote(_section: str, marker: dict[str, Any]) -> dict[str, Any] | None:
nonlocal cursor
position = cursor
cursor += 1
if position not in offending_indices:
return None
stats["promoted"] += 1
return {**marker, "ttl": CACHE_TTL_1H}
system, messages, tools, _ = walk_cache_control(system, messages, tools, _promote)
logger.warning(
"event=cache_control_ttl_order request_id=%s repair=promote_to_1h promoted=%d "
"first_short_section=%s first_long_section=%s; a 5m breakpoint preceded a 1h one, "
"which Anthropic rejects outright",
request_id,
stats["promoted"],
stats["first_short_section"],
stats["first_long_section"],
)
return system, messages, tools, stats
def log_cache_breakpoints(
*,
request_id: str | None,

354
tests/test_cache_control_ttl_order.py Normal file
View file

@ -0,0 +1,354 @@
"""The forwarded body must satisfy Anthropic's cache_control TTL ordering.
Anthropic evaluates cache breakpoints in one global walk -- ``tools``, then
``system``, then ``messages`` -- and rejects the whole request when a
``ttl="1h"`` marker appears after a 5-minute one (a bare ``{"type":
"ephemeral"}`` marker *is* 5m)::
400 messages.15.content.1.cache_control.ttl: a ttl='1h' cache_control block
must not come after a ttl='5m' cache_control block.
Headroom rewrites markers in several independent places, section by section,
and until #2939 nothing checked the rule that spans them. The failure is a dead
turn rather than a silent cost regression, so it needs tests that pin both
repair directions and, just as importantly, pin that a legal request is passed
through by identity.
"""
from typing import Any
import pytest
from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
from headroom.proxy.helpers import (
cache_control_ttl_lane,
cache_control_ttl_lanes,
enforce_cache_control_ttl_order,
inject_tool_search_deferral,
)
TTL_1H: dict[str, Any] = {"type": "ephemeral", "ttl": "1h"}
BARE: dict[str, Any] = {"type": "ephemeral"}
TTL_5M: dict[str, Any] = {"type": "ephemeral", "ttl": "5m"}
def _text(text: str, marker: dict[str, Any] | None = None) -> dict[str, Any]:
block: dict[str, Any] = {"type": "text", "text": text}
if marker is not None:
block["cache_control"] = marker
return block
def _msg(*blocks: dict[str, Any], role: str = "user") -> dict[str, Any]:
return {"role": role, "content": list(blocks)}
def _markers(system: Any, messages: Any, tools: Any) -> list[dict[str, Any]]:
"""Every marker in Anthropic's evaluation order."""
found: list[dict[str, Any]] = []
for tool in tools or []:
if isinstance(tool, dict) and isinstance(tool.get("cache_control"), dict):
found.append(tool["cache_control"])
for block in system or []:
if isinstance(block, dict) and isinstance(block.get("cache_control"), dict):
found.append(block["cache_control"])
for msg in messages or []:
if isinstance(msg.get("cache_control"), dict):
found.append(msg["cache_control"])
for block in msg.get("content") or []:
if not isinstance(block, dict):
continue
if isinstance(block.get("cache_control"), dict):
found.append(block["cache_control"])
for sub in block.get("content") or []:
if isinstance(sub, dict) and isinstance(sub.get("cache_control"), dict):
found.append(sub["cache_control"])
return found
def _is_legal(system: Any, messages: Any, tools: Any) -> bool:
"""Reimplements the API's rule independently of the code under test."""
seen_short = False
for marker in _markers(system, messages, tools):
lane = cache_control_ttl_lane(marker)
if lane == "5m":
seen_short = True
elif lane == "1h" and seen_short:
return False
return True
# ---------------------------------------------------------------------------
# Lane classification
@pytest.mark.parametrize(
("marker", "expected"),
[
({"type": "ephemeral"}, "5m"),
({"type": "ephemeral", "ttl": "5m"}, "5m"),
({"type": "ephemeral", "ttl": "1h"}, "1h"),
({"type": "ephemeral", "ttl": "24h"}, "other"),
("not-a-dict", "other"),
],
)
def test_lane_classification(marker: Any, expected: str) -> None:
# A bare marker must read as 5m, not "unknown": every ordinary Claude Code
# request sends bare markers, and calling those unknown would either mask
# real violations or invent imaginary ones.
assert cache_control_ttl_lane(marker) == expected
def test_lanes_survey_covers_all_three_sections() -> None:
lanes = cache_control_ttl_lanes(
[_text("sys", BARE)],
[_msg(_text("hi", TTL_1H))],
[{"name": "read", "cache_control": {"type": "ephemeral", "ttl": "7d"}}],
)
assert lanes == {"5m", "1h", "other"}
# ---------------------------------------------------------------------------
# Repair 1: lane containment -- the /btw case from #2939
def test_replayed_1h_is_stripped_when_client_asked_for_5m() -> None:
# Claude Code's `/btw` forks the conversation as a "side question", which is
# not on its 1h allowlist: the fork's tools/system breakpoints are bare 5m
# and it does not send the extended-cache-ttl beta header. Headroom's
# overlay of the previous turn's forwarded bytes drags a 1h marker into
# messages behind them, which is exactly the reported 400.
tools = [{"name": "read", "cache_control": dict(BARE)}]
system = [_text("sys", dict(BARE))]
messages = [_msg(_text("old"), _text("replayed", dict(TTL_1H)))]
system, messages, tools, stats = enforce_cache_control_ttl_order(
system, messages, tools, client_uses_1h=False
)
assert stats["violation"] is True
assert stats["demoted"] == 1
assert stats["first_long_section"] == "messages"
assert _markers(system, messages, tools) == [BARE, BARE, BARE], (
"the leaked 1h ttl should be dropped, leaving the marker itself in place"
)
assert _is_legal(system, messages, tools)
def test_containment_keeps_non_ttl_marker_fields() -> None:
# Claude Code also sends `scope` on its markers; only the ttl is at fault.
scoped = {"type": "ephemeral", "ttl": "1h", "scope": "global"}
_, messages, _, stats = enforce_cache_control_ttl_order(
None, [_msg(_text("x", scoped))], None, client_uses_1h=False
)
assert stats["demoted"] == 1
assert messages[0]["content"][0]["cache_control"] == {
"type": "ephemeral",
"scope": "global",
}
def test_client_1h_is_never_stripped() -> None:
system = [_text("sys", dict(TTL_1H))]
messages = [_msg(_text("hi", dict(TTL_1H)))]
out_system, out_messages, out_tools, stats = enforce_cache_control_ttl_order(
system, messages, None, client_uses_1h=True
)
assert stats["violation"] is False
assert out_system is system and out_messages is messages and out_tools is None
# ---------------------------------------------------------------------------
# Repair 2: ordering -- the #2767 case
def test_5m_in_tools_before_1h_in_messages_is_promoted() -> None:
# A transform downgraded the tools breakpoint while the client's message
# breakpoints are still 1h. Promoting restores what the client asked for;
# demoting would throw away 1h caching it is already paying for.
tools = [{"name": "read", "cache_control": dict(BARE)}]
messages = [_msg(_text("hi", dict(TTL_1H)))]
_, messages, tools, stats = enforce_cache_control_ttl_order(
None, messages, tools, client_uses_1h=True
)
assert stats["promoted"] == 1
assert stats["first_short_section"] == "tools"
assert stats["first_long_section"] == "messages"
assert tools[0]["cache_control"] == TTL_1H
assert messages[0]["content"][0]["cache_control"] == TTL_1H
assert _is_legal(None, messages, tools)
def test_5m_in_system_before_1h_in_messages_is_promoted() -> None:
system = [_text("sys", dict(TTL_5M))]
messages = [_msg(_text("hi", dict(TTL_1H)))]
system, messages, _, stats = enforce_cache_control_ttl_order(
system, messages, None, client_uses_1h=True
)
assert stats["first_short_section"] == "system"
assert system[0]["cache_control"] == TTL_1H
def test_violation_within_messages_is_promoted() -> None:
messages = [
_msg(_text("a", dict(BARE))),
_msg(_text("b"), _text("c", dict(TTL_1H))),
]
_, messages, _, stats = enforce_cache_control_ttl_order(
None, messages, None, client_uses_1h=True
)
assert stats["promoted"] == 1
assert messages[0]["content"][0]["cache_control"] == TTL_1H
def test_nested_tool_result_markers_participate() -> None:
# tool_result carries its own content list; a marker hiding in there is
# still a breakpoint the API walks, so it must count for the ordering.
messages = [
_msg(
{
"type": "tool_result",
"tool_use_id": "t1",
"content": [_text("inner", dict(BARE))],
}
),
_msg(_text("later", dict(TTL_1H))),
]
_, messages, _, stats = enforce_cache_control_ttl_order(
None, messages, None, client_uses_1h=True
)
assert stats["promoted"] == 1
assert messages[0]["content"][0]["content"][0]["cache_control"] == TTL_1H
def test_only_markers_before_the_last_1h_are_promoted() -> None:
# A 5m marker AFTER every 1h one is legal and must be left alone -- that is
# the ordering the API documents, not something to normalise away.
messages = [
_msg(_text("a", dict(BARE))),
_msg(_text("b", dict(TTL_1H))),
_msg(_text("c", dict(BARE))),
]
_, messages, _, stats = enforce_cache_control_ttl_order(
None, messages, None, client_uses_1h=True
)
assert stats["promoted"] == 1
assert [m["content"][0]["cache_control"] for m in messages] == [TTL_1H, TTL_1H, BARE]
# ---------------------------------------------------------------------------
# Pass-through cases
@pytest.mark.parametrize(
"markers",
[
pytest.param([TTL_1H, TTL_1H], id="all-1h"),
pytest.param([BARE, BARE], id="all-5m"),
pytest.param([TTL_1H, BARE], id="1h-then-5m"),
pytest.param([], id="no-markers"),
],
)
def test_legal_requests_are_returned_by_identity(markers: list[dict[str, Any]]) -> None:
messages = [_msg(_text(f"m{i}", dict(m))) for i, m in enumerate(markers)] or [_msg(_text("m"))]
out_system, out_messages, out_tools, stats = enforce_cache_control_ttl_order(
None, messages, None, client_uses_1h=True
)
assert stats["violation"] is False
assert out_messages is messages, "a legal body must not be rebuilt"
assert out_system is None and out_tools is None
def test_unknown_ttl_is_left_alone() -> None:
# Mirrors TtlOrderingWalk::observe in headroom-core: a TTL lane we don't
# model takes no part in the rule and is never rewritten.
messages = [_msg(_text("a", {"type": "ephemeral", "ttl": "24h"})), _msg(_text("b", dict(BARE)))]
_, out, _, stats = enforce_cache_control_ttl_order(None, messages, None, client_uses_1h=False)
assert stats["violation"] is False
assert out is messages
def test_kill_switch_disables_the_guard(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setenv("HEADROOM_CACHE_CONTROL_TTL_GUARD", "0")
tools = [{"name": "read", "cache_control": dict(BARE)}]
messages = [_msg(_text("hi", dict(TTL_1H)))]
_, out_messages, out_tools, stats = enforce_cache_control_ttl_order(
None, messages, tools, client_uses_1h=True
)
assert stats["violation"] is False
assert out_messages is messages and out_tools is tools
# ---------------------------------------------------------------------------
# Tool sort must not reorder markers
def _tools(count: int, marked: dict[int, dict[str, Any]] | None = None) -> list[dict[str, Any]]:
out: list[dict[str, Any]] = []
for i in range(count):
# Names descend so an alphabetical sort is guaranteed to reorder them.
tool: dict[str, Any] = {"name": f"tool_{count - i:02d}", "input_schema": {}}
if marked and i in marked:
tool["cache_control"] = dict(marked[i])
out.append(tool)
return out
def test_tool_sort_is_skipped_when_a_tool_carries_a_marker() -> None:
# A breakpoint on a tool means "cache through here"; sorting changes which
# tools are inside that prefix, and with two TTLs it can put the 1h marker
# behind the 5m one. The Rust proxy already refuses for the same reason.
tools = _tools(4, {1: TTL_1H, 3: BARE})
assert AnthropicHandlerMixin._sort_tools_deterministically(tools) is tools
def test_tool_sort_still_sorts_unmarked_tools() -> None:
tools = _tools(4)
ordered = AnthropicHandlerMixin._sort_tools_deterministically(tools)
assert [t["name"] for t in ordered] == sorted(t["name"] for t in tools), (
"clients that mark no tools must keep the deterministic ordering they rely on"
)
def test_tool_sort_would_have_created_the_violation() -> None:
# Pins the hazard itself: without the guard, the alphabetical sort moves the
# 5m-marked tool ahead of the 1h-marked one, which is a 400 on its own.
tools = _tools(4, {1: TTL_1H, 3: BARE})
assert _is_legal(None, [], tools)
assert not _is_legal(None, [], sorted(tools, key=AnthropicHandlerMixin._tool_sort_key))
# ---------------------------------------------------------------------------
# End-to-end regression for #2939 / #2767
def test_deferral_downgrade_then_guard_yields_a_legal_body() -> None:
# The #2767 shape: 13 tools, a 1h marker on one deferred tool and a bare
# marker on a LATER deferred tool. inject_tool_search_deferral keeps the
# last marker it stripped, so the tools prefix lands at 5m while the
# client's message breakpoints are still 1h -- a 400. The guard repairs it.
tools: list[dict[str, Any]] = [{"name": "read", "description": "core", "input_schema": {}}]
for i in range(12):
tool: dict[str, Any] = {"name": f"rare_{i}", "description": "rare", "input_schema": {}}
if i == 4:
tool["cache_control"] = dict(TTL_1H)
if i == 9:
tool["cache_control"] = dict(BARE)
tools.append(tool)
messages = [_msg(_text("history")), _msg(_text("newest", dict(TTL_1H)))]
deferred = inject_tool_search_deferral(tools)
assert deferred is not tools, "fixture no longer triggers the deferral"
assert not _is_legal(None, messages, deferred), (
"expected the downgraded tools breakpoint to make the body illegal"
)
_, messages, deferred, stats = enforce_cache_control_ttl_order(
None, messages, deferred, client_uses_1h=True
)
assert stats["violation"] is True
assert _is_legal(None, messages, deferred)