fix(cache): preserve cache_control ttl when re-anchoring a breakpoint (#2651)

## Description

`normalize_message_cache_control` deliberately reuses the client's
marker verbatim so an explicit `cache_control.ttl` (e.g. `"1h"`)
survives breakpoint consolidation instead of silently downgrading to the
5-minute default (#2375).

Two other sites also strip a breakpoint and re-place it, and both
hardcoded a bare `{"type": "ephemeral"}` — undoing that guarantee.

A downgrade is invisible: the request still succeeds, and the cost shows
up later as a full prefix re-write on every idle gap past 5 minutes.
Measured over 10,409 local Claude Code API requests, cache writes are
**6.1% of raw input tokens but 44.8% of the price-weighted input bill**
(5m write 1.25x vs read 0.1x), and **89% of those write tokens are
re-writes of content cached one request earlier**. Honoring a 1h TTL
when the client asks for it is the cheapest thing we can do about that.

Closes #

## 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/transforms/read_maturation.py` — `relocate_cache_breakpoint`
now carries the stripped marker forward when re-anchoring before the
held-Read region. This is the one that mattered most: it runs **after**
`normalize_message_cache_control` in the Anthropic handler
(`anthropic.py:1747` vs `:1642`), so it had the final say — a 1h client
with read maturation enabled was being downgraded to 5m.
- `headroom/proxy/helpers.py` — `inject_tool_search_deferral` keeps the
dropped marker when moving the tools-array breakpoint off a now-deferred
tool onto the last resident real tool.
- Both fall back to a bare ephemeral only when the client sent no ttl,
and neither invents a breakpoint where none existed.
- `headroom/transforms/compression_policy.py` — comment only. Notes that
`CACHE_WRITE_MULTIPLIER` is hardcoded to the 5m tier (1.25x), so a
client already on 1h caching (2.0x) has its mutations gated with a ~40%
under-stated write penalty. Harmless while the net-cost gate stays
default-off (`HEADROOM_NET_COST_POLICY`); names the plumbing needed if
it is ever enabled.

Both changed code paths sit behind off-by-default flags
(`HEADROOM_READ_MATURATION`, `HEADROOM_TOOL_SEARCH`), so this is a
latent-bug fix with **no default behavior change**.

## Testing

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

### Test Output

```text
$ python -m pytest tests/test_cache_ttl_preserved.py tests/test_read_maturation.py \
    tests/test_read_maturation_handler_nobust.py tests/test_cache_control_move_bust.py -q
tests/test_cache_ttl_preserved.py .....                                  [ 12%]
tests/test_read_maturation.py ......................                     [ 67%]
tests/test_read_maturation_handler_nobust.py ...                         [ 75%]
tests/test_cache_control_move_bust.py ..........                         [100%]
============================= 40 passed in 15.52s ==============================

$ ruff check headroom/ tests/ --exclude headroom/dashboard/templates
All checks passed!

$ mypy headroom
Success: no issues found in 509 source files
```

Broader regression sweep over every cache/breakpoint-adjacent suite:

```text
$ python -m pytest tests/ -q -k "read_maturation or tool_search or cache_control or prefix_tracker or ttl_preserved"
204 passed, 10145 deselected in 59.95s
```

## Real Behavior Proof

- **Environment:** macOS 25.4.0 (arm64), Python 3.12.6, pytest 9.0.2,
branched from `main` at e530de5a.
- **Exact command / steps:** verified the new tests actually fail
without the fix, rather than passing vacuously:
  ```
$ git stash push -- headroom/transforms/read_maturation.py
headroom/proxy/helpers.py
  $ python -m pytest tests/test_cache_ttl_preserved.py -q
  ```
- **Observed result:** exactly the two TTL-preservation tests fail, with
the downgrade visible in the assertion:
  ```text
  E   assert [{'type': 'ephemeral'}] == [{'ttl': '1h'... 'ephemeral'}]
E At index 0 diff: {'type': 'ephemeral'} != {'type': 'ephemeral', 'ttl':
'1h'}
FAILED
tests/test_cache_ttl_preserved.py::test_read_maturation_reanchor_keeps_ttl
FAILED
tests/test_cache_ttl_preserved.py::test_tool_search_deferral_keeps_ttl
========================= 2 failed, 3 passed in 0.56s
=========================
  ```
The other three pass either way, which is correct: they pin the 5m
default and the "don't invent a breakpoint" case. Restored with `git
stash pop`; all 5 pass again.
- **Not tested:** no live Anthropic request was made with `ttl: "1h"` —
both changed paths are behind off-by-default flags, and the corpus I
measured contains only 15 requests that ever used 1h TTL, so the 2.0x
write multiplier cited above is from Anthropic's price list, not
observed traffic. The `compression_policy.py` change is a comment and
has no runtime effect.

## 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`

Docs: N/A — no user-facing surface changes. The behavior being fixed (an
explicit client `cache_control.ttl` is preserved) is what the existing
`normalize_message_cache_control` docstring already promises; these two
sites were violating it.

## Additional Notes

**Scope deliberately kept to the fixes.** An earlier draft also added a
`HEADROOM_CACHE_LONGEVITY` flag that paired the existing cold-prefix
recompaction with an adaptive 5m→1h TTL upgrade for sessions observed
losing a warm prefix. That was dropped: a 1h write costs 2.0x vs 1.25x,
so it is a bet that a session idles often enough to repay the premium,
and the TTL lever is Anthropic-only (OpenAI/Codex cache automatically
with no TTL knob). It carried more side effects than the ~16% it
modelled was worth. The recompaction half already exists behind
`HEADROOM_COLD_RECOMPACT` and needs no new code.

**Follow-up worth considering separately:** the headline compression
savings figure is cache-blind — `cost.py:965-976` destructures the
cache-write price and discards it (`_cw_price`), and the savings-percent
denominator at `cost.py:568-570` includes the write premium while the
numerator does not, so a compression-induced cache bust *inflates*
reported savings. Given cache writes are ~45% of the effective input
bill, that seems worth its own issue.

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
 
 

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

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
 

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))) == []