feat(cache): provider-agnostic cache-mode delta + cc-agnostic prefix comparison (headroomlabs-ai#1868)

## Description

<!-- Briefly explain the change and why it is needed. -->

Closes #

## Type of Change

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

- 

## Testing

<!-- Check what you actually ran, then paste the real command output
below. -->

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

### Test Output

```text
# Paste relevant command output or artifact links here
```

## Real Behavior Proof

- Environment:
- Exact command / steps:
- Observed result:
- Not tested:

## Review Readiness

- [ ] I have performed a self-review
- [ ] This PR is ready for human review

## Checklist

- [ ] My code follows the project's style guidelines
- [ ] I have performed a self-review of my code
- [ ] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [ ] My changes generate no new warnings
- [ ] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

Add screenshots to help explain your changes.

## Additional Notes

<!-- Mention any N/A checklist items, tradeoffs, follow-ups, or
maintainer context. -->

Author: chopratejas

headroom/agent_savings.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import logging
+import os
 from dataclasses import dataclass, replace
 from typing import Protocol
 
@@ -278,6 +279,19 @@ def proxy_pipeline_kwargs(config: object) -> dict[str, object]:
     if smart_crusher_with_compaction is not None:
         kwargs["smart_crusher_with_compaction"] = bool(smart_crusher_with_compaction)
 
+    # Lower the block-compression char floor (default 500) so modest tool outputs
+    # are eligible for the LOSSY path too. Matters in cache mode, where the only
+    # compressible content each turn is a single (often small) delta observation;
+    # a 500-char floor buckets most of them as "small" (skipped). Env-gated so it
+    # only changes behavior when explicitly set; lossless folding has no floor and
+    # is unaffected.
+    _min_chars_block = os.environ.get("HEADROOM_MIN_CHARS_FOR_BLOCK")
+    if _min_chars_block:
+        try:
+            kwargs["min_chars_for_block_compression"] = int(_min_chars_block)
+        except ValueError:
+            pass
+
     return kwargs
 
 

headroom/cache/prefix_tracker.py

@@ -128,6 +128,134 @@ def _strip_cache_control(obj: Any) -> Any:
     return obj
 
 
+# Keys that carry NO semantic payload for the model — transport / caching-directive
+# / telemetry / client-routing annotations that clients attach and vary turn-to-turn.
+# Grounded in provider API docs (Anthropic Messages, OpenAI Chat+Responses, Bedrock
+# Converse) + client-library field inventories (litellm, Vercel AI SDK, opencode,
+# Claude Code, Cline). Dropped from the cross-turn prefix-equality key ONLY.
+#
+# NOTE ON SAFETY: this projection is a COMPARISON KEY, never a source to rebuild
+# forwarded bytes — the cache-stable-delta path always forwards the previously
+# forwarded bytes + the raw appended delta. So dropping these can't deprive the
+# model. What we must NOT do is drop a *semantic* field (that would mask a real
+# divergence and replay a stale prefix), which is why: (1) reasoning SIGNATURES are
+# NOT in this set (Anthropic 400s if a thinking block is altered/missing, and a
+# present/absent flip is a real divergence we want to detect); (2) tool inputs /
+# arguments / json payloads are treated as OPAQUE and compared verbatim (see
+# _OPAQUE_PAYLOAD_KEYS) so a user key that happens to be named "index"/"state" is
+# never stripped from inside a tool call.
+_NON_SEMANTIC_KEYS = frozenset(
+    {
+        # cache-breakpoint markers (moved to the newest block every turn)
+        "cache_control",  # Anthropic (per-block)
+        "cachePoint",  # Bedrock (per-block content block)
+        # litellm unified-message / tool annotations
+        "caller",  # litellm programmatic-tool tag on tool_use
+        "provider_specific_fields",
+        "reasoning_content",  # litellm display echo (the paired signature is separate)
+        "reasoning_items",
+        "annotations",  # citation/display metadata
+        # OpenAI response echoes that can ride on assistant messages
+        "system_fingerprint",
+        "service_tier",
+        # Vercel AI SDK / opencode part transport
+        "providerMetadata",
+        "providerOptions",
+        "callProviderMetadata",
+        "state",
+        "providerExecuted",
+        "synthetic",
+        "ignored",
+        # streaming-assembly artifact
+        "index",
+    }
+)
+
+# Values under these keys are opaque semantic payloads (tool-call input, OpenAI
+# stringified arguments, Bedrock tool_result json). They are compared VERBATIM — we
+# never recurse into them to strip "noise" keys, because arbitrary user data there
+# may legitimately contain keys that collide with _NON_SEMANTIC_KEYS (e.g. an
+# `input` of {"state": "CA", "index": 3}). Recursing would corrupt the comparison.
+_OPAQUE_PAYLOAD_KEYS = frozenset({"input", "arguments", "json"})
+
+
+def _canonicalize_for_prefix_compare(obj: Any) -> Any:
+    """Representation-agnostic canonical form for cross-turn prefix equality.
+
+    Providers accept several *equivalent* encodings for the same message, and real
+    clients vary them turn-to-turn; a raw-dict prefix compare then fails spuriously
+    and drops cache mode to raw (uncompressed) forwarding. This normalizes ONLY
+    representation:
+      * drops non-semantic annotation / cache-directive / telemetry keys
+        (_NON_SEMANTIC_KEYS) at any message/block level;
+      * wraps a bare string ``content`` into ``[{"type": "text", "text": ...}]``
+        (Anthropic's string sugar, which litellm flips per turn);
+      * leaves tool ``input`` / ``arguments`` / ``json`` payloads verbatim
+        (_OPAQUE_PAYLOAD_KEYS) so user data is never corrupted;
+      * KEEPS all real content (text, tool name/input, tool_result content, reasoning
+        signatures, ids) so two messages canonicalize-equal iff they are semantically
+        identical.
+
+    Used ONLY as a comparison key for the cache-stable delta path; the original,
+    unmodified messages are always what gets forwarded.
+    """
+    if isinstance(obj, dict):
+        out: dict[str, Any] = {}
+        for key, value in obj.items():
+            if key in _NON_SEMANTIC_KEYS:
+                continue
+            if key in _OPAQUE_PAYLOAD_KEYS:
+                out[key] = value  # verbatim — do not recurse into user payloads
+            elif key == "content" and isinstance(value, str):
+                out[key] = [{"type": "text", "text": value}]
+            else:
+                out[key] = _canonicalize_for_prefix_compare(value)
+        return out
+    if isinstance(obj, list):
+        canon = [_canonicalize_for_prefix_compare(value) for value in obj]
+        # Drop blocks that projected to {} — a pure cache-directive content block
+        # (e.g. Bedrock {"cachePoint": {...}}) whose only key was non-semantic. Left
+        # in place it would be an empty-dict entry, so a directive block moving
+        # position across turns would spuriously fail the length/order compare.
+        return [value for value in canon if value != {}]
+    return obj
+
+
+def extract_cache_stable_delta(
+    current_messages: list[dict[str, Any]],
+    previous_original_messages: list[dict[str, Any]] | None,
+    previous_forwarded_messages: list[dict[str, Any]] | None,
+) -> tuple[list[dict[str, Any]], list[dict[str, Any]]] | None:
+    """Return ``(stable_forwarded_prefix, appended_delta_messages)`` when the current
+    request append-only-extends the previous one, else ``None``.
+
+    Provider-agnostic delta engine for cache mode. "Append-only" is decided by comparing
+    the *canonicalized* prefix (:func:`_canonicalize_for_prefix_compare`, which ignores
+    per-turn transport / cache-directive / client-annotation noise across
+    Anthropic / OpenAI / Bedrock and the common clients), so a moved cache marker or
+    shape churn does not spuriously collapse cache mode to raw forwarding. On a match the
+    caller replays the byte-identical previously-forwarded prefix and compresses ONLY the
+    appended delta.
+
+    This is a COMPARISON + slice only: the returned prefix is the previously-forwarded
+    bytes verbatim and the delta is the raw appended messages — never a rebuild from the
+    canonical projection — so the projection dropping non-semantic fields is safe.
+    """
+    if not previous_original_messages or previous_forwarded_messages is None:
+        return None
+    prefix_len = len(previous_original_messages)
+    if len(current_messages) < prefix_len:
+        return None
+    if _canonicalize_for_prefix_compare(
+        current_messages[:prefix_len]
+    ) != _canonicalize_for_prefix_compare(previous_original_messages):
+        return None
+    return (
+        copy.deepcopy(previous_forwarded_messages),
+        copy.deepcopy(current_messages[prefix_len:]),
+    )
+
+
 def overlay_cached_prefix(
     optimized_messages: list[dict[str, Any]],
     current_original_messages: list[dict[str, Any]],
@@ -168,12 +296,17 @@ def overlay_cached_prefix(
     if len(current_original_messages) < n or len(optimized_messages) < n:
         return optimized_messages
     # Append-only guard on CONTENT ONLY: the frozen region must be the same
-    # messages we cached. Compare with cache_control stripped — clients move that
-    # breakpoint to the newest message each turn, so a raw dict compare would
-    # spuriously fail whenever a marker lands in the frozen prefix, skip the
-    # replay, and bust the cache (the residual busts observed after the first
-    # fix). Content stability is what the provider's prefix cache actually keys on.
-    if _strip_cache_control(current_original_messages[:n]) != _strip_cache_control(prev_orig):
+    # messages we cached. Compare with the shared canonicalizer (not just
+    # cache_control-stripping) so the guard is robust to ALL per-turn transport /
+    # annotation churn — cache_control movement (Anthropic), litellm `caller`,
+    # provider_specific_fields, streaming `index`, string<->block content shape,
+    # etc. — across providers/clients. Content stability is what the provider's
+    # prefix cache actually keys on; a coarser cache_control-only strip let other
+    # clients' noise spuriously fail the guard, skip the replay, and bust. This
+    # helps every handler that shares overlay_cached_prefix (Anthropic + OpenAI).
+    if _canonicalize_for_prefix_compare(
+        current_original_messages[:n]
+    ) != _canonicalize_for_prefix_compare(prev_orig):
         return optimized_messages
     # Replay the cached (compressed) prefix byte-identical; keep this turn's tail.
     return list(prev_fwd) + list(optimized_messages[n:])
@@ -563,6 +696,22 @@ def _estimate_message_tokens(messages: list[dict[str, Any]]) -> list[int]:
                             chars += len(text)
             else:
                 chars = 0
+            # OpenAI function-calling: the assistant's command lives in the
+            # top-level `tool_calls` (or legacy `function_call`) field, NOT in
+            # `content` (which is empty/None on a tool-call turn). Anthropic puts
+            # the equivalent in a `tool_use` content BLOCK (counted above), but
+            # the OpenAI shape was never counted here. That under-counted every
+            # tool-based assistant turn to ~0, so the frozen-prefix estimate
+            # overshot the real cache boundary and froze the NEWEST delta — which
+            # is why OpenAI/Kimi (fireworks) tool harnesses got ~zero compression
+            # while text/back-tick harnesses (command in `content`) compressed.
+            for tc in msg.get("tool_calls") or []:
+                if isinstance(tc, dict):
+                    fn = tc.get("function") or {}
+                    chars += len(str(fn.get("name", ""))) + len(str(fn.get("arguments", "")))
+            fc = msg.get("function_call")
+            if isinstance(fc, dict):
+                chars += len(str(fc.get("name", ""))) + len(str(fc.get("arguments", "")))
             # Add overhead for role, block structure, etc.
             chars += 20
             counts.append(max(1, int(chars / 3.5)))

headroom/proxy/cost.py

@@ -738,6 +738,21 @@ def record_tokens(
             uncached_tokens: Non-cached input tokens from API response usage.
             output_tokens: Output tokens from API response usage.
         """
+        # Post-guard invariant (all providers): Headroom never forwards a request
+        # larger than the original (handlers revert any inflation before sending),
+        # so compression savings are >= 0 by construction. A negative here is an
+        # intermediate/hook token-count artifact that never reached the model;
+        # clamp it so `total_tokens_removed` reflects actually-forwarded bytes
+        # instead of surfacing spurious negatives (verified clean on the wire).
+        if tokens_saved < 0:
+            import logging as _lg
+
+            _lg.getLogger(__name__).debug(
+                "record_tokens: clamping negative tokens_saved=%d to 0 for %s (artifact; wire not inflated)",
+                tokens_saved,
+                model,
+            )
+            tokens_saved = 0
         self._tokens_saved_by_model[model] = (
             self._tokens_saved_by_model.get(model, 0) + tokens_saved
         )

headroom/proxy/handlers/anthropic.py

@@ -362,17 +362,20 @@ def _extract_cache_stable_delta(
         Safe means the prior original request is an exact message-prefix of the
         current original request. This lets us replay the exact forwarded bytes
         for historical context and only transform newly appended message suffixes.
+
+        The append-only check ignores per-turn transport / cache-directive / client
+        annotation noise (cache_control moved to the newest block, litellm caller,
+        provider_specific_fields, streaming index, string<->block content shape, …) via
+        the shared canonicalizer, so that churn doesn't spuriously drop cache mode to raw
+        forwarding. Delegates to the provider-agnostic engine in prefix_tracker so
+        OpenAI / Bedrock share one implementation.
         """
-        if not previous_original_messages or previous_forwarded_messages is None:
-            return None
-        prefix_len = len(previous_original_messages)
-        if len(current_messages) < prefix_len:
-            return None
-        if current_messages[:prefix_len] != previous_original_messages:
-            return None
-        return (
-            copy.deepcopy(previous_forwarded_messages),
-            copy.deepcopy(current_messages[prefix_len:]),
+        from headroom.cache.prefix_tracker import extract_cache_stable_delta
+
+        return extract_cache_stable_delta(
+            current_messages,
+            previous_original_messages,
+            previous_forwarded_messages,
         )
 
     @staticmethod
@@ -1342,21 +1345,63 @@ class _DeferredCompressionResult:
                                     optimized_messages = messages
                                     optimized_tokens = tokenizer.count_messages(optimized_messages)
                                 else:
+                                    # Compress the delta, with two cache-mode adjustments:
+                                    #
+                                    # fix-5: strip the client's transient cache_control marker so
+                                    #   the router's per-block "never compress an explicit cache
+                                    #   key" guard (content_router.py:4006) doesn't skip the ONLY
+                                    #   compressible content every turn (route_counts had
+                                    #   cache_control_protected == the whole delta -> 0%). In cache
+                                    #   mode that marker is NOT the real forwarded breakpoint: the
+                                    #   compressed delta is frozen + replayed verbatim next turn and
+                                    #   normalize_message_cache_control (AFTER compression, below)
+                                    #   owns the single forwarded breakpoint. Cache-safety is
+                                    #   enforced post-compression, not by protecting the delta.
+                                    #
+                                    # fix-6: the delta is a lone tool_result whose tool_use (tool
+                                    #   NAME + call args) lives in the frozen prefix. Passing only
+                                    #   the delta to the router leaves tool_name="" so
+                                    #   _bash_search_fold (lossless grep/rg folding, no size floor),
+                                    #   per-tool bias, and relevance-query enrichment all degrade.
+                                    #   Pass the FULL current messages with frozen_message_count =
+                                    #   prefix length: _build_tool_name_map scans ALL messages (the
+                                    #   delta resolves its tool_name from the prefix's tool_use) but
+                                    #   the compression loop only touches indices >= frozen count,
+                                    #   so ONLY the delta is compressed. Splice the compressed delta
+                                    #   onto the byte-stable forwarded prefix.
+                                    from headroom.cache.prefix_tracker import _strip_cache_control
+
+                                    # Compression context = the EXACT forwarded (cached) prefix
+                                    # + the stripped delta, with the prefix frozen. Using the
+                                    # forwarded prefix (not the original) keeps _build_tool_name_map
+                                    # AND cross-turn dedup consistent with what is actually cached:
+                                    # dedup can only reference bytes that are truly present in the
+                                    # forwarded context, so no pointer can dangle. The prefix is
+                                    # frozen (never compressed) and we discard the router's copy of
+                                    # it below, so the forwarded prefix stays byte-identical to last
+                                    # turn -> append-only -> no bust.
+                                    prefix_n = len(stable_forwarded_prefix)
+                                    compression_input = list(stable_forwarded_prefix) + list(
+                                        _strip_cache_control(delta_messages)
+                                    )
                                     result = await self._run_compression_in_executor(
                                         lambda: self.anthropic_pipeline.apply(
-                                            messages=delta_messages,
+                                            messages=compression_input,
                                             model=model,
                                             model_limit=context_limit,
-                                            context=extract_user_query(delta_messages),
-                                            frozen_message_count=0,
+                                            context=extract_user_query(compression_input),
+                                            frozen_message_count=prefix_n,
                                             biases=biases,
                                             request_id=request_id,
                                             compression_policy=compression_policy,
                                             **proxy_pipeline_kwargs(self.config),
                                         ),
                                         timeout=COMPRESSION_TIMEOUT_SECONDS,
                                     )
-                                    optimized_messages = stable_forwarded_prefix + result.messages
+                                    # Only the delta was eligible for compression (prefix frozen);
+                                    # forward the byte-identical cached prefix + the compressed delta.
+                                    compressed_delta = result.messages[prefix_n:]
+                                    optimized_messages = stable_forwarded_prefix + compressed_delta
                                     transforms_applied = result.transforms_applied
                                     pipeline_timing = result.timing
                                     optimized_tokens = tokenizer.count_messages(optimized_messages)