From 7ebda67ef65fe82803c7fb729c509a1451165f26 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Sat, 18 Jul 2026 16:50:05 -0700 Subject: [PATCH] feat(transforms): add compressed signal + dispatch code_aware/html/diff via registry (#2400) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## 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 | 21 +++- headroom/transforms/content_router.py | 127 ++++++++++++++----- tests/test_builtin_compressor_adapters.py | 50 ++++++++ tests/test_router_registry_dispatch.py | 138 +++++++++++++++++++++ 4 files changed, 303 insertions(+), 33 deletions(-) diff --git a/headroom/transforms/compressor_registry.py b/headroom/transforms/compressor_registry.py index 3cc99f5ac..ab8c617f7 100644 --- a/headroom/transforms/compressor_registry.py +++ b/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. + """ ... diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 0b8ad9364..701da93f3 100644 --- a/headroom/transforms/content_router.py +++ b/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_().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_().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" diff --git a/tests/test_builtin_compressor_adapters.py b/tests/test_builtin_compressor_adapters.py index 97b5f9a3c..b30ae58c5 100644 --- a/tests/test_builtin_compressor_adapters.py +++ b/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 ─────────────────────────── diff --git a/tests/test_router_registry_dispatch.py b/tests/test_router_registry_dispatch.py index 1157df429..2332c0efe 100644 --- a/tests/test_router_registry_dispatch.py +++ b/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 = "

hello world

" + 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 = "" + 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