diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py
index afc312151..29e69f136 100644
--- a/headroom/transforms/content_router.py
+++ b/headroom/transforms/content_router.py
@@ -44,6 +44,7 @@ import re
 import sys
 import threading
 import time
+from collections.abc import Callable
 from concurrent.futures import ThreadPoolExecutor
 from dataclasses import dataclass, field, replace
 from enum import Enum
@@ -203,40 +204,210 @@ _BUILTIN_COMPRESSOR_DESCRIPTORS: tuple[CompressorDescriptor, ...] = (
 )
 
 
-class _BuiltinCompressorEntry:
-    """Registry adapter exposing a built-in's metadata under the Compressor protocol.
+# ── Built-in Compressor adapters (registry delegation, additive) ──────────────
+# Each built-in registry entry delegates ``compress`` to the SAME underlying
+# built-in method the content router invokes in ``_apply_strategy_to_content`` —
+# reached through the router's own ``_get_*`` getter so config flows through
+# identically. The adapters are ADDITIVE: the router still dispatches built-ins
+# via its existing if/elif and never routes a request through the registry, so
+# they change no routing. Each invoker takes the owning router plus the pure-data
+# :class:`CompressInput` and returns the compressed string, or ``None`` when the
+# built-in is unavailable / not applicable to this str input (→ passthrough).
+_BuiltinInvoke = Callable[["ContentRouter", CompressInput], "str | None"]
 
-    The router dispatches built-ins through its own if/elif — never through the
-    registry — so ``compress`` is a guard that must not run. This type exists only
-    so the built-ins are name-addressable in the shared :class:`CompressorRegistry`
-    inventory next to discovered third-party compressors.
+
+def _adapter_bias(inp: CompressInput) -> float:
+    """Compression bias for a built-in call — the router's dispatch default (1.0).
+
+    Callers may override via ``budget['bias']`` (the router passes ``bias`` on the
+    request path); anything non-numeric falls back to the 1.0 default.
+    """
+    try:
+        return float(inp.budget.get("bias", 1.0))
+    except (TypeError, ValueError):
+        return 1.0
+
+
+def _invoke_smart_crusher(router: ContentRouter, inp: CompressInput) -> str | None:
+    crusher = router._get_smart_crusher()
+    if crusher is None:
+        return None
+    # ``_get_*`` getters are typed ``Any``; pin the result to the contract type.
+    compressed: str = crusher.crush(
+        inp.content, query=inp.query, bias=_adapter_bias(inp)
+    ).compressed
+    return compressed
+
+
+def _invoke_code_aware(router: ContentRouter, inp: CompressInput) -> str | None:
+    compressor = router._get_code_compressor()
+    if compressor is None:
+        return None
+    language = inp.config.get("language")
+    result = compressor.compress(inp.content, language=language, context=inp.query)
+    compressed: str = result.compressed
+    return compressed
+
+
+def _invoke_search(router: ContentRouter, inp: CompressInput) -> str | None:
+    compressor = router._get_search_compressor()
+    if compressor is None:
+        return None
+    result = compressor.compress(inp.content, context=inp.query, bias=_adapter_bias(inp))
+    compressed: str = result.compressed
+    return compressed
+
+
+def _invoke_log(router: ContentRouter, inp: CompressInput) -> str | None:
+    compressor = router._get_log_compressor()
+    if compressor is None:
+        return None
+    compressed: str = compressor.compress(inp.content, bias=_adapter_bias(inp)).compressed
+    return compressed
+
+
+def _invoke_tabular(router: ContentRouter, inp: CompressInput) -> str | None:
+    compressor = router._get_tabular_compressor()
+    if compressor is None:
+        return None
+    result = compressor.compress(inp.content, context=inp.query, bias=_adapter_bias(inp))
+    compressed: str = result.compressed
+    return compressed
+
+
+def _invoke_config(router: ContentRouter, inp: CompressInput) -> str | None:
+    compressor = router._get_config_compressor()
+    if compressor is None:
+        return None
+    result = compressor.compress(inp.content, context=inp.query, bias=_adapter_bias(inp))
+    compressed: str = result.compressed
+    return compressed
+
+
+def _invoke_html(router: ContentRouter, inp: CompressInput) -> str | None:
+    extractor = router._get_html_extractor()
+    if extractor is None:
+        return None
+    # ``.extracted`` may be None/empty when nothing extracts; the caller maps
+    # that to passthrough, matching the router's HTML branch.
+    extracted: str | None = extractor.extract(inp.content).extracted
+    return extracted
+
+
+def _invoke_kompress(router: ContentRouter, inp: CompressInput) -> str | None:
+    # The router dispatches KOMPRESS through ``_try_ml_compressor`` (size gate,
+    # tag protection, background load, marker policy), so the adapter delegates
+    # to the SAME method to stay byte-identical to the router's kompress path.
+    compressed, _tokens = router._try_ml_compressor(inp.content, inp.query, None)
+    return compressed
+
+
+def _invoke_image(router: ContentRouter, inp: CompressInput) -> str | None:
+    # The image built-in (``ImageCompressor``) compresses image blocks inside
+    # message dicts via ``ImageCompressor.compress(messages)`` /
+    # ``optimize_images_in_messages`` — it is NOT dispatched through
+    # ``_apply_strategy_to_content`` and never operates on str content. The
+    # str-based CompressInput/CompressOutput contract has no faithful image
+    # delegation (str content is never image data), so this is a documented
+    # passthrough rather than a fabricated compression.
+    return None
+
+
+def _invoke_passthrough(router: ContentRouter, inp: CompressInput) -> str | None:
+    # Defensive default for a descriptor without a registered invoker.
+    return None
+
+
+#: Built-in descriptor name → the invoker that runs it via the router's getter.
+_BUILTIN_COMPRESSOR_INVOKERS: dict[str, _BuiltinInvoke] = {
+    "smart_crusher": _invoke_smart_crusher,
+    "kompress": _invoke_kompress,
+    "code_aware": _invoke_code_aware,
+    "search": _invoke_search,
+    "log": _invoke_log,
+    "tabular": _invoke_tabular,
+    "config": _invoke_config,
+    "html": _invoke_html,
+    "image": _invoke_image,
+}
+
+
+class _BuiltinCompressorEntry:
+    """Registry adapter running a built-in via the router's existing dispatch path.
+
+    ``compress`` delegates to the SAME underlying built-in method the content
+    router invokes in ``_apply_strategy_to_content`` (obtained through the
+    router's ``_get_*`` getter so config flows through), then maps the built-in's
+    native result onto the pure-data :class:`CompressOutput` contract.
+
+    ADDITIVE by construction: the router still dispatches built-ins through its
+    own if/elif and never routes a request through the registry, so these entries
+    change no routing. ``_resolve_active_external_compressors`` filters them out
+    of the opt-in external-dispatch path *by type*, so the class name is load-
+    bearing. Constructed lazily/cheaply — it stores only the descriptor, the
+    owning router, and the invoke callable; no built-in is instantiated until
+    ``compress`` runs.
+
+    ``recoverable`` is always ``{}``: the built-ins embed CCR retrieval markers
+    in the compressed content and mirror ``hash -> original`` into the CCR store
+    as a side effect of their own ``compress`` call (which this adapter invokes),
+    rather than returning a recovery map on their result object.
     """
 
-    def __init__(self, descriptor: CompressorDescriptor) -> None:
+    def __init__(
+        self,
+        descriptor: CompressorDescriptor,
+        router: ContentRouter | None = None,
+        invoke: _BuiltinInvoke | None = None,
+    ) -> None:
         self._descriptor = descriptor
+        self._router = router
+        self._invoke = (
+            invoke
+            if invoke is not None
+            else _BUILTIN_COMPRESSOR_INVOKERS.get(descriptor.name, _invoke_passthrough)
+        )
 
     @property
     def descriptor(self) -> CompressorDescriptor:
         return self._descriptor
 
-    def compress(self, inp: CompressInput) -> CompressOutput:  # pragma: no cover - guard
-        raise NotImplementedError(
-            f"built-in compressor {self._descriptor.name!r} is dispatched by the "
-            "content router's built-in path, not through the registry"
+    def compress(self, inp: CompressInput) -> CompressOutput:
+        tokens_before = _estimate_tokens(inp.content)
+        compressed: str | None = None
+        if self._router is not None:
+            compressed = self._invoke(self._router, inp)
+        if compressed is None:
+            # Built-in unavailable, no bound router, or not applicable to this
+            # str input → passthrough (never blank out or expand a block).
+            compressed = inp.content
+        return CompressOutput(
+            content=compressed,
+            tokens_before=tokens_before,
+            tokens_after=_estimate_tokens(compressed),
+            lossless=self._descriptor.lossless,
+            markers=[],
+            recoverable={},
+            warnings=[],
         )
 
 
-def _build_compressor_registry() -> CompressorRegistry:
-    """Build the router's compressor registry: built-in inventory + discovery.
+def _build_compressor_registry(router: ContentRouter | None = None) -> CompressorRegistry:
+    """Build the router's compressor registry: built-in adapters + discovery.
 
-    Registers a metadata-only entry for each built-in, then runs opt-in
-    discovery of ``headroom.compressor`` entry points. Discovery never invokes
-    ``compress`` and is fail-open (a broken third-party package is logged and
-    skipped), so constructing this registry cannot change request handling.
+    Registers a delegating adapter for each built-in (bound to ``router`` so
+    ``compress`` runs the built-in through the router's own getter), then runs
+    opt-in discovery of ``headroom.compressor`` entry points. Discovery never
+    invokes ``compress`` and is fail-open (a broken third-party package is logged
+    and skipped). Building the registry has no side effects: adapters instantiate
+    nothing until ``compress`` is called, and the router still dispatches built-
+    ins via its own if/elif, so constructing this registry cannot change request
+    handling. When ``router`` is ``None`` the adapters have nothing to delegate to
+    and ``compress`` is an inert passthrough.
     """
     registry = CompressorRegistry()
     for descriptor in _BUILTIN_COMPRESSOR_DESCRIPTORS:
-        registry.register(_BuiltinCompressorEntry(descriptor))
+        registry.register(_BuiltinCompressorEntry(descriptor, router))
     # External compressors register under distinct names; a name collision with
     # a built-in is skipped fail-open (replace=False) so a third-party package
     # can never shadow a built-in's inventory entry.
@@ -1541,7 +1712,7 @@ class ContentRouter(Transform):
         # follow-up. Failure to build it must never break the router, so it is
         # fail-open to an empty registry.
         try:
-            self.compressor_registry: CompressorRegistry = _build_compressor_registry()
+            self.compressor_registry: CompressorRegistry = _build_compressor_registry(self)
         except Exception as exc:  # noqa: BLE001 - inventory is non-critical
             logger.debug("compressor registry unavailable: %s", exc)
             self.compressor_registry = CompressorRegistry()
diff --git a/tests/test_builtin_compressor_adapters.py b/tests/test_builtin_compressor_adapters.py
new file mode 100644
index 000000000..97b5f9a3c
--- /dev/null
+++ b/tests/test_builtin_compressor_adapters.py
@@ -0,0 +1,256 @@
+"""Differential/faithfulness tests for the built-in Compressor adapters.
+
+Each built-in registry entry now delegates ``compress`` to the SAME underlying
+built-in method the content router invokes in ``_apply_strategy_to_content``
+(reached through the router's own ``_get_*`` getter so config flows through
+identically). These tests feed representative content per content type and assert
+the adapter output matches what the built-in produces DIRECTLY, and that every
+built-in registry entry exposes a working (non-raising) ``compress``.
+
+Guardrails honored:
+  * Kompress is mocked — no real ONNX/HF model inference (which hangs here).
+  * Only the additive registry path is exercised; the router's dispatch
+    (``_apply_strategy_to_content``) is never called, so these prove the adapter
+    capability in isolation without touching routing.
+"""
+
+from __future__ import annotations
+
+import json
+from types import SimpleNamespace
+
+import pytest
+
+from headroom.transforms.compressor_registry import CompressInput, CompressOutput
+from headroom.transforms.content_router import (
+    _BUILTIN_COMPRESSOR_DESCRIPTORS,
+    ContentRouter,
+    ContentRouterConfig,
+    _BuiltinCompressorEntry,
+    _estimate_tokens,
+)
+
+
+def _router() -> ContentRouter:
+    # enable_code_aware defaults off (that flag gates ROUTING, not the getter);
+    # turn it on so nothing about the code getter differs in this test env. The
+    # adapters use the getters directly regardless of the enable flags.
+    return ContentRouter(ContentRouterConfig(enable_code_aware=True))
+
+
+def _entry(router: ContentRouter, name: str) -> _BuiltinCompressorEntry:
+    entry = router.compressor_registry.get(name)
+    assert isinstance(entry, _BuiltinCompressorEntry), name
+    return entry
+
+
+def _assert_output_contract(
+    out: CompressOutput, inp: CompressInput, entry: _BuiltinCompressorEntry
+) -> None:
+    assert isinstance(out, CompressOutput)
+    # Token counts use the router's own estimator, over the pure-data content.
+    assert out.tokens_before == _estimate_tokens(inp.content)
+    assert out.tokens_after == _estimate_tokens(out.content)
+    # lossless mirrors the descriptor; adapters emit no markers/warnings and no
+    # recovery map (built-ins mirror hash -> original into the CCR store as a
+    # side effect of their own compress call, not on the returned result).
+    assert out.lossless == entry.descriptor.lossless
+    assert out.markers == []
+    assert out.recoverable == {}
+    assert out.warnings == []
+
+
+# ─────────────────────── differential (per built-in) ─────────────────────────
+
+
+def test_smart_crusher_adapter_matches_builtin() -> None:
+    router = _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
+    entry = _entry(router, "smart_crusher")
+    inp = CompressInput(content=content, content_type="application/json", query="q")
+    out = entry.compress(inp)
+    assert out.content == direct
+    assert out.tokens_after < out.tokens_before  # representative JSON actually shrinks
+    _assert_output_contract(out, inp, entry)
+
+
+def test_tabular_adapter_matches_builtin() -> None:
+    router = _router()
+    content = "id,name,status,score\n" + "\n".join(f"{i},row{i},ok,{i * 3}" for i in range(50))
+    direct = router._get_tabular_compressor().compress(content, context="q", bias=1.0).compressed
+    entry = _entry(router, "tabular")
+    inp = CompressInput(content=content, content_type="text/csv", query="q")
+    out = entry.compress(inp)
+    assert out.content == direct
+    _assert_output_contract(out, inp, entry)
+
+
+def test_log_adapter_matches_builtin() -> None:
+    router = _router()
+    content = (
+        "\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))
+    )
+    direct = router._get_log_compressor().compress(content, bias=1.0).compressed
+    entry = _entry(router, "log")
+    inp = CompressInput(content=content, content_type="text/x-log", query="")
+    out = entry.compress(inp)
+    assert out.content == direct
+    assert out.tokens_after < out.tokens_before  # repetitive log actually shrinks
+    _assert_output_contract(out, inp, entry)
+
+
+def test_search_adapter_matches_builtin() -> None:
+    router = _router()
+    content = "\n".join(f"src/file{i}.py:{i}: def func{i}(): return {i}" for i in range(30))
+    direct = router._get_search_compressor().compress(content, context="func", bias=1.0).compressed
+    entry = _entry(router, "search")
+    inp = CompressInput(content=content, content_type="text/x-search-results", query="func")
+    out = entry.compress(inp)
+    assert out.content == direct
+    _assert_output_contract(out, inp, entry)
+
+
+def test_config_adapter_matches_builtin() -> None:
+    router = _router()
+    content = "\n".join(f"key{i} = value{i}" for i in range(30)) + "\n# comment\n\n# another\n"
+    direct = router._get_config_compressor().compress(content, context="q", bias=1.0).compressed
+    entry = _entry(router, "config")
+    inp = CompressInput(content=content, content_type="text/x-config", query="q")
+    out = entry.compress(inp)
+    assert out.content == direct
+    _assert_output_contract(out, inp, entry)
+
+
+def test_code_aware_adapter_matches_builtin() -> None:
+    router = _router()
+    content = (
+        "def foo(x):\n"
+        "    # a comment\n"
+        "    return x + 1\n\n\n"
+        "class Bar:\n"
+        "    def baz(self):\n"
+        "        return 42\n"
+    )
+    compressor = router._get_code_compressor()
+    if compressor is None:
+        pytest.skip("code compressor (tree-sitter) unavailable in this environment")
+    direct = compressor.compress(content, language=None, context="q").compressed
+    entry = _entry(router, "code_aware")
+    inp = CompressInput(content=content, content_type="text/x-code", query="q")
+    out = entry.compress(inp)
+    assert out.content == direct
+    _assert_output_contract(out, inp, entry)
+
+
+def test_html_adapter_matches_builtin() -> None:
+    router = _router()
+    content = (
+        "<html><head><title>T</title></head><body><nav>menu</nav>"
+        "<article><h1>Hi</h1><p>Hello world, this is the article body content "
+        "that trafilatura should extract from the surrounding chrome.</p></article>"
+        "</body></html>"
+    )
+    extractor = router._get_html_extractor()
+    if extractor is None:
+        pytest.skip("html extractor (trafilatura) unavailable in this environment")
+    direct = extractor.extract(content).extracted
+    entry = _entry(router, "html")
+    inp = CompressInput(content=content, content_type="text/html", query="")
+    out = entry.compress(inp)
+    # Adapter maps empty/None extraction to passthrough, matching the router.
+    assert out.content == (direct if direct is not None else content)
+    _assert_output_contract(out, inp, entry)
+
+
+def test_kompress_adapter_maps_mocked_result(monkeypatch: pytest.MonkeyPatch) -> None:
+    router = _router()
+    # Mock the underlying kompress compressor so NO real ONNX/HF inference runs.
+    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
+    entry = _entry(router, "kompress")
+    inp = CompressInput(content=content, content_type="text/plain", query="")
+    out = entry.compress(inp)
+    # The adapter delegates to _try_ml_compressor (the router's kompress path);
+    # with no protected tags the result is the fake's compressed text verbatim.
+    assert out.content == "KOMPRESSED::" + content
+    # ... which is exactly what the router's own kompress dispatch produces.
+    assert out.content == router._try_ml_compressor(content, "", None)[0]
+    _assert_output_contract(out, inp, entry)
+
+
+def test_kompress_adapter_passthrough_when_ml_disabled() -> None:
+    # With ML disabled the router's kompress path is a passthrough (no model load
+    # / network), so the adapter returns the content unchanged, never raising.
+    router = ContentRouter(ContentRouterConfig(enable_kompress=False))
+    content = "plain text with nothing special to compress"
+    entry = _entry(router, "kompress")
+    inp = CompressInput(content=content, content_type="text/plain", query="")
+    out = entry.compress(inp)
+    assert out.content == content
+    _assert_output_contract(out, inp, entry)
+
+
+def test_image_adapter_is_documented_passthrough() -> None:
+    # ImageCompressor operates on message image blocks (list[dict]) via
+    # ImageCompressor.compress(messages), not on the str-based CompressInput
+    # contract, and images are never routed through _apply_strategy_to_content.
+    # There is no faithful str -> str delegation, so the adapter passes str
+    # content through unchanged (documented, never a fabricated compression).
+    router = _router()
+    entry = _entry(router, "image")
+    content = "arbitrary str content that is never an image payload"
+    inp = CompressInput(content=content, content_type="image/png", query="")
+    out = entry.compress(inp)
+    assert out.content == content
+    _assert_output_contract(out, inp, entry)
+
+
+# ──────────────────────── registry-wide invariants ───────────────────────────
+
+
+def test_every_builtin_registry_entry_has_working_compress() -> None:
+    # ML disabled so kompress is a passthrough (no model load / network); every
+    # other built-in returns unchanged on this non-matching plain-text block.
+    router = ContentRouter(ContentRouterConfig(enable_kompress=False, enable_code_aware=True))
+    names = {d.name for d in _BUILTIN_COMPRESSOR_DESCRIPTORS}
+    assert names, "expected built-in descriptors to be registered"
+    content = "hello world\nsecond line\n"
+    for name in sorted(names):
+        entry = _entry(router, name)
+        out = entry.compress(CompressInput(content=content, content_type="text/plain"))
+        assert isinstance(out, CompressOutput), name
+        assert isinstance(out.content, str), name
+        assert out.tokens_before == _estimate_tokens(content), name
+        assert out.tokens_after == _estimate_tokens(out.content), name
+        assert out.lossless == entry.descriptor.lossless, name
+
+
+def test_adapters_never_expand_or_blank_representative_content() -> None:
+    # Adapter outputs must be usable blocks: never blank when input is non-blank,
+    # never longer than the input (mirrors the router's own external-dispatch
+    # guards, computed here from the built-in's real output).
+    router = _router()
+    cases = {
+        "smart_crusher": json.dumps([{"id": i, "v": i} for i in range(30)]),
+        "log": "\n".join("repeated identical log line" for _ in range(40)),
+        "search": "\n".join(f"a/b{i}.py:{i}: match here" for i in range(30)),
+        "config": "\n".join(f"k{i}=v{i}" for i in range(30)),
+    }
+    for name, content in cases.items():
+        entry = _entry(router, name)
+        out = entry.compress(CompressInput(content=content, content_type="text/plain", query=""))
+        assert out.content.strip(), name
+        assert len(out.content) <= len(content), name
diff --git a/tests/test_compressor_selection.py b/tests/test_compressor_selection.py
index a2eb7b8ba..7f01d4ec0 100644
--- a/tests/test_compressor_selection.py
+++ b/tests/test_compressor_selection.py
@@ -229,12 +229,28 @@ def test_registry_inventory_does_not_enable_selection() -> None:
     assert registry.active(None) == []
 
 
-def test_builtin_entry_compress_is_a_guard() -> None:
+def test_builtin_entry_compress_delegates_via_router() -> None:
+    # The built-in entries now expose a WORKING (non-raising) compress that
+    # delegates to the router's own dispatch path. kompress with ML disabled is a
+    # passthrough (no model load), proving the entry runs without raising.
+    router = ContentRouter(ContentRouterConfig(enable_kompress=False))
+    entry = router.compressor_registry.get("kompress")
+    assert entry is not None
+    out = entry.compress(CompressInput(content="hello world", content_type="text/plain"))
+    assert isinstance(out, CompressOutput)
+    assert out.content == "hello world"  # ML disabled → passthrough, never raises
+
+
+def test_builtin_entry_without_router_is_inert_passthrough() -> None:
+    # A registry built with no bound router (module-level inventory use) has
+    # nothing to delegate to, so compress is an inert passthrough — still working
+    # (non-raising), never a fabricated result.
     registry = _build_compressor_registry()
     entry = registry.get("kompress")
     assert entry is not None
-    with pytest.raises(NotImplementedError):
-        entry.compress(CompressInput(content="x", content_type="text/plain"))
+    out = entry.compress(CompressInput(content="x", content_type="text/plain"))
+    assert isinstance(out, CompressOutput)
+    assert out.content == "x"
 
 
 def test_discovery_merges_external_compressor(monkeypatch: pytest.MonkeyPatch) -> None: