From a21766a673aebaa44190572ef8fb7b7bc138aad9 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Sun, 2 Aug 2026 09:59:19 -0700 Subject: [PATCH] feat(compress): accept config.frozen_message_count on /v1/compress MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Callers that resend a growing conversation every turn — agent loops, and the Strands plugin in particular — cannot currently keep a stable prompt-cache prefix through /v1/compress. The router compresses older messages more aggressively as the conversation grows, so their bytes change and the provider's cache misses from the first rewritten message onward, trading a 90% read discount for nothing. Measured before this change, compressing the same conversation at increasing lengths: the first 16 messages were byte-identical at 4 and 8 turns, then drifted at every tool-result message from 16 turns on. TransformPipeline and ContentRouter already honour frozen_message_count — it was only missing from the endpoint's config parsing, so HTTP callers had no way to reach it. Pinning still lets cross-message transforms such as dedup *read* the prefix; it only forbids rewriting it. protect_recent guards the opposite end of the list and cannot express this. Co-Authored-By: Claude Opus 5 (1M context) --- headroom/proxy/handlers/openai.py | 33 ++++++++ tests/test_proxy_compress_endpoint.py | 104 ++++++++++++++++++++++++++ 2 files changed, 137 insertions(+) diff --git a/headroom/proxy/handlers/openai.py b/headroom/proxy/handlers/openai.py index f17ecbba7..22803d234 100644 --- a/headroom/proxy/handlers/openai.py +++ b/headroom/proxy/handlers/openai.py @@ -8354,6 +8354,13 @@ class OpenAIHandlerMixin: Any other ``config.mode`` value is a 400 (see ``COMPRESS_MODES``). + ``config.frozen_message_count`` pins a prefix: the first N messages are + returned byte-for-byte unchanged, while still being visible to cross-message + transforms such as dedup. Callers that resend a growing conversation each turn + should set it to the number of messages the provider has already cached, so + compression does not rewrite the prefix and bust that cache. Must be a + non-negative integer; anything else is a 400. + Returns compressed messages + metrics. """ from fastapi.responses import JSONResponse @@ -8473,6 +8480,30 @@ class OpenAIHandlerMixin: target_ratio = compress_config.get("target_ratio") protect_recent = compress_config.get("protect_recent") protect_analysis_context = compress_config.get("protect_analysis_context") + # Leading messages already in the provider's prompt cache. Callers that + # resend a growing conversation every turn (agent loops) need to pin the + # prefix they have already paid for: without it the router compresses old + # messages harder as the conversation grows, so their bytes change and the + # cache misses from that point on. `protect_recent` guards the other end of + # the list and cannot express this. + frozen_message_count = compress_config.get("frozen_message_count") + if frozen_message_count is not None and ( + isinstance(frozen_message_count, bool) + or not isinstance(frozen_message_count, int) + or frozen_message_count < 0 + ): + return JSONResponse( + status_code=400, + content={ + "error": { + "type": "invalid_request", + "message": ( + f"Invalid config.frozen_message_count: {frozen_message_count!r}. " + "Expected a non-negative integer." + ), + } + }, + ) # Mode selection. Default is marker-free (see _no_ccr_pipeline): # no caller of this route can resolve a CCR marker unless it opts in # with mode="ccr", which restores the full marker + store behaviour. @@ -8510,6 +8541,8 @@ class OpenAIHandlerMixin: pipeline_kwargs["protect_recent"] = int(protect_recent) if protect_analysis_context is not None: pipeline_kwargs["protect_analysis_context"] = bool(protect_analysis_context) + if frozen_message_count is not None: + pipeline_kwargs["frozen_message_count"] = frozen_message_count # Offload the CPU-bound pipeline to the bounded compression executor # (mirrors the request handlers above). Running apply() inline blocked diff --git a/tests/test_proxy_compress_endpoint.py b/tests/test_proxy_compress_endpoint.py index ac1be3c10..31c6162f0 100644 --- a/tests/test_proxy_compress_endpoint.py +++ b/tests/test_proxy_compress_endpoint.py @@ -706,3 +706,107 @@ class TestCompressEndpointDoesNotBlockLoop: resp = await asyncio.wait_for(compress, timeout=5) assert resp.status_code == 200 assert resp.json()["tokens_saved"] == 5 + + +class TestCompressEndpointFrozenMessageCount: + """``config.frozen_message_count`` pins a prefix the provider has already cached. + + Callers that resend a growing conversation every turn (agent loops, the Strands + plugin) need the leading messages to come back byte-for-byte identical. Without + this the router compresses old messages harder as the conversation grows, their + bytes change, and the provider's prompt cache misses from that point on — turning + compression into a net cost. ``protect_recent`` guards the other end of the list + and cannot express it. + """ + + @staticmethod + def _conversation(turns: int) -> list[dict]: + log = "\n".join( + f"2026-07-31 12:00:{n:02d} INFO worker={n} req=r{n} took {n}ms" for n in range(60) + ) + messages: list[dict] = [] + for i in range(turns): + messages += [ + {"role": "user", "content": f"step {i}"}, + {"role": "assistant", "content": f"reading log {i}\n{log}"}, + ] + return messages + + def test_pinned_prefix_is_returned_byte_for_byte(self, client): + messages = self._conversation(12) + response = client.post( + "/v1/compress", + json={ + "messages": messages, + "model": "gpt-4", + "config": {"compress_user_messages": True, "frozen_message_count": 8}, + }, + ) + assert response.status_code == 200 + assert response.json()["messages"][:8] == messages[:8] + + def test_the_unpinned_tail_is_still_compressed(self, client): + messages = self._conversation(12) + body = {"messages": messages, "model": "gpt-4", "config": {"compress_user_messages": True}} + full = client.post("/v1/compress", json=body).json() + pinned = client.post( + "/v1/compress", + json={**body, "config": {**body["config"], "frozen_message_count": 8}}, + ).json() + + # Pinning must not disable compression outright — only exempt the prefix. + assert pinned["messages"][8:] != messages[8:], "tail was left uncompressed" + assert pinned["tokens_after"] >= full["tokens_after"], "pinning should compress no harder" + + def test_a_pinned_prefix_does_not_drift_as_the_conversation_grows(self, client): + """The regression this field exists to prevent.""" + short, long = self._conversation(6), self._conversation(24) + config = {"compress_user_messages": True, "frozen_message_count": 12} + + a = client.post( + "/v1/compress", json={"messages": short, "model": "gpt-4", "config": config} + ).json() + b = client.post( + "/v1/compress", json={"messages": long, "model": "gpt-4", "config": config} + ).json() + + assert a["messages"][:12] == b["messages"][:12], ( + "prefix was re-rendered as the conversation grew" + ) + + @pytest.mark.parametrize("value", ["8", -1, 3.5, True, [8], {"n": 8}]) + def test_invalid_values_return_400(self, client, value): + response = client.post( + "/v1/compress", + json={ + "messages": [{"role": "user", "content": "hello"}], + "model": "gpt-4", + "config": {"frozen_message_count": value}, + }, + ) + assert response.status_code == 400 + data = response.json() + assert data["error"]["type"] == "invalid_request" + assert "frozen_message_count" in data["error"]["message"] + + @pytest.mark.parametrize("value", [0, 1, 10_000], ids=["zero", "one", "beyond-the-list"]) + def test_valid_values_are_accepted(self, client, value): + """0 means "pin nothing"; a count past the end simply pins everything.""" + response = client.post( + "/v1/compress", + json={ + "messages": [{"role": "user", "content": "hello"}], + "model": "gpt-4", + "config": {"frozen_message_count": value}, + }, + ) + assert response.status_code == 200 + + def test_unset_is_unchanged_behaviour(self, client): + messages = self._conversation(4) + body = {"messages": messages, "model": "gpt-4", "config": {"compress_user_messages": True}} + without = client.post("/v1/compress", json=body).json() + explicit_zero = client.post( + "/v1/compress", json={**body, "config": {**body["config"], "frozen_message_count": 0}} + ).json() + assert without["messages"] == explicit_zero["messages"]