fix(cache): mirror client cache_control positions instead of single-marker consolidation

Preserve client cache-control breakpoint positions.

headroom/cache/compression_store.py

@@ -52,6 +52,10 @@ DEFAULT_CCR_TTL_SECONDS = 1800  # session-scale; override via HEADROOM_CCR_TTL_S
 CCR_TTL_SECONDS_ENV = "HEADROOM_CCR_TTL_SECONDS"
 
 _RETRIEVAL_LOG_PREVIEW_CHARS = 4096
+# Previews carry verbatim tool-result content (post-redaction), which makes
+# proxy.log too sensitive for users to share in bug reports. Set to
+# 0/false/no/off to log byte counts only.
+PAYLOAD_PREVIEW_ENV = "HEADROOM_LOG_PAYLOAD_PREVIEW"
 _SECRET_KEY_VALUE_RE = re.compile(
     r"(?i)\b([A-Z0-9_-]*(?:API[_-]?KEY|TOKEN|SECRET|PASSWORD|CREDENTIAL|AUTH)[A-Z0-9_-]*)"
     r"(\s*[:=]\s*)([\"']?)([^\"'\s,}]+)"
@@ -108,7 +112,21 @@ def _redact_retrieval_log_payload(payload: str) -> str:
     return _API_KEY_VALUE_RE.sub("sk-[REDACTED]", redacted)
 
 
+def _payload_preview_enabled() -> bool:
+    raw = os.environ.get(PAYLOAD_PREVIEW_ENV)
+    if raw is None:
+        return True
+    return raw.strip().lower() not in ("0", "false", "no", "off")
+
+
 def _payload_for_retrieval_log(payload: str) -> dict[str, Any]:
+    if not _payload_preview_enabled():
+        return {
+            "payload_chars": len(payload),
+            "payload_preview_chars": 0,
+            "payload_truncated": len(payload) > 0,
+            "payload_preview": "",
+        }
     redacted = _redact_retrieval_log_payload(payload)
     preview = redacted[:_RETRIEVAL_LOG_PREVIEW_CHARS]
     truncated = len(redacted) > len(preview)

headroom/cache/prefix_tracker.py

@@ -633,9 +633,31 @@ def _breakpoint_index(
     return relation.stable_prefix_blocks - 1
 
 
+def _client_marker_positions(
+    client_messages: list[dict[str, Any]],
+) -> list[tuple[int, int, dict[str, Any]]]:
+    """(message index, block index, marker) for every CLIENT cache_control.
+
+    Block-level, not one-per-message: clients mark multiple blocks within a
+    single long message (Claude Code does this on 1-2-message requests with a
+    large first message), and the ~20-block lookback applies within a message
+    just as it does across messages. Only block-style content carries markers.
+    """
+    positions: list[tuple[int, int, dict[str, Any]]] = []
+    for i, msg in enumerate(client_messages):
+        content = msg.get("content") if isinstance(msg, dict) else None
+        if not isinstance(content, list):
+            continue
+        for bi, b in enumerate(content):
+            if isinstance(b, dict) and isinstance(b.get("cache_control"), dict):
+                positions.append((i, bi, b["cache_control"]))
+    return positions
+
+
 def normalize_message_cache_control(
     messages: list[dict[str, Any]],
     previous_forwarded_messages: list[dict[str, Any]] | None = None,
+    client_messages: list[dict[str, Any]] | None = None,
 ) -> list[dict[str, Any]]:
     """Own message-level cache_control placement so breakpoints stay bounded.
 
@@ -645,25 +667,35 @@ def normalize_message_cache_control(
     hard-errors at **>4 cache_control blocks total** (system + tools + messages),
     so on a long conversation the accumulation eventually 400s.
 
-    Fix: strip EVERY message-level cache_control and re-place a **single**
-    ephemeral breakpoint. Pure append-only growth keeps it on the newest block,
-    which both reads the prior write and writes the appended tail. If last turn's
-    counterpart proves that the tail was rewritten, it is placed at the end of
-    the byte-stable leading run instead. One breakpoint caches the prefix up to
-    it, and — because the provider's cache key is message CONTENT, not marker
-    presence (moving the
-    breakpoint forward is the documented client pattern and it hits) — stripping
-    and re-placing markers never busts. system/tools breakpoints live outside
-    ``messages`` and are left untouched (they still count toward the 4 limit, so
-    holding messages to one breakpoint leaves room for them).
+    Fix: strip EVERY message-level cache_control, then re-place markers at the
+    positions the CLIENT's current request marks (``client_messages``). The
+    client's positions are load-bearing, not redundant: Anthropic resolves each
+    breakpoint by walking back **at most ~20 content blocks** for a prior cache
+    entry, and agentic clients (Claude Code) keep a marker on the previous
+    turn's newest message precisely so the new turn's write can chain to the
+    old entry. Collapsing to a single newest-block marker breaks that chain
+    whenever one turn adds >20 blocks (typical for tool-heavy turns): the
+    lookback misses, the entire message history silently re-bills as cache
+    creation, and a marker anchored short of the final block leaves the tail
+    billing as fully uncached input. Mirroring the client's positions bounds
+    accumulation identically (the client manages its own 4-marker budget) while
+    preserving its read/write chaining.
 
-    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).
+    The provider's cache key is message CONTENT, not marker presence (moving
+    the breakpoint forward is the documented client pattern and it hits), so
+    stripping replay leftovers and re-placing markers never busts. system/tools
+    breakpoints live outside ``messages`` and are left untouched.
 
-    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.
+    Headroom owns WHICH BLOCK carries each marker; the client owns the message
+    positions and the marker values, so an explicit ``ttl`` (e.g. ``"1h"``)
+    survives per position instead of silently downgrading (#2375). The newest
+    position uses stable-run anchoring for proven rewritten tails; earlier
+    positions go on their message's last block.
+
+    Without ``client_messages`` (or when the transformed list no longer aligns
+    with it), falls back to the legacy single-marker consolidation. 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]] = []
@@ -690,22 +722,75 @@ def normalize_message_cache_control(
                 last_block_idx = i
         else:
             out.append(msg)
-    # Re-place exactly one breakpoint on the last block-style message.
-    if last_block_idx >= 0:
-        msg = out[last_block_idx]
-        content = list(msg["content"])
-        marker = dict(last_marker) if last_marker else {"type": "ephemeral"}
-        breakpoint_index = _breakpoint_index(
-            content, msg, last_block_idx, previous_forwarded_messages
-        )
+
+    def _place(
+        target_idx: int,
+        marker: dict[str, Any],
+        *,
+        anchor: bool,
+        block_idx: int | None = None,
+    ) -> bool:
+        msg = out[target_idx]
+        content = msg.get("content")
+        if not isinstance(content, list) or not content:
+            return False
+        content = list(content)
+        if block_idx is not None and 0 <= block_idx < len(content):
+            # Transforms can shift block indices (e.g. a dropped thinking
+            # block); a slightly-off placement still lands on a stable block
+            # in the same message, which is harmless — markers are not part
+            # of the provider's cache key.
+            breakpoint_index = block_idx
+        elif anchor:
+            breakpoint_index = _breakpoint_index(
+                content, msg, target_idx, previous_forwarded_messages
+            )
+        else:
+            breakpoint_index = len(content) - 1
         # Anthropic content blocks are dictionaries, but callers can still
-        # supply mixed list content.  The newest block is known to be a dict
-        # from the scan above; fall back to it rather than attempting ``**``
-        # on a scalar stable-boundary element.
+        # supply mixed list content. Fall back to the newest block rather than
+        # attempting ``**`` on a scalar stable-boundary element, and skip the
+        # message entirely when even that is not a dict.
         if not isinstance(content[breakpoint_index], dict):
             breakpoint_index = len(content) - 1
-        content[breakpoint_index] = {**content[breakpoint_index], "cache_control": marker}
-        out[last_block_idx] = {**msg, "content": content}
+        if not isinstance(content[breakpoint_index], dict):
+            return False
+        content[breakpoint_index] = {**content[breakpoint_index], "cache_control": dict(marker)}
+        out[target_idx] = {**msg, "content": content}
+        return True
+
+    # Preferred: mirror the client's marker positions 1:1, block-level. The
+    # transform pipeline preserves message count, so index alignment is the
+    # invariant; fall back to legacy consolidation if it ever does not hold,
+    # or when the client marked nothing (legacy still places one so the
+    # prefix caches). The newest client marker keeps stable-run anchoring
+    # when the client placed it on its message's final block (intent: "cache
+    # through the end"); an explicit mid-message marker is honored verbatim.
+    if client_messages is not None and len(client_messages) == len(messages):
+        positions = _client_marker_positions(client_messages)
+        if positions:
+            placed_any = False
+            newest_mi, newest_bi, _ = positions[-1]
+            newest_client_content = client_messages[newest_mi].get("content")
+            newest_on_final_block = (
+                isinstance(newest_client_content, list)
+                and newest_bi == len(newest_client_content) - 1
+            )
+            for mi, bi, marker in positions:
+                is_newest = (mi, bi) == (newest_mi, newest_bi)
+                if is_newest and newest_on_final_block:
+                    placed = _place(mi, marker, anchor=True)
+                else:
+                    placed = _place(mi, marker, anchor=False, block_idx=bi)
+                placed_any = placed or placed_any
+            if placed_any or changed:
+                return out
+            return messages
+
+    # Legacy: re-place exactly one breakpoint on the last block-style message.
+    if last_block_idx >= 0:
+        marker = dict(last_marker) if last_marker else {"type": "ephemeral"}
+        _place(last_block_idx, marker, anchor=True)
         changed = True
     return out if changed else messages
 

headroom/proxy/handlers/anthropic.py

@@ -290,7 +290,16 @@ class AnthropicHandlerMixin:
             new_content: list[dict[str, Any]] = []
             appended = False
             for block in content:
-                if not appended and isinstance(block, dict) and block.get("type") == "text":
+                if (
+                    not appended
+                    and isinstance(block, dict)
+                    and block.get("type") == "text"
+                    # Never mutate a block carrying the client's cache
+                    # breakpoint: the client re-sends the original bytes
+                    # next turn, so any injection here busts the prefix
+                    # cache at this message from then on.
+                    and "cache_control" not in block
+                ):
                     existing = block.get("text", "")
                     new_content.append({**block, "text": existing + "\n\n" + context_text})
                     appended = True
@@ -736,6 +745,16 @@ class AnthropicHandlerMixin:
             if input_event.tools is not None:
                 body["tools"] = input_event.tools
 
+            # Snapshot the client's cache_control breakpoints before any
+            # 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
+
+            inbound_breakpoints = count_cache_breakpoints(
+                body.get("system"), messages, body.get("tools")
+            )
+
             # Validate message array size
             if len(messages) > MAX_MESSAGE_ARRAY_LENGTH:
                 await _finalize_pre_upstream()
@@ -1661,12 +1680,19 @@ class AnthropicHandlerMixin:
 
             # Own cache_control placement: the client moves the breakpoint each
             # turn and the overlay replays past markers, so they accumulate ~1/turn
-            # and Anthropic hard-errors at >4. Strip message-level markers and keep
-            # one breakpoint. Pure appends advance it to the newest block; a
-            # proven rewritten tail anchors it at the byte-stable boundary so
-            # that the same prefix is readable next turn. Applied last so the
+            # and Anthropic hard-errors at >4. Strip message-level markers and
+            # re-place them at the CLIENT's current positions: Anthropic resolves
+            # each breakpoint with a ~20-content-block lookback, and the client's
+            # previous-message marker is the read anchor that lets a big turn's
+            # write chain to the prior entry. Collapsing to one newest-block
+            # marker breaks that chain on tool-heavy turns (silent full re-write)
+            # and can leave the tail billing uncached. Applied last so the
             # forwarded AND recorded (next_forwarded) messages stay bounded.
-            _norm = normalize_message_cache_control(optimized_messages, previous_forwarded_messages)
+            _norm = normalize_message_cache_control(
+                optimized_messages,
+                previous_forwarded_messages,
+                client_messages=original_client_messages,
+            )
             if _norm is not optimized_messages:
                 optimized_messages = _norm
 
@@ -2971,6 +2997,16 @@ class AnthropicHandlerMixin:
                         "upstream request for server-side retrieval handling"
                     )
 
+                from headroom.proxy.helpers import log_cache_breakpoints
+
+                log_cache_breakpoints(
+                    request_id=request_id,
+                    inbound=inbound_breakpoints,
+                    outbound=count_cache_breakpoints(
+                        body.get("system"), body.get("messages"), body.get("tools")
+                    ),
+                )
+
                 if stream and not buffered_stream_ccr:
                     self.pipeline_extensions.emit(
                         PipelineStage.POST_SEND,

headroom/proxy/helpers.py

@@ -347,6 +347,112 @@ def log_outbound_request(
     )
 
 
+def count_cache_breakpoints(
+    system: Any,
+    messages: Any,
+    tools: Any,
+) -> dict[str, int]:
+    """Count client ``cache_control`` breakpoints per request section.
+
+    Besides raw counts, records how far from the END of the message list the
+    last marker sits (``last_marker_tail`` = messages after the last marked
+    one). A dropped or backward-moved final breakpoint — the signature of a
+    "large uncached tail next to a healthy cache read" billing regression —
+    shows up as ``last_marker_tail`` growing between inbound and outbound.
+    Nested markers inside ``tool_result`` list content are counted too, so a
+    transform that rewrites sub-blocks can't lose one invisibly.
+    """
+    system_count = 0
+    if isinstance(system, list):
+        system_count = sum(1 for b in system if isinstance(b, dict) and "cache_control" in b)
+
+    tools_count = 0
+    if isinstance(tools, list):
+        tools_count = sum(1 for t in tools if isinstance(t, dict) and "cache_control" in t)
+
+    message_count = 0
+    messages_total = 0
+    last_marker_index = -1
+    if isinstance(messages, list):
+        message_count = len(messages)
+        for i, msg in enumerate(messages):
+            if not isinstance(msg, dict):
+                continue
+            found = 1 if "cache_control" in msg else 0
+            content = msg.get("content")
+            if isinstance(content, list):
+                for block in content:
+                    if not isinstance(block, dict):
+                        continue
+                    if "cache_control" in block:
+                        found += 1
+                    inner = block.get("content")
+                    if isinstance(inner, list):
+                        found += sum(
+                            1 for sub in inner if isinstance(sub, dict) and "cache_control" in sub
+                        )
+            if found:
+                messages_total += found
+                last_marker_index = i
+
+    last_marker_tail = message_count - 1 - last_marker_index if last_marker_index >= 0 else -1
+    return {
+        "system": system_count,
+        "tools": tools_count,
+        "messages": messages_total,
+        "total": system_count + tools_count + messages_total,
+        "message_count": message_count,
+        "last_marker_tail": last_marker_tail,
+    }
+
+
+def log_cache_breakpoints(
+    *,
+    request_id: str | None,
+    inbound: dict[str, int],
+    outbound: dict[str, int],
+) -> None:
+    """One structured line per request: client breakpoints in vs forwarded out.
+
+    Per realignment build constraints: every cache-affecting decision is
+    logged. Escalates to WARNING when the forwarded request has fewer
+    breakpoints than the client sent, or the last marker moved further from
+    the end of the message list — either one silently un-caches the tail.
+    """
+    dropped = outbound["total"] < inbound["total"]
+    tail_grew = (
+        inbound["last_marker_tail"] >= 0
+        and outbound["last_marker_tail"] != inbound["last_marker_tail"]
+        and (
+            outbound["last_marker_tail"] < 0
+            or outbound["last_marker_tail"] > inbound["last_marker_tail"]
+        )
+    )
+    log = logger.warning if (dropped or tail_grew) else logger.info
+    log(
+        "event=cache_breakpoints request_id=%s "
+        "in_total=%d out_total=%d in_system=%d out_system=%d "
+        "in_tools=%d out_tools=%d in_messages=%d out_messages=%d "
+        "in_msg_count=%d out_msg_count=%d in_last_tail=%d out_last_tail=%d "
+        "dropped=%s tail_grew=%s",
+        request_id or "",
+        inbound["total"],
+        outbound["total"],
+        inbound["system"],
+        outbound["system"],
+        inbound["tools"],
+        outbound["tools"],
+        inbound["messages"],
+        outbound["messages"],
+        inbound["message_count"],
+        outbound["message_count"],
+        inbound["last_marker_tail"],
+        outbound["last_marker_tail"],
+        "true" if dropped else "false",
+        "true" if tail_grew else "false",
+    )
+
+
 def log_memory_injection(
     *,
     request_id: str,

tests/test_cache_breakpoint_diagnostics.py

@@ -0,0 +1,180 @@
+"""Tests for cache_control breakpoint diagnostics and log-privacy switches.
+
+Covers the three pieces added for the uncached-tail investigation:
+- ``count_cache_breakpoints`` / ``log_cache_breakpoints`` (proxy helpers)
+- the ``HEADROOM_LOG_PAYLOAD_PREVIEW`` kill switch (compression store)
+- the injection guard that keeps proactive expansion out of breakpointed blocks
+"""
+
+from __future__ import annotations
+
+import logging
+
+from headroom.cache.compression_store import _payload_for_retrieval_log
+from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
+from headroom.proxy.helpers import count_cache_breakpoints, log_cache_breakpoints
+
+_CC = {"cache_control": {"type": "ephemeral"}}
+
+
+def _claude_code_style_request() -> tuple[list[dict], list[dict], list[dict]]:
+    """System/messages/tools shaped like a real Claude Code request."""
+    system = [
+        {"type": "text", "text": "You are Claude Code."},
+        {"type": "text", "text": "project instructions", **_CC},
+    ]
+    tools = [
+        {"name": "Bash", "input_schema": {}},
+        {"name": "Read", "input_schema": {}, **_CC},
+    ]
+    messages = [
+        {"role": "user", "content": [{"type": "text", "text": "hi", **_CC}]},
+        {"role": "assistant", "content": [{"type": "text", "text": "ack"}]},
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "tool_result",
+                    "tool_use_id": "t1",
+                    "content": [{"type": "text", "text": "big output"}],
+                    **_CC,
+                }
+            ],
+        },
+    ]
+    return system, messages, tools
+
+
+def test_count_cache_breakpoints_counts_all_sections() -> None:
+    system, messages, tools = _claude_code_style_request()
+    stats = count_cache_breakpoints(system, messages, tools)
+    assert stats["system"] == 1
+    assert stats["tools"] == 1
+    assert stats["messages"] == 2
+    assert stats["total"] == 4
+    assert stats["message_count"] == 3
+    assert stats["last_marker_tail"] == 0  # last message carries a marker
+
+
+def test_count_cache_breakpoints_counts_nested_tool_result_markers() -> None:
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {
+                    "type": "tool_result",
+                    "tool_use_id": "t1",
+                    "content": [{"type": "text", "text": "out", **_CC}],
+                }
+            ],
+        }
+    ]
+    stats = count_cache_breakpoints("plain system string", messages, None)
+    assert stats["system"] == 0
+    assert stats["tools"] == 0
+    assert stats["messages"] == 1
+    assert stats["last_marker_tail"] == 0
+
+
+def test_count_cache_breakpoints_tail_tracks_last_marker() -> None:
+    messages = [
+        {"role": "user", "content": [{"type": "text", "text": "a", **_CC}]},
+        {"role": "assistant", "content": [{"type": "text", "text": "b"}]},
+        {"role": "user", "content": [{"type": "text", "text": "c"}]},
+    ]
+    stats = count_cache_breakpoints(None, messages, None)
+    assert stats["last_marker_tail"] == 2
+    assert count_cache_breakpoints(None, [], None)["last_marker_tail"] == -1
+
+
+def test_log_cache_breakpoints_warns_on_dropped_marker(caplog) -> None:
+    system, messages, tools = _claude_code_style_request()
+    inbound = count_cache_breakpoints(system, messages, tools)
+    # Transform "lost" the final breakpoint: strip it from the last message.
+    stripped = [dict(m) for m in messages]
+    stripped[2] = {
+        "role": "user",
+        "content": [{"type": "tool_result", "tool_use_id": "t1", "content": "compressed"}],
+    }
+    outbound = count_cache_breakpoints(system, stripped, tools)
+    with caplog.at_level(logging.INFO, logger="headroom.proxy"):
+        log_cache_breakpoints(request_id="r1", inbound=inbound, outbound=outbound)
+    [record] = caplog.records
+    assert record.levelno == logging.WARNING
+    assert "dropped=true" in record.getMessage()
+    assert "tail_grew=true" in record.getMessage()
+
+
+def test_log_cache_breakpoints_info_when_preserved(caplog) -> None:
+    system, messages, tools = _claude_code_style_request()
+    stats = count_cache_breakpoints(system, messages, tools)
+    with caplog.at_level(logging.INFO, logger="headroom.proxy"):
+        log_cache_breakpoints(request_id="r1", inbound=stats, outbound=stats)
+    [record] = caplog.records
+    assert record.levelno == logging.INFO
+    assert "dropped=false" in record.getMessage()
+
+
+def test_payload_preview_disabled_omits_content(monkeypatch) -> None:
+    monkeypatch.setenv("HEADROOM_LOG_PAYLOAD_PREVIEW", "0")
+    payload = "secret file contents: api_key=sk-abcdefghijklmnop"
+    event = _payload_for_retrieval_log(payload)
+    assert event["payload_preview"] == ""
+    assert event["payload_preview_chars"] == 0
+    assert event["payload_chars"] == len(payload)
+    assert event["payload_truncated"] is True
+
+
+def test_payload_preview_enabled_by_default(monkeypatch) -> None:
+    monkeypatch.delenv("HEADROOM_LOG_PAYLOAD_PREVIEW", raising=False)
+    event = _payload_for_retrieval_log("hello world")
+    assert event["payload_preview"] == "hello world"
+
+
+def test_append_context_skips_breakpointed_text_block() -> None:
+    messages = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "breakpointed", **_CC},
+                {"type": "text", "text": "free"},
+            ],
+        }
+    ]
+    result = AnthropicHandlerMixin._append_context_to_latest_non_frozen_user_turn(
+        messages, "CTX", frozen_message_count=0
+    )
+    blocks = result[0]["content"]
+    assert blocks[0]["text"] == "breakpointed"  # untouched
+    assert blocks[1]["text"].endswith("CTX")
+
+
+def test_append_context_no_eligible_block_returns_unchanged() -> None:
+    messages = [
+        {
+            "role": "user",
+            "content": [{"type": "text", "text": "breakpointed", **_CC}],
+        }
+    ]
+    result = AnthropicHandlerMixin._append_context_to_latest_non_frozen_user_turn(
+        messages, "CTX", frozen_message_count=0
+    )
+    assert result == messages
+
+
+def test_count_cache_breakpoints_tolerates_malformed_shapes() -> None:
+    messages = [
+        "not-a-dict",
+        {"role": "user", "content": ["scalar-block", {"type": "text", "text": "x", **_CC}]},
+        {"role": "user", "content": "plain string"},
+    ]
+    stats = count_cache_breakpoints("system-as-string", messages, "tools-as-string")
+    assert stats["system"] == 0
+    assert stats["tools"] == 0
+    assert stats["messages"] == 1
+    assert stats["message_count"] == 3
+    assert stats["last_marker_tail"] == 1
+
+    empty = count_cache_breakpoints(None, None, None)
+    assert empty["total"] == 0
+    assert empty["message_count"] == 0

tests/test_cache_control_move_bust.py

@@ -224,3 +224,177 @@ def test_normalize_ttl_survives_many_turns():
         conv = normalize_message_cache_control(conv)
         assert _markers(conv) == 1
         assert conv[-1]["content"][-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}
+
+
+# ── fix-4: mirror the CLIENT's marker positions (20-block lookback chain) ────
+# Anthropic resolves each breakpoint by walking back at most ~20 content
+# blocks. Agentic clients keep a marker on the previous turn's newest message
+# as the read anchor; collapsing to a single newest-block marker breaks the
+# chain on tool-heavy turns. With client_messages provided, normalize must
+# keep exactly the client's positions.
+
+
+def test_normalize_mirrors_client_marker_positions():
+    client = [
+        B("user", "a", cc=True),
+        B("assistant", "b"),
+        B("user", "c", cc=True),  # read anchor (previous newest)
+        B("assistant", "d"),
+        B("user", "e", cc=True),  # newest
+    ]
+    # Forwarded form: replay leftovers piled markers onto other messages too.
+    merged = [
+        B("user", "a", cc=True),
+        B("assistant", "b", cc=True),
+        B("user", "c", cc=True),
+        B("assistant", "d", cc=True),
+        B("user", "e", cc=True),
+    ]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 3  # exactly the client's three, not one
+    for idx in (0, 2, 4):
+        assert "cache_control" in out[idx]["content"][-1], idx
+    for idx in (1, 3):
+        assert _markers([out[idx]]) == 0, idx
+    assert _strip_cache_control(out) == _strip_cache_control(merged)
+
+
+def test_normalize_mirror_preserves_per_position_ttl():
+    client = [B_ttl("user", "a", "1h"), B("user", "b", cc=True)]
+    merged = [B("user", "a", cc=True), B("user", "b", cc=True)]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert out[0]["content"][-1]["cache_control"] == {"type": "ephemeral", "ttl": "1h"}
+    assert out[1]["content"][-1]["cache_control"] == {"type": "ephemeral"}
+
+
+def test_normalize_mirror_bounded_across_many_turns():
+    """Client moves its pair of markers forward; forwarded stays at client count."""
+    conv = []
+    for t in range(1, 12):
+        conv = conv + [B("user", f"turn-{t}")]
+        # Client marks the newest and second-newest marked position (CC pattern).
+        client = [dict(m) for m in conv]
+        client[-1] = B("user", f"turn-{t}", cc=True)
+        if len(client) >= 2:
+            client[-2] = B(client[-2]["role"], client[-2]["content"][0]["text"], cc=True)
+        # Forwarded side accumulated replay leftovers everywhere.
+        merged = [B(m["role"], m["content"][0]["text"], cc=True) for m in client]
+        out = normalize_message_cache_control(merged, client_messages=client)
+        assert _markers(out) == min(2, len(client))
+        assert "cache_control" in out[-1]["content"][-1]
+
+
+def test_normalize_falls_back_when_counts_mismatch():
+    client = [B("user", "a", cc=True)]  # transform changed message count
+    merged = [B("user", "a", cc=True), B("user", "b", cc=True)]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 1  # legacy consolidation
+    assert "cache_control" in out[-1]["content"][-1]
+
+
+def test_normalize_falls_back_when_client_has_no_markers():
+    client = [B("user", "a"), B("user", "b")]
+    merged = [B("user", "a", cc=True), B("user", "b")]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 1  # legacy: still place one so the prefix caches
+    assert "cache_control" in out[-1]["content"][-1]
+
+
+def test_normalize_mirror_skips_string_content_positions():
+    # Client marked message 0; forwarded counterpart is string-content (cannot
+    # carry a marker) — the position is skipped, the rest still mirror.
+    client = [B("user", "a", cc=True), B("user", "b", cc=True)]
+    merged = [{"role": "user", "content": "a"}, B("user", "b", cc=True)]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 1
+    assert "cache_control" in out[1]["content"][-1]
+
+
+# ── fix-5: block-level mirroring (multiple client markers in one message) ────
+
+
+def test_normalize_mirrors_multiple_markers_within_one_message():
+    """A 2-message request where the client marks TWO blocks of message 0
+    (Claude Code's pattern on large first messages) keeps all three markers."""
+    big = {
+        "role": "user",
+        "content": [
+            {"type": "text", "text": "part-1", "cache_control": {"type": "ephemeral"}},
+            {"type": "text", "text": "part-2"},
+            {"type": "text", "text": "part-3", "cache_control": {"type": "ephemeral"}},
+        ],
+    }
+    client = [big, B("user", "follow-up", cc=True)]
+    # Forwarded form: an extra replay leftover on message 1's sibling... use
+    # identical structure with markers everywhere to prove selective stripping.
+    merged = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "part-1", "cache_control": {"type": "ephemeral"}},
+                {"type": "text", "text": "part-2", "cache_control": {"type": "ephemeral"}},
+                {"type": "text", "text": "part-3", "cache_control": {"type": "ephemeral"}},
+            ],
+        },
+        B("user", "follow-up", cc=True),
+    ]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 3
+    assert "cache_control" in out[0]["content"][0]
+    assert "cache_control" not in out[0]["content"][1]
+    assert "cache_control" in out[0]["content"][2]
+    assert "cache_control" in out[1]["content"][-1]
+    assert _strip_cache_control(out) == _strip_cache_control(merged)
+
+
+def test_normalize_mirror_clamps_out_of_range_block_index():
+    # Client marked block 2; forwarded message only has 1 block (transform
+    # merged content) — marker falls back to the last block, not dropped.
+    client = [
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "a"},
+                {"type": "text", "text": "b"},
+                {"type": "text", "text": "c", "cache_control": {"type": "ephemeral"}},
+            ],
+        },
+        B("user", "tail", cc=True),
+    ]
+    merged = [B("user", "abc-merged"), B("user", "tail", cc=True)]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 2
+    assert "cache_control" in out[0]["content"][-1]
+    assert "cache_control" in out[1]["content"][-1]
+
+
+def test_normalize_mirror_scalar_only_content_is_left_unchanged():
+    """Client marks a message whose forwarded counterpart carries only scalar
+    blocks: nothing can hold a marker and nothing was stripped, so the input
+    comes back unchanged (identity, not a copy)."""
+    client = [B("user", "a", cc=True)]
+    merged = [{"role": "user", "content": ["scalar-only"]}]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert out is merged
+    assert _markers(out) == 0
+
+
+def test_normalize_mirror_scalar_target_still_strips_leftovers():
+    # Message 0's forwarded content is scalar-only (marker unplaceable) but a
+    # replay leftover on message 1 still gets stripped, and message 1 keeps
+    # its client marker.
+    client = [B("user", "a", cc=True), B("user", "b", cc=True)]
+    merged = [
+        {"role": "user", "content": ["scalar-only"]},
+        {
+            "role": "user",
+            "content": [
+                {"type": "text", "text": "left-over", "cache_control": {"type": "ephemeral"}},
+                {"type": "text", "text": "b"},
+            ],
+        },
+    ]
+    out = normalize_message_cache_control(merged, client_messages=client)
+    assert _markers(out) == 1
+    assert "cache_control" not in out[1]["content"][0]
+    assert "cache_control" in out[1]["content"][-1]