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

## Problem

Callers that resend a growing conversation every turn — agent loops
generally, and an in-progress Strands plugin specifically — cannot 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. That trades a 90% read discount for nothing.

Measured before this change, compressing the same conversation at
increasing lengths and comparing the first 16 messages against the
4-turn baseline:

```
turns  msgs   first 16 still byte-identical?
    4    16   16/16  OK
    8    32   16/16  OK
   16    64   12/16  DRIFT at [2, 6, 10, 14]
   32   128   12/16  DRIFT at [2, 6, 10, 14]
   64   256   12/16  DRIFT at [2, 6, 10, 14]
```

Indices 2/6/10/14 are the tool-result messages.

## Fix

`TransformPipeline` and `ContentRouter` already honour
`frozen_message_count` — `content_router.py` skips any message below the
index, and `pipeline.py` even logs *"freezing first N/M messages (prefix
cached by provider)"*. It was simply missing from this endpoint's
`config` parsing, so no HTTP caller could reach it.

This adds it alongside the existing `compress_user_messages` /
`target_ratio` / `protect_recent` / `protect_analysis_context` options.
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.

Invalid values return 400, matching the existing `config.mode`
validation. `bool` is rejected explicitly, since `isinstance(True, int)`
is `True` in Python and a JSON `true` silently becoming
`frozen_message_count=1` would be a nasty surprise.

## Verified against a live proxy

```
sent 64 messages, pinned first 32 -> returned unchanged: True
                     unpinned tail still compressed: True

simulated agent loop carrying the compressed prefix forward:
   4 turns: prefix of   0 held      16 turns: prefix of  48 held
   8 turns: prefix of  16 held      24 turns: prefix of  64 held
  12 turns: prefix of  32 held      32 turns: prefix of  96 held
  prefix never drifted across the whole run: True
```

## Tests

13 new tests in `TestCompressEndpointFrozenMessageCount`, including the
regression itself: compress a 6-turn and a 24-turn conversation with the
same pin and assert the prefix is identical. 48/48 pass in
`test_proxy_compress_endpoint.py` (35 pre-existing, unchanged). ruff and
ruff-format clean.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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