From e36439a9411bf7fc93b4a5dceac50aa4570a6105 Mon Sep 17 00:00:00 2001 From: julienguarino <43437111+julienguarino@users.noreply.github.com> Date: Wed, 8 Jul 2026 23:24:15 +0200 Subject: [PATCH] fix(proxy): compress Anthropic user text blocks when enabled (#1875) ## 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 --- headroom/transforms/content_router.py | 16 ++-- tests/test_content_router_user_blocks.py | 108 +++++++++++++++++++++++ 2 files changed, 117 insertions(+), 7 deletions(-) create mode 100644 tests/test_content_router_user_blocks.py diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 2f6e46838..2b5d73edf 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -4360,14 +4360,16 @@ class ContentRouter(Transform): # Role-based gate for `text` blocks. Tool/function roles are tool # outputs and compress freely; assistant defaults to skip (cache # safety) with explicit opt-in; unknown roles default to skip. - if (skip_user and role == "user") or (skip_system and role in {"system", "developer"}): - protect_text_blocks = True - elif role == "assistant" and not compress_assistant_text_blocks: - protect_text_blocks = True - elif role not in ("assistant", "tool", "function"): - protect_text_blocks = True - else: + if role == "user": + protect_text_blocks = skip_user + elif role in {"system", "developer"}: + protect_text_blocks = skip_system + elif role == "assistant": + protect_text_blocks = not compress_assistant_text_blocks + elif role in ("tool", "function"): protect_text_blocks = False + else: + protect_text_blocks = True for block in content_blocks: if not isinstance(block, dict): diff --git a/tests/test_content_router_user_blocks.py b/tests/test_content_router_user_blocks.py new file mode 100644 index 000000000..41db62ddf --- /dev/null +++ b/tests/test_content_router_user_blocks.py @@ -0,0 +1,108 @@ +"""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"]