fix(headroom): don't let trailing cache_control make compression a no-op

litellm/compression/compress.py

@@ -215,11 +215,35 @@ def _message_has_cache_control(message: Mapping[str, object]) -> bool:
 
 
 def _cached_prefix_indices(messages: Sequence[Mapping[str, object]]) -> tuple[int, ...]:
-    last_breakpoint: Final = max(
-        (index for index, msg in enumerate(messages) if _message_has_cache_control(msg)),
-        default=-1,
+    """Indices covered by the Anthropic prompt-cache prefix that must not be rewritten.
+
+    A ``cache_control`` breakpoint pins the provider's cache prefix to the exact
+    bytes of every row up to it. Mid-history breakpoints therefore protect that
+    prefix from compression.
+
+    A breakpoint on the *final* message is different: clients such as Claude Code
+    put one on the trailing turn on every request as a write marker for the *next*
+    call. Treating that trailing marker as the prefix end protects the entire
+    conversation (``protected == len(messages)``), leaves nothing compressible,
+    and makes the Headroom guardrail a silent no-op (#42939). When the trailing
+    row is marked, use the previous breakpoint if one exists; if it is the only
+    breakpoint, it does not establish a stable cached prefix yet, so return empty.
+    """
+    breakpoints: Final = [  # mutable-ok: preserve indices for trailing-marker handling
+        index for index, msg in enumerate(messages) if _message_has_cache_control(msg)
+    ]
+    if not breakpoints:
+        return ()
+    prefix_end: Final = (
+        breakpoints[-2]
+        if breakpoints[-1] == len(messages) - 1 and len(breakpoints) >= 2
+        else -1
+        if breakpoints[-1] == len(messages) - 1
+        else breakpoints[-1]
     )
-    return tuple(range(last_breakpoint + 1))
+    if prefix_end < 0:
+        return ()
+    return tuple(range(prefix_end + 1))
 
 
 def get_protected_indices(messages: Sequence[Mapping[str, object]]) -> tuple[int, ...]:
@@ -228,7 +252,8 @@ def get_protected_indices(messages: Sequence[Mapping[str, object]]) -> tuple[int
     - All system messages
     - The last user message
     - The last assistant message
-    - Every message up to and including the last one carrying an Anthropic cache_control breakpoint
+    - Every message up to and including the last *stable* Anthropic cache_control
+      breakpoint (a trailing-turn-only marker is ignored; see ``_cached_prefix_indices``)
 
     The last user message is what the model is being asked to act on right now,
     so compressing it replaces the live instruction with a marker. Compression

litellm/proxy/guardrails/guardrail_hooks/headroom/headroom.py

@@ -807,6 +807,14 @@ async def apply_guardrail(
         protected_indices: Final = _protected_indices(messages, raw_retrieve_call_ids)
         compressible: Final = [m for i, m in enumerate(messages) if i not in protected_indices]
         if not compressible:
+            # Match the other early returns: leave a breadcrumb when compression
+            # is a no-op so Cost Optimization / DEBUG is not left with a silent
+            # "0 tokens compressed" and no explanation (#42939).
+            verbose_proxy_logger.debug(
+                "Headroom: nothing compressible (protected=%s/%s); skipping compression",
+                len(protected_indices),
+                len(messages),
+            )
             return inputs
 
         model: Final = self.headroom_model or request_data.get("model")

tests/test_litellm/proxy/guardrails/guardrail_hooks/test_headroom.py

@@ -2571,6 +2571,43 @@ async def test_mid_history_cache_control_row_is_never_sent_for_compression(guard
     assert result["structured_messages"][3] == cached_row
 
 
+TRAILING_CACHE_CONTROL_MESSAGES = [
+    {
+        "role": "system",
+        "content": [
+            {"type": "text", "text": "You are Claude Code. " + "S" * 200, "cache_control": {"type": "ephemeral"}}
+        ],
+    },
+    {"role": "user", "content": "old question " + "Q" * 2000},
+    {
+        "role": "assistant",
+        "content": "Reading the file now.",
+        "tool_calls": [{"id": "old_1", "type": "function", "function": {"name": "Read", "arguments": "{}"}}],
+    },
+    {"role": "tool", "tool_call_id": "old_1", "content": "large file body " + "F" * 5000},
+    {"role": "user", "content": "follow-up " + "U" * 500},
+    {
+        "role": "assistant",
+        "content": "Done.",
+        "tool_calls": [{"id": "new_1", "type": "function", "function": {"name": "Bash", "arguments": "{}"}}],
+    },
+    {"role": "tool", "tool_call_id": "new_1", "content": "volatile tail " + "T" * 500},
+    {"role": "user", "content": "live instruction", "cache_control": {"type": "ephemeral"}},
+]
+
+
+@pytest.mark.asyncio
+async def test_trailing_cache_control_still_sends_mid_history_for_compression(guardrail: HeadroomGuardrail):
+    """Claude Code's trailing write marker must not make Headroom a no-op (#42939)."""
+    wire, result = await _wire_and_result(guardrail, TRAILING_CACHE_CONTROL_MESSAGES)
+
+    # Older tool results / user turns stay compressible; only the live exchange is held back.
+    assert any(row.get("tool_call_id") == "old_1" for row in wire)
+    assert any(row.get("role") == "user" and "old question" in str(row.get("content", "")) for row in wire)
+    # Live turn stays protected / not rewritten.
+    assert result["structured_messages"][-1] == TRAILING_CACHE_CONTROL_MESSAGES[-1]
+
+
 # ---------------------------------------------------------------------------
 # #38558: a client that runs its own tool loop (e.g. Claude Code via the MCP
 # gateway) executes headroom_retrieve and echoes the recovered original content

tests/unit/compression/test_compress.py

@@ -217,3 +217,45 @@ def test_compress_keeps_part_level_cache_control_row_verbatim():
     assert result["messages"][0] == pinned
     assert result["messages"][2] != stale_log
     assert len(result["cache"]) >= 1
+
+
+def test_trailing_cache_control_alone_does_not_protect_whole_conversation():
+    """Claude Code marks the live turn every request. That trailing write marker
+    must not pin the entire history as a cached prefix (#42939)."""
+    messages = [
+        {"role": "user", "content": "old question"},
+        {"role": "assistant", "content": "old answer"},
+        {"role": "tool", "tool_call_id": "t1", "content": "large file body"},
+        {"role": "user", "content": "live instruction", "cache_control": {"type": "ephemeral"}},
+    ]
+
+    protected = sorted(get_protected_indices(messages))
+
+    assert protected == [1, 3]
+    assert 0 not in protected
+    assert 2 not in protected
+
+
+def test_trailing_cache_control_defers_to_earlier_breakpoint():
+    """System (or mid-history) breakpoint stays the prefix end; the trailing
+    live-turn marker is ignored so mid-history after that earlier breakpoint
+    remains compressible (#42939)."""
+    messages = [
+        {
+            "role": "system",
+            "content": [
+                {"type": "text", "text": "You are Claude Code.", "cache_control": {"type": "ephemeral"}}
+            ],
+        },
+        {"role": "user", "content": "old question"},
+        {"role": "assistant", "content": "old answer"},
+        {"role": "tool", "tool_call_id": "t1", "content": "large file body"},
+        {"role": "user", "content": "live instruction", "cache_control": {"type": "ephemeral"}},
+    ]
+
+    protected = sorted(get_protected_indices(messages))
+
+    # System (0) from the earlier breakpoint + last assistant (2) + last user (4).
+    assert protected == [0, 2, 4]
+    assert 1 not in protected
+    assert 3 not in protected