diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 92041aa0c..44207332f 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -78,6 +78,7 @@ from .content_detector import ( ) from .content_detector import detect_content_type as _regex_detect_content_type from .error_detection import content_has_strong_error_indicators +from .lossless_provider import get_lossless_provider from .mixed_content import ContentSection, mixed_content_indicators from .relevance_split import build_relevance_query, plan_relevance_split @@ -5187,7 +5188,20 @@ class ContentRouter(Transform): Always safe to run (information-preserving) so there is no feature gate. Never raises. """ - if not isinstance(content, str) or len(content) < 200: + if not isinstance(content, str): + return None + provider = get_lossless_provider() + if provider is not None: + try: + # A registered provider is authoritative for excluded-tool + # compaction; fall back to the built-in folds only if it raises. + return provider(content) + except Exception: # noqa: BLE001 + logger.debug( + "lossless provider failed; using built-in compaction", + exc_info=True, + ) + if len(content) < 200: return None try: from .lossless_compaction import compact_lossless diff --git a/headroom/transforms/lossless_provider.py b/headroom/transforms/lossless_provider.py new file mode 100644 index 000000000..36e3d3ece --- /dev/null +++ b/headroom/transforms/lossless_provider.py @@ -0,0 +1,41 @@ +"""Pluggable provider for information-preserving compaction of protected output. + +Excluded ("protected") tool results are kept out of *lossy* compression for +accuracy; the content router still applies reversible/data-preserving folds to +them via :meth:`ContentRouter._lossless_compact_excluded`. This module lets an +external extension supply that compaction instead of the built-in folds — the +same open-core pattern as the ``proxy_extension`` / ``compressor`` seams. + +Contract — ``provider(content: str) -> tuple[compacted: str, kind: str] | None``: + +* ``compacted`` MUST be information-preserving — byte-recoverable, or + data-lossless for structured data (same guarantee the built-in path gives). + Return ``None`` to leave the content unchanged. +* The provider MUST be deterministic and depend only on ``content`` (no + cross-message state), so the proxy's prefix cache stays byte-stable across + turns. + +A registered provider is *authoritative*: when one is set the router does not run +its built-in folds — it falls back to the built-in only if the provider raises. +Default is ``None`` → the router uses its built-in folds, unchanged. +""" + +from __future__ import annotations + +from collections.abc import Callable + +#: ``content -> (compacted, kind) | None``. +LosslessProvider = Callable[[str], "tuple[str, str] | None"] + +_provider: LosslessProvider | None = None + + +def set_lossless_provider(provider: LosslessProvider | None) -> None: + """Register (or clear, with ``None``) the lossless compaction provider.""" + global _provider + _provider = provider + + +def get_lossless_provider() -> LosslessProvider | None: + """Return the registered provider, or ``None`` if the built-in should run.""" + return _provider diff --git a/tests/test_lossless_excluded_compaction.py b/tests/test_lossless_excluded_compaction.py index c8a8f8c76..5872299fe 100644 --- a/tests/test_lossless_excluded_compaction.py +++ b/tests/test_lossless_excluded_compaction.py @@ -22,6 +22,10 @@ from headroom.providers import OpenAIProvider from headroom.tokenizer import Tokenizer from headroom.transforms.content_router import ContentRouter, ContentRouterConfig from headroom.transforms.lossless_compaction import expand_runs, search_unheading, strip_ansi +from headroom.transforms.lossless_provider import ( + get_lossless_provider, + set_lossless_provider, +) GREP = "".join( f"src/module_{f}.py:{ln * 3}:matched occurrence with some real content here\n" @@ -117,3 +121,40 @@ def test_pipeline_minifies_json_read(tokenizer): def test_pipeline_leaves_source_read_untouched(tokenizer): out, _ = _run(CODE, "read", tokenizer) assert out == CODE + + +# --- pluggable lossless provider seam --------------------------------------- + + +@pytest.fixture(autouse=True) +def _reset_provider(): + """Never leak a registered provider between tests.""" + yield + set_lossless_provider(None) + + +def test_default_no_provider_uses_builtin(): + # Unset (default) → built-in folds run; GREP compacts via search-heading. + assert get_lossless_provider() is None + out, kind = _compact(GREP) + assert kind == "search" and search_unheading(out) == GREP + + +def test_registered_provider_is_authoritative(): + # A registered provider fully owns excluded-tool compaction; the built-in + # search fold does NOT run (we'd get "search", not our sentinel). + set_lossless_provider(lambda content: ("<>", "custom")) + assert _compact(GREP) == ("<>", "custom") + # Authoritative on None too: provider says "leave it" → no built-in fallback. + set_lossless_provider(lambda content: None) + assert _compact(GREP) is None + + +def test_provider_exception_falls_back_to_builtin(): + def boom(content): + raise RuntimeError("provider blew up") + + set_lossless_provider(boom) + # Falls back to the built-in fold rather than crashing or passing through raw. + out, kind = _compact(GREP) + assert kind == "search" and search_unheading(out) == GREP