fix(proxy): honor force_kompress routing profile (#996)

## Description

Honor the proxy savings profile's `force_kompress` setting all the way
through the Anthropic proxy path.

`HEADROOM_SAVINGS_PROFILE=agent-90` already resolves to
`force_kompress=True`, but `ContentRouter` still paid for the full
auto-detection path before selecting Kompress. On long Claude Code /
tool-output conversations this can hang inside the detection/router path
before any `Transform content_router` line is emitted. This change makes
the forced-Kompress path skip unused strategy detection during
compression, while still preserving recent-code protection via the
lightweight regex detector.

This also passes `proxy_pipeline_kwargs(self.config)` through Anthropic
batch requests so batch traffic receives the same savings-profile knobs
as normal Anthropic messages.

Refs #946

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] 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

- Skip `is_mixed_content()` / `_detect_content()` when runtime
`force_kompress` is set and route directly to
`CompressionStrategy.KOMPRESS`.
- Keep forced-Kompress recent-code protection, but use
`_regex_detect_content_type()` instead of the full router detection
chain.
- Read `_runtime_force_kompress` defensively in `ContentRouter.apply()`
so regular `ContentRouter()` instances keep the normal content-detection
path.
- Pass proxy savings-profile kwargs into Anthropic batch compression.
- Add regression tests for forced-Kompress routing, normal routing,
recent-code protection, and Anthropic batch profile propagation.
- Update `CHANGELOG.md`.

## Testing

- [ ] Unit tests pass (`pytest`)
- [ ] Linting passes (`ruff check .`)
- [ ] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

### Test Output

```text
$ ruff check headroom/transforms/content_router.py tests/test_transforms_content_router.py
All checks passed!

$ ruff format --check headroom/transforms/content_router.py tests/test_transforms_content_router.py
2 files already formatted

$ pytest tests/test_transforms_content_router.py::test_force_kompress_bypasses_content_detection \
    tests/test_transforms_content_router.py::test_normal_compress_path_still_uses_content_detection \
    tests/test_transforms_content_router.py::test_force_kompress_apply_uses_lightweight_detection \
    tests/test_transforms_content_router.py::test_force_kompress_apply_lightweight_detection_protects_recent_code \
    tests/test_proxy_anthropic_cache_stability.py::test_batch_optimization_passes_savings_profile_kwargs \
    tests/test_bundled_tools_savings.py -q
============================= test session starts =============================
platform win32 -- Python 3.13.1, pytest-9.0.3, pluggy-1.6.0
rootdir: E:\work\code\third-party\headroom
configfile: pyproject.toml
plugins: anyio-4.12.1, langsmith-0.8.0, asyncio-1.3.0, cov-7.0.0, timeout-2.4.0
collected 11 items

tests\test_transforms_content_router.py ....                             [ 36%]
tests\test_proxy_anthropic_cache_stability.py .                          [ 45%]
tests\test_bundled_tools_savings.py ....ss                               [100%]

======================== 9 passed, 2 skipped in 9.77s =========================
```

Full-suite attempt status on Windows / Python 3.13 after installing
missing local test dependencies and bundled tools (`fastembed`,
`socksio`, `pytest-timeout`, `difft`, `scc`, cached HF models with
offline env vars):

