feat(transforms): make built-in compressors real Compressor implementations (adapters) (#2391)

## 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.
This commit is contained in:
Tejas Chopra 2026-07-18 12:59:49 -07:00 committed by GitHub
parent 6cdfd3f64d
commit 981616c60e
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 465 additions and 22 deletions

View file

@ -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()

View file

@ -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

View file

@ -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: