fix(cache): preserve client cache_control ttl when consolidating breakpoints (#2382)

## Description

`normalize_message_cache_control()` consolidates message-level
`cache_control` breakpoints (strip all, re-place exactly one) to stay
under Anthropic's 4-block limit. The re-placed marker was hardcoded to
`{"type": "ephemeral"}`, so a client using 1-hour caching
(`cache_control: {"type": "ephemeral", "ttl": "1h"}`) was silently
downgraded to the 5-minute default on every consolidated turn — no
error, no signal, just quietly worse cache economics.

Fix: track the newest client marker while stripping, and re-place **that
marker verbatim** (a copy). Headroom keeps owning *where* the breakpoint
goes; the client keeps owning *what it says*. Older replayed markers
don't win — if the client's newest marker has no `ttl`, we don't
resurrect a stale `1h` (covered by a dedicated regression test).

Fixes #2375.

## Type of Change

- [x] Bug fix (silent 1h→5m cache downgrade)

## Changes Made

- `headroom/cache/prefix_tracker.py`:
`normalize_message_cache_control()` records the last marker dict seen in
message order and re-places a copy of it instead of a hardcoded
`{"type": "ephemeral"}`; docstring documents the ownership split.
- `tests/test_cache_control_move_bust.py`: 3 new tests — ttl preserved,
newest-marker-wins over stale ttls, ttl survives an 8-turn conversation
loop.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff` + `mypy`, CI-pinned settings)
- [x] Reproduced the bug first (2 new tests failed on the old code),
then verified the fix

### Test Output

```text
$ .venv/bin/python -m pytest tests/test_cache_control_move_bust.py -q
10 passed

# Before the fix, the two new ttl tests fail exactly as #2375 describes:
#   FAILED ...::test_normalize_preserves_ttl_of_newest_marker
#   FAILED ...::test_normalize_ttl_survives_many_turns

$ ruff check headroom/cache/prefix_tracker.py tests/test_cache_control_move_bust.py   # All checks passed!
$ ruff format --check <both files>                                                     # already formatted
$ mypy headroom/cache/prefix_tracker.py --ignore-missing-imports                       # Success: no issues
```

## Real Behavior Proof

- Environment: macOS (Darwin), Python in a uv venv, branch
`fix/cache-control-ttl-preserve` off `main` (`56c7d4a5`).
- Exact command / steps: drove `normalize_message_cache_control`
directly with a 2-message conversation whose marker carries `ttl: "1h"`,
printed the re-placed marker before/after the fix, and ran the new
regression tests against the unfixed code first.
- Observed result: before — output marker `{'type': 'ephemeral'}` (ttl
silently dropped); after — output marker `{'type': 'ephemeral', 'ttl':
'1h'}` with marker count still exactly 1 (the ≤4-block guarantee is
untouched).
- Not tested: a live Anthropic round-trip asserting
`cache_creation.ephemeral_1h_input_tokens` (needs a billed API call);
the marker dict forwarded on the wire is what the assertion pins.

## 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 — N/A
(docstring updated)
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md — N/A

## Additional Notes

- The `test_normalize_newest_marker_wins_over_stale_ttl` test also
guards against over-fixing (e.g. "any 1h seen anywhere wins"), which
would pin users to 1h pricing after they switch back to the default.

headroom/cache/prefix_tracker.py

@@ -379,16 +379,29 @@ def normalize_message_cache_control(
     ``messages`` and are left untouched (they still count toward the 4 limit, so
     holding messages to one breakpoint leaves room for them).
 
+    Headroom owns WHERE the breakpoint goes; the client still owns WHAT it says:
+    the re-placed marker reuses the newest client marker verbatim, so an explicit
+    ``ttl`` (e.g. ``"1h"``) survives consolidation instead of silently
+    downgrading to the 5-minute default (#2375).
+
     Only block-style (list) content can carry cache_control; string content is
     left as-is. Returns the input unchanged when there is nothing to normalize.
     """
     changed = False
     out: list[dict[str, Any]] = []
     last_block_idx = -1
+    last_marker: dict[str, Any] | None = None
     for i, msg in enumerate(messages):
         content = msg.get("content") if isinstance(msg, dict) else None
         if isinstance(content, list):
-            had = any(isinstance(b, dict) and "cache_control" in b for b in content)
+            had = False
+            for b in content:
+                if isinstance(b, dict) and "cache_control" in b:
+                    had = True
+                    # The newest marker in message order is the client's current
+                    # intent (older ones are replay leftovers) — keep it.
+                    if isinstance(b["cache_control"], dict):
+                        last_marker = b["cache_control"]
             stripped = [
                 {k: v for k, v in b.items() if k != "cache_control"} if isinstance(b, dict) else b
                 for b in content
@@ -403,7 +416,8 @@ def normalize_message_cache_control(
     if last_block_idx >= 0:
         msg = out[last_block_idx]
         content = list(msg["content"])
-        content[-1] = {**content[-1], "cache_control": {"type": "ephemeral"}}
+        marker = dict(last_marker) if last_marker else {"type": "ephemeral"}
+        content[-1] = {**content[-1], "cache_control": marker}
         out[last_block_idx] = {**msg, "content": content}
         changed = True
     return out if changed else messages

tests/test_cache_control_move_bust.py

@@ -188,3 +188,39 @@ def test_normalize_is_noop_when_no_block_markers():
     # places exactly one breakpoint (so the prefix gets cached), content stable
     assert _markers(out) == 1
     assert _strip_cache_control(out) == _strip_cache_control(plain)
+
+
+# ── fix-3 (#2375): consolidation must not silently drop the client's ttl ─────
+
+
+def B_ttl(role, text, ttl):
+    """Block-style message whose marker carries an explicit ttl (1h caching)."""
+    blk = {"type": "text", "text": text, "cache_control": {"type": "ephemeral", "ttl": ttl}}
+    return {"role": role, "content": [blk]}
+
+
+def test_normalize_preserves_ttl_of_newest_marker():
+    """A 1h-ttl client must not be silently downgraded to the 5m default."""
+    msgs = [B("user", "a", cc=True), B_ttl("user", "b", "1h")]
+    out = normalize_message_cache_control(msgs)
+    assert _markers(out) == 1
+    assert out[-1]["content"][-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}
+
+
+def test_normalize_newest_marker_wins_over_stale_ttl():
+    # Older replayed markers still carry 1h, but the client's NEWEST marker has
+    # no ttl — the client switched back to the default; don't resurrect 1h.
+    msgs = [B_ttl("user", "a", "1h"), B_ttl("assistant", "b", "1h"), B("user", "c", cc=True)]
+    out = normalize_message_cache_control(msgs)
+    assert _markers(out) == 1
+    assert out[-1]["content"][-1]["cache_control"] == {"type": "ephemeral"}
+
+
+def test_normalize_ttl_survives_many_turns():
+    """The #2375 scenario: ttl held for one turn, gone on every later turn."""
+    conv = []
+    for t in range(1, 8):
+        conv = conv + [B_ttl("user", f"turn-{t}", "1h")]  # client always asks 1h
+        conv = normalize_message_cache_control(conv)
+        assert _markers(conv) == 1
+        assert conv[-1]["content"][-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}