feat(transforms): dispatch smart_crusher via the compressor registry (defer kompress/text ML boundary) (#2404)

## What
Final increment of the adapter phase (builds on #2391/#2399/#2400).
Flips the **SMART_CRUSHER** primary `.crush()` invocation in
`_apply_strategy_to_content` to registry-resolved dispatch, following
the CODE_AWARE/HTML pattern. The shared SmartCrusher→Kompress→Log
fallback block is unchanged.

## Byte-identical (SMART_CRUSHER)
The `smart_crusher` adapter delegates to the same
`_get_smart_crusher().crush(content, query=context, bias=bias)` (same
cached getter, same method), so `output.content == result.compressed`;
the branch recomputes the same `_estimate_tokens` metric; the `if
crusher:` guard and the entire fallback chain / `strategy_chain` /
`decision_reason` mutations are preserved verbatim.

## Deferred — KOMPRESS and TEXT (honest contract limitation)
The `kompress` adapter can't reproduce the direct
`_try_ml_compressor(content, context, question)` byte-for-byte, for two
independent reasons:
1. **`question` is dropped** — the adapter hardcodes `None`, so QA-aware
compression content would diverge.
2. **Token count differs** — the historical branch returns Kompress's
own `compressed_tokens` (a word count taken *before* the CCR marker is
appended), while the registry path recomputes `_estimate_tokens` over
the marker-augmented output. Structurally different numbers whenever
Kompress actually compresses.

Flipping them would require evolving the adapter/`CompressOutput`
contract (forward `question`; carry the compressor's own token count),
which is a separate change and would touch the ML boundary — so they're
left byte-for-byte here.

## Testing
New `tests/test_router_registry_smartcrusher.py`: SMART_CRUSHER success
(differential vs a real crush), query/bias forwarding, Kompress-fallback
(`[smart_crusher, kompress]`) and Log-fallback (`[smart_crusher,
kompress, log]`) chains; plus KOMPRESS/TEXT tests that *pin the deferral
facts* (token mismatch + `question` forwarding). Offline suite: 94
passed; ruff (0.15.17) + mypy clean. Full content-router CI suite is the
authoritative byte-identical gate.

No new config/env. Reversibility gate, external dispatch (#2388),
default behavior unchanged.

headroom/transforms/content_router.py

@@ -2956,12 +2956,26 @@ class ContentRouter(Transform):
                     crusher = self._get_smart_crusher()
                     if crusher:
                         compressor_name = type(crusher).__name__
-                        result = crusher.crush(content, query=context, bias=bias)
+                        # Registry-resolved dispatch: the built-in "smart_crusher"
-                        compressed, compressed_tokens = (
+                        # adapter delegates to this same getter + ``.crush(...)``
-                            result.compressed,
+                        # with the SAME query (``context``) and bias, so on a real
-                            _estimate_tokens(result.compressed),
+                        # compression the content is byte-identical to the historical
+                        # direct call and the token metric (``_estimate_tokens`` of
+                        # the crush output) is unchanged. SmartCrusher's ``.crush``
+                        # always returns a string (``compressed=True``), so — inside
+                        # this ``if crusher`` guard where the adapter's own getter
+                        # returns the same cached crusher — ``compressed`` is always
+                        # set exactly as the direct call did. The ``output`` /
+                        # ``output.compressed`` guard mirrors the CODE_AWARE/HTML
+                        # flip and only leaves ``compressed`` None in the defensive
+                        # not-registered case (never reachable for a built-in).
+                        output = self._registry_compress(
+                            "smart_crusher", strategy, content, context, bias
                         )
-                        decision_reason = "smart_crusher"
+                        if output is not None and output.compressed:
+                            compressed = output.content
+                            compressed_tokens = _estimate_tokens(output.content)
+                            decision_reason = "smart_crusher"
                         # Fallback to Kompress (and possibly Log) is
                         # handled by the unified post-strategy block below
                         # — no inline fallback here to avoid duplicate

tests/test_router_registry_dispatch.py

@@ -10,9 +10,12 @@ direct dispatch: same compressed content, same token count, same single-entry
 
 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)
+"registry dispatch == old dispatch". SMART_CRUSHER is now also flipped (its
-are asserted unchanged: they still route through their bespoke paths (fallback
+``.crush`` primary invocation goes through the registry while the shared
-chain / the ``_try_ml_compressor`` ML boundary), not the registry.
+Kompress→Log fallback block stays direct); the KOMPRESS/TEXT ML boundary
+(``_try_ml_compressor``) remains DEFERRED and is asserted unchanged. See
+``test_router_registry_smartcrusher.py`` for the full SMART_CRUSHER fallback-chain
+and KOMPRESS/TEXT differential coverage.
 
 Offline guardrails:
   * No real ML/ONNX/HF inference — the deferred KOMPRESS path is mocked.
@@ -167,10 +170,13 @@ def test_config_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch)
 # ─────────────────────────── deferred (unchanged) ────────────────────────────
 
 
-def test_smart_crusher_deferred_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+def test_smart_crusher_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
-    # SMART_CRUSHER is DEFERRED (its branch feeds a Kompress→Log fallback chain in
+    # SMART_CRUSHER is now FLIPPED: its primary ``.crush`` invocation routes
-    # the shared post-strategy block), so it still dispatches via the direct
+    # through the registry "smart_crusher" adapter (which delegates to the SAME
-    # crusher, not the registry. A JSON array shrinks, so the chain stays single.
+    # getter + ``.crush(query=..., bias=...)``), while the shared Kompress→Log
+    # fallback block stays direct. Registry dispatch must be byte-identical to the
+    # historical direct crush. A JSON array shrinks, so no fallback fires and the
+    # chain stays single.
     router = _router()
     _isolate_branch(monkeypatch, router)
     content = json.dumps(

tests/test_router_registry_smartcrusher.py

@@ -0,0 +1,252 @@
+"""Byte-identical differential tests for the SMART_CRUSHER / KOMPRESS / TEXT flip.
+
+PR-C2 flips the PRIMARY compressor invocation of SMART_CRUSHER to the compressor
+registry (``_registry_compress("smart_crusher", ...)``), while the shared
+post-strategy Kompress -> Log fallback block stays a direct dispatch. KOMPRESS and
+TEXT are DEFERRED: they are the ``_try_ml_compressor`` ML boundary, and the
+``kompress`` built-in adapter (a) hardcodes ``question=None`` (dropping the real
+QA-aware ``question`` argument) and (b) recomputes the token count via
+``_estimate_tokens`` instead of returning ``_try_ml_compressor``'s tuple token
+count (Kompress's own word-count ``compressed_tokens``, computed pre-CCR-marker),
+so a registry round-trip cannot reproduce either the content (when a question is
+supplied) or the token metric byte-for-byte. These tests pin both facts.
+
+Every path asserts registry-dispatch output == the historical direct-dispatch
+output (content, token metric, ``strategy_chain``, and — via the recorded call
+args — the query/bias/question that flow through).
+
+Offline guardrails:
+  * No real ML/ONNX/HF inference — the ML boundary is mocked at
+    ``_try_ml_compressor`` (SMART_CRUSHER fallback tests) or ``_get_kompress``
+    (KOMPRESS/TEXT tests, which exercise the real ``_try_ml_compressor`` with a
+    fake in-memory model).
+  * SmartCrusher itself is pure-Python (no ML), so the SMART_CRUSHER success path
+    runs the real crusher on a small JSON array.
+  * ``relevance_split`` / ``lossless_then_lossy`` off and STAGE 0
+    (``_lossless_first``) neutralized so the if/elif branch under test is the
+    terminal path. All of that is 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, ``lossless_then_lossy`` off, and
+    ``ccr_inject_marker`` off make the compressed output deterministic and
+    marker-free; the same config is applied to the direct reference and the
+    dispatch router, so the differential holds regardless.
+    """
+    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 (``_lossless_first``) so the if/elif branch is exercised.
+
+    ``_lossless_first`` is shared, unchanged code (the flip only touches the branch
+    body), so forcing it to a no-op isolates what the flip actually changed.
+    """
+    monkeypatch.setattr(router, "_lossless_first", lambda content, strategy: (content, None))
+
+
+# A JSON array the real SmartCrusher shrinks (so the fallback chain stays single).
+_JSON = json.dumps([{"id": i, "status": "ok", "level": "INFO", "value": i * 2} for i in range(40)])
+
+
+# ─────────────────────── SMART_CRUSHER: flipped (registry) ────────────────────
+
+
+def test_smart_crusher_success_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    # SMART_CRUSHER-SUCCEEDS: the flip routes the primary ``.crush`` through the
+    # registry "smart_crusher" adapter, which delegates to the SAME getter +
+    # ``.crush(query=..., bias=...)``. Differential: registry dispatch == a direct
+    # crush on an independent router. A real shrink means no fallback fires.
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    context, bias = "q", 1.0
+    direct = _router()._get_smart_crusher().crush(_JSON, query=context, bias=bias).compressed
+    out, tokens, chain = router._apply_strategy_to_content(
+        _JSON, CompressionStrategy.SMART_CRUSHER, context, bias=bias
+    )
+    assert out == direct
+    assert tokens == _estimate_tokens(direct)
+    assert chain == [CompressionStrategy.SMART_CRUSHER.value]
+    assert len(out) < len(_JSON)
+
+
+def test_smart_crusher_forwards_query_and_bias(monkeypatch: pytest.MonkeyPatch) -> None:
+    # The flip must forward content, query (==context), and bias to ``.crush``
+    # unchanged. A fake crusher records the exact call args, proving the registry
+    # adapter's ``query=inp.query`` / ``bias=budget['bias']`` reproduce the direct
+    # ``crush(content, query=context, bias=bias)`` call.
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    seen: dict[str, object] = {}
+
+    def _crush(content: str, query: str = "", bias: float = 1.0) -> SimpleNamespace:
+        seen.update(content=content, query=query, bias=bias)
+        return SimpleNamespace(compressed="CRUSHED " + content[:5])
+
+    monkeypatch.setattr(router, "_get_smart_crusher", lambda: SimpleNamespace(crush=_crush))
+    # Sentinel ML: if the flip WRONGLY dropped into the Kompress fallback we'd see
+    # KOMPRESS appended to the chain; the chain assertion catches it.
+    monkeypatch.setattr(
+        router, "_try_ml_compressor", lambda *a, **k: ("KOMPRESS_SENTINEL", 999_999)
+    )
+
+    content = "x" * 200
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.SMART_CRUSHER, "myquery", bias=0.7
+    )
+    assert out == "CRUSHED " + content[:5]
+    assert tokens == _estimate_tokens("CRUSHED " + content[:5])
+    assert chain == [CompressionStrategy.SMART_CRUSHER.value]
+    assert seen == {"content": content, "query": "myquery", "bias": 0.7}
+
+
+def test_smart_crusher_kompress_fallback_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    # SMART_CRUSHER-NO-SHRINK -> KOMPRESS: the crusher returns the content
+    # unchanged (no net saving), so the SHARED, UNCHANGED post-strategy block runs
+    # the Kompress fallback. The flip only sets ``compressed``/``compressed_tokens``
+    # on entry — identically to the direct crush — so the fallback fires exactly as
+    # before: chain [smart_crusher, kompress], Kompress output adopted.
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    monkeypatch.setattr(
+        router,
+        "_get_smart_crusher",
+        lambda: SimpleNamespace(crush=lambda c, query="", bias=1.0: SimpleNamespace(compressed=c)),
+    )
+    # Kompress shrinks (fallback_tokens < compressed_tokens) so it is adopted.
+    monkeypatch.setattr(router, "_try_ml_compressor", lambda c, ctx, q: ("KOMPRESSED::" + c, 3))
+    out, tokens, chain = router._apply_strategy_to_content(
+        _JSON, CompressionStrategy.SMART_CRUSHER, "ctx", bias=1.0
+    )
+    assert out == "KOMPRESSED::" + _JSON
+    assert tokens == 3
+    assert chain == [
+        CompressionStrategy.SMART_CRUSHER.value,
+        CompressionStrategy.KOMPRESS.value,
+    ]
+
+
+def test_smart_crusher_log_fallback_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    # SMART_CRUSHER-NO-SHRINK -> KOMPRESS-NO-SHRINK -> LOG: crusher passes content
+    # through, Kompress fails to shrink (fallback_tokens NOT < compressed_tokens),
+    # and — because the content is valid JSON and the log compressor is enabled —
+    # the last-ditch Log fallback runs and shrinks. All of that is the shared,
+    # unchanged block; the flip only feeds it identical entry state. Chain
+    # [smart_crusher, kompress, log].
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    monkeypatch.setattr(
+        router,
+        "_get_smart_crusher",
+        lambda: SimpleNamespace(crush=lambda c, query="", bias=1.0: SimpleNamespace(compressed=c)),
+    )
+    # Kompress returns content unchanged with a huge token count -> NOT a shrink,
+    # so the else branch (Log fallback) is taken.
+    monkeypatch.setattr(router, "_try_ml_compressor", lambda c, ctx, q: (c, 10**9))
+    monkeypatch.setattr(
+        router,
+        "_get_log_compressor",
+        lambda: SimpleNamespace(
+            compress=lambda c, bias=1.0: SimpleNamespace(compressed="LOG_FOLDED")
+        ),
+    )
+    out, tokens, chain = router._apply_strategy_to_content(
+        _JSON, CompressionStrategy.SMART_CRUSHER, "ctx", bias=1.0
+    )
+    assert out == "LOG_FOLDED"
+    assert tokens == _estimate_tokens("LOG_FOLDED")
+    assert chain == [
+        CompressionStrategy.SMART_CRUSHER.value,
+        CompressionStrategy.KOMPRESS.value,
+        CompressionStrategy.LOG.value,
+    ]
+
+
+# ───────────────────── KOMPRESS / TEXT: deferred (ML boundary) ─────────────────
+
+
+def _fake_kompress(seen: dict[str, object]) -> SimpleNamespace:
+    """In-memory Kompress model: records compress kwargs, no ONNX/HF."""
+
+    def _compress(text: str, **kwargs: object) -> SimpleNamespace:
+        seen.update(kwargs)
+        # ``compressed_tokens`` is the model's OWN word-count (7), deliberately
+        # unequal to ``_estimate_tokens`` of the output — this is exactly the
+        # value the registry adapter would discard, which is why the branch is
+        # deferred.
+        return SimpleNamespace(compressed="KOMPRESSED::" + text, compressed_tokens=7)
+
+    return SimpleNamespace(
+        is_ready=lambda: True,
+        ensure_background_load=lambda: None,
+        compress=_compress,
+    )
+
+
+def test_kompress_deferred_ml_path_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    seen: dict[str, object] = {}
+    monkeypatch.setattr(router, "_get_kompress", lambda: _fake_kompress(seen))
+
+    content = "some plain text that the ML model would compress. " * 4
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.KOMPRESS, "ctx", question="my question", bias=1.0
+    )
+    assert out == "KOMPRESSED::" + content
+    assert chain == [CompressionStrategy.KOMPRESS.value]
+    # Token count is the model's tuple value (7), NOT _estimate_tokens(out) — the
+    # exact divergence that makes the registry round-trip non-byte-identical.
+    assert tokens == 7
+    assert tokens != _estimate_tokens(out)
+    # The real ``question`` is forwarded (the kompress adapter would pass None).
+    assert seen["question"] == "my question"
+    # Still the bespoke ML path, byte-identical to a direct _try_ml_compressor call.
+    direct, direct_tokens = router._try_ml_compressor(content, "ctx", "my question")
+    assert out == direct
+    assert tokens == direct_tokens
+
+
+def test_text_deferred_ml_path_unchanged(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    _isolate_branch(monkeypatch, router)
+    seen: dict[str, object] = {}
+    monkeypatch.setattr(router, "_get_kompress", lambda: _fake_kompress(seen))
+
+    content = "plain prose the text strategy sends straight to kompress. " * 4
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.TEXT, "ctx", question="q2", bias=1.0
+    )
+    assert out == "KOMPRESSED::" + content
+    assert chain == [CompressionStrategy.TEXT.value]
+    assert tokens == 7
+    assert tokens != _estimate_tokens(out)
+    assert seen["question"] == "q2"
+    direct, direct_tokens = router._try_ml_compressor(content, "ctx", "q2")
+    assert out == direct
+    assert tokens == direct_tokens