diff --git a/headroom/cache/compression_store.py b/headroom/cache/compression_store.py index 72e6ca34f..bbda697cd 100644 --- a/headroom/cache/compression_store.py +++ b/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) diff --git a/headroom/cache/prefix_tracker.py b/headroom/cache/prefix_tracker.py index 178300295..8157e9e9a 100644 --- a/headroom/cache/prefix_tracker.py +++ b/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 diff --git a/headroom/proxy/handlers/anthropic.py b/headroom/proxy/handlers/anthropic.py index 8e12401b7..059fdb8f0 100644 --- a/headroom/proxy/handlers/anthropic.py +++ b/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, diff --git a/headroom/proxy/helpers.py b/headroom/proxy/helpers.py index 26e360552..31cbe4635 100644 --- a/headroom/proxy/helpers.py +++ b/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, diff --git a/tests/test_cache_breakpoint_diagnostics.py b/tests/test_cache_breakpoint_diagnostics.py new file mode 100644 index 000000000..03786dc31 --- /dev/null +++ b/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 diff --git a/tests/test_cache_control_move_bust.py b/tests/test_cache_control_move_bust.py index 0b74ff92d..3816c7b38 100644 --- a/tests/test_cache_control_move_bust.py +++ b/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]