github-featured/chopratejas--headroom
2
0
Fork 0
Code Issues Pull Requests Actions 3 Deployments Models Agent Notes Packages Projects Releases Wiki Activity

fix(proxy): compress cache-mode cold starts and tag prefix-mismatch passthrough (#2365)

Browse Source 
## Description

Since v0.31.0 shipped cache mode as the default (68676daa), users report
the dashboard showing "Optimization ENABLED" while every request
forwards with 0 tokens saved — Before Compression == After Compression
even on 100k-token requests (#2357).

Root cause: in the cache-mode branch of `handle_anthropic_messages`,
when `_extract_cache_stable_delta` returns `None` the handler silently
sets `optimized_messages = messages`. That single fall-through covers
two very different cases:

1. **Session cold start** — no previous turn recorded for the session.
Every fresh proxy session, including a resumed 100k-token Claude Code
transcript, was forwarded raw. Until session identity stabilized (#2193
helped), this could be *every* turn, i.e. compression literally never
ran.
2. **Mid-session prefix mismatch** — the client rewrote history. This
passthrough is intentional (replaying a rewritten transcript risks
per-turn cache busts) but was invisible: no tag, no log, so
`optimize:true` + 0 savings looked like a broken product.

This PR: (1) cold starts now run the same full-message compression as
non-cache modes — there is no provider cache prefix to protect yet, and
the compressed output is recorded as the forwarded messages so later
turns replay it byte-identically through the existing stable-delta path
(append-only cache safety preserved); (2) the mismatch passthrough is
kept but tagged `passthrough_reason=cache_mode_prefix_mismatch` and
logged, mirroring the existing `pre_upstream_backpressure` inline-tag
pattern.

Fixes #2357

## Type of Change

- [x] Bug fix (non-breaking change which fixes an issue)
- [ ] New feature (non-breaking change which adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to not work as expected)
- [ ] Documentation update
- [ ] Refactoring (no functional changes)

## Changes Made

- `headroom/proxy/handlers/anthropic.py` (cache-mode branch only): split
the `delta is None` fall-through. When
`prefix_tracker.get_last_original_messages()` is empty (cold start), run
the full pipeline via `_run_compression_in_executor` (same call shape as
the non-cache branch) and append a `cache_mode:cold_start_full`
transform marker. When a previous turn exists but the delta is `None`
(prefix mismatch), keep the conservative passthrough but set
`tags["passthrough_reason"] = "cache_mode_prefix_mismatch"` and log it.
`CompressionDecision` untouched — it is the frozen pre-pipeline gate;
inline tags are the established mechanism for mid-pipeline passthrough
reasons.
- `tests/test_cache_mode_cold_start.py` (new): handler-level tests using
the same dummy-handler harness as `tests/test_cold_start_fast_pass.py`,
with `mode="cache"`. Cold start → pipeline invoked once, compressed form
forwarded upstream, no passthrough tag. Prefix mismatch → pipeline not
invoked, original bytes forwarded unmodified, outcome tags carry
`passthrough_reason=cache_mode_prefix_mismatch`.

## Testing

- [x] Unit tests pass locally
- [x] Lint/format/type checks pass locally

```
$ python -m pytest tests/test_cache_mode_cold_start.py tests/test_cache_mode_delta_marker.py tests/test_cache_prefix_overlay.py tests/test_cache/test_prefix_tracker.py tests/test_token_headroom_mode.py tests/test_cold_start_fast_pass.py tests/test_anthropic_pre_upstream_backpressure.py tests/test_handler_outcome_tag_invariant.py -q
146 passed

$ ruff check headroom/proxy/handlers/anthropic.py tests/test_cache_mode_cold_start.py
All checks passed!
$ ruff format --check headroom/proxy/handlers/anthropic.py tests/test_cache_mode_cold_start.py
2 files already formatted
$ mypy headroom --ignore-missing-imports
(no error lines)
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.13.11, this branch; real
`AnthropicHandlerMixin.handle_anthropic_messages` driven end-to-end
through a FastAPI `Request` with upstream stubbed at `_retry_request`
(harness identical to the existing
`tests/test_cold_start_fast_pass.py`), `mode="cache"`.
- Exact command / steps: `python -m pytest
tests/test_cache_mode_cold_start.py -q`, plus a manual run of the
mismatch scenario with `logging.basicConfig(level=INFO)`.
- Observed result: Cold start: `anthropic_pipeline.apply` invoked once
and the forwarded upstream body contains the compressed tool_result
content (previously: forwarded raw with zero pipeline invocations).
Mismatch: bytes forwarded unmodified, and the proxy log now emits
`[req-...] Compression skipped: reason=cache_mode_prefix_mismatch` with
the same reason present in `RequestOutcome.tags["passthrough_reason"]`
(previously: nothing).
- Not tested: a live multi-turn session against the real Anthropic API
measuring `cache_read_input_tokens` across turns (the byte-identical
replay contract the cold-start path relies on is the same one exercised
by the existing stable-delta tests in
`tests/test_cache_mode_delta_marker.py` and
`tests/test_cache_prefix_overlay.py`, all green).

## Review Readiness

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

---------

Co-authored-by: Claude Fable 5 <noreply@anthropic.com>
Co-authored-by: JD Davis <jd@jds-macbook-air.tail2a279.ts.net>
This commit is contained in:
Parideboy
2026-08-12 06:55:11 +02:00
committed by GitHub
parent f669149769
commit aaeba0a319
3 changed files with 333 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files
headroom/proxy/handlers/anthropic.py
+67 -6
View File
@@ -1152,6 +1152,9 @@ class AnthropicHandlerMixin:
previous_original_messages = prefix_tracker.get_last_original_messages()
previous_forwarded_messages = prefix_tracker.get_last_forwarded_messages()
frozen_message_count = prefix_tracker.get_frozen_message_count()
# Pre-strict-override tracker truth: >0 only when a provider cache
# prefix was actually confirmed (or restored) for this session.
tracker_frozen_count = frozen_message_count
# Idle gap since the previous turn's response, snapshotted at fetch
# (before get_or_create bumped the access clock). Forwarded to the
# pipeline so the net-cost/TTL gate (HEADROOM_NET_COST_POLICY=1) can
@@ -1556,11 +1559,54 @@ class AnthropicHandlerMixin:
optimized_tokens = tokenizer.count_messages(optimized_messages)
transforms_applied = _cold_transforms
else:
delta = self._extract_cache_stable_delta(
original_client_messages,
previous_original_messages,
previous_forwarded_messages,
)
if not previous_original_messages and tracker_frozen_count == 0:
# Session cold start: nothing has been forwarded for
# this session yet and no frozen prefix survives from
# a prior process (frozen_message_count > 0 means a
# provider cache prefix may already exist upstream —
# e.g. a resumed session restored from the CCR
# compression cache — so it must stay passthrough),
# hence there is no provider cache prefix to protect — run the same full-message
# compression as non-cache modes (issue #2357: the
# previous silent passthrough here meant cache mode
# never compressed anything until a stable delta
# appeared, which for resumed 100k-token transcripts
# was never). The compressed output is recorded as
# the forwarded messages below, and later turns
# replay that prefix byte-identically via the
# stable-delta path, so append-only cache safety is
# preserved.
delta = None
async with stage_timer.measure("compression_first_stage"):
result = await self._run_compression_in_executor(
lambda: self.anthropic_pipeline.apply(
messages=messages,
model=model,
model_limit=context_limit,
context=extract_user_query(messages),
frozen_message_count=frozen_message_count,
biases=biases,
request_id=request_id,
compression_policy=compression_policy,
**proxy_pipeline_kwargs(self.config),
),
timeout=COMPRESSION_TIMEOUT_SECONDS,
)
if result.messages != messages:
optimized_messages = result.messages
transforms_applied = list(result.transforms_applied) + [
"cache_mode:cold_start_full"
]
pipeline_timing = result.timing
original_tokens = result.tokens_before
optimized_tokens = result.tokens_after
else:
delta = self._extract_cache_stable_delta(
original_client_messages,
previous_original_messages,
previous_forwarded_messages,
)
if delta is not None:
stable_forwarded_prefix, delta_messages = delta
if delta_messages:
@@ -1628,11 +1674,26 @@ class AnthropicHandlerMixin:
else:
optimized_messages = stable_forwarded_prefix
optimized_tokens = tokenizer.count_messages(optimized_messages)
else:
elif previous_original_messages:
# Conservative rule for cache mode:
# only replay exact stable message-prefix extensions.
# In-message append rewriting is deferred until we can
# prove it is perfectly replayable across future turns.
# Tag the passthrough so the dashboard can explain 0
# savings instead of silently reporting "optimization
# enabled" with Before == After (issue #2357).
tags["passthrough_reason"] = "cache_mode_prefix_mismatch"
logger.info(
"[%s] Compression skipped: reason=cache_mode_prefix_mismatch",
request_id,
)
optimized_messages = messages
optimized_tokens = original_tokens
elif tracker_frozen_count > 0:
# Cold start with a frozen prefix restored from a
# prior process: the provider cache may already hold
# that prefix, so forward unmodified.
tags["passthrough_reason"] = "cache_mode_frozen_cold_start"
optimized_messages = messages
optimized_tokens = original_tokens
tests/test_cache_mode_cold_start.py
+239
View File
@@ -0,0 +1,239 @@
"""Issue #2357: cache mode must not silently forward everything uncompressed.
Cache mode compresses only the inter-turn delta against the previously
forwarded prefix. Before this fix, both delta-miss cases fell through to a
silent passthrough:
- session cold start (no previous turn recorded) — including a resumed
100k-token transcript, so a fresh proxy never compressed anything while the
dashboard kept reporting "optimization enabled";
- mid-session prefix mismatch (client rewrote history) — an intentional
passthrough, but with no tag or log explaining the 0 savings.
Now a cold start runs the same full-message compression as non-cache modes
(there is no provider cache prefix to protect yet), and the mismatch
passthrough is tagged with ``passthrough_reason=cache_mode_prefix_mismatch``.
"""
from __future__ import annotations
import json
from types import SimpleNamespace
from unittest.mock import MagicMock
import anyio
from fastapi import Request
from headroom.config import TransformResult
from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
from headroom.proxy.models import ProxyConfig
_COMPRESSED_TEXT = "compressed tool output"
class _DummyTokenizer:
def count_messages(self, messages) -> int:
return json.dumps(messages).count(" ") + 1
def count_text(self, text: str) -> int:
return max(1, text.count(" ") + 1)
class _DummyMetrics:
async def record_request(self, **kwargs):
return None
async def record_stage_timings(self, path, timings):
return None
async def record_failed(self, **kwargs):
return None
def record_compression_failed(self, reason: str) -> None:
return None
async def record_rate_limited(self, **kwargs):
return None
class _ResponseStub:
status_code = 200
headers: dict[str, str] = {}
content = b'{"id":"msg_1","type":"message","role":"assistant","content":[],"usage":{"input_tokens":1,"output_tokens":1}}'
def json(self):
return {
"id": "msg_1",
"type": "message",
"role": "assistant",
"content": [],
"usage": {"input_tokens": 1, "output_tokens": 1},
}
def _fake_pipeline_apply(messages, model, **kwargs):
compressed = []
for msg in messages:
new = dict(msg)
if msg.get("role") == "user" and isinstance(msg.get("content"), list):
new["content"] = [
{**part, "content": _COMPRESSED_TEXT}
if isinstance(part, dict) and part.get("type") == "tool_result"
else part
for part in msg["content"]
]
compressed.append(new)
return TransformResult(
messages=compressed,
tokens_before=1000,
tokens_after=100,
transforms_applied=["read_lifecycle:stale:test.py"],
)
class _DummyAnthropicHandler(AnthropicHandlerMixin):
ANTHROPIC_API_URL = "https://api.anthropic.com"
def __init__(self, previous_messages: list | None = None) -> None:
self.rate_limiter = None
self.metrics = _DummyMetrics()
self.config = ProxyConfig(
optimize=True,
image_optimize=False,
retry_max_attempts=1,
retry_base_delay_ms=1,
retry_max_delay_ms=1,
connect_timeout_seconds=10,
mode="cache",
cache_enabled=False,
rate_limit_enabled=False,
fallback_enabled=False,
fallback_provider=None,
prefix_freeze_enabled=False,
memory_enabled=False,
)
self.usage_reporter = None
self.anthropic_provider = SimpleNamespace(get_context_limit=lambda model: 200_000)
self.anthropic_pipeline = SimpleNamespace(apply=MagicMock(side_effect=_fake_pipeline_apply))
self.anthropic_backend = None
self.cost_tracker = None
self.memory_handler = None
self.cache = None
self.security = None
self.ccr_context_tracker = None
self.ccr_injector = None
self.ccr_response_handler = None
self.ccr_feedback = None
self.ccr_batch_processor = None
self.ccr_mcp_server = None
self.traffic_learner = None
self.tool_injector = None
self.read_lifecycle_manager = None
self.logger = SimpleNamespace(log=lambda *a, **k: None)
self.request_logger = self.logger
self.usage_observer = None
self.image_compressor = None
prev = previous_messages or []
tracker = MagicMock()
tracker.get_frozen_message_count.return_value = 0
tracker.get_last_original_messages.return_value = prev
tracker.get_last_forwarded_messages.return_value = prev
tracker._cached_token_count = 0
tracker.classify_cache_miss.return_value = SimpleNamespace(is_miss=False)
self.session_tracker_store = SimpleNamespace(
compute_session_id=lambda *a, **k: "sess-1",
get_or_create=lambda *a, **k: tracker,
resolve_tracker=lambda *a, **k: tracker,
)
self._background_compression_enabled = False
self.recorded_tags: dict = {}
async def _record_request_outcome(self, outcome) -> None:
self.recorded_tags = dict(outcome.tags or {})
async def _run_compression_in_executor(self, fn, timeout):
return fn()
async def _next_request_id(self) -> str:
return "req-cache-cold-start-test"
async def _retry_request(self, method, url, headers, body, **_kwargs):
self.captured_body = body
return _ResponseStub()
def _build_request(body: dict) -> Request:
payload = json.dumps(body).encode("utf-8")
async def receive():
return {"type": "http.request", "body": payload, "more_body": False}
scope = {
"type": "http",
"asgi": {"version": "3.0"},
"http_version": "1.1",
"method": "POST",
"scheme": "https",
"path": "/v1/messages",
"raw_path": b"/v1/messages",
"query_string": b"",
"headers": [(b"authorization", b"Bearer sk-ant-api-test")],
"client": ("127.0.0.1", 12345),
"server": ("testserver", 443),
}
return Request(scope, receive)
_TOOL_RESULT_MESSAGE = {
"role": "user",
"content": [
{
"type": "tool_result",
"tool_use_id": "toolu_1",
"content": "verbose stale tool output " * 200,
}
],
}
def test_cache_mode_cold_start_compresses_full_request(monkeypatch):
import headroom.tokenizers as _tk
monkeypatch.setattr(_tk, "get_tokenizer", lambda model: _DummyTokenizer())
handler = _DummyAnthropicHandler(previous_messages=[])
request = _build_request(
{"model": "claude-3-5-sonnet-latest", "messages": [_TOOL_RESULT_MESSAGE]}
)
anyio.run(handler.handle_anthropic_messages, request)
# The full pipeline ran and its output was forwarded upstream.
assert handler.anthropic_pipeline.apply.call_count == 1
forwarded = handler.captured_body["messages"]
assert forwarded[0]["content"][0]["content"] == _COMPRESSED_TEXT
# No passthrough tag: compression genuinely ran.
assert "passthrough_reason" not in handler.recorded_tags
def test_cache_mode_prefix_mismatch_passes_through_with_tag(monkeypatch):
import headroom.tokenizers as _tk
monkeypatch.setattr(_tk, "get_tokenizer", lambda model: _DummyTokenizer())
# A previous turn exists but is NOT a prefix of the current request.
previous = [{"role": "user", "content": [{"type": "text", "text": "totally different"}]}]
handler = _DummyAnthropicHandler(previous_messages=previous)
original_text = "verbose stale tool output " * 200
request = _build_request(
{"model": "claude-3-5-sonnet-latest", "messages": [_TOOL_RESULT_MESSAGE]}
)
anyio.run(handler.handle_anthropic_messages, request)
# Conservative passthrough preserved: bytes forwarded unmodified...
assert handler.anthropic_pipeline.apply.call_count == 0
forwarded = handler.captured_body["messages"]
assert forwarded[0]["content"][0]["content"] == original_text
# ...but now visibly tagged instead of silent.
assert handler.recorded_tags["passthrough_reason"] == "cache_mode_prefix_mismatch"
tests/test_proxy_anthropic_cache_stability.py
+27 -3
View File
@@ -17,6 +17,15 @@ from headroom.proxy.handlers.anthropic import AnthropicHandlerMixin
from headroom.proxy.server import ProxyConfig, create_app
def _force_compression(monkeypatch) -> None: # noqa: ANN001
decision = SimpleNamespace(should_compress=True, passthrough_reason=None)
decision.apply_to_tags = lambda tags: None
monkeypatch.setattr(
"headroom.proxy.handlers.anthropic.CompressionDecision.decide",
lambda **kwargs: decision,
)
class _FakePrefixTracker:
def __init__(self, frozen_count: int):
self._frozen_count = frozen_count
@@ -994,7 +1003,8 @@ def test_token_mode_does_not_force_freeze_all_previous_turns() -> None:
assert captured["frozen_message_count"] >= 0
def test_cache_mode_restores_frozen_prefix_if_transform_mutates_history() -> None:
def test_cache_mode_restores_frozen_prefix_if_transform_mutates_history(monkeypatch) -> None:
_force_compression(monkeypatch)
captured = {}
with _make_proxy_client() as client:
proxy = client.app.state.proxy
@@ -1013,6 +1023,11 @@ def test_cache_mode_restores_frozen_prefix_if_transform_mutates_history() -> Non
{"role": "assistant", "content": "turn1-assistant"},
{"role": "user", "content": "current turn"},
]
# Mid-session: the first two messages were already forwarded last turn,
# so they form the byte-stable cached prefix the handler must replay
# even if a transform tries to mutate them.
fake_tracker._last_original_messages = original_messages[:2]
fake_tracker._last_forwarded_messages = original_messages[:2]
def _fake_apply(**kwargs):
mutated = list(kwargs["messages"])
@@ -1064,7 +1079,11 @@ def test_cache_mode_restores_frozen_prefix_if_transform_mutates_history() -> Non
assert sent_messages[1] == original_messages[1]
def test_cache_mode_does_not_forward_latest_turn_rewrites() -> None:
def test_cache_mode_cold_start_forwards_pipeline_rewrites(monkeypatch) -> None:
_force_compression(monkeypatch)
# Issue #2357: on a true session cold start (no prior forwarded messages,
# no frozen prefix) there is no provider cache to protect, so the full
# pipeline output is forwarded and recorded for byte-identical replay.
captured = {}
with _make_proxy_client() as client:
proxy = client.app.state.proxy
@@ -1129,7 +1148,12 @@ def test_cache_mode_does_not_forward_latest_turn_rewrites() -> None:
)
assert response.status_code == 200
assert captured["body"]["messages"] == original_messages
sent_messages = captured["body"]["messages"]
assert sent_messages[:2] == original_messages[:2]
assert sent_messages[2]["content"] == "REWRITTEN_CURRENT_TURN"
# Recorded as forwarded so later turns replay this prefix verbatim
# (the tracker may append the assistant reply after the forwarded turns).
assert fake_tracker._last_forwarded_messages[: len(sent_messages)] == sent_messages
def test_cache_mode_reuses_prior_forwarded_prefix_and_compresses_only_new_suffix() -> None: