From fc9c63f18c1a8414b62ced8b2dd54ad1fe4d1c14 Mon Sep 17 00:00:00 2001 From: Tejas Chopra Date: Sat, 18 Jul 2026 15:39:13 -0700 Subject: [PATCH] refactor(transforms): dispatch simple built-in strategies via the compressor registry (#2399) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit ## What Second increment of the adapter phase (builds on #2391). Flips the content router's per-strategy dispatch in `_apply_strategy_to_content` from the hardcoded if/elif to **registry-resolved** — but only for the *clean, single-compressor* strategies: **SEARCH, LOG, TABULAR, CONFIG**. Each resolves its compressor by name from `compressor_registry` and runs it over the pure-data `CompressInput`/`CompressOutput` contract via a shared `_registry_compress_content` helper, then maps back to the branch's exact historical return shape. ## Byte-identical by construction - The built-in adapter delegates to the SAME `_get_()` getter + method with the same args (`context`→query, `bias`→budget), so returned content is identical to the old direct call. - Each flipped branch **keeps its `enable_*` gate and `_get_*` availability guard** — so the built-in-unavailable → passthrough behavior is preserved and the adapter's `None`→content collapse is never reached. - Each branch **recomputes its token count with its own historical metric** (`_estimate_tokens` for search/log/tabular; `len(split())` for config). - `content_type` in `CompressInput` is inert (built-ins don't consume it), so it can't shift output. ## Deferred (left byte-for-byte as-is) — and why - **CODE_AWARE** — has a Kompress/ML fallback chain (`compressed is None` → `_try_ml_compressor`, plus a `lossless_then_lossy` no-shrink retry) that mutates `strategy`/`strategy_chain`. Not a clean single call. - **HTML** — uses `.extract().extracted` (different shape) and relies on `None` extraction falling through to bottom passthrough (`[html, passthrough]`); the adapter's `None`→content collapse would change the chain. Not byte-identical through the entry. - **SMART_CRUSHER** (fallback chain), **KOMPRESS/TEXT** (ML boundary), **PASSTHROUGH**, **DIFF** — untouched per plan. The reversibility gate, external-compressor dispatch (#2388), and default (nothing-selected) behavior are unchanged. No new config/env. ## Testing New `tests/test_router_registry_dispatch.py` (6 tests): differential test per flipped strategy asserting registry-dispatch output == old direct-dispatch output (content + branch token metric + `[strategy]` chain), plus assertions that deferred SMART_CRUSHER and KOMPRESS are unchanged. Offline suite: 78 passed; ruff + mypy clean. The broad content-router suite (HF-Hub/ONNX) is deferred to CI — **that full suite is the authoritative byte-identical gate for the flipped strategies.** --- headroom/transforms/content_router.py | 85 ++++++++-- tests/test_router_registry_dispatch.py | 208 +++++++++++++++++++++++++ 2 files changed, 276 insertions(+), 17 deletions(-) create mode 100644 tests/test_router_registry_dispatch.py diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 29e69f136..0b8ad9364 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -2671,6 +2671,48 @@ 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_content( + self, + name: str, + strategy: CompressionStrategy, + content: str, + context: str, + bias: float, + ) -> 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 + 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 + the SAME arguments (``context`` as the query, ``bias`` via the budget), so + the returned content is byte-identical to the direct call. + + Callers keep their own ``if self.config.enable_:`` gate and ``_get_*`` + 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. + """ + entry = self.compressor_registry.get(name) + if entry 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( self, content: str, @@ -2876,11 +2918,13 @@ class ContentRouter(Transform): compressor = self._get_search_compressor() if compressor: compressor_name = type(compressor).__name__ - result = compressor.compress(content, context=context, bias=bias) - compressed, compressed_tokens = ( - result.compressed, - _estimate_tokens(result.compressed), + # Registry-resolved dispatch: the built-in "search" adapter + # delegates to this same getter+method, so the content is + # byte-identical to the historical direct call. + compressed = self._registry_compress_content( + "search", strategy, content, context, bias ) + compressed_tokens = _estimate_tokens(compressed) decision_reason = "search_compressor" elif strategy == CompressionStrategy.LOG: @@ -2888,15 +2932,17 @@ class ContentRouter(Transform): compressor = self._get_log_compressor() if compressor: compressor_name = type(compressor).__name__ - result = compressor.compress(content, bias=bias) + # Registry-resolved dispatch: the built-in "log" adapter + # delegates to this same getter+method, so the content is + # byte-identical to the historical direct call. + compressed = self._registry_compress_content( + "log", strategy, content, context, bias + ) # Use the same word-count metric the rest of the # router uses; `compressed_line_count` is in # lines, not tokens — recording it here made # ratios meaningless against `original_tokens`. - compressed, compressed_tokens = ( - result.compressed, - _estimate_tokens(result.compressed), - ) + compressed_tokens = _estimate_tokens(compressed) decision_reason = "log_compressor" elif strategy == CompressionStrategy.TABULAR: @@ -2904,11 +2950,13 @@ class ContentRouter(Transform): compressor = self._get_tabular_compressor() if compressor: compressor_name = type(compressor).__name__ - result = compressor.compress(content, context=context, bias=bias) - compressed, compressed_tokens = ( - result.compressed, - _estimate_tokens(result.compressed), + # Registry-resolved dispatch: the built-in "tabular" adapter + # delegates to this same getter+method, so the content is + # byte-identical to the historical direct call. + compressed = self._registry_compress_content( + "tabular", strategy, content, context, bias ) + compressed_tokens = _estimate_tokens(compressed) decision_reason = "tabular_compressor" elif strategy == CompressionStrategy.CONFIG: @@ -2916,11 +2964,14 @@ class ContentRouter(Transform): compressor = self._get_config_compressor() if compressor: compressor_name = type(compressor).__name__ - result = compressor.compress(content, context=context, bias=bias) - compressed, compressed_tokens = ( - result.compressed, - len(result.compressed.split()), + # Registry-resolved dispatch: the built-in "config" adapter + # delegates to this same getter+method, so the content is + # byte-identical to the historical direct call. Keep the + # branch's own whitespace-split token metric. + compressed = self._registry_compress_content( + "config", strategy, content, context, bias ) + compressed_tokens = len(compressed.split()) decision_reason = "config_compressor" elif strategy == CompressionStrategy.DIFF: diff --git a/tests/test_router_registry_dispatch.py b/tests/test_router_registry_dispatch.py new file mode 100644 index 000000000..1157df429 --- /dev/null +++ b/tests/test_router_registry_dispatch.py @@ -0,0 +1,208 @@ +"""Byte-identical differential tests for registry-resolved built-in dispatch. + +The content router now dispatches the SIMPLE built-in strategies (CONFIG, LOG, +SEARCH, TABULAR) through the compressor registry instead of a hardcoded direct +``self._get_*().compress(...)`` call in ``_apply_strategy_to_content``. Each +built-in adapter delegates to the SAME ``_get_*`` getter+method with the SAME +arguments, so registry-resolved dispatch must be byte-identical to the historical +direct dispatch: same compressed content, same token count, same single-entry +``strategy_chain``. + +Each FLIPPED strategy has a differential test comparing the router's dispatch +output to the built-in's direct output obtained via its ``_get_*`` getter — i.e. +"registry dispatch == old dispatch". DEFERRED strategies (SMART_CRUSHER, KOMPRESS) +are asserted unchanged: they still route through their bespoke paths (fallback +chain / the ``_try_ml_compressor`` ML boundary), not the registry. + +Offline guardrails: + * No real ML/ONNX/HF inference — the deferred KOMPRESS path is mocked. + * The flipped strategies shrink their representative content, so no zero-savings + Kompress fallback fires (that would touch the ML boundary and append KOMPRESS + to the chain). + * ``lossless_then_lossy`` and ``relevance_split`` are off so the if/elif branch + is the terminal path; STAGE 0 (``_lossless_first``) is neutralized so search/ + log folds don't return before the branch. Both are shared, unchanged code. + * The broad ``content_router``/``compression`` -k selection is NOT exercised + (it hangs on HF-Hub/ONNX). +""" + +from __future__ import annotations + +import json +from types import SimpleNamespace + +import pytest + +from headroom.transforms.content_router import ( + CompressionStrategy, + ContentRouter, + ContentRouterConfig, + _estimate_tokens, +) + + +def _router() -> ContentRouter: + """Router whose if/elif branch is the terminal dispatch path. + + ``relevance_split`` off (no LOG/SEARCH relevance split) and + ``lossless_then_lossy`` off (no lossy layer on top of a strategy result) so a + successful strategy result returns directly. ``ccr_inject_marker`` off makes + the compressed output deterministic and marker-free; it is applied identically + to the direct reference and the dispatch router, so the differential holds + regardless of its value. + """ + return ContentRouter( + ContentRouterConfig( + relevance_split=False, + lossless_then_lossy=False, + ccr_inject_marker=False, + ) + ) + + +def _isolate_branch(monkeypatch: pytest.MonkeyPatch, router: ContentRouter) -> None: + """Neutralize STAGE 0 so the if/elif branch under test is exercised. + + ``_lossless_first`` runs unconditionally and can fold search/log content, + returning before the if/elif. It is shared, unchanged code (the flip only + touches the branch bodies), so forcing it to a no-op isolates what the flip + actually changed without altering the branch semantics. + """ + monkeypatch.setattr(router, "_lossless_first", lambda content, strategy: (content, None)) + + +# Representative content per type. The flipped strategies must SHRINK this so the +# fallback-eligible strategies (TABULAR/CONFIG) don't trip the zero-savings +# Kompress fallback (which would append KOMPRESS to the chain). +_SEARCH = "\n".join(f"src/file{i}.py:{i}: def func{i}(): return {i}" for i in range(30)) +_LOG = ( + "\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)) +) +# A markdown table the tabular compressor actually shrinks (schema-fold), and a +# repetitive INI the config compressor actually shrinks (block-fold) — so these +# fallback-eligible strategies produce a real token saving and the shared +# zero-savings Kompress fallback does NOT fire (chain stays single-entry). +_TABULAR = "| id | name | status | score |\n|----|------|--------|-------|\n" + "\n".join( + f"| {i} | row{i} | ok | {i * 3} |" for i in range(60) +) +_CONFIG = "\n".join( + f"[section_{i}]\nname = svc{i}\ntimeout = 30\nretries = 3\nverbose = false\nregion = us-east-1" + for i in range(40) +) + + +# ───────────────────────── flipped (registry dispatch) ──────────────────────── + + +def test_search_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None: + router = _router() + _isolate_branch(monkeypatch, router) + context, bias = "func", 1.0 + # OLD dispatch reference: same getter + method the branch used before the flip. + direct = ( + _router()._get_search_compressor().compress(_SEARCH, context=context, bias=bias).compressed + ) + out, tokens, chain = router._apply_strategy_to_content( + _SEARCH, CompressionStrategy.SEARCH, context, bias=bias + ) + assert out == direct + assert tokens == _estimate_tokens(direct) + assert chain == [CompressionStrategy.SEARCH.value] + + +def test_log_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None: + router = _router() + _isolate_branch(monkeypatch, router) + bias = 1.0 + direct = _router()._get_log_compressor().compress(_LOG, bias=bias).compressed + out, tokens, chain = router._apply_strategy_to_content( + _LOG, CompressionStrategy.LOG, "", bias=bias + ) + assert out == direct + assert tokens == _estimate_tokens(direct) + assert chain == [CompressionStrategy.LOG.value] + + +def test_tabular_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None: + router = _router() + _isolate_branch(monkeypatch, router) + context, bias = "q", 1.0 + direct = ( + _router() + ._get_tabular_compressor() + .compress(_TABULAR, context=context, bias=bias) + .compressed + ) + out, tokens, chain = router._apply_strategy_to_content( + _TABULAR, CompressionStrategy.TABULAR, context, bias=bias + ) + assert out == direct + assert tokens == _estimate_tokens(direct) + # Fallback-eligible, but a real shrink means no zero-savings Kompress fallback. + assert chain == [CompressionStrategy.TABULAR.value] + assert len(out) < len(_TABULAR) + + +def test_config_router_dispatch_matches_direct(monkeypatch: pytest.MonkeyPatch) -> None: + router = _router() + _isolate_branch(monkeypatch, router) + context, bias = "q", 1.0 + direct = ( + _router()._get_config_compressor().compress(_CONFIG, context=context, bias=bias).compressed + ) + out, tokens, chain = router._apply_strategy_to_content( + _CONFIG, CompressionStrategy.CONFIG, context, bias=bias + ) + assert out == direct + # CONFIG's historical metric is len(text.split()), NOT _estimate_tokens; the + # flip must preserve that exact metric. + assert tokens == len(direct.split()) + assert chain == [CompressionStrategy.CONFIG.value] + assert len(out) < len(_CONFIG) + + +# ─────────────────────────── deferred (unchanged) ──────────────────────────── + + +def test_smart_crusher_deferred_unchanged(monkeypatch: pytest.MonkeyPatch) -> None: + # SMART_CRUSHER is DEFERRED (its branch feeds a Kompress→Log fallback chain in + # the shared post-strategy block), so it still dispatches via the direct + # crusher, not the registry. A JSON array shrinks, so the chain stays single. + router = _router() + _isolate_branch(monkeypatch, 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 + out, _tokens, chain = router._apply_strategy_to_content( + content, CompressionStrategy.SMART_CRUSHER, "q", bias=1.0 + ) + assert out == direct + assert chain == [CompressionStrategy.SMART_CRUSHER.value] + + +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 + # real ONNX/HF inference runs, and assert the router still routes through + # _try_ml_compressor rather than the registry. + router = _router() + _isolate_branch(monkeypatch, router) + 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 + out, _tokens, chain = router._apply_strategy_to_content( + content, CompressionStrategy.KOMPRESS, "", bias=1.0 + ) + assert out == "KOMPRESSED::" + content + assert chain == [CompressionStrategy.KOMPRESS.value] + # Still the bespoke ML path (unchanged), not a registry round-trip. + assert out == router._try_ml_compressor(content, "", None)[0]