```text
tests/test_adapter_hooks.py: 29 passed, 2 failed
  - sqlite:///C:\... and jsonl:///C:\... URLs are parsed into invalid \C:\... paths on Windows.

tests/test_cache/test_client_integration.py: 16 failed
  - Same Windows URL path parsing issue.

tests/test_cli/test_wrap_helpers.py: 29 passed, then KeyboardInterrupt during read/cleanup.

tests/test_memory tests/test_storage:
  - Collection/run receives KeyboardInterrupt in this Windows environment.
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.13.1, `headroom-ai` v0.25.0,
Anthropic proxy on `127.0.0.1:8787`, Kompress ONNX backend,
`HEADROOM_SAVINGS_PROFILE=agent-90`,
`HEADROOM_COMPRESS_USER_MESSAGES=1`, `HEADROOM_MIN_TOKENS=120`.
- Exact command / steps: started the proxy with the local launcher, sent
a long `/v1/messages` request with a fake upstream token, and inspected
`/livez`, `/stats?include_config=true`, and
`~/.headroom/logs/proxy.log`.
- Observed result: request returned promptly with the expected upstream
auth failure after local compression, and logs showed the compression
ran before forwarding:

```text
/livez healthy
/v1/messages completed in ~3005ms with expected upstream 401
Transform content_router: 1900 -> 193 tokens (saved 1707) [1328.4ms]
Pipeline complete: 1907 -> 200 tokens (saved 1707, 89.5% reduction)
UPSTREAM_ERROR ... compressed=yes transforms=['router:kompress:0.06'] original_tokens=1886 optimized_tokens=119
PERF ... tok_before=1886 tok_after=119 tok_saved=1767 transforms=router:kompress:0.06
/stats tokens.saved = 1767
/stats compressions_by_strategy = {"kompress": 1}
```

- Not tested: full upstream CI matrix, full `uv run pytest`, full `ruff
check .`, `mypy headroom`, real Anthropic success response with a valid
upstream token, and Anthropic batch against the live upstream. The
Anthropic batch change is covered by a local handler regression test.

## 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
- [ ] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [ ] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Screenshots (if applicable)

N/A

## Additional Notes

This PR is ready for human review. The patch is scoped to the
forced-Kompress profile path and does not change the default
auto-routing behavior when `force_kompress` is false.

The latest `PR Governance / template` check passes after the readiness
checkbox update. A later `PR Governance / label` run currently fails
while trying to execute `.github/scripts/pr-health-labels.py` from the
base checkout; that file is missing on the checked-out base ref, so this
appears to be a governance workflow issue rather than a
PR-template/content failure in this branch.

CHANGELOG.md

@@ -34,6 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 * **proxy:** route native Bedrock `/model/{id}/converse` requests to the upstream Converse endpoint instead of the hard-coded `/invoke` action — the non-streaming handler now resolves the action from the inbound path, matching the streaming handler ([#999](https://github.com/chopratejas/headroom/pull/999)).
 * **proxy:** preserve byte-faithful `/v1/messages` forwarding when Anthropic tool arrays are already canonical, and only canonicalize-and-mutate tool lists when sorting changes ordering ([#1042](https://github.com/chopratejas/headroom/issues/1042)).
 * **ccr:** make retrieval store TTL configurable with `HEADROOM_CCR_TTL_SECONDS`, expose the effective TTL in `/v1/retrieve/stats`, and distinguish expired retrievals from missing hashes.
+* **proxy:** make `force_kompress` skip ContentRouter auto-detection during compression and pass savings-profile kwargs through Anthropic batch requests.
 * **proxy:** add native Bedrock `/model/{id}/converse-stream` route and forward it through the existing streaming EventStream/SSE pipeline.
 * **wrap (codex):** fix `headroom wrap codex` producing a `config.toml` with duplicate top-level `model_provider` / `openai_base_url` keys (TOML-spec error) when the user had already configured their own provider. The injector now rewrites pre-existing top-level `model_provider` and `openai_base_url` lines in place — the previous value is kept in a `# was: …` trailing comment — instead of unconditionally prepending a duplicate, so `codex` can start against the proxy. The pre-wrap snapshot mechanism continues to byte-for-byte restore the original file on `headroom unwrap codex`.
 * **wrap:** isolate wrapped proxy subprocess stdout/stderr into `proxy-stdio.log`, so `proxy.log` remains the canonical rotating runtime log and Windows rollover failures from `RotatingFileHandler` are no longer blocked by wrapper stdio handles ([#1184](https://github.com/chopratejas/headroom/issues/1184)).

headroom/proxy/handlers/anthropic.py

@@ -2695,6 +2695,7 @@ class AnthropicHandlerMixin:
                         context=extract_user_query(messages),
                         frozen_message_count=frozen_message_count,
                         request_id=request_id,
+                        **proxy_pipeline_kwargs(self.config),
                     )
 
                     optimized_messages = result.messages

headroom/transforms/content_router.py

@@ -1104,15 +1104,18 @@ class ContentRouter(Transform):
                 routing_log=[],
             )
         else:
-            # Determine strategy from content analysis
+            # Determine strategy from content analysis. When runtime settings
-            mixed = is_mixed_content(content)
+            # force Kompress, skip the full router detection path so large
-            detection = _detect_content(content)
+            # proxy payloads do not pay for an unused strategy decision.
             force_kompress = bool(getattr(self, "_runtime_force_kompress", False))
