From fd9ddaa238dfc4691383d82cdeff84cc5885f195 Mon Sep 17 00:00:00 2001 From: gglucass Date: Tue, 14 Jul 2026 12:34:41 +0200 Subject: [PATCH] =?UTF-8?q?fix(proxy):=20cold-start=20fast=20pass=20?= =?UTF-8?q?=E2=80=94=20defer=20only=20Kompress,=20not=20the=20whole=20pipe?= =?UTF-8?q?line=20(#2073)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## Description Since #1850, the freeze path forwards a session's provider-cached prefix byte-identical — so a session is permanently locked to whatever form its cold start put in the provider cache. That fix is correct (it stopped token-mode cache busting measured at +41% cost), but it interacts badly with off-path background compression (#1171): when `HEADROOM_BACKGROUND_COMPRESSION=1` defers a cold-start-large request (frozen=0, ≥50k tokens), the ENTIRE pipeline is deferred and the raw transcript is forwarded, cached, and frozen. The background job's results can never be applied afterward (doing so would rewrite the frozen prefix), so the session forfeits its compression savings for its lifetime. Field data (same day, same session, A/B across a version boundary): ~15k tokens/turn saved when the cold start compressed synchronously vs 0/turn forever when it deferred. Notably, the recurring savings came from `read_lifecycle` stale-read drops completing in ~300ms — deferral throws away sub-second lossless wins to avoid a 30s Kompress pass. Only the Kompress ML stage can blow the request budget (the #1171 cascade). This PR splits the two: - The deferral branch now runs the pipeline synchronously with a new `skip_kompress=True` per-call kwarg — everything except the ML stage — under a bounded budget, and forwards the pruned form. The provider caches (and #1850 freezes) the *compressed* transcript, so the cheap savings persist for the session's lifetime. - The full pipeline (Kompress included) still goes to the background job, unchanged, keyed against the original messages so its content-hash results remain reusable at future cache-miss boundaries. - Fail-open: on fast-pass timeout or error, the request forwards uncompressed exactly as before this change. ## Type of Change - [x] 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 - `headroom/transforms/content_router.py`: new per-call `skip_kompress` runtime kwarg (follows the existing `_runtime_force_kompress` pattern). Gates only the Kompress deep-path call site; units routed there take the identical fallback used when the model isn't ready. Wins over `force_kompress`. - `headroom/proxy/helpers.py`: `COLD_START_FAST_PASS_TIMEOUT_SECONDS` (env `HEADROOM_COLD_START_FAST_PASS_TIMEOUT_SECONDS`, default 10s), documented next to `COMPRESSION_TIMEOUT_SECONDS`. - `headroom/proxy/handlers/anthropic.py`: the background-deferral branch runs the fast pass synchronously, stores its result in the session `CompressionCache`, forwards the pruned messages, and tags `deferred:kompress_background` (or `deferred:dropped` when the enqueue was dropped). On failure it constructs the same `_DeferredCompressionResult` as before. The Anthropic handler is the only deferral site (OpenAI/Gemini handlers don't defer). - `tests/test_transforms/test_content_router.py`: `skip_kompress` never invokes the ML stage and wins over `force_kompress` (mirrors the existing `force_kompress` test). - `tests/test_cold_start_fast_pass.py`: handler-level tests — exactly one synchronous `skip_kompress=True` pass, the background job runs the full pipeline, the forwarded body carries the fast-pass form, fast-pass results land in the compression cache; and the fail-open path (executor timeout → original messages forwarded, background job still queued). - `CHANGELOG.md`: Bug Fixes entry. ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [ ] Manual testing performed ### Test Output ```text $ uv run --frozen --extra dev pytest tests/test_cold_start_fast_pass.py -v tests/test_cold_start_fast_pass.py::test_cold_start_runs_fast_pass_and_defers_only_kompress PASSED tests/test_cold_start_fast_pass.py::test_fast_pass_failure_falls_back_to_full_deferral PASSED ============================== 2 passed in 0.28s =============================== $ uv run --frozen --extra dev pytest tests/test_proxy/test_background_compression.py \ tests/test_proxy/test_phase3_byte_identity.py tests/test_anthropic_stage_timings.py \ tests/test_transforms/test_content_router.py tests/test_anthropic_pre_upstream_backpressure.py ======================== 90 passed, 1 warning in 10.47s ======================== $ uv run --frozen --extra dev mypy headroom/proxy/handlers/anthropic.py headroom/proxy/helpers.py headroom/transforms/content_router.py Success: no issues found in 3 source files $ ruff check && ruff format --check All checks passed! / 5 files already formatted ``` ## Real Behavior Proof - Environment: macOS 15 (darwin 24.6.0), Python 3.10 via `uv run --frozen --extra dev`; field logs from a production desktop deployment (Python 3.12, `HEADROOM_MODE=token`, `HEADROOM_BACKGROUND_COMPRESSION=1`, subscription auth policy). - Exact command / steps: compared per-request PERF log lines for the same Claude Code session served by 0.30.0-lineage (sync cold start) vs 0.31.0-lineage (deferred cold start) on the same day. - Observed result: deferred-cold-start sessions log `tok_saved=0` on every subsequent turn with `Pipeline: freezing first 281/284 messages`; sync-cold-start sessions log `tok_saved=15526-18791` per turn with `read_lifecycle:stale` transforms at `opt_ms≈300`. - Not tested: this patch has not run against a live proxy yet (behavior verified at the handler-test level); `ruff`/`mypy` scoped to changed files. ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [x] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [x] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) N/A — proxy pipeline change, no UI. ## Additional Notes - Companion to #2057 (nested tool_result image token counting) and #2058 (new-content-relative savings rate) — all three came out of the same investigation into near-zero reported savings on long 1M-context Claude Code sessions. - Deliberate scope cuts: the OpenAI/Gemini handlers don't have a deferral branch, so nothing to change there; the background job is left keyed to original messages (not the fast-pass output) so its cached results match client-resent bytes at future cache-miss boundaries. - Timeout leak caveat is documented in code: a fast-pass timeout briefly leaks an executor worker, but without the ML stage the pass is bounded by routing + statistical crushers (observed 5-8s worst case on multi-M-token counted transcripts). 🤖 Generated with [Claude Code](https://claude.com/claude-code) --------- Co-authored-by: Claude Fable 5 Co-authored-by: Tejas Chopra Co-authored-by: JerrettDavis --- CHANGELOG.md | 1 + headroom/proxy/handlers/anthropic.py | 75 ++++- headroom/proxy/helpers.py | 14 + headroom/transforms/content_router.py | 8 +- tests/test_cold_start_fast_pass.py | 287 +++++++++++++++++++ tests/test_transforms/test_content_router.py | 56 ++++ 6 files changed, 434 insertions(+), 7 deletions(-) create mode 100644 tests/test_cold_start_fast_pass.py diff --git a/CHANGELOG.md b/CHANGELOG.md index 673dae2b4..d6a69a647 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -109,6 +109,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Bug Fixes +* **proxy:** run a cold-start fast pass before background-compression deferral so byte-identical freeze doesn't lock sessions to the uncompressed transcript. Since #1850, a session's provider-cached prefix is frozen in whatever form its cold start forwarded; deferring the WHOLE pipeline (`HEADROOM_BACKGROUND_COMPRESSION=1`, frozen=0, ≥50k tokens) therefore cached the raw transcript and forfeited the session's compression savings for its lifetime — including sub-second lossless wins like `read_lifecycle` stale-read drops, observed in the field as sessions permanently stuck at 0 savings. The deferral branch now runs the pipeline synchronously with the new `skip_kompress=True` kwarg (everything except the Kompress ML stage — the only stage that can blow the request budget per #1171) under a bounded fast-pass budget (`HEADROOM_COLD_START_FAST_PASS_TIMEOUT_SECONDS`, default 10s), forwards the pruned form, and defers only Kompress to the background job (tagged `deferred:kompress_background`). Fail-open: on fast-pass timeout/error the request forwards uncompressed exactly as before. Units routed to Kompress under `skip_kompress` take the same fallback as when the model isn't ready. * **ccr:** detect `read_lifecycle` stale/superseded markers in the retrieve-tool injector so they stay redeemable. Those markers (`[Read content stale: … Retrieve original: hash=]`) store the original bytes in the CCR store under a valid hash, but none of `CCRToolInjector`'s patterns matched them — every pattern required the word "compressed" or the `< 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 + + 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}, + } + + +class _RecordingBackgroundCompressor: + def __init__(self) -> None: + self.enqueued: list[tuple[str, object, object]] = [] + + def enqueue(self, key, compress, store) -> bool: + self.enqueued.append((key, compress, store)) + return True + + +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) -> 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="token", + 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 + self.session_tracker_store = SimpleNamespace( + compute_session_id=lambda *a, **k: "sess-1", + get_or_create=lambda *a, **k: SimpleNamespace( + get_frozen_message_count=lambda: 0, + get_last_original_messages=lambda: [], + get_last_forwarded_messages=lambda: [], + record_request=lambda *a, **k: None, + ), + ) + # Cold-start deferral wiring under test. + self._background_compression_enabled = True + self._background_compression_min_tokens = 1 + self._background_compressor = _RecordingBackgroundCompressor() + self.executor_calls: list[float] = [] + + async def _run_compression_in_executor(self, fn, timeout): + self.executor_calls.append(timeout) + return fn() + + async def _next_request_id(self) -> str: + return "req-fastpass-test" + + def _extract_tags(self, headers): + return {} + + async def _retry_request(self, method, url, headers, body, **_kwargs): + self.captured_body = body + return _ResponseStub() + + def _get_compression_cache(self, session_id): + self.comp_cache_updates: list[tuple] = getattr(self, "comp_cache_updates", []) + return SimpleNamespace( + apply_cached=lambda m: m, + compute_frozen_count=lambda m: 0, + mark_stable_from_messages=lambda *a, **k: None, + should_defer_compression=lambda h: False, + mark_stable=lambda h: None, + content_hash=lambda c: "h", + update_from_result=lambda *a: self.comp_cache_updates.append(a), + _cache={}, + _stable_hashes=set(), + ) + + +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) + + +def test_cold_start_runs_fast_pass_and_defers_only_kompress(monkeypatch): + import headroom.tokenizers as _tk + + monkeypatch.setattr(_tk, "get_tokenizer", lambda model: _DummyTokenizer()) + + handler = _DummyAnthropicHandler() + request = _build_request( + { + "model": "claude-3-5-sonnet-latest", + "messages": [ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": "verbose stale tool output " * 200, + } + ], + }, + ], + } + ) + + anyio.run(handler.handle_anthropic_messages, request) + + # The fast pass ran synchronously with the ML stage disabled. + assert handler.executor_calls, "fast pass never ran through the executor" + sync_calls = [ + c for c in handler.anthropic_pipeline.apply.call_args_list if c.kwargs.get("skip_kompress") + ] + assert len(sync_calls) == 1, "expected exactly one synchronous skip_kompress pass" + + # The full pipeline (kompress included) went to the background queue, + # keyed against the ORIGINAL messages for content-hash reuse. + assert len(handler._background_compressor.enqueued) == 1 + _key, bg_compress, _store = handler._background_compressor.enqueued[0] + bg_compress() + bg_calls = [ + c + for c in handler.anthropic_pipeline.apply.call_args_list + if not c.kwargs.get("skip_kompress") + ] + assert len(bg_calls) == 1, "background job must run the full pipeline" + + # The FORWARDED body carries the fast-pass form — that is what the + # provider caches and the byte-identical freeze locks in. + forwarded = handler.captured_body["messages"] + assert forwarded[0]["content"][0]["content"] == _COMPRESSED_TEXT + + # Fast-pass results were stored in the compression cache. + assert handler.comp_cache_updates + + +def test_fast_pass_failure_falls_back_to_full_deferral(monkeypatch): + import headroom.tokenizers as _tk + + monkeypatch.setattr(_tk, "get_tokenizer", lambda model: _DummyTokenizer()) + + handler = _DummyAnthropicHandler() + + async def _boom(fn, timeout): + raise TimeoutError("fast pass exceeded budget") + + handler._run_compression_in_executor = _boom # type: ignore[method-assign] + + original_text = "verbose stale tool output " * 200 + request = _build_request( + { + "model": "claude-3-5-sonnet-latest", + "messages": [ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_1", + "content": original_text, + } + ], + }, + ], + } + ) + + anyio.run(handler.handle_anthropic_messages, request) + + # Fail-open: original messages forwarded, background job still queued. + forwarded = handler.captured_body["messages"] + assert forwarded[0]["content"][0]["content"] == original_text + assert len(handler._background_compressor.enqueued) == 1 diff --git a/tests/test_transforms/test_content_router.py b/tests/test_transforms/test_content_router.py index a3318d24a..45ef3aea2 100644 --- a/tests/test_transforms/test_content_router.py +++ b/tests/test_transforms/test_content_router.py @@ -204,6 +204,62 @@ def test_force_kompress_routes_anthropic_tool_result_to_targeted_kompress( assert captured["target_ratio"] == 0.10 +def test_skip_kompress_routes_around_ml_stage(router, tokenizer, monkeypatch): + """skip_kompress (cold-start fast pass) must never invoke the Kompress ML + stage — units that would route there take the same fallback as when the + model isn't ready — and wins over force_kompress.""" + calls: list[str] = [] + + class FakeKompress: + def is_ready(self) -> bool: + return True + + def ensure_background_load(self) -> None: + pass + + def compress(self, content, **kwargs): + calls.append(content) + compressed = " ".join(content.split()[:20]) + " Retrieve more: hash=deadbeef" + return SimpleNamespace( + compressed=compressed, + compressed_tokens=len(compressed.split()), + ) + + monkeypatch.setattr(router, "_get_kompress", lambda: FakeKompress()) + tool_content = " ".join( + f'{{"file":"src/module_{i}.py","line":{i},"text":"repeated search payload"}}' + for i in range(160) + ) + messages = [ + { + "role": "user", + "content": [ + { + "type": "tool_result", + "tool_use_id": "toolu_search_1", + "content": tool_content, + } + ], + } + ] + + result = router.apply( + messages, + tokenizer, + force_kompress=True, + skip_kompress=True, + target_ratio=0.10, + compress_user_messages=True, + min_tokens_to_compress=10, + read_protection_window=0, + ) + + assert calls == [] + assert result.transforms_applied != ["router:tool_result:kompress"] + # The pass still completes and returns a well-formed message list. + assert result.messages[0]["content"][0]["content"] + + def test_anthropic_tool_result_lossy_without_marker_stays_verbatim(router, tokenizer, monkeypatch): """Reversibility gate (#1307): a lossy Kompress result on a tool_result block with no CCR retrieval marker is unrecoverable, so the router must keep the