fix(kompress): accept ccr_original on the remote compressor (#3162)

## Description

From a user's proxy log (Copilot Chat 0.61.0 on Windows, VS Code
1.133.0, Headroom 0.36.x). This appears on **every single request**:

```
WARNING Kompress failed: RemoteKompressCompressor.compress() got an
        unexpected keyword argument 'ccr_original'
INFO    [router] route_counts={'ratio_too_high': 1, 'cache_miss': 1} compressed=0 frozen=1 msgs=2
INFO    Transform content_router: 1611 -> 1611 tokens (saved 0) [48.3ms]
INFO    PERF model=... tok_before=1623 tok_after=1623 tok_saved=0 tool_saved=0 savings=none
```

`RemoteKompressCompressor`'s module docstring promises the class
"mirrors `KompressCompressor`'s public surface (`is_ready` / `preload` /
`ensure_background_load` / `compress`), so it is a drop-in at the
ContentRouter seam". That promise lapsed — the local `compress` gained a
`ccr_original` keyword and the remote one did not.

`ContentRouter._try_ml_compressor` passes `ccr_original` whenever custom
tags are protected. The comment there reads:

> Only set it when tags were protected so callers/compressors that don't
accept the kwarg are unaffected on the common path.

That assumption is wrong. The remote compressor **is** affected: the
call raises `TypeError`, which the surrounding broad `except Exception`
catches and downgrades to `logger.warning("Kompress failed: %s", e)`.
The request then forwards uncompressed and the proxy reports success.

**The blast radius is the entire deployment, not one request.**
`_get_kompress` returns the remote compressor *ahead of* every local
path, so on any install with `HEADROOM_KOMPRESS_ENDPOINT` set —
precisely the sandboxed/enterprise deployment this class exists to serve
— ML compression was silently disabled while every dashboard read
"working, 0 tokens saved".

Closes #

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Changes Made

Two parts, because fixing only the crash would leave the bug
`ccr_original` exists to prevent:

- **Accept the keyword** on `RemoteKompressCompressor.compress`, so the
seam contract actually holds.
- **Honor it** — store the pre-protection text in CCR rather than the
placeholder intermediate, so a later full retrieval returns the real
block instead of `{{HEADROOM_TAG_N}}`. The endpoint's own
`original_tokens` describes `content`, so when an override is supplied
the stored text is counted locally; the common path (no override) keeps
the endpoint's count exactly as before.
- **A signature-compatibility test** over the two `compress` methods, so
this drift cannot recur silently. It compares *public* keywords only —
`_deadline_started_at` is underscore-prefixed and only ever passed by
`kompress_compressor` to itself on its recursive batch path, never
across the seam.

## Testing

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

### Test Output

```text
$ pytest tests/test_remote_kompress_dropin.py -q
8 passed in 0.25s

# Same file against pre-fix code (git stash) — reproduces the reported error:
3 failed, 5 passed
  FAILED test_remote_compress_accepts_every_local_keyword
  FAILED test_passing_ccr_original_no_longer_raises
  FAILED test_ccr_stores_the_pre_protection_text_not_the_placeholder
  E  TypeError: RemoteKompressCompressor.compress() got an unexpected
     keyword argument 'ccr_original'

$ pytest tests/ -q -k "kompress or content_router"
411 passed, 9 skipped

$ pytest tests/ -q          # this branch
6 failed, 11381 passed, 587 skipped in 446.31s

All 6 also fail on clean origin/main, same machine — pre-existing, not regressions:
  test_graceful_shutdown.py::test_run_server_installs_cancelled_error_filter
  test_learn/test_integration.py::TestCodexIntegration::test_full_pipeline
  test_release_workflows.py::test_no_native_tls_in_wheel_build_tree
  test_providers/test_deepseek.py::...v4_flash_litellm_pricing
  test_providers/test_deepseek.py::...v4_pro_litellm_pricing
  test_providers/test_deepseek.py::...cost_per_token_resolves_deepseek_v4_flash
(verified by stashing this branch and running test_deepseek.py: 3 failed, 17 passed)

$ ruff check headroom/
All checks passed!

$ mypy headroom/transforms/kompress_remote.py
Success: no issues found in 1 source file
```

