feat(proxy): select built-in compressors via --compressor + registry inventory (#2373)

## What
- Adds an opt-in `--compressor` / `HEADROOM_COMPRESSORS` selection that
narrows the active built-in compressors, mapped onto the existing
`ContentRouterConfig` `enable_*` flags at the proxy config seam.
Recognized names: `smart_crusher, kompress, code_aware, search, log,
tabular, config, html, image`; `"*"` selects all.
- Builds a name-addressable compressor registry in `ContentRouter`: a
metadata-only descriptor per built-in plus opt-in discovery of
`headroom.compressor` entry points (the seam added in #2370).

## Why
Built-in compressors were only reachable through a hardcoded if/elif;
there was no supported way to select a subset (7 of the `enable_*` flags
had no external surface) or to see the built-ins as a name-addressable
set alongside third-party ones.

## Behavior change
**None by default.** `--compressor` unset (the default) leaves every
`enable_*` flag at its dataclass default, so the request path is
byte-identical to today. The registry is inventory-only — built-ins are
still constructed and dispatched by the existing if/elif;
`_BuiltinCompressorEntry.compress` deliberately raises (never called),
and registry construction is fail-open. Routing an external compressor
*through* the pipeline is a deliberate follow-up.

## How
- `server.py`: `BUILTIN_COMPRESSOR_FLAGS` map +
`_apply_compressor_selection(router_config, compressors)` (no-op when
`None`/empty; runs before the `disable_kompress` override so that stays
authoritative).
- `models.py`: `ProxyConfig.compressors: set[str] | None = None`.
- `cli/proxy.py`: `--compressor` (repeatable, comma-split,
`HEADROOM_COMPRESSORS`), mirroring `--proxy-extension`.
- `content_router.py`: built-in descriptors +
`_build_compressor_registry()` (register built-ins, then fail-open
`discover()`), exposed as `self.compressor_registry`. Dispatch
unchanged.

## Testing
`tests/test_compressor_selection.py` — 23 tests: selection mapping
(None/empty/whitespace = byte-identical defaults, single/multi/wildcard,
external-only disables built-ins, unrecognized ignored), `ProxyConfig`
field, registry inventory (descriptors cover the 9 names, valid cost
tiers, router exposes registry, inventory doesn't auto-activate,
built-in `compress` guard, discovery merges external, fail-open on
discovery error). Local: 23 passed; ruff + mypy clean on changed files.
Full suite runs in CI.

Stacks conceptually on #2370 (registry seam); rebased onto `main` after
that merged.

---------

Co-authored-by: JerrettDavis <mxjerrett@gmail.com>

headroom/cli/proxy.py

@@ -351,6 +351,20 @@ def dashboard(port: int, no_open: bool) -> None:
         "every discovered extension. Env: HEADROOM_PROXY_EXTENSIONS."
     ),
 )