-            strategy = (
+            if force_kompress:
-                CompressionStrategy.KOMPRESS
+                mixed = False
-                if force_kompress
+                detection = DetectionResult(ContentType.PLAIN_TEXT, 1.0, {})
-                else self._determine_strategy(content)
+                strategy = CompressionStrategy.KOMPRESS
-            )
+            else:
+                mixed = is_mixed_content(content)
+                detection = _detect_content(content)
+                strategy = self._determine_strategy(content)
             if debug_enabled:
                 _log_router_debug(
                     "content_router_input",
@@ -2569,8 +2572,14 @@ class ContentRouter(Transform):
                 route_counts["error_protected"] += 1
                 continue
 
-            # Detect content type for protection decisions
+            # Detect content type for protection decisions. Even when the
-            detection = _detect_content(content)
+            # runtime strategy is forced to Kompress, keep code-protection
+            # checks but use the lightweight regex detector instead of the
+            # full router chain.
+            force_kompress = bool(getattr(self, "_runtime_force_kompress", False))
+            detection = (
+                _regex_detect_content_type(content) if force_kompress else _detect_content(content)
+            )
             is_code = detection.content_type == ContentType.SOURCE_CODE
 
             # Protection 2: Don't compress recent CODE

tests/test_proxy_anthropic_cache_stability.py

@@ -685,6 +685,71 @@ def test_batch_optimization_freezes_previous_turns_only() -> None:
         ]
 
 
+def test_batch_optimization_passes_savings_profile_kwargs() -> None:
+    captured = {}
+    with _make_proxy_client() as client:
+        proxy = client.app.state.proxy
+        proxy.config.optimize = True
+        proxy.config.mode = "token"
+        proxy.config.savings_profile = "agent-90"
+        proxy.config.ccr_inject_tool = False
+
+        def _fake_apply(**kwargs):
+            captured["pipeline_kwargs"] = kwargs
+            return SimpleNamespace(
+                messages=kwargs["messages"],
+                transforms_applied=[],
+                timing={},
+                tokens_before=100,
+                tokens_after=80,
+                waste_signals=None,
+            )
+
+        proxy.anthropic_pipeline.apply = _fake_apply
+
+        async def _fake_retry(method, url, headers, body, stream=False, **kwargs):  # noqa: ANN001
+            captured["body"] = body
+            return httpx.Response(
+                200,
+                json={
+                    "id": "msgbatch_profile",
+                    "type": "message_batch",
+                    "processing_status": "in_progress",
+                    "request_counts": {
+                        "processing": 1,
+                        "succeeded": 0,
+                        "errored": 0,
+                        "canceled": 0,
+                    },
+                },
+            )
+
+        proxy._retry_request = _fake_retry
+
+        response = client.post(
+            "/v1/messages/batches",
+            headers={"x-api-key": "test-key", "anthropic-version": "2023-06-01"},
+            json={
+                "requests": [
+                    {
+                        "custom_id": "req-1",
+                        "params": {
+                            "model": "claude-sonnet-4-6",
+                            "max_tokens": 128,
+                            "messages": [{"role": "user", "content": "compress me"}],
+                        },
+                    }
+                ]
+            },
+        )
+
+        assert response.status_code == 200
+        pipeline_kwargs = captured["pipeline_kwargs"]
+        assert pipeline_kwargs["force_kompress"] is True
+        assert pipeline_kwargs["target_ratio"] == 0.10
+        assert pipeline_kwargs["compress_user_messages"] is True
+
+
 def test_token_mode_does_not_force_freeze_all_previous_turns() -> None:
     captured = {}
     with _make_proxy_client() as client:

tests/test_transforms_content_router.py

