fix(proxy): keep large compression results on the critical path (#296) (#1352)

## Description

In Anthropic token mode, compression appears to complete in the
transform pipeline (`proxy.log` shows `Pipeline complete: ... saved N
tokens`), but ~30s later the proxy times out in
`compression_first_stage` and forwards the **original** uncompressed
request — so `/stats` and `recent_requests` show `tokens_saved: 0`,
`savings_percent: 0.0`, `transforms_applied: []`,
`optimization_latency_ms: ~31,000`. It starts once a compacted Claude
Code transcript grows to ~367k–425k input tokens.

Root cause: after the pipeline finishes, `TransformPipeline.apply` runs
a **telemetry-only** waste-signal re-parse of the *original* messages
(`parse_messages`) on the critical path. On a
several-hundred-thousand-token transcript that diagnostic parse can take
tens of seconds and blow the Anthropic compression timeout — so the
already-computed compression result is discarded and the proxy fails
open with the original request.

Fix: skip waste-signal detection above
`MAX_WASTE_SIGNAL_DETECTION_TOKENS` (100k). The diagnostic never changes
the compression result, so skipping it on huge requests keeps the result
on the critical path. Smaller requests are unaffected.

(The earlier diagnostics PRs #303/#304 — both merged — added the
`request_id`/exception-type logging that made this root cause visible.
This is the focused follow-up fix.)

Closes #296

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

- `headroom/transforms/pipeline.py`: gate waste-signal detection on
`tokens_before <= waste_signal_token_limit` (default
`MAX_WASTE_SIGNAL_DETECTION_TOKENS = 100_000`, overridable via kwarg);
above the limit, log a debug line and skip. Extracted the "saved enough"
predicate and a named `_MIN_TOKENS_SAVED_FOR_WASTE_SIGNALS` constant
(was a bare `100`).
- `tests/test_transforms/test_pipeline_waste_signal_limit.py`: new
regression test — above the limit the waste-signal parse is skipped and
the compression result is preserved; below the limit it still runs.
- `CHANGELOG.md`: Unreleased → Bug Fixes entry.

## Testing

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

### Test Output

```text
$ uv run pytest tests/test_canonical_pipeline.py tests/test_proxy_anthropic_compression_diagnostics.py tests/test_transforms/test_pipeline_waste_signal_limit.py -q
12 passed in 35.97s

$ uv run ruff check headroom/transforms/pipeline.py tests/test_transforms/test_pipeline_waste_signal_limit.py
All checks passed!

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

#### TDD verification (RED → GREEN)

RED — new test with the prod fix reverted (waste-signal detection still
runs on the large request):
```text
E   AssertionError: waste-signal parse must be skipped above the limit
    assert True is False
1 failed, 1 passed in 0.17s
```
(The 1 passing on red is the below-limit no-regression guard.)

GREEN — with the fix applied:
```text
2 passed in 0.12s
```

## Real Behavior Proof

- Environment: Linux, Python 3.13, headroom @ this branch.
- Exact command / steps: drive `TransformPipeline.apply` with a stub
transform that compresses and a tracked `parse_messages`, sizing the
request above vs below the limit:
- `tokens_before=200_000`, limit `100_000` → `parse_messages` is **not**
called; the result still carries `transforms_applied=['test:shrink']`
and `tokens_after < tokens_before` (pre-fix: `parse_messages` ran, which
is the slow step the timeout killed, discarding this result).
- `tokens_before=10_000`, limit `100_000` → `parse_messages` **is**
called (diagnostic preserved for normal requests).
- Observed result: above the limit the compression result reaches the
caller without the diagnostic parse that caused the timeout; below the
limit behavior is unchanged.
- Not tested: the live multi-hundred-k-token Claude Code session against
Anthropic that originally tripped the wall-clock timeout (needs a real
large transcript + provider); the causal chain (slow `parse_messages` on
the critical path → timeout → discard) is covered deterministically by
the unit test.

## 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] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

The limit is overridable per-call via the `waste_signal_token_limit`
kwarg, so callers that want the diagnostic on larger requests can opt
back in. Waste-signal data is telemetry only (OTel metrics) — it never
affects the compressed output sent upstream.

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Co-authored-by: JD Davis <mxjerrett@gmail.com>

CHANGELOG.md

@@ -34,6 +34,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ### Bug Fixes
 
+* **proxy:** stop discarding a finished compression on very large requests. After the transform pipeline completed, a telemetry-only waste-signal re-parse of the *original* messages ran on the critical path; on huge Claude Code transcripts (~400k tokens) that parse could exceed the Anthropic compression timeout, so the proxy failed open and forwarded the uncompressed request despite "Pipeline complete" logging real savings (`tokens_saved: 0`, `transforms_applied: []`, ~31s latency). Waste-signal detection is now skipped above `MAX_WASTE_SIGNAL_DETECTION_TOKENS` (100k) so the compression result stays on the critical path ([#296](https://github.com/chopratejas/headroom/issues/296)).
 * **codex:** retag existing Codex threads when `headroom init` injects the `headroom` provider, so Codex Desktop history stays visible. Codex filters its sidebar/search by the active `model_provider`; the init path set `model_provider = "headroom"` without retagging, so existing native `openai` threads disappeared from the menu (data was never deleted, only hidden). `_ensure_codex_provider` now reconciles thread tags openai→headroom, matching what the install and `wrap` paths already do; `headroom unwrap codex` handles the revert direction ([#961](https://github.com/chopratejas/headroom/issues/961)).
 * **install:** stop duplicating the container ENTRYPOINT in the `persistent-docker` runtime command. The published image already runs `headroom proxy` as its ENTRYPOINT, but `build_runtime_command` re-added `headroom proxy` after the image name, so the container ran `headroom proxy headroom proxy --host 0.0.0.0 …` and Click aborted with "Got unexpected extra arguments (headroom proxy)" — the deployment never became ready and rollback left nothing running. The runtime command now appends only the proxy flags ([#833](https://github.com/chopratejas/headroom/issues/833)).
 * **code:** keep Python `from __future__` imports before executable code during AST compression and validate compressed Python with `compile(..., "exec")` so compile-time syntax rules are enforced ([#1233](https://github.com/chopratejas/headroom/issues/1233)).

headroom/transforms/pipeline.py

@@ -30,6 +30,18 @@ if TYPE_CHECKING:
 
 logger = logging.getLogger(__name__)
 
+# Waste-signal detection re-parses the *original* messages for telemetry only
+# (it never changes the compression result). On very large transcripts that
+# extra parse can take tens of seconds and blow the Anthropic compression
+# timeout, making the proxy fail open and discard an already-computed
+# compression (#296). Skip the diagnostic above this size to keep the result
+# on the critical path.
+MAX_WASTE_SIGNAL_DETECTION_TOKENS = 100_000
+
+# A token saving below this is treated as noise — waste-signal detection only
+# runs when compression saved more than this many tokens.
+_MIN_TOKENS_SAVED_FOR_WASTE_SIGNALS = 100
+
 _N = TypeVar("_N", int, float)
 
 
@@ -220,6 +232,9 @@ class TransformPipeline:
         """
         record_metrics = kwargs.pop("record_metrics", True)
         waste_messages = kwargs.pop("waste_messages", None)
+        waste_signal_token_limit = int(
+            kwargs.pop("waste_signal_token_limit", MAX_WASTE_SIGNAL_DETECTION_TOKENS)
+        )
         tokenizer = self._get_tokenizer(model)
         provider_name = self._provider_name()
 
@@ -439,7 +454,21 @@ class TransformPipeline:
             # pass a richer waste_messages list that is parsed instead — it is
             # telemetry-only and never transformed.
             waste_signals: WasteSignals | None = None
-            if tokens_before > tokens_after and (tokens_before - tokens_after) > 100:
+            saved_enough = (
+                tokens_before > tokens_after
+                and (tokens_before - tokens_after) > _MIN_TOKENS_SAVED_FOR_WASTE_SIGNALS
+            )
+            if saved_enough and tokens_before > waste_signal_token_limit:
+                # Telemetry-only re-parse would risk the compression timeout on a
+                # request this large (#296); skip it and keep the result.
+                logger.debug(
+                    "%sSkipping waste-signal detection for %d-token request "
+                    "(limit=%d) to keep the compression result on the critical path",
+                    log_prefix,
+                    tokens_before,
+                    waste_signal_token_limit,
+                )
+            elif saved_enough:
                 try:
                     from ..parser import parse_messages
 

tests/test_transforms/test_pipeline_waste_signal_limit.py

@@ -0,0 +1,95 @@
+"""Waste-signal detection must not discard a finished compression (#296).
+
+On very large Claude Code transcripts the telemetry-only waste-signal re-parse
+of the *original* messages can take tens of seconds and blow the Anthropic
+compression timeout, making the proxy fail open and forward the original
+request even though compression already succeeded. The pipeline now skips that
+diagnostic above ``MAX_WASTE_SIGNAL_DETECTION_TOKENS`` so the compression
+result stays on the critical path.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from headroom.config import HeadroomConfig, TransformResult
+from headroom.transforms.base import Transform
+from headroom.transforms.pipeline import TransformPipeline
+
+
+class _FakeTokenizer:
+    """Reports a fixed token count for the original messages so the test can
+    drive ``tokens_before`` above or below the waste-signal limit."""
+
+    def __init__(self, before: int, after: int) -> None:
+        self._before = before
+        self._after = after
+
+    def count_messages(self, messages: list[dict[str, Any]]) -> int:
+        # The compressed message carries the marker "compressed".
+        if any(m.get("content") == "compressed" for m in messages):
+            return self._after
+        return self._before
+
+    def count_text(self, text: Any) -> int:
+        return len(str(text))
+
+
+class _ShrinkTransform(Transform):
+    name = "test_shrink"
+
+    def apply(
+        self, messages: list[dict[str, Any]], tokenizer: Any, **kwargs: Any
+    ) -> TransformResult:
+        optimized = [dict(m) for m in messages]
+        optimized[-1] = {**optimized[-1], "content": "compressed"}
+        return TransformResult(
+            messages=optimized,
+            tokens_before=tokenizer.count_messages(messages),
+            tokens_after=tokenizer.count_messages(optimized),
+            transforms_applied=["test:shrink"],
+        )
+
+
+def _run(monkeypatch, *, before: int, after: int, limit: int):
+    """Run the pipeline with a stub transform; return (result, parse_called)."""
+    pipeline = TransformPipeline(HeadroomConfig())
+    pipeline.transforms = [_ShrinkTransform()]
+    monkeypatch.setattr(pipeline, "_get_tokenizer", lambda _model: _FakeTokenizer(before, after))
+
+    parse_called = False
+
+    def _tracked_parse_messages(*args: Any, **kwargs: Any):
+        nonlocal parse_called
+        parse_called = True
+        return [], {}, None
+
+    monkeypatch.setattr("headroom.parser.parse_messages", _tracked_parse_messages)
+
+    messages = [{"role": "user", "content": "x" * 1000}]
+    result = pipeline.apply(
+        messages,
+        model="claude-3-5-sonnet",
+        model_limit=1_000_000,
+        record_metrics=False,
+        waste_signal_token_limit=limit,
+    )
+    return result, parse_called
+
+
+def test_large_request_skips_waste_signal_and_keeps_compression(monkeypatch):
+    """Above the limit, waste-signal detection is skipped but the compression
+    result is preserved (the bug discarded it via the timeout)."""
+    result, parse_called = _run(monkeypatch, before=200_000, after=180_000, limit=100_000)
+
+    assert parse_called is False, "waste-signal parse must be skipped above the limit"
+    assert "test:shrink" in result.transforms_applied
+    assert result.tokens_after < result.tokens_before
+    assert result.messages[-1]["content"] == "compressed"
+
+
+def test_small_request_still_runs_waste_signal_detection(monkeypatch):
+    """Below the limit, the diagnostic still runs (no behavior change)."""
+    _result, parse_called = _run(monkeypatch, before=10_000, after=5_000, limit=100_000)
+
+    assert parse_called is True, "waste-signal parse must still run below the limit"