## Real Behavior Proof

- **Environment:** macOS, Python 3.12.13, branch on `origin/main` @
`a3821378`.
- **Exact command / steps:** drove `RemoteKompressCompressor.compress`
with the exact kwargs `ContentRouter._try_ml_compressor` builds when
`protected` is truthy (`context`, `question`, `target_ratio`,
`allow_download`, `ccr_original`), against a stubbed HTTP client.
- **Observed result:** pre-fix that call raises `TypeError: ...
unexpected keyword argument 'ccr_original'` — byte-identical to the
user's log line. Post-fix it returns a `KompressResult`, and CCR
receives the pre-protection text (`"HEADROOM_TAG" not in stored`) with a
token count matching what was stored.
- **Not tested:** no live remote Kompress endpoint was contacted; the
HTTP client is stubbed. The end-to-end path through a running proxy
against a real `HEADROOM_KOMPRESS_ENDPOINT` has not been exercised here.

## Runtime Rollout Safety

- **Rollout-managed feature(s):** none. Affects deployments with
`HEADROOM_KOMPRESS_ENDPOINT` set.
- **Minimum rollout channel:** n/a.
- **Stable/default behavior changed:** for remote-Kompress deployments,
compression starts working again where it previously no-op'd.
Deployments without the endpoint set are untouched — they never reach
this class.
- **Kill switch / disable path:** unchanged
(`HEADROOM_KOMPRESS_ENDPOINT` unset, or `kompress_model="disabled"`).
- **Unsafe override required:** none.
- **Qualification impact:** the remote compressor's fail-open contract
is unchanged — a bad endpoint still passes content through verbatim.
- **Rollback path:** revert; behavior returns to silently-disabled
compression on remote deployments.

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

Co-authored-by: Tejas Chopra <tejas@Tejass-MacBook-Pro.local>
Co-authored-by: Claude Opus 5 <noreply@anthropic.com>

headroom/transforms/kompress_remote.py

@@ -179,7 +179,26 @@ class RemoteKompressCompressor:
         target_ratio: float | None = None,
         *,
         allow_download: bool = True,
+        ccr_original: str | None = None,
     ) -> KompressResult:
+        """Compress via the remote endpoint.
+
+        ``ccr_original`` mirrors :meth:`KompressCompressor.compress`: text to
+        store in CCR instead of ``content``, used when ``content`` is a
+        tag-protected placeholder intermediate ({{HEADROOM_TAG_N}}). It is
+        accepted here because this class promises to be a DROP-IN for the local
+        compressor at the ContentRouter seam (see the module docstring) — and it
+        was not. ContentRouter passes the kwarg whenever custom tags are
+        protected, so on any deployment with HEADROOM_KOMPRESS_ENDPOINT set,
+        every such request raised
+
+            TypeError: RemoteKompressCompressor.compress() got an unexpected
+            keyword argument 'ccr_original'
+
+        which ContentRouter caught with a broad ``except Exception`` and logged
+        as ``Kompress failed: ...``. Compression silently degraded to zero on the
+        whole deployment while the proxy kept reporting success.
+        """
         n_words = len(content.split())
         if n_words < _MIN_WORDS:
             return self._passthrough(content, n_words)
@@ -216,12 +235,24 @@ class RemoteKompressCompressor:
         # store the mapping + append the retrieval marker here — same policy and
         # marker format as KompressCompressor.compress.
         if self.config.enable_ccr and result.compression_ratio < _CCR_RATIO_GATE:
