fix(proxy): time-cap the compression timeout-debt quarantine (#2360) (#2412)

## Description

Fixes #2360.

The proxy runs compression on a bounded thread-pool executor with a
per-request deadline. Because Python cannot preempt a worker after its
`asyncio.wait_for` times out, the code quarantines new compression while
a timed-out worker is still running (`_compression_timed_out_in_flight >
0`), to avoid piling more work onto a saturating executor.

The gap: that counter only decrements when the worker finally exits. A
worker that **never returns** — a hung or pathological compression of a
large frame — keeps the counter above zero forever, so the quarantine
stays open permanently and every subsequent compression raises
`CompressionQuarantinedError`. On Codex WS this is exactly what #2360
reports: one 5s timeout, then Token Savings pinned at ~0% with no
recovery, even though the machine is fine.

The "parity with direct upstream" nature of the accounting was correct;
the only missing piece is an upper bound on how long a single stuck
worker may hold the quarantine.

## Fix

Add a time cap on the quarantine:

- A deadline (`_compression_quarantine_deadline`) is (re)armed on every
fresh timeout, to `now + HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS`
(default **60s**).
- The gate quarantines only while `timed_out_in_flight > 0` **and** `now
< deadline`. Once the deadline lapses with no new timeouts, the worker
is presumed leaked/abandoned and compression resumes. The release is
counted once (a `"released"` quarantine metric + a warning), and the
deadline is cleared so it is not re-counted on every later request.
- The bounded executor still caps thread growth, and any new timeout
re-arms the quarantine, so ongoing genuine slowness keeps quarantining
while a single hung worker cannot pin it forever.

This preserves the original protection (a burst of slow compressions
still quarantines) while guaranteeing recovery.

## 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/proxy/server.py`: add `_compression_quarantine_deadline` /
`_compression_quarantine_max_seconds` (from
`HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS`, default 60s) and
`_compression_quarantine_releases`; arm the deadline when timeout debt
is recorded; release the quarantine (once) in the gate when the deadline
lapses.
- `tests/test_platform_stabilization_functional.py`: add a test that a
standing timed-out worker quarantines within the cap and releases
(running compression again, counted once) past it.

## 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
$ uvx ruff@0.15.17 check headroom/proxy/server.py tests/test_platform_stabilization_functional.py
All checks passed!
$ uvx mypy@1.20.2 --ignore-missing-imports headroom/proxy/server.py
Success: no issues found in 1 source file
```

## Real Behavior Proof

- Environment: Windows 11, Python 3.12, `uvx ruff@0.15.17` / `uvx
mypy@1.20.2`. A full `pytest` here imports the ML stack and OOMs this
box, so I modeled the gate/deadline state machine with a dependency-free
script and left the added `create_app` test to CI.
- Exact command / steps: simulated a standing timed-out worker, then
exercised the gate at times within the cap, past the cap, and after a
fresh timeout, plus a normal worker exit.
- Observed result: within the cap the gate quarantines (raises); past
the cap it releases exactly once and then lets compression run; a new
timeout re-arms the quarantine; a normal worker exit clears the debt.
Matching the added handler test (`_run_compression_in_executor` raises
`CompressionQuarantinedError` within the cap and returns the callable's
result past it, with `_compression_quarantine_releases == 1`).
- Not tested: a live Codex WS session hanging a real worker; the added
test drives `_run_compression_in_executor` directly with the quarantine
state set.

## 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
- [ ] 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
- [ ] New and existing unit tests pass locally with my changes
- [x] I have updated the CHANGELOG.md if applicable

## Additional Notes

The default cap (60s) is deliberately well above a normal
slow-but-completing compression so the original saturation protection is
unchanged in practice; it only ever fires for a worker that has run far
past its deadline. Tunable via
`HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS`. The `"released"`
quarantine metric and a one-time warning make the recovery observable.
The "unit tests pass locally" box is unchecked because the added
`create_app` test imports the ML stack (OOM on this box); it runs under
the normal CI job, and the state machine is verified by the standalone
proof above.

headroom/proxy/server.py

@@ -1156,6 +1156,19 @@ class HeadroomProxy(
         self._compression_timed_out_in_flight_max: int = 0
         self._compression_quarantine_activations: int = 0
         self._compression_quarantine_skips: int = 0
+        # Time cap on the timeout-debt quarantine. Python cannot preempt a
+        # worker, so one that never returns (a hung/pathological compression)
+        # would keep ``_compression_timed_out_in_flight > 0`` forever and pin the
+        # quarantine open — zeroing out ALL further compression (#2360). The
+        # deadline is (re)set on every fresh timeout; once it lapses with no new
+        # timeouts the leaked worker is presumed abandoned and compression
+        # resumes. The bounded executor still caps thread growth, and any new
+        # timeout re-arms the quarantine.
+        self._compression_quarantine_deadline: float = 0.0
+        self._compression_quarantine_max_seconds: float = _get_env_float(
+            "HEADROOM_COMPRESSION_QUARANTINE_MAX_SECONDS", 60.0
+        )
+        self._compression_quarantine_releases: int = 0
         self._compression_metrics_lock = threading.Lock()
 
         # Backend for Anthropic API (direct, LiteLLM, or any-llm)
@@ -1413,12 +1426,36 @@ class HeadroomProxy(
             ``asyncio.TimeoutError`` subclass) if a prior timed-out worker is
             still running.
         """
+        now = time.monotonic()
         with self._compression_metrics_lock:
             timed_out_in_flight = self._compression_timed_out_in_flight
-            if timed_out_in_flight > 0:
+            quarantined = timed_out_in_flight > 0 and now < self._compression_quarantine_deadline
+            # Debt outlived the cap: presume the worker leaked/hung and stop
+            # blocking on it. Count the release once per lapse (while debt stands
+            # and the deadline has passed) so operators can see it happened.
+            released = (
+                timed_out_in_flight > 0
+                and not quarantined
+                and self._compression_quarantine_deadline > 0.0
+            )
+            if quarantined:
                 self._compression_quarantine_skips += 1
+            if released:
+                self._compression_quarantine_releases += 1
+                # Clear the deadline so the release is recorded once, not on
+                # every subsequent request until the worker (maybe never) exits.
+                self._compression_quarantine_deadline = 0.0
 
-        if timed_out_in_flight > 0:
+        if released:
+            self.metrics.record_compression_quarantine("released")
+            logger.warning(
+                "Compression quarantine released after %.0fs cap; %d timed-out "
+                "worker(s) presumed leaked. Compression resumes.",
+                self._compression_quarantine_max_seconds,
+                timed_out_in_flight,
+            )
+
+        if quarantined:
             self.metrics.record_compression_quarantine("skipped")
             raise CompressionQuarantinedError(
                 f"compression quarantined: {timed_out_in_flight} timed-out worker(s) still running"
@@ -1457,6 +1494,12 @@ class HeadroomProxy(
                 self._compression_timed_out_in_flight_max,
                 self._compression_timed_out_in_flight,
             )
+            # (Re)arm the quarantine time cap on every fresh timeout, so ongoing
+            # slowness keeps quarantining while a single leaked worker cannot
+            # hold it past the cap (#2360).
+            self._compression_quarantine_deadline = (
+                time.monotonic() + self._compression_quarantine_max_seconds
+            )
             state["timeout_debt_recorded"] = True
             if was_clear:
                 self._compression_quarantine_activations += 1

tests/test_platform_stabilization_functional.py

@@ -168,3 +168,35 @@ def test_v1_compress_real_json_tool_payload_reduces_tokens(monkeypatch) -> None:
     assert body["tokens_saved"] > 0
     assert body["compression_ratio"] < 1.0
     assert body["transforms_applied"], body
+
+
+def test_compression_quarantine_releases_after_time_cap(monkeypatch) -> None:
+    """A leaked/hung timed-out worker must not pin the quarantine open forever:
+    once the time cap lapses, compression resumes and the release is counted
+    once (#2360)."""
+    import asyncio
+
+    from headroom.proxy.server import CompressionQuarantinedError
+
+    monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
+    app = create_app(_proxy_config())
+    proxy = app.state.proxy
+
+    # Simulate a timed-out worker that is still running (debt standing).
+    proxy._compression_timed_out_in_flight = 1
+
+    # Within the cap: new compression is quarantined.
+    proxy._compression_quarantine_deadline = time.monotonic() + 1000.0
+    with pytest.raises(CompressionQuarantinedError):
+        asyncio.run(proxy._run_compression_in_executor(lambda: "unused", timeout=5.0))
+    assert proxy._compression_quarantine_releases == 0
+
+    # Past the cap: the worker is presumed leaked and compression runs again;
+    # the release is recorded once.
+    proxy._compression_quarantine_deadline = time.monotonic() - 1.0
+    assert asyncio.run(proxy._run_compression_in_executor(lambda: "ran", timeout=5.0)) == "ran"
+    assert proxy._compression_quarantine_releases == 1
+
+    # A subsequent request is not re-counted and still runs.
+    assert asyncio.run(proxy._run_compression_in_executor(lambda: "ok", timeout=5.0)) == "ok"
+    assert proxy._compression_quarantine_releases == 1