perf(content_router): dedupe content detection (#2419)

## Description

ContentRouter ran the native content detector two to three times on
identical content, on the hottest path in the proxy (every compressed
message, every request). This cuts it to once.

`_detect_content` isn't cheap and isn't memoized. It strips a detection
envelope, runs the Rust/Magika ONNX classifier, then several regex
passes. `compress()` ran it once for debug logging that's off by
default, then `_determine_strategy()` recomputed it (plus
`is_mixed_content`) on the same content. That's twice per `compress()`,
and three times on the `apply()` cache-miss path.

Closes: N/A (no filed issue, surfaced by an internal
contribution-backlog audit).

## Type of Change

- [ ] 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
- [x] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

- `compress()` computes `is_mixed_content` and `_detect_content` once,
then threads both into `_determine_strategy` through new optional params
(`mixed`, `detection`).
- `_determine_strategy` uses the passed values when present, and
computes them itself when they're `None`. Its one private caller
changes. Any other caller keeps working.
- Added `tests/test_content_router_detection_dedup.py`. One test asserts
`compress()` detects exactly once (it fails before the fix at `assert 2
== 1`). The other asserts the threaded result routes the same as the
recomputed one across content types.
- Updated two existing `_determine_strategy` test doubles to take the
new kwargs.

## 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
$ pytest tests/test_content_router_detection_dedup.py tests/test_transforms_content_router.py \
         tests/test_transforms/test_content_router.py tests/test_transforms_content_detection.py -q
135 passed in 8.98s

$ pytest tests/test_transforms/ tests/test_content_router_*.py tests/test_router_*.py \
         tests/test_lossless_excluded_compaction.py -q
423 passed, 62 skipped in 54.93s

$ ruff check .
All checks passed!

$ mypy headroom
Success: no issues found in 505 source files
```

## Real Behavior Proof

- Environment: macOS (Darwin), Python 3.13, headroom worktree on this
branch off `upstream/main`, `HF_HUB_OFFLINE=1
LITELLM_LOCAL_MODEL_COST_MAP=true`. A counter wraps the real
`_detect_content` and delegates to it, so real routing and compression
run.
- Exact command / steps: run the real router over one representative
message and count `_detect_content` calls on the fixed tree, then `git
stash` the source and count again on the unfixed tree. Covered
`router.compress(blob)` and `router.apply([tool_msg])`.
- Observed result: `compress()` dropped from 2 detection calls to 1, and
`apply()` dropped from 3 to 2, on the same input with the same routing
strategy (`text`) and the same output. The once-only test flips from
`assert 2 == 1` before to passing after.
- Not tested: production Magika ONNX timing. This dev env has no
onnxruntime, so the detector ran its regex fallback tier, which makes
the saved cost a floor, not a ceiling. I also scoped out the Tier B
extension (threading the `apply()` Pass-1 detection into `compress()`)
on purpose.

## 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
- [ ] 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 did **not** edit `CHANGELOG.md` — it is generated by
release-please from my Conventional Commit PR title (a CI guard enforces
this)

## Screenshots (if applicable)

N/A

## Additional Notes

Scope is the default routing path. `force_kompress` already uses the
cheaper regex detector, so it never paid the redundant native cost.
`_compress_mixed` re-detects per split section, but that's different
content (sub-sections), so it's out of scope.

The `apply()` Pass-1 detection stays. It gates the `is_code` protection
check for every message, including cache hits that never reach
`compress()`. Threading it into `compress()` would widen a shared task
tuple and change the public `compress()` signature, all for a
cache-miss-only save, so I left it as a possible follow-up.

Doc checklist item is N/A (internal perf dedup, no user-facing docs
change). This is a Python-only change, so the first push will use
`--no-verify` for the known `ci-precheck` Rust-latency bench flake
(`classify_under_10us_per_call`), which runs clean in CI.

headroom/transforms/content_router.py

@@ -2121,7 +2121,7 @@ class ContentRouter(Transform):
             else:
                 mixed = is_mixed_content(content)
                 detection = _detect_content(content)
-                strategy = self._determine_strategy(content)
+                strategy = self._determine_strategy(content, mixed=mixed, detection=detection)
             if debug_enabled:
                 _log_router_debug(
                     "content_router_input",
@@ -2229,17 +2229,37 @@ class ContentRouter(Transform):
         except Exception as e:  # pragma: no cover - defensive
             logger.debug("CompressionObserver raised (non-fatal): %s", e)
 
-    def _determine_strategy(self, content: str) -> CompressionStrategy:
+    def _determine_strategy(
+        self,
+        content: str,
+        mixed: bool | None = None,
+        detection: DetectionResult | None = None,
+    ) -> CompressionStrategy:
         """Determine the compression strategy from content analysis.
 
         Args:
             content: Content to analyze.
+            mixed: Precomputed ``is_mixed_content(content)`` when the caller
+                already has it. ``compress`` runs it on this exact content one
+                line before calling here; recomputed only when ``None``.
+            detection: Precomputed ``_detect_content(content)`` when the caller
+                already has it. The native Rust/Magika pass is the router's
+                hottest per-message cost — reuse it instead of a second
+                identical detection; recomputed only when ``None``.
 
         Returns:
             Selected compression strategy.
         """
+        # Reuse the caller's analysis when supplied — ``compress`` already ran
+        # both on this exact content, and re-running is_mixed_content/
+        # _detect_content on identical bytes is the router's hottest wasted cost.
+        if mixed is None:
+            mixed = is_mixed_content(content)
+        if detection is None:
+            detection = _detect_content(content)
+
         # 1. Check for mixed content
-        if is_mixed_content(content):
+        if mixed:
             # 2. Verify with the native detector: ``is_mixed_content`` uses
             # cheap regex heuristics that produce false positives on source
             # code.  Python files with dict/list literals (``{``, ``[`` at
@@ -2250,13 +2270,11 @@ class ContentRouter(Transform):
             # wasting latency on splitting without any compression.
             # When the native magika detector confidently says SOURCE_CODE,
             # trust it over the regex heuristics.
-            detection = _detect_content(content)
             if detection.content_type == ContentType.SOURCE_CODE and detection.confidence >= 0.8:
                 return self._strategy_from_detection(detection)
             return CompressionStrategy.MIXED
 
-        # 2. Detect content type from content itself
-        detection = _detect_content(content)
+        # 2. Not mixed — map the detected type straight to a strategy.
         return self._strategy_from_detection(detection)
 
     def _strategy_from_detection(self, detection: Any) -> CompressionStrategy:

tests/test_content_router_detection_dedup.py

@@ -0,0 +1,96 @@
+"""Regression guards for the content-detection dedup on the router hot path.
+
+``ContentRouter.compress`` used to run the native ``_detect_content`` twice on
+identical content: once for (default-off) debug logging, then again inside
+``_determine_strategy``.  The native Rust/Magika pass is the router's hottest
+per-message cost, so ``compress`` now runs it once and threads the result into
+``_determine_strategy``.  These tests pin the call count at one and prove the
+threaded result routes identically to the recomputed one.
+"""
+
+from __future__ import annotations
+
+import pytest
+
+import headroom.transforms.content_router as content_router_module
+from headroom.transforms.content_detector import ContentType, DetectionResult
+from headroom.transforms.content_router import (
+    CompressionStrategy,
+    ContentRouter,
+    ContentRouterConfig,
+    RouterCompressionResult,
+)
+
+
+@pytest.fixture(autouse=True)
+def _reset_detect_module_state(monkeypatch: pytest.MonkeyPatch) -> None:
+    # The native-detector circuit breaker (#575) is process-wide; keep it from
+    # leaking across tests (mirrors tests/test_transforms_content_router.py).
+    monkeypatch.setattr(content_router_module, "_detect_native_unhealthy", False)
+    monkeypatch.setattr(content_router_module, "_detect_backend_warned", False)
+    monkeypatch.setattr(content_router_module, "_detect_panic_warned", False)
+
+
+def _count_detects(monkeypatch: pytest.MonkeyPatch, result: DetectionResult) -> dict[str, int]:
+    """Replace the module-level ``_detect_content`` with a deterministic counter.
+
+    Both call sites (``compress`` and ``_determine_strategy``) resolve the name
+    from the module namespace at call time, so a single patch counts every call.
+    """
+    calls = {"n": 0}
+
+    def _counting(content: str) -> DetectionResult:
+        calls["n"] += 1
+        return result
+
+    monkeypatch.setattr(content_router_module, "_detect_content", _counting)
+    return calls
+
+
+def test_compress_runs_detection_once(monkeypatch: pytest.MonkeyPatch) -> None:
+    # Before the dedup this was 2 — compress ran _detect_content for debug
+    # logging, then _determine_strategy ran it again on identical bytes. The
+    # fix threads the single result through, so it must now be exactly 1.
+    router = ContentRouter(ContentRouterConfig(prefer_code_aware_for_code=False))
+    calls = _count_detects(monkeypatch, DetectionResult(ContentType.PLAIN_TEXT, 1.0, {}))
+    monkeypatch.setattr(content_router_module, "is_mixed_content", lambda content: False)
+    # Stub the downstream compressor — this test isolates detection, not output.
+    monkeypatch.setattr(
+        router,
+        "_compress_pure",
+        lambda *a, **k: RouterCompressionResult(
+            compressed="x", original="x", strategy_used=CompressionStrategy.TEXT
+        ),
+    )
+
+    router.compress("a representative plain-text blob that is comfortably non-empty")
+
+    assert calls["n"] == 1
+
+
+@pytest.mark.parametrize(
+    ("mixed", "detected"),
+    [
+        (False, ContentType.SOURCE_CODE),
+        (False, ContentType.JSON_ARRAY),
+        (False, ContentType.BUILD_OUTPUT),
+        (False, ContentType.PLAIN_TEXT),
+        (True, ContentType.SOURCE_CODE),  # mixed regex hit, native says code -> trust native
+        (True, ContentType.PLAIN_TEXT),  # genuinely mixed
+    ],
+)
+def test_determine_strategy_threaded_matches_recomputed(
+    monkeypatch: pytest.MonkeyPatch, mixed: bool, detected: ContentType
+) -> None:
+    # Threading precomputed (mixed, detection) must pick the SAME strategy as
+    # letting _determine_strategy recompute them: the dedup changes cost, not
+    # routing.
+    router = ContentRouter(ContentRouterConfig(prefer_code_aware_for_code=False))
+    detection = DetectionResult(detected, 1.0, {})
+    monkeypatch.setattr(content_router_module, "is_mixed_content", lambda content: mixed)
+    monkeypatch.setattr(content_router_module, "_detect_content", lambda content: detection)
+
+    recomputed = router._determine_strategy("payload")
+    threaded = router._determine_strategy("payload", mixed=mixed, detection=detection)
+
+    assert threaded is recomputed

tests/test_transforms_content_router.py

@@ -323,10 +323,14 @@ def test_content_router_strategy_and_compress_paths(monkeypatch: pytest.MonkeyPa
     monkeypatch.setattr(router, "_compress_mixed", lambda *args, **kwargs: mixed_result)
     monkeypatch.setattr(router, "_compress_pure", lambda *args, **kwargs: pure_result)
 
-    monkeypatch.setattr(router, "_determine_strategy", lambda content: CompressionStrategy.MIXED)
+    monkeypatch.setattr(
+        router, "_determine_strategy", lambda content, **_kwargs: CompressionStrategy.MIXED
+    )
     assert router.compress("mixed") is mixed_result
 
-    monkeypatch.setattr(router, "_determine_strategy", lambda content: CompressionStrategy.TEXT)
+    monkeypatch.setattr(
+        router, "_determine_strategy", lambda content, **_kwargs: CompressionStrategy.TEXT
+    )
     assert router.compress("pure") is pure_result
     assert router.compress("   ").strategy_used is CompressionStrategy.PASSTHROUGH