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 <julien.guarino@fashiondata.io>

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):

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