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