-            cache_key = store_kompress_in_ccr(content, compressed, result.original_tokens)
+            # Store the PRE-protection text when the caller supplied it. ``content``
+            # may be the tag-protected placeholder intermediate, and storing that
+            # makes a later full retrieval hand back {{HEADROOM_TAG_N}} instead of
+            # the real block — the exact loss ``ccr_original`` exists to prevent.
+            # Same resolution order as KompressCompressor.compress.
+            ccr_source = ccr_original if ccr_original is not None else content
+            # The endpoint's ``original_tokens`` describes ``content``, so it does
+            # not describe a different ``ccr_source``; count that one locally.
+            # Unchanged on the common path where no override was passed.
+            ccr_source_tokens = (
+                len(ccr_source.split()) if ccr_original is not None else result.original_tokens
+            )
+            cache_key = store_kompress_in_ccr(ccr_source, compressed, ccr_source_tokens)
             if cache_key:
                 result.cache_key = cache_key
                 # Report the source line span so a reader can tell content was
                 # compressed away rather than absent (#2586).
-                source_lines = content.count("\n") + 1
+                source_lines = ccr_source.count("\n") + 1
                 line_word = "line" if source_lines == 1 else "lines"
                 result.compressed += (
                     f"\n[{result.original_tokens} items compressed to "

tests/test_remote_kompress_dropin.py

@@ -0,0 +1,189 @@
+"""RemoteKompressCompressor must really be a drop-in for KompressCompressor.
+
+Its module docstring promises the class "mirrors KompressCompressor's public
+surface (``is_ready`` / ``preload`` / ``ensure_background_load`` / ``compress``),
+so it is a drop-in at the ContentRouter seam". That promise silently lapsed:
+the local ``compress`` gained a ``ccr_original`` keyword and the remote one did
+not.
+
+ContentRouter passes ``ccr_original`` whenever custom tags are protected. On any
+deployment with ``HEADROOM_KOMPRESS_ENDPOINT`` set — which is exactly the
+sandboxed/enterprise install the remote compressor exists for — every such
+request raised
+
+    TypeError: RemoteKompressCompressor.compress() got an unexpected keyword
+    argument 'ccr_original'
+
+ContentRouter caught it with a broad ``except Exception`` and logged
+``Kompress failed: ...`` at WARNING. The request then forwarded uncompressed
+with ``tok_saved=0`` and the proxy reported success, so the deployment lost ALL
+ML compression while every dashboard read "working, 0 saved".
+
+From a field log (Copilot Chat on Windows, 0.36.x), on every single request:
+
+    WARNING Kompress failed: RemoteKompressCompressor.compress() got an
+            unexpected keyword argument 'ccr_original'
+    INFO    [router] route_counts={...} compressed=0 frozen=1 msgs=2
+    INFO    PERF ... tok_before=1623 tok_after=1623 tok_saved=0 savings=none
+"""
+
+from __future__ import annotations
+
+import inspect
+
+import pytest
+
+from headroom.transforms.kompress_compressor import KompressCompressor
+from headroom.transforms.kompress_remote import RemoteKompressCompressor
+
+
+def _kwargs(fn) -> set[str]:
+    """Public keywords only.
+
+    ``_deadline_started_at`` is underscore-prefixed and only ever passed by
+    kompress_compressor to itself on its recursive batch path — it never crosses
+    the ContentRouter seam, so it is genuinely private and not part of the
+    drop-in contract.
+    """
+    return {
+        name
+        for name, p in inspect.signature(fn).parameters.items()
+        if name != "self"
+        and not name.startswith("_")
+        and p.kind in (p.POSITIONAL_OR_KEYWORD, p.KEYWORD_ONLY)
+    }
+
+
+# --------------------------------------------------------------------------- #
+# The contract
+# --------------------------------------------------------------------------- #
+def test_remote_compress_accepts_every_local_keyword() -> None:
+    """The drift guard. This is what would have caught the reported bug."""
+    local = _kwargs(KompressCompressor.compress)
+    remote = _kwargs(RemoteKompressCompressor.compress)
+
+    missing = local - remote
+    assert not missing, (
+        f"RemoteKompressCompressor.compress is missing {sorted(missing)}. "
+        "ContentRouter calls both through one seam, so a keyword the local "
+        "compressor accepts and the remote one does not becomes a TypeError "
+        "that ContentRouter swallows into a warning — silently disabling "
+        "compression for the whole deployment."
+    )
+
+
+@pytest.mark.parametrize("method", ["is_ready", "preload", "ensure_background_load", "compress"])
+def test_the_promised_public_surface_exists(method: str) -> None:
+    assert callable(getattr(RemoteKompressCompressor, method, None))
+
+
+# --------------------------------------------------------------------------- #
+# The reported failure, end to end through the real call shape
+# --------------------------------------------------------------------------- #
+class _FakeResponse:
+    status_code = 200
+
+    def __init__(self, payload: dict) -> None:
+        self._payload = payload
+
+    def raise_for_status(self) -> None:
+        return None
+
+    def json(self) -> dict:
+        return self._payload
+
+
+class _FakeClient:
+    def __init__(self, payload: dict) -> None:
+        self._payload = payload
+        self.calls: list[dict] = []
+
+    def post(self, url, headers=None, json=None):  # noqa: A002, ANN001
+        self.calls.append(json or {})
+        return _FakeResponse(self._payload)
+
+    def close(self) -> None:
+        return None
+
+
+def _compressor(monkeypatch, *, enable_ccr: bool, payload: dict):
+    monkeypatch.setenv("HEADROOM_KOMPRESS_ENDPOINT", "https://ml.example.invalid")
+    c = RemoteKompressCompressor("https://ml.example.invalid")
+    c._client = _FakeClient(payload)  # type: ignore[assignment]
+    c.config.enable_ccr = enable_ccr
+    return c
+
+
+ORIGINAL = "real secret block " * 20
+PLACEHOLDER = "{{HEADROOM_TAG_0}} " * 20
+
+
+def test_passing_ccr_original_no_longer_raises(monkeypatch) -> None:
+    """The bug itself: this call is what ContentRouter makes."""
+    c = _compressor(
+        monkeypatch,
+        enable_ccr=False,
+        payload={"compressed": "short", "compression_ratio": 0.2},
+    )
+
+    result = c.compress(
+        PLACEHOLDER,
+        context="",
+        question=None,
+        target_ratio=0.5,
+        allow_download=False,
+        ccr_original=ORIGINAL,
+    )
+
+    assert result.compressed == "short"
+
+
+def test_ccr_stores_the_pre_protection_text_not_the_placeholder(monkeypatch) -> None:
+    """Fixing only the TypeError would leave retrieval returning a placeholder."""
+    stored: dict = {}
+
+    def _fake_store(original, compressed, original_tokens):  # noqa: ANN001
+        stored["original"] = original
+        stored["tokens"] = original_tokens
+        return "cafebabe"
+
+    monkeypatch.setattr("headroom.transforms.kompress_remote.store_kompress_in_ccr", _fake_store)
+    c = _compressor(
+        monkeypatch,
+        enable_ccr=True,
+        payload={"compressed": "short", "compression_ratio": 0.2},
+    )
+
+    result = c.compress(PLACEHOLDER, ccr_original=ORIGINAL)
+
+    assert stored["original"] == ORIGINAL
+    assert "HEADROOM_TAG" not in stored["original"]
+    # Token count describes what was actually stored, not the placeholder.
+    assert stored["tokens"] == len(ORIGINAL.split())
+    assert result.cache_key == "cafebabe"
+
+
+def test_the_common_path_without_an_override_is_unchanged(monkeypatch) -> None:
+    stored: dict = {}
+
+    def _fake_store(original, compressed, original_tokens):  # noqa: ANN001
+        stored["original"] = original
+        stored["tokens"] = original_tokens
+        return "d00d"
+
+    monkeypatch.setattr("headroom.transforms.kompress_remote.store_kompress_in_ccr", _fake_store)
+    c = _compressor(
+        monkeypatch,
+        enable_ccr=True,
+        payload={
+            "compressed": "short",
+            "compression_ratio": 0.2,
+            "original_tokens": 999,
+        },
+    )
+
+    c.compress(ORIGINAL)
+
+    assert stored["original"] == ORIGINAL
+    # Still the endpoint's own count when no override was supplied.
+    assert stored["tokens"] == 999