diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index 5125c0a01..ac608f4b9 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/headroom/proxy/handlers/anthropic.py @@ -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, diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py index 31cbe4635..cdf7a2021 100644 --- a/headroom/proxy/helpers.py +++ b/headroom/proxy/helpers.py @@ -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, diff --git a/tests/test_cache_control_ttl_order.py b/tests/test_cache_control_ttl_order.py new file mode 100644 index 000000000..74bf146f6 --- /dev/null +++ b/tests/test_cache_control_ttl_order.py @@ -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)