@@ -277,6 +277,150 @@ def test_content_router_strategy_and_compress_paths(monkeypatch: pytest.MonkeyPa
     assert router.compress("   ").strategy_used is CompressionStrategy.PASSTHROUGH
 
 
+def test_force_kompress_bypasses_content_detection(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = ContentRouter()
+    router._runtime_force_kompress = True
+    pure_result = RouterCompressionResult(
+        compressed="pure",
+        original="pure",
+        strategy_used=CompressionStrategy.KOMPRESS,
+    )
+
+    monkeypatch.setattr(
+        content_router_module,
+        "is_mixed_content",
+        lambda content: (_ for _ in ()).throw(AssertionError("mixed detection called")),
+    )
+    monkeypatch.setattr(
+        content_router_module,
+        "_detect_content",
+        lambda content: (_ for _ in ()).throw(AssertionError("content detection called")),
+    )
+    monkeypatch.setattr(router, "_determine_strategy", lambda content: CompressionStrategy.MIXED)
+    monkeypatch.setattr(router, "_compress_pure", lambda *args, **kwargs: pure_result)
+
+    assert router.compress("large tool output") is pure_result
+
+
+def test_normal_compress_path_still_uses_content_detection(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    router = ContentRouter()
+    calls = {"mixed": 0, "detect": 0}
+    pure_result = RouterCompressionResult(
+        compressed="pure",
+        original="pure",
+        strategy_used=CompressionStrategy.TEXT,
+    )
+
+    def _fake_mixed(content: str) -> bool:
+        calls["mixed"] += 1
+        return False
+
+    def _fake_detect(content: str) -> DetectionResult:
+        calls["detect"] += 1
+        return DetectionResult(ContentType.PLAIN_TEXT, 1.0, {})
+
+    monkeypatch.setattr(content_router_module, "is_mixed_content", _fake_mixed)
+    monkeypatch.setattr(content_router_module, "_detect_content", _fake_detect)
+    monkeypatch.setattr(router, "_compress_pure", lambda *args, **kwargs: pure_result)
+
+    assert router.compress("plain text") is pure_result
+    assert calls["mixed"] > 0
+    assert calls["detect"] > 0
+
+
+def test_force_kompress_apply_uses_lightweight_detection(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class FakeTokenizer:
+        def count_text(self, text: str) -> int:
+            return len(text.split())
+
+    router = ContentRouter(ContentRouterConfig(protect_recent_code=2))
+    content = " ".join(["plain text payload"] * 80)
+
+    monkeypatch.setattr(
+        content_router_module,
+        "_detect_content",
+        lambda content: (_ for _ in ()).throw(AssertionError("content detection called")),
+    )
+    monkeypatch.setattr(
+        content_router_module,
+        "_regex_detect_content_type",
+        lambda content: DetectionResult(ContentType.PLAIN_TEXT, 1.0, {}),
+    )
+    monkeypatch.setattr(
+        router,
+        "compress",
+        lambda content, context="", bias=1.0: RouterCompressionResult(
+            compressed="compressed",
+            original=content,
+            strategy_used=CompressionStrategy.KOMPRESS,
+            routing_log=[
+                RoutingDecision(
+                    content_type=ContentType.PLAIN_TEXT,
+                    strategy=CompressionStrategy.KOMPRESS,
+                    original_tokens=len(content.split()),
+                    compressed_tokens=1,
+                )
+            ],
+        ),
+    )
+
+    result = router.apply(
+        [{"role": "tool", "content": content}],
+        FakeTokenizer(),
+        force_kompress=True,
+        min_tokens_to_compress=10,
+        protect_recent=2,
+    )
+
+    assert result.messages[0]["content"] == "compressed"
+
+
+def test_force_kompress_apply_lightweight_detection_protects_recent_code(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    class FakeTokenizer:
+        def count_text(self, text: str) -> int:
+            return len(text.split())
+
+    router = ContentRouter(ContentRouterConfig(protect_recent_code=2))
+    content = "\n".join(
+        [
+            "def generated_function(value):",
+            "    if value:",
+            "        return str(value)",
+        ]
+        * 40
+    )
+
+    monkeypatch.setattr(
+        content_router_module,
+        "_detect_content",
+        lambda content: (_ for _ in ()).throw(AssertionError("content detection called")),
+    )
+    monkeypatch.setattr(
+        router,
+        "compress",
+        lambda *args, **kwargs: (_ for _ in ()).throw(
+            AssertionError("recent code should be protected")
+        ),
+    )
+
+    result = router.apply(
+        [{"role": "tool", "content": content}],
+        FakeTokenizer(),
+        force_kompress=True,
+        min_tokens_to_compress=10,
+        protect_recent=2,
+    )
+
+    assert result.messages[0]["content"] == content
+    assert result.transforms_applied == ["router:protected:recent_code"]
+
+
 def test_content_router_mixed_pure_apply_and_toin(monkeypatch: pytest.MonkeyPatch) -> None:
     router = ContentRouter()
     mixed_content = "\n".join(["before", "```python", "print('x')", "```", "after"])