mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-10 14:27:00 -04:00
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>
This commit is contained in:
Tejas Chopra
GitHub
parent
46da91b2f1
commit
2797099bec
2 changed files with 137 additions and 0 deletions
|
|
@ -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
|
||||
|
|
|
|||
|
|
@ -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"]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue