feat(transforms): add compressed signal + dispatch code_aware/html/diff via registry (#2400)

## What
Third increment of the adapter phase (builds on #2391/#2399). Adds a
`compressed: bool` field to `CompressOutput` and uses it to flip the
**fallback/passthrough** strategies — CODE_AWARE and HTML (and DIFF
where clean) — to registry-resolved dispatch, byte-identically.

## The contract addition (the enabling piece)
`CompressOutput.compressed: bool = True` — lets a compressor signal
**passthrough** (did-not-compress, `content` is the original unchanged)
vs a real result. This is what the router's `None`-driven
fallback/passthrough branches needed to move to the registry without
changing behavior. Default `True`, so existing and external compressors
are unaffected.

## How (byte-identical)
A new `_registry_compress` helper returns the `CompressOutput` (or
`None` when the built-in is unavailable, preserving the `_get_*` guard's
passthrough). The flipped branches map that back to their historical
`compressed is None` semantics:
- **CODE_AWARE:** a passthrough (`not output.compressed` / `None`) sets
local `compressed = None`, so the existing `_try_ml_compressor` Kompress
fallback + `lossless_then_lossy` no-shrink retry +
`strategy`/`strategy_chain` mutations run **verbatim**.
- **HTML:** a `None`/passthrough falls through to the bottom passthrough
exactly as before (`strategy_chain == [html, passthrough]`).

## Deferred
SMART_CRUSHER, KOMPRESS, TEXT, PASSTHROUGH — the
SmartCrusher→Kompress→Log fallback chain + the ML boundary — are the
next (final) increment, left byte-for-byte here. Reversibility gate,
external dispatch (#2388), default behavior unchanged. No new
config/env.

## Testing
`tests/test_router_registry_dispatch.py` +
`tests/test_builtin_compressor_adapters.py` extended: differential tests
for CODE_AWARE (success AND None→Kompress-fallback with matching
`strategy_chain`, ML mocked), HTML (success AND None→`[html,
passthrough]`), and the adapter `compressed=False`-on-None mapping.
Offline suite: 88 passed; ruff + mypy clean. The full content-router
suite in CI is the authoritative byte-identical gate.

headroom/transforms/compressor_registry.py

@@ -105,13 +105,23 @@ class CompressOutput:
     """Pure-data output from :meth:`Compressor.compress`.
 
     Attributes:
-        content: The compressed content.
+        content: The compressed content (or the original content unchanged when
+            ``compressed`` is ``False``).
         tokens_before: Token count of the input content.
         tokens_after: Token count of the compressed content.
         lossless: Whether this particular result is losslessly reversible.
         markers: Marker strings describing what was applied (e.g. for routing).
         recoverable: ``hash -> original`` map for recovering dropped content.
         warnings: Non-fatal warning strings emitted during compression.
+        compressed: Whether the compressor actually compressed/extracted the
+            content. ``True`` (the default) means compression/extraction was
+            applied and :attr:`content` is the transformed result; ``False``
+            means the compressor did not compress — a passthrough — and
+            :attr:`content` is the original input unchanged. Defaults to ``True``
+            so existing and external compressors that do not set it are
+            unaffected (treated as having compressed). Callers can read this to
+            distinguish a real (possibly no-shrink) result from a passthrough and
+            run their own fallback on a passthrough.
     """
 
     content: str
@@ -121,6 +131,7 @@ class CompressOutput:
     markers: list[str] = field(default_factory=list)
     recoverable: dict[str, str] = field(default_factory=dict)
     warnings: list[str] = field(default_factory=list)
+    compressed: bool = True
 
 
 @runtime_checkable
@@ -133,7 +144,13 @@ class Compressor(Protocol):
         ...
 
     def compress(self, inp: CompressInput) -> CompressOutput:
-        """Compress ``inp`` and return a :class:`CompressOutput`."""
+        """Compress ``inp`` and return a :class:`CompressOutput`.
+
+        A compressor that does not compress the input (a passthrough) should
+        return ``CompressOutput(content=inp.content, compressed=False, ...)``;
+        the ``compressed`` flag defaults to ``True`` so a compressor that always
+        transforms need not set it.
+        """
         ...
 
 

headroom/transforms/content_router.py

@@ -374,21 +374,28 @@ class _BuiltinCompressorEntry:
 
     def compress(self, inp: CompressInput) -> CompressOutput:
         tokens_before = _estimate_tokens(inp.content)
-        compressed: str | None = None
+        raw: 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
+            raw = self._invoke(self._router, inp)
+        # A ``None`` from the invoker means the built-in did not compress — it
+        # was unavailable, had no bound router, or was not applicable to this str
+        # input (e.g. HTML extraction found nothing). Report that with
+        # ``compressed=False`` and pass the ORIGINAL content through unchanged
+        # (never blank out or expand a block) so a caller can run its own
+        # fallback exactly as the historical direct call did on a ``None``
+        # result. A non-``None`` result is a real compression → ``compressed``
+        # stays True and byte-identical to before.
+        did_compress = raw is not None
+        content = raw if raw is not None else inp.content
         return CompressOutput(
-            content=compressed,
+            content=content,
             tokens_before=tokens_before,
-            tokens_after=_estimate_tokens(compressed),
+            tokens_after=_estimate_tokens(content),
             lossless=self._descriptor.lossless,
             markers=[],
             recoverable={},
             warnings=[],
+            compressed=did_compress,
         )
 
 
@@ -2671,6 +2678,50 @@ 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(
+        self,
+        name: str,
+        strategy: CompressionStrategy,
+        content: str,
+        context: str,
+        bias: float,
+        config: dict[str, Any] | None = None,
+    ) -> CompressOutput | None:
+        """Compress ``content`` with a built-in via the registry, full output.
+
+        Resolves the built-in named ``name`` from :attr:`compressor_registry` and
+        runs it over the pure-data :class:`CompressInput` contract, returning the
+        adapter's :class:`CompressOutput` — including its ``compressed`` flag,
+        which reports whether the built-in actually compressed (``True``) or
+        passed the content through unchanged (``False``, e.g. the built-in
+        returned ``None`` / HTML extraction found nothing). Returns ``None`` only
+        when the built-in is not registered (defensive; the inventory is always
+        registered by ``_build_compressor_registry``).
+
+        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, and any
+        per-strategy ``config`` such as ``language`` for code_aware), so on a
+        real compression the returned content is byte-identical to the direct
+        call. Callers read ``.compressed`` to reproduce the historical
+        ``compressed is None`` fallback/passthrough branches exactly.
+        """
+        entry = self.compressor_registry.get(name)
+        if entry is None:
+            return None
+        return entry.compress(
+            CompressInput(
+                content=content,
+                content_type=_CONTENT_TYPE_TO_MIME.get(
+                    self._content_type_from_strategy(strategy), "text/plain"
+                ),
+                query=context,
+                config=config or {},
+                budget={"bias": bias},
+            )
+        )
+
     def _registry_compress_content(
         self,
         name: str,
@@ -2681,8 +2732,7 @@ class ContentRouter(Transform):
     ) -> 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
+        Thin wrapper over :meth:`_registry_compress` returning just 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
@@ -2693,24 +2743,14 @@ class ContentRouter(Transform):
         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.
+        return shape is unchanged. When the built-in is not registered
+        (defensive), falls back to the unchanged content.
         """
-        entry = self.compressor_registry.get(name)
-        if entry is None:
+        output = self._registry_compress(name, strategy, content, context, bias)
+        if output 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(
@@ -2854,12 +2894,26 @@ class ContentRouter(Transform):
                     compressor = self._get_code_compressor()
                     if compressor:
                         compressor_name = type(compressor).__name__
-                        result = compressor.compress(content, language=language, context=context)
-                        compressed, compressed_tokens = (
-                            result.compressed,
-                            len(result.compressed.split()),
+                        # Registry-resolved dispatch: the built-in "code_aware"
+                        # adapter delegates to this same getter+method with the
+                        # language passed through ``config``, so on a real
+                        # compression the content is byte-identical to the
+                        # historical direct call. If the adapter did NOT compress
+                        # (``compressed=False``), leave the local ``compressed``
+                        # None so the EXISTING Kompress fallback below runs
+                        # exactly as today.
+                        output = self._registry_compress(
+                            "code_aware",
+                            strategy,
+                            content,
+                            context,
+                            bias,
+                            config={"language": language},
                         )
-                        decision_reason = "code_aware"
+                        if output is not None and output.compressed:
+                            compressed = output.content
+                            compressed_tokens = len(output.content.split())
+                            decision_reason = "code_aware"
                 if compressed is None:
                     # Fallback to Kompress
                     compressed, compressed_tokens = self._try_ml_compressor(
@@ -2990,8 +3044,19 @@ class ContentRouter(Transform):
                     extractor = self._get_html_extractor()
                     if extractor:
                         compressor_name = type(extractor).__name__
-                        result = extractor.extract(content)
-                        compressed = result.extracted
+                        # Registry-resolved dispatch: the built-in "html" adapter
+                        # delegates to this same getter + extract(). It reports
+                        # ``compressed=False`` (and returns the original content)
+                        # when nothing extracts, so we collapse that to
+                        # ``compressed = None`` and the branch falls through to
+                        # the bottom passthrough exactly as the historical
+                        # ``result.extracted is None`` path (chain
+                        # ``[html, passthrough]``). A real extraction is
+                        # byte-identical to the historical ``result.extracted``.
+                        output = self._registry_compress("html", strategy, content, context, bias)
+                        compressed = (
+                            output.content if output is not None and output.compressed else None
+                        )
                         # Estimate tokens from extracted text (simple word count)
                         compressed_tokens = _estimate_tokens(compressed) if compressed else 0
                         decision_reason = "html_extractor"

tests/test_builtin_compressor_adapters.py

@@ -218,6 +218,56 @@ def test_image_adapter_is_documented_passthrough() -> None:
     _assert_output_contract(out, inp, entry)
 
 
+# ───────────────────── compressed-signal (did-compress flag) ─────────────────
+
+
+def test_default_compress_output_flag_is_true() -> None:
+    # Additive contract: `compressed` defaults True so existing/external
+    # compressors that never set it are unaffected (treated as having compressed).
+    out = CompressOutput(content="x", tokens_before=1, tokens_after=1, lossless=True)
+    assert out.compressed is True
+
+
+def test_adapter_reports_compressed_true_when_builtin_returns_content() -> None:
+    # A built-in that returns real content → compressed=True and that content.
+    descriptor = _BUILTIN_COMPRESSOR_DESCRIPTORS[0]
+    entry = _BuiltinCompressorEntry(descriptor, router=object(), invoke=lambda r, inp: "SHRUNK")
+    out = entry.compress(
+        CompressInput(content="a longer original block", content_type="text/plain")
+    )
+    assert out.compressed is True
+    assert out.content == "SHRUNK"
+
+
+def test_adapter_reports_compressed_false_when_builtin_returns_none() -> None:
+    # A built-in that returns None (unavailable / not applicable to this str
+    # input) → compressed=False and the ORIGINAL content passed through unchanged.
+    descriptor = _BUILTIN_COMPRESSOR_DESCRIPTORS[0]
+    entry = _BuiltinCompressorEntry(descriptor, router=object(), invoke=lambda r, inp: None)
+    out = entry.compress(CompressInput(content="original", content_type="text/plain"))
+    assert out.compressed is False
+    assert out.content == "original"
+
+
+def test_adapter_reports_compressed_false_when_no_router() -> None:
+    # No bound router → nothing to delegate to → compressed=False, passthrough.
+    descriptor = _BUILTIN_COMPRESSOR_DESCRIPTORS[0]
+    entry = _BuiltinCompressorEntry(descriptor, router=None)
+    out = entry.compress(CompressInput(content="original", content_type="text/plain"))
+    assert out.compressed is False
+    assert out.content == "original"
+
+
+def test_image_builtin_adapter_reports_not_compressed() -> None:
+    # The image built-in never compresses str content (documented passthrough) →
+    # compressed=False, original content unchanged.
+    router = _router()
+    entry = _entry(router, "image")
+    out = entry.compress(CompressInput(content="not an image payload", content_type="image/png"))
+    assert out.compressed is False
+    assert out.content == "not an image payload"
+
+
 # ──────────────────────── registry-wide invariants ───────────────────────────
 
 

tests/test_router_registry_dispatch.py

@@ -34,6 +34,7 @@ from types import SimpleNamespace
 import pytest
 
 from headroom.transforms.content_router import (
+    _BUILTIN_COMPRESSOR_DESCRIPTORS,
     CompressionStrategy,
     ContentRouter,
     ContentRouterConfig,
@@ -183,6 +184,143 @@ def test_smart_crusher_deferred_unchanged(monkeypatch: pytest.MonkeyPatch) -> No
     assert chain == [CompressionStrategy.SMART_CRUSHER.value]
 
 
+def _fallback_router() -> ContentRouter:
+    """Router with CODE_AWARE routing enabled and the if/elif branch terminal.
+
+    Same isolation as :func:`_router` (no relevance split, no lossy layer, markers
+    off) but with ``enable_code_aware=True`` since that flag gates CODE_AWARE
+    ROUTING (default off) rather than the getter. HTML routing is on by default.
+    """
+    return ContentRouter(
+        ContentRouterConfig(
+            relevance_split=False,
+            lossless_then_lossy=False,
+            ccr_inject_marker=False,
+            enable_code_aware=True,
+        )
+    )
+
+
+# ─────────────── flipped fallback strategies: CODE_AWARE / HTML ───────────────
+
+
+def test_code_aware_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    # CODE_AWARE-SUCCEEDS: the flip routes through the registry "code_aware"
+    # adapter, which delegates to the SAME getter+method with language via config.
+    # A deterministic fake getter keeps this offline (no tree-sitter) and records
+    # the exact call args, proving content/language/context flow through unchanged.
+    router = _fallback_router()
+    _isolate_branch(monkeypatch, router)
+    seen: dict[str, object] = {}
+    shrunk = "def foo(): ...  # compressed body"
+
+    def _fake_compress(content: str, language: object = None, context: str = "") -> SimpleNamespace:
+        seen.update(content=content, language=language, context=context)
+        return SimpleNamespace(compressed=shrunk)
+
+    monkeypatch.setattr(
+        router, "_get_code_compressor", lambda: SimpleNamespace(compress=_fake_compress)
+    )
+    # Sentinel ML: if the flip WRONGLY dropped into a 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 = "def foo():\n    " + "x = 1\n    " * 40 + "return x\n"
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.CODE_AWARE, "q", language="python", bias=1.0
+    )
+    assert out == shrunk
+    # CODE_AWARE's historical token metric is len(compressed.split()), NOT _estimate_tokens.
+    assert tokens == len(shrunk.split())
+    assert chain == [CompressionStrategy.CODE_AWARE.value]
+    # The flip forwarded content, language, and context through the registry adapter.
+    assert seen == {"content": content, "language": "python", "context": "q"}
+
+
+def test_code_aware_unavailable_falls_back_to_kompress(monkeypatch: pytest.MonkeyPatch) -> None:
+    # CODE_AWARE-RETURNS-NONE: with the code compressor UNAVAILABLE (tree-sitter
+    # missing) the branch's local `compressed` stays None, so the EXISTING inline
+    # Kompress fallback runs — the flip touches none of that logic. Mock
+    # _try_ml_compressor so NO real ML runs, and assert the SAME strategy_chain
+    # ([code_aware, kompress]) the historical direct dispatch produced.
+    router = _fallback_router()
+    _isolate_branch(monkeypatch, router)
+    monkeypatch.setattr(router, "_get_code_compressor", lambda: None)
+    monkeypatch.setattr(
+        router,
+        "_try_ml_compressor",
+        lambda content, ctx, q: ("KOMPRESSED::" + content, 3),
+    )
+    content = "def foo():\n    return 1\n"
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.CODE_AWARE, "ctx", language=None, bias=1.0
+    )
+    assert out == "KOMPRESSED::" + content
+    assert tokens == 3
+    assert chain == [CompressionStrategy.CODE_AWARE.value, CompressionStrategy.KOMPRESS.value]
+
+
+def test_html_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None:
+    # HTML-EXTRACT-SUCCEEDS: the flip routes through the registry "html" adapter,
+    # which delegates to the SAME getter + extract(). A deterministic fake getter
+    # keeps this offline (no trafilatura) and records the call arg.
+    router = _fallback_router()
+    _isolate_branch(monkeypatch, router)
+    extracted = "Extracted article body text that trafilatura would return."
+    seen: dict[str, object] = {}
+
+    def _fake_extract(content: str) -> SimpleNamespace:
+        seen["content"] = content
+        return SimpleNamespace(extracted=extracted)
+
+    monkeypatch.setattr(
+        router, "_get_html_extractor", lambda: SimpleNamespace(extract=_fake_extract)
+    )
+    content = "<html><body><article><p>hello world</p></article></body></html>"
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.HTML, "", bias=1.0
+    )
+    assert out == extracted
+    assert tokens == _estimate_tokens(extracted)
+    assert chain == [CompressionStrategy.HTML.value]
+    assert seen == {"content": content}
+
+
+def test_html_extract_none_falls_through_to_passthrough(monkeypatch: pytest.MonkeyPatch) -> None:
+    # HTML-EXTRACT-NONE: when extraction yields None the adapter reports
+    # compressed=False, the branch's local `compressed` collapses to None, and the
+    # function falls through to the bottom passthrough exactly as the historical
+    # `result.extracted is None` path — chain [html, passthrough], content verbatim.
+    router = _fallback_router()
+    _isolate_branch(monkeypatch, router)
+    monkeypatch.setattr(
+        router,
+        "_get_html_extractor",
+        lambda: SimpleNamespace(extract=lambda content: SimpleNamespace(extracted=None)),
+    )
+    # Sentinel ML so we'd notice if the None path wrongly reached a lossy compressor.
+    monkeypatch.setattr(router, "_try_ml_compressor", lambda *a, **k: ("KOMPRESS_SENTINEL", 1))
+    content = "<html><body><script>no extractable article body</script></body></html>"
+    out, tokens, chain = router._apply_strategy_to_content(
+        content, CompressionStrategy.HTML, "", bias=1.0
+    )
+    assert out == content
+    assert tokens == _estimate_tokens(content)
+    assert chain == [CompressionStrategy.HTML.value, CompressionStrategy.PASSTHROUGH.value]
+
+
+def test_diff_deferred_no_registry_entry() -> None:
+    # DIFF is DEFERRED: there is no "diff" built-in adapter/descriptor in the
+    # registry inventory, so registry resolution would return None and fall back
+    # to raw content — NOT byte-identical. It stays on its direct dispatch until a
+    # diff built-in adapter lands (PR-A scope).
+    router = _router()
+    assert router.compressor_registry.get("diff") is None
+    assert "diff" not in {d.name for d in _BUILTIN_COMPRESSOR_DESCRIPTORS}
+
+
 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