refactor(transforms): dispatch simple built-in strategies via the compressor registry (#2399)

## What
Second increment of the adapter phase (builds on #2391). Flips the
content router's per-strategy dispatch in `_apply_strategy_to_content`
from the hardcoded if/elif to **registry-resolved** — but only for the
*clean, single-compressor* strategies: **SEARCH, LOG, TABULAR, CONFIG**.
Each resolves its compressor by name from `compressor_registry` and runs
it over the pure-data `CompressInput`/`CompressOutput` contract via a
shared `_registry_compress_content` helper, then maps back to the
branch's exact historical return shape.

## Byte-identical by construction
- The built-in adapter delegates to the SAME `_get_<name>()` getter +
method with the same args (`context`→query, `bias`→budget), so returned
content is identical to the old direct call.
- Each flipped branch **keeps its `enable_*` gate and `_get_*`
availability guard** — so the built-in-unavailable → passthrough
behavior is preserved and the adapter's `None`→content collapse is never
reached.
- Each branch **recomputes its token count with its own historical
metric** (`_estimate_tokens` for search/log/tabular; `len(split())` for
config).
- `content_type` in `CompressInput` is inert (built-ins don't consume
it), so it can't shift output.

## Deferred (left byte-for-byte as-is) — and why
- **CODE_AWARE** — has a Kompress/ML fallback chain (`compressed is
None` → `_try_ml_compressor`, plus a `lossless_then_lossy` no-shrink
retry) that mutates `strategy`/`strategy_chain`. Not a clean single
call.
- **HTML** — uses `.extract().extracted` (different shape) and relies on
`None` extraction falling through to bottom passthrough (`[html,
passthrough]`); the adapter's `None`→content collapse would change the
chain. Not byte-identical through the entry.
- **SMART_CRUSHER** (fallback chain), **KOMPRESS/TEXT** (ML boundary),
**PASSTHROUGH**, **DIFF** — untouched per plan.

The reversibility gate, external-compressor dispatch (#2388), and
default (nothing-selected) behavior are unchanged. No new config/env.

## Testing
New `tests/test_router_registry_dispatch.py` (6 tests): differential
test per flipped strategy asserting registry-dispatch output == old
direct-dispatch output (content + branch token metric + `[strategy]`
chain), plus assertions that deferred SMART_CRUSHER and KOMPRESS are
unchanged. Offline suite: 78 passed; ruff + mypy clean. The broad
content-router suite (HF-Hub/ONNX) is deferred to CI — **that full suite
is the authoritative byte-identical gate for the flipped strategies.**

headroom/transforms/content_router.py

@@ -2671,6 +2671,48 @@ class ContentRouter(Transform):
             except Exception as exc:  # noqa: BLE001 - defensive; never break the request
                 logger.debug("external compressor %r: store.store raised (%s)", name, exc)
 
+    def _registry_compress_content(
+        self,
+        name: str,
+        strategy: CompressionStrategy,
+        content: str,
+        context: str,
+        bias: float,
+    ) -> str:
+        """Compress ``content`` with a built-in via the compressor registry.
+
+        Resolves the built-in named ``name`` from :attr:`compressor_registry` and
+        runs it over the pure-data :class:`CompressInput` contract, returning the
+        compressed string. This is the registry-resolved equivalent of the
+        router's historical ``self._get_<name>().compress(...)`` dispatch: the
+        built-in adapter delegates to the SAME ``_get_*`` getter and method with
+        the SAME arguments (``context`` as the query, ``bias`` via the budget), so
+        the returned content is byte-identical to the direct call.
+
+        Callers keep their own ``if self.config.enable_<x>:`` gate and ``_get_*``
+        availability guard (which preserves the built-in-unavailable → passthrough
+        behavior the adapter's None→content collapse would otherwise hide) and
+        recompute the token count with the branch's own metric, so the branch's
+        return shape is unchanged.
+        """
+        entry = self.compressor_registry.get(name)
+        if entry is None:
+            # Built-in inventory is always registered by _build_compressor_registry;
+            # defensive only — fall back to the unchanged content.
+            return content
+        output = entry.compress(
+            CompressInput(
+                content=content,
+                content_type=_CONTENT_TYPE_TO_MIME.get(
+                    self._content_type_from_strategy(strategy), "text/plain"
+                ),
+                query=context,
+                config={},
+                budget={"bias": bias},
+            )
+        )
+        return output.content
+
     def _apply_strategy_to_content(
         self,
         content: str,
@@ -2876,11 +2918,13 @@ class ContentRouter(Transform):
                     compressor = self._get_search_compressor()
                     if compressor:
                         compressor_name = type(compressor).__name__
-                        result = compressor.compress(content, context=context, bias=bias)
-                        compressed, compressed_tokens = (
-                            result.compressed,
-                            _estimate_tokens(result.compressed),
+                        # Registry-resolved dispatch: the built-in "search" adapter
+                        # delegates to this same getter+method, so the content is
+                        # byte-identical to the historical direct call.
+                        compressed = self._registry_compress_content(
+                            "search", strategy, content, context, bias
                         )
+                        compressed_tokens = _estimate_tokens(compressed)
                         decision_reason = "search_compressor"
 
             elif strategy == CompressionStrategy.LOG:
@@ -2888,15 +2932,17 @@ class ContentRouter(Transform):
                     compressor = self._get_log_compressor()
                     if compressor:
                         compressor_name = type(compressor).__name__
-                        result = compressor.compress(content, bias=bias)
+                        # Registry-resolved dispatch: the built-in "log" adapter
+                        # delegates to this same getter+method, so the content is
+                        # byte-identical to the historical direct call.
+                        compressed = self._registry_compress_content(
+                            "log", strategy, content, context, bias
+                        )
                         # Use the same word-count metric the rest of the
                         # router uses; `compressed_line_count` is in
                         # lines, not tokens — recording it here made
                         # ratios meaningless against `original_tokens`.
-                        compressed, compressed_tokens = (
-                            result.compressed,
-                            _estimate_tokens(result.compressed),
-                        )
+                        compressed_tokens = _estimate_tokens(compressed)
                         decision_reason = "log_compressor"
 
             elif strategy == CompressionStrategy.TABULAR:
@@ -2904,11 +2950,13 @@ class ContentRouter(Transform):
                     compressor = self._get_tabular_compressor()
                     if compressor:
                         compressor_name = type(compressor).__name__
-                        result = compressor.compress(content, context=context, bias=bias)
-                        compressed, compressed_tokens = (
-                            result.compressed,
-                            _estimate_tokens(result.compressed),
+                        # Registry-resolved dispatch: the built-in "tabular" adapter
+                        # delegates to this same getter+method, so the content is
+                        # byte-identical to the historical direct call.
+                        compressed = self._registry_compress_content(
+                            "tabular", strategy, content, context, bias
                         )
+                        compressed_tokens = _estimate_tokens(compressed)
                         decision_reason = "tabular_compressor"
 
             elif strategy == CompressionStrategy.CONFIG:
@@ -2916,11 +2964,14 @@ class ContentRouter(Transform):
                     compressor = self._get_config_compressor()
                     if compressor:
                         compressor_name = type(compressor).__name__
-                        result = compressor.compress(content, context=context, bias=bias)
-                        compressed, compressed_tokens = (
-                            result.compressed,
-                            len(result.compressed.split()),
+                        # Registry-resolved dispatch: the built-in "config" adapter
+                        # delegates to this same getter+method, so the content is
+                        # byte-identical to the historical direct call. Keep the
+                        # branch's own whitespace-split token metric.
+                        compressed = self._registry_compress_content(
+                            "config", strategy, content, context, bias
                         )
+                        compressed_tokens = len(compressed.split())
                         decision_reason = "config_compressor"
 
             elif strategy == CompressionStrategy.DIFF:

tests/test_router_registry_dispatch.py

@@ -0,0 +1,208 @@
+"""Byte-identical differential tests for registry-resolved built-in dispatch.
+
+The content router now dispatches the SIMPLE built-in strategies (CONFIG, LOG,
+SEARCH, TABULAR) through the compressor registry instead of a hardcoded direct
+``self._get_*().compress(...)`` call in ``_apply_strategy_to_content``. Each
+built-in adapter delegates to the SAME ``_get_*`` getter+method with the SAME
+arguments, so registry-resolved dispatch must be byte-identical to the historical
+direct dispatch: same compressed content, same token count, same single-entry
+``strategy_chain``.
+
+Each FLIPPED strategy has a differential test comparing the router's dispatch
+output to the built-in's direct output obtained via its ``_get_*`` getter — i.e.
+"registry dispatch == old dispatch". DEFERRED strategies (SMART_CRUSHER, KOMPRESS)
+are asserted unchanged: they still route through their bespoke paths (fallback
+chain / the ``_try_ml_compressor`` ML boundary), not the registry.
+
+Offline guardrails:
+  * No real ML/ONNX/HF inference — the deferred KOMPRESS path is mocked.
+  * The flipped strategies shrink their representative content, so no zero-savings
+    Kompress fallback fires (that would touch the ML boundary and append KOMPRESS
+    to the chain).
+  * ``lossless_then_lossy`` and ``relevance_split`` are off so the if/elif branch
+    is the terminal path; STAGE 0 (``_lossless_first``) is neutralized so search/
+    log folds don't return before the branch. Both are shared, unchanged code.
+  * The broad ``content_router``/``compression`` -k selection is NOT exercised
+    (it hangs on HF-Hub/ONNX).
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from headroom.transforms.content_router import (
+    CompressionStrategy,
+    ContentRouter,
+    ContentRouterConfig,
+    _estimate_tokens,
+)
+
+
+def _router() -> ContentRouter:
+    """Router whose if/elif branch is the terminal dispatch path.
+
+    ``relevance_split`` off (no LOG/SEARCH relevance split) and
+    ``lossless_then_lossy`` off (no lossy layer on top of a strategy result) so a
+    successful strategy result returns directly. ``ccr_inject_marker`` off makes
+    the compressed output deterministic and marker-free; it is applied identically
+    to the direct reference and the dispatch router, so the differential holds
+    regardless of its value.
+    """
+    return ContentRouter(
+        ContentRouterConfig(
+            relevance_split=False,
+            lossless_then_lossy=False,
+            ccr_inject_marker=False,
+        )
+    )
+
+
+def _isolate_branch(monkeypatch: pytest.MonkeyPatch, router: ContentRouter) -> None:
+    """Neutralize STAGE 0 so the if/elif branch under test is exercised.
+
+    ``_lossless_first`` runs unconditionally and can fold search/log content,
+    returning before the if/elif. It is shared, unchanged code (the flip only
+    touches the branch bodies), so forcing it to a no-op isolates what the flip
+    actually changed without altering the branch semantics.
+    """
+    monkeypatch.setattr(router, "_lossless_first", lambda content, strategy: (content, None))
+
+
+# Representative content per type. The flipped strategies must SHRINK this so the
+# fallback-eligible strategies (TABULAR/CONFIG) don't trip the zero-savings
+# Kompress fallback (which would append KOMPRESS to the chain).
+_SEARCH = "\n".join(f"src/file{i}.py:{i}: def func{i}(): return {i}" for i in range(30))
+_LOG = (
+    "\n".join(f"2024-01-01 12:00:{i:02d} INFO task {i}" for i in range(30))
+    + "\n"
+    + "\n".join("identical repeated line" for _ in range(25))
+)
+# A markdown table the tabular compressor actually shrinks (schema-fold), and a
+# repetitive INI the config compressor actually shrinks (block-fold) — so these
+# fallback-eligible strategies produce a real token saving and the shared
+# zero-savings Kompress fallback does NOT fire (chain stays single-entry).
+_TABULAR = "| id | name | status | score |\n|----|------|--------|-------|\n" + "\n".join(
+    f"| {i} | row{i} | ok | {i * 3} |" for i in range(60)
+)
+_CONFIG = "\n".join(
+    f"[section_{i}]\nname = svc{i}\ntimeout = 30\nretries = 3\nverbose = false\nregion = us-east-1"
+    for i in range(40)
+)
+
+
+# ───────────────────────── flipped (registry dispatch) ────────────────────────
+
+
+def test_search_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    context, bias = "func", 1.0
+    # OLD dispatch reference: same getter + method the branch used before the flip.
+    direct = (
+        _router()._get_search_compressor().compress(_SEARCH, context=context, bias=bias).compressed
+    )
+    out, tokens, chain = router._apply_strategy_to_content(
+        _SEARCH, CompressionStrategy.SEARCH, context, bias=bias
+    )
+    assert out == direct
+    assert tokens == _estimate_tokens(direct)
+    assert chain == [CompressionStrategy.SEARCH.value]
+
+
+def test_log_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    bias = 1.0
+    direct = _router()._get_log_compressor().compress(_LOG, bias=bias).compressed
+    out, tokens, chain = router._apply_strategy_to_content(
+        _LOG, CompressionStrategy.LOG, "", bias=bias
+    )
+    assert out == direct
+    assert tokens == _estimate_tokens(direct)
+    assert chain == [CompressionStrategy.LOG.value]
+
+
+def test_tabular_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    context, bias = "q", 1.0
+    direct = (
+        _router()
+        ._get_tabular_compressor()
+        .compress(_TABULAR, context=context, bias=bias)
+        .compressed
+    )
+    out, tokens, chain = router._apply_strategy_to_content(
+        _TABULAR, CompressionStrategy.TABULAR, context, bias=bias
+    )
+    assert out == direct
+    assert tokens == _estimate_tokens(direct)
+    # Fallback-eligible, but a real shrink means no zero-savings Kompress fallback.
+    assert chain == [CompressionStrategy.TABULAR.value]
+    assert len(out) < len(_TABULAR)
+
+
+def test_config_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    context, bias = "q", 1.0
+    direct = (
+        _router()._get_config_compressor().compress(_CONFIG, context=context, bias=bias).compressed
+    )
+    out, tokens, chain = router._apply_strategy_to_content(
+        _CONFIG, CompressionStrategy.CONFIG, context, bias=bias
+    )
+    assert out == direct
+    # CONFIG's historical metric is len(text.split()), NOT _estimate_tokens; the
+    # flip must preserve that exact metric.
+    assert tokens == len(direct.split())
+    assert chain == [CompressionStrategy.CONFIG.value]
+    assert len(out) < len(_CONFIG)
+
+
+# ─────────────────────────── deferred (unchanged) ────────────────────────────
+
+
+def test_smart_crusher_deferred_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    # SMART_CRUSHER is DEFERRED (its branch feeds a Kompress→Log fallback chain in
+    # the shared post-strategy block), so it still dispatches via the direct
+    # crusher, not the registry. A JSON array shrinks, so the chain stays single.
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    content = json.dumps(
+        [{"id": i, "status": "ok", "level": "INFO", "value": i * 2} for i in range(40)]
+    )
+    direct = _router()._get_smart_crusher().crush(content, query="q", bias=1.0).compressed
+    out, _tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.SMART_CRUSHER, "q", bias=1.0
+    )
+    assert out == direct
+    assert chain == [CompressionStrategy.SMART_CRUSHER.value]
+
+
+def test_kompress_deferred_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    # KOMPRESS is DEFERRED (it is the ML boundary — dispatched through
+    # _try_ml_compressor, not a built-in adapter). Mock the underlying model so no
+    # real ONNX/HF inference runs, and assert the router still routes through
+    # _try_ml_compressor rather than the registry.
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    fake = SimpleNamespace(
+        is_ready=lambda: True,
+        ensure_background_load=lambda: None,
+        compress=lambda text, **kwargs: SimpleNamespace(
+            compressed="KOMPRESSED::" + text, compressed_tokens=7
+        ),
+    )
+    monkeypatch.setattr(router, "_get_kompress", lambda: fake)
+    content = "some plain text that the ML model would compress. " * 4
+    out, _tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.KOMPRESS, "", bias=1.0
+    )
+    assert out == "KOMPRESSED::" + content
+    assert chain == [CompressionStrategy.KOMPRESS.value]
+    # Still the bespoke ML path (unchanged), not a registry round-trip.
+    assert out == router._try_ml_compressor(content, "", None)[0]