mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-10 14:27:00 -04:00
e36439a941
## Summary - make Anthropic content-block user text honor compress_user_messages - keep cache_control text blocks protected even when user-message compression is enabled - add regression coverage for user text blocks, default protection, and cache_control protection ## Why The string-message path already compresses role=user content when compress_user_messages=True, but the Anthropic content-block path always fell through to the unknown-role protection branch for role=user. In proxy mode this produced router:noop for large user text blocks even with HEADROOM_COMPRESS_USER_MESSAGES=1 and HEADROOM_FORCE_KOMPRESS_ALL=1. ## Tests - python -m pytest tests/test_content_router_user_blocks.py tests/test_compression_safety_rails.py tests/test_force_kompress_all.py -q Note: running tests/test_agent_savings.py locally still hits an unrelated Rust binding mismatch: SmartCrusherConfig.__new__() got an unexpected keyword argument 'lossless_only'. Co-authored-by: Julien Guarino <julien.guarino@fashiondata.io>
108 lines
3.2 KiB
Python
108 lines
3.2 KiB
Python
"""Content-router coverage for Anthropic user text blocks."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
|
|
from headroom.transforms.content_router import ContentRouter, ContentRouterConfig
|
|
|
|
|
|
class _Tokenizer:
|
|
def count_text(self, text: str) -> int:
|
|
return max(1, len(str(text)) // 4)
|
|
|
|
|
|
def _long_text() -> str:
|
|
return "This user message should be compressible when opted in. " * 80
|
|
|
|
|
|
def test_user_text_block_compresses_when_user_messages_are_enabled() -> None:
|
|
router = ContentRouter(ContentRouterConfig(force_kompress_all=True))
|
|
calls: list[dict[str, Any]] = []
|
|
|
|
def fake_compress_block_content(**kwargs: Any) -> tuple[str | None, bool]:
|
|
calls.append(kwargs)
|
|
kwargs["transforms_applied"].append("router:text_block:fake")
|
|
return "COMPRESSED", True
|
|
|
|
router._compress_block_content = fake_compress_block_content # type: ignore[method-assign]
|
|
|
|
result = router.apply(
|
|
[
|
|
{
|
|
"role": "user",
|
|
"content": [{"type": "text", "text": _long_text()}],
|
|
}
|
|
],
|
|
_Tokenizer(),
|
|
compress_user_messages=True,
|
|
force_kompress=True,
|
|
target_ratio=0.10,
|
|
)
|
|
|
|
assert len(calls) == 1
|
|
assert result.messages[0]["content"][0]["text"] == "COMPRESSED"
|
|
assert result.transforms_applied == ["router:text_block:fake"]
|
|
|
|
|
|
def test_user_text_block_stays_protected_by_default() -> None:
|
|
router = ContentRouter(ContentRouterConfig(force_kompress_all=True))
|
|
calls: list[dict[str, Any]] = []
|
|
text = _long_text()
|
|
|
|
def fake_compress_block_content(**kwargs: Any) -> tuple[str | None, bool]:
|
|
calls.append(kwargs)
|
|
return "COMPRESSED", True
|
|
|
|
router._compress_block_content = fake_compress_block_content # type: ignore[method-assign]
|
|
|
|
result = router.apply(
|
|
[
|
|
{
|
|
"role": "user",
|
|
"content": [{"type": "text", "text": text}],
|
|
}
|
|
],
|
|
_Tokenizer(),
|
|
force_kompress=True,
|
|
target_ratio=0.10,
|
|
)
|
|
|
|
assert calls == []
|
|
assert result.messages[0]["content"][0]["text"] == text
|
|
assert result.transforms_applied == ["router:noop"]
|
|
|
|
|
|
def test_user_cache_control_text_block_stays_protected_when_enabled() -> None:
|
|
router = ContentRouter(ContentRouterConfig(force_kompress_all=True))
|
|
calls: list[dict[str, Any]] = []
|
|
text = _long_text()
|
|
|
|
def fake_compress_block_content(**kwargs: Any) -> tuple[str | None, bool]:
|
|
calls.append(kwargs)
|
|
return "COMPRESSED", True
|
|
|
|
router._compress_block_content = fake_compress_block_content # type: ignore[method-assign]
|
|
|
|
result = router.apply(
|
|
[
|
|
{
|
|
"role": "user",
|
|
"content": [
|
|
{
|
|
"type": "text",
|
|
"text": text,
|
|
"cache_control": {"type": "ephemeral"},
|
|
}
|
|
],
|
|
}
|
|
],
|
|
_Tokenizer(),
|
|
compress_user_messages=True,
|
|
force_kompress=True,
|
|
target_ratio=0.10,
|
|
)
|
|
|
|
assert calls == []
|
|
assert result.messages[0]["content"][0]["text"] == text
|
|
assert result.transforms_applied == ["router:noop"]
|