+@click.option(
+    "--compressor",
+    "compressor",
+    multiple=True,
+    envvar="HEADROOM_COMPRESSORS",
+    help=(
+        "Restrict the active built-in compressors to the named set (opt-in). "
+        "Repeat the flag or pass a comma-separated list; recognized names are "
+        "smart_crusher, kompress, code_aware, search, log, tabular, config, "
+        "html, image. Unselected built-ins are disabled; '*' selects all. "
+        "Omit to keep every compressor enabled (default). "
+        "Env: HEADROOM_COMPRESSORS."
+    ),
+)
 @click.option(
     "--no-subscription-tracking",
     is_flag=True,
@@ -926,6 +940,7 @@ def proxy(
     lossless: bool,
     no_ccr_proactive_expansion: bool,
     proxy_extension: tuple[str, ...],
+    compressor: tuple[str, ...],
     no_subscription_tracking: bool,
     subscription_poll_interval: int | None,
     retry_max_attempts: int | None,
@@ -1203,6 +1218,13 @@ def proxy(
             [part.strip() for chunk in proxy_extension for part in chunk.split(",") if part.strip()]
             or None
         ),
+        # Same flatten-and-split shape as proxy_extensions, but a set: order
+        # and duplicates don't matter for a selection. None when nothing was
+        # supplied, which leaves every built-in compressor enabled.
+        compressors=(
+            {part.strip() for chunk in compressor for part in chunk.split(",") if part.strip()}
+            or None
+        ),
         subscription_tracking_enabled=not no_subscription_tracking,
         subscription_poll_interval_s=(
             subscription_poll_interval if subscription_poll_interval is not None else 300

headroom/proxy/models.py

@@ -321,6 +321,17 @@ class ProxyConfig:
     # CLI: --proxy-extension <name1,name2>; env: HEADROOM_PROXY_EXTENSIONS.
     proxy_extensions: list[str] | None = None
 
+    # Compressor selection (opt-in narrowing of the built-in compressor set).
+    # None (the default) leaves EVERY built-in compressor enabled — byte-
+    # identical to today. When a set is given, only the named recognized
+    # built-ins {smart_crusher, kompress, code_aware, search, log, tabular,
+    # config, html, image} stay enabled and the rest are disabled at the
+    # ContentRouterConfig `enable_*` seam; `"*"` enables all. Names that are
+    # not recognized built-ins are ignored here (reserved for the
+    # `headroom.compressor` registry). CLI: --compressor <name1,name2>
+    # (repeatable); env: HEADROOM_COMPRESSORS.
+    compressors: set[str] | None = None
+
     # Fallback
     fallback_enabled: bool = False
     fallback_provider: str | None = None

headroom/proxy/server.py

@@ -654,6 +654,49 @@ def _provider_httpx_client_options(
     return config.http2 and not config.http_proxy, client_kwargs
 
 
+# Recognized built-in compressor names → the `ContentRouterConfig` `enable_*`
+# flag that gates each one. This is the whole selection surface: a `--compressor`
+# selection is mapped onto these existing flags with ZERO new dispatch logic, so
+# the router's if/elif built-in dispatch stays byte-identical. Names outside this
+# map (e.g. a third-party `headroom.compressor` entry point) are intentionally
+# ignored here — they belong to the registry, not the built-in enable_* seam.
+BUILTIN_COMPRESSOR_FLAGS: dict[str, str] = {
+    "smart_crusher": "enable_smart_crusher",
+    "kompress": "enable_kompress",
+    "code_aware": "enable_code_aware",
+    "search": "enable_search_compressor",
+    "log": "enable_log_compressor",
+    "tabular": "enable_tabular_compressor",
+    "config": "enable_config_compressor",
+    "html": "enable_html_extractor",
+    "image": "enable_image_optimizer",
+}
+
+
+def _apply_compressor_selection(
+    router_config: ContentRouterConfig,
+    compressors: set[str] | None,
+) -> None:
+    """Narrow the built-in compressor set on ``router_config`` in place.
+
+    ``compressors is None`` (the default) is a no-op: every ``enable_*`` flag
+    keeps its dataclass default, so behavior is byte-identical to today. When a
+    selection is given, each recognized built-in in :data:`BUILTIN_COMPRESSOR_FLAGS`
+    is enabled iff it (or the wildcard ``"*"``) was selected, and disabled
+    otherwise. Unrecognized names are ignored (reserved for the compressor
+    registry). This maps a selection onto the existing flags without adding any
+    dispatch logic.
+    """
+    if compressors is None:
+        return
+    selected = {name.strip() for name in compressors if name.strip()}
+    if not selected:
+        return
+    select_all = "*" in selected
+    for name, flag in BUILTIN_COMPRESSOR_FLAGS.items():
+        setattr(router_config, flag, select_all or name in selected)
+
+
 class HeadroomProxy(
     StreamingMixin,
     AnthropicHandlerMixin,
@@ -753,6 +796,12 @@ class HeadroomProxy(
             force_kompress_all=config.force_kompress_all,
             lossless=config.lossless,
         )
+        # Compressor selection (opt-in). None keeps every built-in enabled
+        # (default, byte-identical to today); a selection maps the recognized
+        # built-in names onto the `enable_*` flags just constructed above.
+        # Runs BEFORE the disable_kompress override below so that flag stays
+        # authoritative for turning Kompress off.
+        _apply_compressor_selection(router_config, config.compressors)
         # No-CCR lossless mode: compress tool outputs with format-native
         # lossless compaction and marker-free SmartCrusher, and suppress every
         # retrieval marker + the retrieve-tool injection so no MCP round-trip is

headroom/settings_store.py

@@ -355,6 +355,16 @@ SETTINGS: tuple[SettingField, ...] = (
         help="Disable (false) or force-enable (true) Kompress for the OpenAI/Codex pipeline only. Unset = inherit.",
         tier="advanced",
     ),
+    SettingField(
+        "HEADROOM_COMPRESSORS",
+        "compressors",
+        "Enabled compressors",
+        "Compression",
+        "csv-list",
+        default=None,
+        help="Comma-separated opt-in built-in compressor names ('*' enables all built-ins).",
+        tier="advanced",
+    ),
     # --- CCR (experimental read-maturation) ---
     SettingField(
         "HEADROOM_READ_MATURATION",

headroom/transforms/content_router.py

@@ -62,6 +62,12 @@ from ..tokenizer import Tokenizer
 from ..tokenizers.estimator import EstimatingTokenCounter
 from . import mixed_content as _mixed_content
 from .base import Transform
+from .compressor_registry import (
+    CompressInput,
+    CompressorDescriptor,
+    CompressorRegistry,
+    CompressOutput,
+)
 from .content_detector import (
     ContentType,
     DetectionResult,
@@ -120,6 +126,124 @@ def _router_debug_dumps(value: Any) -> str:
     return json.dumps(value, ensure_ascii=False, default=str, separators=(",", ":"))
 
 
+# ── Built-in compressor inventory (registry metadata only) ────────────────────
+# Declarative capability metadata for each built-in compressor. The registry is
+# an *inventory*: built-ins are still constructed and dispatched by the router's
+# existing if/elif in `_apply_strategy_to_content` — these descriptors change no
+# routing. They give the (opt-in) `headroom.compressor` registry a name-
+# addressable view of what ships in-tree, alongside any third-party compressors
+# discovered from the entry-point group. The content_types / lossless /
+# cost_tier / recoverable fields are declarative (they describe the built-in's
+# typical behavior in the default CCR configuration) and are not read on the
+# request hot path today.
+_BUILTIN_COMPRESSOR_DESCRIPTORS: tuple[CompressorDescriptor, ...] = (
+    CompressorDescriptor(
+        name="smart_crusher",
+        content_types=["application/json"],
+        lossless=False,
+        cost_tier="fast",
+        recoverable=True,
+    ),
+    CompressorDescriptor(
+        name="kompress",
+        content_types=["text/plain"],
+        lossless=False,
+        cost_tier="ml",
+        recoverable=True,
+    ),
+    CompressorDescriptor(
+        name="code_aware",
+        content_types=["text/x-code"],
+        lossless=False,
+        cost_tier="fast",
+        recoverable=True,
+    ),
+    CompressorDescriptor(
+        name="search",
+        content_types=["text/x-search-results"],
+        lossless=True,
+        cost_tier="fast",
+        recoverable=True,
+    ),
+    CompressorDescriptor(
+        name="log",
+        content_types=["text/x-log"],
+        lossless=False,
+        cost_tier="fast",
+        recoverable=True,
+    ),
+    CompressorDescriptor(
+        name="tabular",
+        content_types=["text/csv"],
+        lossless=False,
+        cost_tier="fast",
+        recoverable=True,
+    ),
+    CompressorDescriptor(
+        name="config",
+        content_types=["text/x-config"],
+        lossless=False,
+        cost_tier="fast",
+        recoverable=False,
+    ),
+    CompressorDescriptor(
+        name="html",
+        content_types=["text/html"],
+        lossless=False,
+        cost_tier="fast",
+        recoverable=False,
+    ),
+    CompressorDescriptor(
+        name="image",
+        content_types=["image/*"],
+        lossless=False,
+        cost_tier="ml",
+        recoverable=False,
+    ),
+)
+
+
+class _BuiltinCompressorEntry:
+    """Registry adapter exposing a built-in's metadata under the Compressor protocol.
+
+    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 __init__(self, descriptor: CompressorDescriptor) -> None:
+        self._descriptor = descriptor
+
+    @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 _build_compressor_registry() -> CompressorRegistry:
+    """Build the router's compressor registry: built-in inventory + 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.
+    """
+    registry = CompressorRegistry()
+    for descriptor in _BUILTIN_COMPRESSOR_DESCRIPTORS:
+        registry.register(_BuiltinCompressorEntry(descriptor))
+    # 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.
+    registry.discover()
+    return registry
+
+
 def _tool_call_args_text(raw: Any) -> str:
     """Compact, query-usable text from a tool call's args.
 
@@ -1360,6 +1484,18 @@ class ContentRouter(Transform):
             self.config.smart_crusher_lossless_only = True
         self._observer = observer
 
+        # Name-addressable compressor inventory: built-in metadata + opt-in
+        # discovery of `headroom.compressor` entry points. Inventory only —
+        # built-ins are still constructed and dispatched by the if/elif below,
+        # so this changes no routing. Exposed for selection/routing wiring in a
+        # 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()
+        except Exception as exc:  # noqa: BLE001 - inventory is non-critical
+            logger.debug("compressor registry unavailable: %s", exc)
+            self.compressor_registry = CompressorRegistry()
+
         # Lazy-loaded compressors
         self._code_compressor: Any = None
         self._smart_crusher: Any = None

tests/test_compressor_selection.py

@@ -0,0 +1,254 @@
+"""Tests for compressor selection (swap-OUT built-ins) + registry inventory.
+
+Covers the two behavior-safe halves of the router/registry integration:
+
+  1. Selection surface: ``ProxyConfig.compressors`` mapped onto the router's
+     ``enable_*`` flags at the proxy seam (``_apply_compressor_selection``).
+     Default (``None``) must leave every flag at its dataclass default so the
+     request path is byte-identical to today.
+  2. Registry inventory: ``ContentRouter`` builds a ``CompressorRegistry`` with a
+     metadata-only entry for every built-in plus opt-in entry-point discovery,
+     without changing dispatch.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+import pytest
+
+from headroom.proxy.models import ProxyConfig
+from headroom.proxy.server import (
+    BUILTIN_COMPRESSOR_FLAGS,
+    _apply_compressor_selection,
+)
+from headroom.transforms import compressor_registry as cr_module
+from headroom.transforms.compressor_registry import (
+    CompressInput,
+    CompressorDescriptor,
+    CompressOutput,
+)
+from headroom.transforms.content_router import (
+    _BUILTIN_COMPRESSOR_DESCRIPTORS,
+    ContentRouter,
+    ContentRouterConfig,
+    _build_compressor_registry,
+)
+
+# The nine recognized built-in selection names.
+_BUILTIN_NAMES = {
+    "smart_crusher",
+    "kompress",
+    "code_aware",
+    "search",
+    "log",
+    "tabular",
+    "config",
+    "html",
+    "image",
+}
+
+
+def _enable_flags(config: ContentRouterConfig) -> dict[str, bool]:
+    return {flag: getattr(config, flag) for flag in BUILTIN_COMPRESSOR_FLAGS.values()}
+
+
+# ─────────────────────────── selection mapping ───────────────────────────────
+
+
+def test_flag_map_covers_the_nine_recognized_names() -> None:
+    assert set(BUILTIN_COMPRESSOR_FLAGS) == _BUILTIN_NAMES
+
+
+def test_flag_map_targets_real_config_fields() -> None:
+    fields = {f.name for f in dataclasses.fields(ContentRouterConfig)}
+    for flag in BUILTIN_COMPRESSOR_FLAGS.values():
+        assert flag in fields, f"{flag} is not a ContentRouterConfig field"
+
+
+def test_none_selection_is_a_noop_byte_identical_defaults() -> None:
+    """Default (no --compressor) must leave every enable_* at its default."""
+    baseline = _enable_flags(ContentRouterConfig())
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, None)
+    assert _enable_flags(config) == baseline
+
+
+def test_empty_selection_is_a_noop() -> None:
+    baseline = _enable_flags(ContentRouterConfig())
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, set())
+    assert _enable_flags(config) == baseline
+
+
+def test_whitespace_only_selection_is_a_noop() -> None:
+    baseline = _enable_flags(ContentRouterConfig())
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {"", "  "})
+    assert _enable_flags(config) == baseline
+
+
+def test_single_selection_enables_only_that_builtin() -> None:
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {"kompress"})
+    flags = _enable_flags(config)
+    assert flags["enable_kompress"] is True
+    for name, flag in BUILTIN_COMPRESSOR_FLAGS.items():
+        if name != "kompress":
+            assert flags[flag] is False, f"{flag} should be disabled"
+
+
+def test_multi_selection_enables_exactly_those() -> None:
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {"smart_crusher", "log"})
+    flags = _enable_flags(config)
+    enabled = {name for name, flag in BUILTIN_COMPRESSOR_FLAGS.items() if flags[flag]}
+    assert enabled == {"smart_crusher", "log"}
+
+
+def test_selecting_code_aware_enables_it_even_though_it_defaults_off() -> None:
+    # enable_code_aware defaults to False; an explicit selection turns it on.
+    config = ContentRouterConfig()
+    assert config.enable_code_aware is False
+    _apply_compressor_selection(config, {"code_aware"})
+    assert config.enable_code_aware is True
+
+
+def test_wildcard_enables_all_builtins() -> None:
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {"*"})
+    assert all(_enable_flags(config).values())
+
+
+def test_selection_strips_whitespace() -> None:
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {" kompress ", " search"})
+    flags = _enable_flags(config)
+    enabled = {name for name, flag in BUILTIN_COMPRESSOR_FLAGS.items() if flags[flag]}
+    assert enabled == {"kompress", "search"}
+
+
+def test_only_external_name_disables_all_builtins() -> None:
+    # Selecting only a non-builtin (an external/registry name) disables every
+    # recognized built-in — the opt-in "exactly these" contract.
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {"my_external_compressor"})
+    assert not any(_enable_flags(config).values())
+
+
+def test_unrecognized_names_are_ignored_but_recognized_still_apply() -> None:
+    config = ContentRouterConfig()
+    _apply_compressor_selection(config, {"kompress", "does_not_exist"})
+    flags = _enable_flags(config)
+    assert flags["enable_kompress"] is True
+    # No crash / no spurious flag creation for the unknown name.
+
+
+# ─────────────────────────── ProxyConfig field ───────────────────────────────
+
+
+def test_proxyconfig_compressors_defaults_to_none() -> None:
+    assert ProxyConfig().compressors is None
+
+
+def test_proxyconfig_accepts_compressor_set() -> None:
+    config = ProxyConfig(compressors={"kompress", "smart_crusher"})
+    assert config.compressors == {"kompress", "smart_crusher"}
+
+
+# ─────────────────────────── registry inventory ──────────────────────────────
+
+
+def test_builtin_descriptors_cover_the_nine_names() -> None:
+    names = {d.name for d in _BUILTIN_COMPRESSOR_DESCRIPTORS}
+    assert names == _BUILTIN_NAMES
+
+
+def test_builtin_descriptor_names_match_selection_flag_map() -> None:
+    names = {d.name for d in _BUILTIN_COMPRESSOR_DESCRIPTORS}
+    assert names == set(BUILTIN_COMPRESSOR_FLAGS)
+
+
+def test_builtin_descriptors_have_valid_cost_tiers() -> None:
+    for d in _BUILTIN_COMPRESSOR_DESCRIPTORS:
+        assert d.cost_tier in cr_module.COST_TIERS
+
+
+def test_build_registry_registers_all_builtins() -> None:
+    registry = _build_compressor_registry()
+    assert set(registry.names()) >= _BUILTIN_NAMES
+
+
+def test_content_router_exposes_populated_registry() -> None:
+    router = ContentRouter(ContentRouterConfig())
+    assert set(router.compressor_registry.names()) >= _BUILTIN_NAMES
+
+
+def test_registry_inventory_does_not_enable_selection() -> None:
+    # Inventory is metadata only: with no selection resolved, nothing is active.
+    registry = _build_compressor_registry()
+    assert registry.active(None) == []
+
+
+def test_builtin_entry_compress_is_a_guard() -> None:
+    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"))
+
+
+def test_discovery_merges_external_compressor(monkeypatch: pytest.MonkeyPatch) -> None:
+    """A discovered `headroom.compressor` entry point joins the built-in inventory."""
+
+    class _FakeExternal:
+        @property
+        def descriptor(self) -> CompressorDescriptor:
+            return CompressorDescriptor(
+                name="fake_external",
+                content_types=["text/plain"],
+                lossless=True,
+                cost_tier="fast",
+                recoverable=False,
+            )
+
+        def compress(self, inp: CompressInput) -> CompressOutput:
+            return CompressOutput(
+                content=inp.content,
+                tokens_before=1,
+                tokens_after=1,
+                lossless=True,
+            )
+
+    class _FakeEntry:
+        name = "fake_external"
+
+        def load(self) -> type[_FakeExternal]:
+            return _FakeExternal
+
+    def _fake_entry_points(*, group: str) -> list[_FakeEntry]:
+        assert group == cr_module.ENTRY_POINT_GROUP
+        return [_FakeEntry()]
+
+    monkeypatch.setattr(cr_module.importlib.metadata, "entry_points", _fake_entry_points)
+
+    registry = _build_compressor_registry()
+    assert "fake_external" in registry.names()
+    assert set(registry.names()) >= _BUILTIN_NAMES
+    # The external one is selectable/active; built-ins are inventory-only.
+    active = registry.active({"fake_external"})
+    assert [c.descriptor.name for c in active] == ["fake_external"]
+
+
+def test_router_construction_is_unchanged_by_default_registry(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    # Even if discovery raises, the router still constructs with an empty-but-
+    # present registry (fail-open) — never breaking the request path.
+    def _boom(*, group: str) -> list[object]:
+        raise RuntimeError("discovery blew up")
+
+    monkeypatch.setattr(cr_module.importlib.metadata, "entry_points", _boom)
+    router = ContentRouter(ContentRouterConfig())
+    # discover() itself is fail-open, so built-ins are still registered.
+    assert set(router.compressor_registry.names()) >= _BUILTIN_NAMES