mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
## Description `read_lifecycle.apply()` already supports a frozen message prefix (`frozen_message_count`) — stale-Read replacements inside the prefix are skipped so compression never rewrites messages the provider's prompt cache has anchored. But only the proxy handlers can pass it: `ContentRouter` reads it from transform kwargs, `CompressConfig` has no such field, and the public `compress()` never forwards it. Library-mode callers that manage their own conversation loop (SDK integrations, offline evaluation, sidecar scoring) therefore can't stop transforms from rewriting already-sent history. On cached Anthropic traffic that's expensive: every byte after the first rewritten one stops billing as a 0.1× cache read and re-bills as a cache write (1.25× at the 5-minute TTL, 2× at the 1-hour TTL) — measured on live coding-agent traffic, retroactive stale-Read rewrites were the dominant cache-bust source once tool injection went session-sticky (PR-B7). Relates to #809 (cache-bust economics discussion); does not close it. ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [ ] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made - `CompressConfig.frozen_message_count: int = 0` — documented field; default `0` preserves existing behavior exactly. - `compress()` forwards it through `pipeline.apply()` to the transforms, matching what the proxy handlers already do. - `compress()` docstring: added to the kwargs shorthand list. - CHANGELOG entry under Unreleased → Features. - Four tests in `tests/test_compress_api.py` (`TestFrozenMessageCount`). ## Testing - [x] Unit tests pass (`pytest`) - [x] Linting passes (`ruff check .`) - [x] Type checking passes (`mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text $ uv run pytest tests/test_compress_api.py tests/test_transforms/test_read_lifecycle.py \ tests/test_compression_safety_rails.py tests/test_compress_failure.py -q 59 passed, 1 warning in 3.05s $ uv run ruff check headroom/compress.py tests/test_compress_api.py All checks passed! $ uv run mypy headroom Success: no issues found in 471 source files ``` ## Real Behavior Proof - Environment: Linux, Python 3.12.3, this branch installed via `uv sync --extra dev` - Exact command / steps: build an Anthropic-format conversation with a stale Read (file read at message 2, edited at message 3), then: ```python r0 = compress(msgs, model="claude-sonnet-4-5-20250929") r5 = compress(msgs, model="claude-sonnet-4-5-20250929", frozen_message_count=5) ``` - Observed result: without frozen prefix the stale Read is rewritten; with frozen_message_count=5 the Read remains byte-identical. ```text without frozen prefix: stale Read rewritten: True transforms: ['read_lifecycle:stale:/app/config.py'] with frozen_message_count=5: Read byte-identical: True transforms: [] ``` - Not tested: proxy-mode code paths (untouched — they already pass `frozen_message_count` their own way); Rust crates (untouched). ## Review Readiness - [x] I have performed a self-review - [x] This PR is ready for human review ## Checklist - [x] My code follows the project's style guidelines - [x] I have performed a self-review of my code - [x] I have commented my code, particularly in hard-to-understand areas - [x] I have made corresponding changes to the documentation - [x] My changes generate no new warnings - [x] I have added tests that prove my fix is effective or that my feature works - [x] New and existing unit tests pass locally with my changes - [x] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) N/A — library API change, no UI. ## Additional Notes Default `0` makes this a strict superset of current behavior — no caller sees any change without opting in. The motivation data comes from a proxy-side measurement tool that prices compression's cache effects on live Anthropic agent traffic (per-request cache-adjusted dollars); happy to share methodology in #809 if useful. --------- Co-authored-by: Claude Fable 5 <noreply@anthropic.com> Co-authored-by: JerrettDavis <mxjerrett@gmail.com> Co-authored-by: Tejas Chopra <chopratejas@gmail.com>
416 lines
15 KiB
Python
416 lines
15 KiB
Python
"""Tests for the one-function compress() API and integrations."""
|
|
|
|
import json
|
|
from dataclasses import replace as _dc_replace
|
|
|
|
import pytest
|
|
|
|
from headroom.compress import CompressConfig, CompressResult, compress
|
|
from headroom.hooks import CompressionHooks
|
|
|
|
try:
|
|
from starlette.applications import Starlette
|
|
from starlette.requests import Request
|
|
from starlette.responses import JSONResponse
|
|
from starlette.routing import Route
|
|
from starlette.testclient import TestClient
|
|
|
|
from headroom.integrations.asgi import CompressionMiddleware
|
|
|
|
HAS_STARLETTE = True
|
|
except ImportError:
|
|
HAS_STARLETTE = False
|
|
|
|
|
|
# =============================================================================
|
|
# Tests: compress() function
|
|
# =============================================================================
|
|
|
|
|
|
class TestCompressFunction:
|
|
def test_empty_messages(self):
|
|
result = compress([], model="test")
|
|
assert result.messages == []
|
|
assert result.tokens_saved == 0
|
|
|
|
def test_small_messages_passthrough(self):
|
|
"""Small messages below compression threshold pass through unchanged."""
|
|
messages = [{"role": "user", "content": "hello"}]
|
|
result = compress(messages, model="gpt-4o")
|
|
assert result.messages[0]["content"] == "hello"
|
|
assert result.tokens_saved == 0
|
|
|
|
def test_returns_compress_result(self):
|
|
result = compress([{"role": "user", "content": "hi"}])
|
|
assert isinstance(result, CompressResult)
|
|
assert hasattr(result, "messages")
|
|
assert hasattr(result, "tokens_saved")
|
|
assert hasattr(result, "compression_ratio")
|
|
assert hasattr(result, "transforms_applied")
|
|
|
|
def test_large_tool_output_compressed(self):
|
|
"""Large JSON tool output should be compressed."""
|
|
big_data = json.dumps(
|
|
[
|
|
{"id": i, "status": "active", "name": f"item_{i}", "value": i * 17}
|
|
for i in range(200)
|
|
]
|
|
)
|
|
messages = [
|
|
{"role": "user", "content": "What are the top items?"},
|
|
{"role": "tool", "content": big_data, "tool_call_id": "call_1"},
|
|
]
|
|
result = compress(messages, model="gpt-4o")
|
|
assert result.tokens_after <= result.tokens_before
|
|
assert len(result.messages) == 2
|
|
|
|
def test_compact_json_counts_tokens_not_whitespace(self):
|
|
"""Compact JSON arrays should still compress under token thresholds."""
|
|
numbers = [42.0 + i * 0.1 for i in range(200)]
|
|
messages = [
|
|
{"role": "system", "content": "You are helpful."},
|
|
{"role": "user", "content": "Show metrics"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "call_1",
|
|
"type": "function",
|
|
"function": {"name": "get_metrics", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call_1", "content": json.dumps(numbers)},
|
|
]
|
|
|
|
result = compress(messages, min_tokens_to_compress=250)
|
|
|
|
assert result.tokens_saved > 0
|
|
assert any(
|
|
transform.startswith("router:smart_crusher") for transform in result.transforms_applied
|
|
)
|
|
|
|
def test_optimize_false_passthrough(self):
|
|
"""optimize=False returns messages unchanged."""
|
|
messages = [{"role": "user", "content": "hello world " * 100}]
|
|
result = compress(messages, optimize=False)
|
|
assert result.messages is messages
|
|
assert result.tokens_saved == 0
|
|
|
|
def test_kwargs_do_not_mutate_caller_config(self):
|
|
"""kwargs must not smuggle their values onto the caller's CompressConfig.
|
|
|
|
Regression: ``compress`` did ``cfg = config or CompressConfig()`` and
|
|
then ``setattr(cfg, key, value)`` for every matching kwarg — so a caller
|
|
who passed ``config=my_cfg, protect_recent=0`` came back to find their
|
|
long-lived ``my_cfg`` silently rewritten. A shared, per-agent config
|
|
was corrupted by every request that overrode a single option.
|
|
"""
|
|
|
|
big_data = json.dumps([{"id": i, "status": "active"} for i in range(200)])
|
|
messages = [
|
|
{"role": "user", "content": "analyze"},
|
|
{"role": "tool", "content": big_data, "tool_call_id": "c1"},
|
|
]
|
|
cfg = CompressConfig(protect_recent=4, target_ratio=0.8)
|
|
snapshot = _dc_replace(cfg)
|
|
|
|
compress(
|
|
messages,
|
|
model="claude-sonnet-4-5-20250929",
|
|
config=cfg,
|
|
protect_recent=0,
|
|
target_ratio=0.2,
|
|
)
|
|
|
|
assert cfg.protect_recent == snapshot.protect_recent, (
|
|
"compress() mutated caller's config.protect_recent via kwargs"
|
|
)
|
|
assert cfg.target_ratio == snapshot.target_ratio, (
|
|
"compress() mutated caller's config.target_ratio via kwargs"
|
|
)
|
|
|
|
def test_with_custom_hooks(self):
|
|
"""Hooks are called when provided."""
|
|
calls = []
|
|
|
|
class TrackingHooks(CompressionHooks):
|
|
def pre_compress(self, messages, ctx):
|
|
calls.append(("pre", len(messages)))
|
|
return messages
|
|
|
|
def compute_biases(self, messages, ctx):
|
|
calls.append(("biases", len(messages)))
|
|
return {}
|
|
|
|
def post_compress(self, event):
|
|
calls.append(("post", event.tokens_saved))
|
|
|
|
big_data = json.dumps([{"id": i, "status": "active"} for i in range(100)])
|
|
messages = [
|
|
{"role": "user", "content": "analyze"},
|
|
{"role": "tool", "content": big_data, "tool_call_id": "c1"},
|
|
]
|
|
compress(messages, hooks=TrackingHooks())
|
|
|
|
assert any(c[0] == "pre" for c in calls)
|
|
assert any(c[0] == "biases" for c in calls)
|
|
|
|
|
|
class TestCompressResultFields:
|
|
def test_fields_populated(self):
|
|
big_data = json.dumps([{"id": i, "type": "log"} for i in range(100)])
|
|
messages = [
|
|
{"role": "user", "content": "summarize"},
|
|
{"role": "tool", "content": big_data, "tool_call_id": "c1"},
|
|
]
|
|
result = compress(messages, model="claude-sonnet-4-5-20250929")
|
|
assert result.tokens_before > 0
|
|
assert result.tokens_after >= 0
|
|
assert result.tokens_saved >= 0
|
|
assert 0.0 <= result.compression_ratio <= 1.0
|
|
|
|
|
|
# =============================================================================
|
|
# Tests: ASGI CompressionMiddleware (requires starlette)
|
|
# =============================================================================
|
|
|
|
|
|
def _make_asgi_app(middleware_kwargs=None):
|
|
"""Create a test ASGI app with CompressionMiddleware."""
|
|
|
|
async def chat_endpoint(request: Request) -> JSONResponse:
|
|
body = await request.json()
|
|
return JSONResponse(
|
|
{
|
|
"model": "gpt-4o",
|
|
"choices": [{"message": {"content": "response"}}],
|
|
"usage": {"prompt_tokens": 10, "completion_tokens": 5},
|
|
"_message_count": len(body.get("messages", [])),
|
|
}
|
|
)
|
|
|
|
async def health(request: Request) -> JSONResponse:
|
|
return JSONResponse({"status": "ok"})
|
|
|
|
app = Starlette(
|
|
routes=[
|
|
Route("/health", health),
|
|
Route("/v1/chat/completions", chat_endpoint, methods=["POST"]),
|
|
Route("/v1/messages", chat_endpoint, methods=["POST"]),
|
|
]
|
|
)
|
|
app.add_middleware(CompressionMiddleware, **(middleware_kwargs or {}))
|
|
return app
|
|
|
|
|
|
@pytest.mark.skipif(not HAS_STARLETTE, reason="starlette not installed")
|
|
class TestASGIMiddleware:
|
|
def test_non_llm_paths_passthrough(self):
|
|
app = _make_asgi_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/health")
|
|
assert resp.status_code == 200
|
|
assert resp.json()["status"] == "ok"
|
|
|
|
def test_small_messages_passthrough(self):
|
|
app = _make_asgi_app()
|
|
client = TestClient(app)
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={"model": "gpt-4o", "messages": [{"role": "user", "content": "hi"}]},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_large_messages_compressed(self):
|
|
"""Large tool output should be compressed by middleware."""
|
|
app = _make_asgi_app()
|
|
client = TestClient(app)
|
|
|
|
big_data = json.dumps([{"id": i, "status": "active"} for i in range(200)])
|
|
resp = client.post(
|
|
"/v1/chat/completions",
|
|
json={
|
|
"model": "gpt-4o",
|
|
"messages": [
|
|
{"role": "user", "content": "analyze"},
|
|
{"role": "tool", "content": big_data, "tool_call_id": "c1"},
|
|
],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_anthropic_path(self):
|
|
"""Works with Anthropic /v1/messages path."""
|
|
app = _make_asgi_app()
|
|
client = TestClient(app)
|
|
resp = client.post(
|
|
"/v1/messages",
|
|
json={
|
|
"model": "claude-sonnet-4-5-20250929",
|
|
"messages": [{"role": "user", "content": "hello"}],
|
|
},
|
|
)
|
|
assert resp.status_code == 200
|
|
|
|
def test_get_requests_passthrough(self):
|
|
"""GET requests to LLM paths pass through."""
|
|
app = _make_asgi_app()
|
|
client = TestClient(app)
|
|
resp = client.get("/v1/chat/completions")
|
|
assert resp.status_code in (200, 405)
|
|
|
|
|
|
# =============================================================================
|
|
# Tests: LiteLLM Callback
|
|
# =============================================================================
|
|
|
|
|
|
class TestLiteLLMCallback:
|
|
def test_callback_imports(self):
|
|
"""Verify the callback can be imported."""
|
|
from headroom.integrations.litellm_callback import HeadroomCallback
|
|
|
|
callback = HeadroomCallback()
|
|
assert callback.total_tokens_saved == 0
|
|
|
|
def test_callback_compresses_messages(self):
|
|
"""Callback compresses messages in pre_call_hook."""
|
|
import asyncio
|
|
|
|
from headroom.integrations.litellm_callback import HeadroomCallback
|
|
|
|
callback = HeadroomCallback()
|
|
|
|
big_data = json.dumps([{"id": i, "status": "active"} for i in range(200)])
|
|
data = {
|
|
"model": "gpt-4o",
|
|
"messages": [
|
|
{"role": "user", "content": "analyze"},
|
|
{"role": "tool", "content": big_data, "tool_call_id": "c1"},
|
|
],
|
|
}
|
|
|
|
result = asyncio.run(callback.async_pre_call_hook("key", data, "completion"))
|
|
assert result is data
|
|
|
|
def test_callback_ignores_non_completion(self):
|
|
"""Non-completion calls are passed through."""
|
|
import asyncio
|
|
|
|
from headroom.integrations.litellm_callback import HeadroomCallback
|
|
|
|
callback = HeadroomCallback()
|
|
data = {"messages": [{"role": "user", "content": "hi"}]}
|
|
|
|
result = asyncio.run(callback.async_pre_call_hook("key", data, "embedding"))
|
|
assert result is data
|
|
|
|
|
|
# =============================================================================
|
|
# Tests: frozen_message_count through library-mode compress()
|
|
# =============================================================================
|
|
|
|
|
|
class TestFrozenMessageCount:
|
|
"""The frozen prefix must be reachable from library mode.
|
|
|
|
Proxy handlers pass frozen_message_count so transforms never rewrite
|
|
messages already anchored in the provider's prompt cache. Library-mode
|
|
callers manage their own conversation loop and need the same control —
|
|
without it, read_lifecycle rewrites sent history and converts cached
|
|
prefix reads into full-price rewrites.
|
|
"""
|
|
|
|
@staticmethod
|
|
def _stale_read_conversation() -> list[dict]:
|
|
"""Anthropic-format conversation with a stale Read: file read early,
|
|
edited later. read_lifecycle should classify the Read as STALE."""
|
|
big_content = "\n".join(f"line {i}: some file content here" for i in range(80))
|
|
return [
|
|
{"role": "user", "content": "read then edit the config"},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"type": "tool_use",
|
|
"id": "t_read",
|
|
"name": "Read",
|
|
"input": {"file_path": "/app/config.py"},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{"type": "tool_result", "tool_use_id": "t_read", "content": big_content}
|
|
],
|
|
},
|
|
{
|
|
"role": "assistant",
|
|
"content": [
|
|
{
|
|
"type": "tool_use",
|
|
"id": "t_edit",
|
|
"name": "Edit",
|
|
"input": {
|
|
"file_path": "/app/config.py",
|
|
"old_string": "old",
|
|
"new_string": "new",
|
|
},
|
|
}
|
|
],
|
|
},
|
|
{
|
|
"role": "user",
|
|
"content": [{"type": "tool_result", "tool_use_id": "t_edit", "content": "ok"}],
|
|
},
|
|
{"role": "assistant", "content": "edited."},
|
|
{"role": "user", "content": "now summarize the change"},
|
|
]
|
|
|
|
@staticmethod
|
|
def _read_result_content(messages: list[dict]) -> str:
|
|
for msg in messages:
|
|
content = msg.get("content")
|
|
if not isinstance(content, list):
|
|
continue
|
|
for block in content:
|
|
if isinstance(block, dict) and block.get("tool_use_id") == "t_read":
|
|
return str(block.get("content"))
|
|
raise AssertionError("t_read tool_result not found")
|
|
|
|
def test_stale_read_rewritten_without_frozen_prefix(self):
|
|
"""Baseline: with no frozen prefix, the stale Read is rewritten."""
|
|
messages = self._stale_read_conversation()
|
|
original = self._read_result_content(messages)
|
|
result = compress(messages, model="claude-sonnet-4-5-20250929")
|
|
assert self._read_result_content(result.messages) != original
|
|
|
|
def test_frozen_prefix_blocks_stale_read_rewrite(self):
|
|
"""frozen_message_count as kwarg: messages inside the frozen prefix
|
|
must come back byte-identical, even though the Read is stale."""
|
|
messages = self._stale_read_conversation()
|
|
original = self._read_result_content(messages)
|
|
result = compress(
|
|
messages,
|
|
model="claude-sonnet-4-5-20250929",
|
|
frozen_message_count=5,
|
|
)
|
|
assert self._read_result_content(result.messages) == original
|
|
|
|
def test_frozen_prefix_via_config_object(self):
|
|
"""frozen_message_count set on CompressConfig behaves identically."""
|
|
messages = self._stale_read_conversation()
|
|
original = self._read_result_content(messages)
|
|
cfg = CompressConfig(frozen_message_count=5)
|
|
result = compress(messages, model="claude-sonnet-4-5-20250929", config=cfg)
|
|
assert self._read_result_content(result.messages) == original
|
|
|
|
def test_frozen_zero_is_legacy_behavior(self):
|
|
"""Explicit 0 matches the default: stale Read gets rewritten."""
|
|
messages = self._stale_read_conversation()
|
|
original = self._read_result_content(messages)
|
|
result = compress(messages, model="claude-sonnet-4-5-20250929", frozen_message_count=0)
|
|
assert self._read_result_content(result.messages) != original
|