diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py index f1096acc5..c926c4fb6 100644 --- a/headroom/proxy/helpers.py +++ b/headroom/proxy/helpers.py @@ -2902,6 +2902,7 @@ def inject_tool_search_deferral( out: list[Any] = [search_tool] deferred = 0 dropped_cache_control = False + dropped_marker: dict[str, Any] | None = None last_resident_real: dict[str, Any] | None = None resident_has_cache_control = False @@ -2928,8 +2929,13 @@ def inject_tool_search_deferral( continue new_tool = dict(tool) new_tool["defer_loading"] = True - if new_tool.pop("cache_control", None) is not None: + _dropped = new_tool.pop("cache_control", None) + if _dropped is not None: dropped_cache_control = True + # Keep the marker itself, not just the fact of it: re-placing a bare + # ephemeral would downgrade a 1h breakpoint to the 5m default. + if isinstance(_dropped, dict): + dropped_marker = _dropped out.append(new_tool) deferred += 1 @@ -2939,7 +2945,9 @@ def inject_tool_search_deferral( # deferred tool and no resident tool carries one, move it to the last # resident real tool (never the search tool, to keep its shape canonical). if dropped_cache_control and not resident_has_cache_control and last_resident_real is not None: - last_resident_real["cache_control"] = {"type": "ephemeral"} + last_resident_real["cache_control"] = ( + dict(dropped_marker) if dropped_marker else {"type": "ephemeral"} + ) return out diff --git a/headroom/transforms/compression_policy.py b/headroom/transforms/compression_policy.py index 626b40bc3..faf54fa60 100644 --- a/headroom/transforms/compression_policy.py +++ b/headroom/transforms/compression_policy.py @@ -58,6 +58,13 @@ _MAX_LOSSY_RATIO_SUBSCRIPTION: float = 0.25 #: Anthropic prompt-cache write multiplier: a ``cache_creation`` token #: costs 1.25x a plain input token (5-minute TTL tier). Input to the #: net-cost mutation formula (#856). Mirrors the Rust ``pub const``. +#: ponytail: hardcoded to the 5m tier. A client on Anthropic's 1h cache +#: (ENABLE_PROMPT_CACHING_1H / cache_control.ttl="1h", which Headroom +#: preserves) writes at 2.0x, so its mutations are gated with a ~40% +#: under-stated write penalty. Harmless while the net-cost gate stays +#: default-off (HEADROOM_NET_COST_POLICY); thread the TTL from +#: cold_prefix.anthropic_cache_ttl_seconds through ContentRouter -> +#: net_mutation_gain if that gate is ever turned on. CACHE_WRITE_MULTIPLIER: float = 1.25 #: Anthropic prompt-cache read multiplier: a ``cache_read`` token costs diff --git a/headroom/transforms/read_maturation.py b/headroom/transforms/read_maturation.py index 7efc653bd..309c369e4 100644 --- a/headroom/transforms/read_maturation.py +++ b/headroom/transforms/read_maturation.py @@ -343,12 +343,18 @@ def relocate_cache_breakpoint( stripped_any = False # 1. Strip breakpoints from the held region [earliest:]. + held_marker: dict[str, Any] | None = None for i in range(earliest, len(out)): msg = out[i] content = msg.get("content") if not isinstance(content, list): continue if any(isinstance(b, dict) and "cache_control" in b for b in content): + for b in content: + # Carry the TTL forward: re-anchoring with a bare ephemeral marker + # would silently downgrade a 1h breakpoint to the 5m default. + if isinstance(b, dict) and isinstance(b.get("cache_control"), dict): + held_marker = b["cache_control"] out[i] = { **msg, "content": [ @@ -371,7 +377,10 @@ def relocate_cache_breakpoint( content = out[i].get("content") if isinstance(content, list) and content and isinstance(content[-1], dict): new_content = list(content) - new_content[-1] = {**new_content[-1], "cache_control": {"type": "ephemeral"}} + new_content[-1] = { + **new_content[-1], + "cache_control": dict(held_marker) if held_marker else {"type": "ephemeral"}, + } out[i] = {**out[i], "content": new_content} break diff --git a/tests/test_cache_ttl_preserved.py b/tests/test_cache_ttl_preserved.py new file mode 100644 index 000000000..4aeabb4ce --- /dev/null +++ b/tests/test_cache_ttl_preserved.py @@ -0,0 +1,72 @@ +"""Re-anchored cache breakpoints must keep the client's TTL. + +``normalize_message_cache_control`` deliberately preserves an explicit +``cache_control.ttl`` so a client on Anthropic's 1h cache isn't silently +downgraded to the 5-minute default (#2375). Two other sites also strip a +breakpoint and re-place it, and both used to hardcode a bare ephemeral marker — +undoing that guarantee. A downgrade is invisible (the request still succeeds) +and costs a full prefix re-write on every gap past 5 minutes, so it needs a test +rather than a comment. +""" + +from typing import Any + +from headroom.proxy.helpers import inject_tool_search_deferral +from headroom.transforms.read_maturation import relocate_cache_breakpoint + +TTL_1H = {"type": "ephemeral", "ttl": "1h"} + + +def _markers(blocks: list[dict[str, Any]]) -> list[dict[str, Any]]: + return [b["cache_control"] for b in blocks if isinstance(b, dict) and "cache_control" in b] + + +def _held(marker: dict[str, Any]) -> list[dict[str, Any]]: + return [ + {"role": "user", "content": [{"type": "text", "text": "keep"}]}, + {"role": "user", "content": [{"type": "text", "text": "held", "cache_control": marker}]}, + ] + + +def test_read_maturation_reanchor_keeps_ttl() -> None: + # Breakpoint sits inside the held-Read region, so it is moved back before it. + out = relocate_cache_breakpoint(_held(TTL_1H), holding_msg_indices=[1]) + assert _markers(out[0]["content"]) == [TTL_1H], "re-anchored breakpoint lost the 1h ttl" + assert _markers(out[1]["content"]) == [], "held region should carry no breakpoint" + + +def test_read_maturation_reanchor_defaults_to_5m() -> None: + out = relocate_cache_breakpoint(_held({"type": "ephemeral"}), holding_msg_indices=[1]) + assert _markers(out[0]["content"]) == [{"type": "ephemeral"}] + + +def _tools(marker: dict[str, Any] | None) -> list[dict[str, Any]]: + # Needs >= _TOOL_SEARCH_MIN_TOOLS (12) to trigger, with one core tool resident + # and the tools-array breakpoint riding on a tool that will be deferred. + tools: list[dict[str, Any]] = [{"name": "read", "description": "core", "input_schema": {}}] + for i in range(12): + t: dict[str, Any] = {"name": f"rare_{i}", "description": "rare", "input_schema": {}} + if marker is not None and i == 11: + t["cache_control"] = marker + tools.append(t) + return tools + + +def _tool_markers(tools: Any) -> list[dict[str, Any]]: + return [t["cache_control"] for t in tools if isinstance(t, dict) and "cache_control" in t] + + +def test_tool_search_deferral_keeps_ttl() -> None: + out = inject_tool_search_deferral(_tools(TTL_1H)) + assert out is not _tools(TTL_1H), "deferral did not apply — fixture no longer triggers it" + assert _tool_markers(out) == [TTL_1H], "tools breakpoint lost the 1h ttl" + + +def test_tool_search_deferral_defaults_to_5m() -> None: + out = inject_tool_search_deferral(_tools({"type": "ephemeral"})) + assert _tool_markers(out) == [{"type": "ephemeral"}] + + +def test_tool_search_deferral_no_breakpoint_adds_none() -> None: + # Nothing was stripped, so nothing should be invented. + assert _tool_markers(inject_tool_search_deferral(_tools(None))) == []