From 981616c60ef04c32b3eb5b51c4f0f4a7ef297ef1 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Sat, 18 Jul 2026 12:59:49 -0700 Subject: [PATCH] feat(transforms): make built-in compressors real Compressor implementations (adapters) (#2391) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Turns each built-in registry entry into a working `Compressor` (the `compressor_registry` contract): `compress(CompressInput) -> CompressOutput` delegates to the same underlying built-in method the content router already invokes in `_apply_strategy_to_content`, reached through the router's own `_get_*` getter so config flows through identically. Token counts use the router's `_estimate_tokens`; `lossless` mirrors the descriptor; `recoverable` is `{}` (built-ins persist CCR recovery to the store as a side effect, not via their return value). Adapted: `smart_crusher, code_aware, search, log, tabular, config, html, kompress`. ## Behavior change **None — additive by construction.** Dispatch, the `_get_*` getters, fallback chains, the reversibility gate, and config are all unchanged. The router still dispatches built-ins via its existing if/elif and never routes a request through the registry; `_resolve_active_external_compressors` filters built-in entries out of the opt-in external-dispatch path *by type* (the class name `_BuiltinCompressorEntry` is load-bearing). A default request is byte-identical: `_active_external_compressors == []`, external dispatch is an inert guard, and adapters are reachable only via `compressor_registry.get()/active()`. ## `image` — documented passthrough (not a guess) `ImageCompressor.compress(messages)` operates on image blocks inside message dicts, not `str` content, and isn't on the `_apply_strategy_to_content` path, so there's no faithful `str→str` delegation. Its adapter is a documented non-raising passthrough rather than a fabricated one. ## Testing `tests/test_builtin_compressor_adapters.py` — differential tests asserting each adapter's output matches the built-in's direct output (JSON→smart_crusher, CSV→tabular, log lines→log, grep→search, config→config, Python→code_aware, HTML→html); kompress is mocked (no ML inference); every registry entry has a working non-raising `compress`. Updated the obsolete guard test in `test_compressor_selection.py`. Offline suite: 72 passed; ruff + mypy clean. (Broad content-router/compression suite deferred to CI — it needs HF-Hub/ONNX model loads.) This is PR-A of the adapter phase (built-ins become Compressor implementations); flipping the router's dispatch to registry-resolved is the follow-up. Builds on #2370/#2371/#2373/#2388. --- headroom/transforms/content_router.py | 209 ++++++++++++++++-- tests/test_builtin_compressor_adapters.py | 256 ++++++++++++++++++++++ tests/test_compressor_selection.py | 22 +- 3 files changed, 465 insertions(+), 22 deletions(-) create mode 100644 tests/test_builtin_compressor_adapters.py 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 = ( + "T" + "

Hi

Hello world, this is the article body content " + "that trafilatura should extract from the surrounding chrome.

" + "" + ) + 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: