feat(compress): accept config.frozen_message_count on /v1/compress

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) <noreply@anthropic.com>

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

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