feat(transforms): pluggable lossless-compaction provider seam (#2433)

## Description

Adds an optional external provider for the information-preserving
compaction of protected (excluded) tool output, mirroring the existing
`proxy_extension` / `compressor` extension seams. Lets an out-of-tree
extension supply its own reversible compaction for excluded tools
without forking the router. Default behavior is unchanged.

Closes #

## Type of Change

- [x] New feature (non-breaking change that adds functionality)

## Changes Made

- New `headroom/transforms/lossless_provider.py`:
`set_lossless_provider` / `get_lossless_provider`. Contract: `content ->
(compacted, kind) | None`, where `compacted` must be byte-recoverable
(or data-lossless for structured data), and the provider must be
deterministic and per-block so the prefix cache stays byte-stable across
turns.
- `ContentRouter._lossless_compact_excluded` consults a registered
provider first and is **authoritative** when one is set; it falls back
to the built-in folds only if the provider raises. With no provider
registered (the default) behavior is byte-for-byte identical to before.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality

### Test Output

```text
$ python -m pytest tests/test_lossless_excluded_compaction.py -q
tests/test_lossless_excluded_compaction.py ...........                   [100%]
11 passed in 0.60s

$ ruff check headroom/transforms/lossless_provider.py headroom/transforms/content_router.py
All checks passed!

$ mypy headroom/transforms/lossless_provider.py headroom/transforms/content_router.py
Success: no issues found in 2 source files
```

## Real Behavior Proof

- Environment: local, Python 3.12 venv,
`ContentRouter(ContentRouterConfig())`.
- Exact command / steps: (1) default — call
`_lossless_compact_excluded(GREP)` with no provider; (2) register
`set_lossless_provider(lambda c: ("<<folded>>","custom"))` and call
again; (3) register a provider that raises.
- Observed result: (1) built-in search-heading fold `("…","search")`;
(2) returns `("<<folded>>","custom")` — provider is authoritative,
built-in not run; provider returning `None` yields `None` (no built-in
fallback); (3) provider exception → falls back to the built-in fold.
Covered by the 3 new tests.
- Not tested: the broad `tests/test_transforms/test_content_router.py`
suite stalls locally on model downloads (HF/ONNX); CI runs it.

## Review Readiness

- [x] I have performed a self-review
- [x] This PR is ready for human review

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [x] My changes generate no new warnings
- [x] I have added tests that prove my feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I did **not** edit `CHANGELOG.md`
This commit is contained in:
Tejas Chopra 2026-07-20 10:44:42 -07:00 committed by GitHub
parent c400f90810
commit a7dcb9e91c
No known key found for this signature in database
GPG key ID: B5690EEEBB952194
3 changed files with 97 additions and 1 deletions

View file

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

View file

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

View file

@ -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: ("<<folded>>", "custom"))
assert _compact(GREP) == ("<<folded>>", "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