fix(kompress): raise the default execution-slot wait (headroomlabs-ai#2456)

## Description

Concurrent Kompress requests currently fail open after a 25 ms
execution-slot wait even though ordinary ONNX inference can hold the
single slot for hundreds of milliseconds. This raises the existing
default wait to 3000 ms while retaining concurrency one, the
`HEADROOM_KOMPRESS_EXECUTION_TIMEOUT_MS` override, the tighter acquire
and request budgets, and passthrough after a genuine timeout.

The reproduction and validated 3000 ms setting come from
headroomlabs-ai#2451

Closes headroomlabs-ai#2451

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

- Raise the default Kompress execution-slot wait from 25 ms to 3000 ms.
- Start the Kompress request deadline at call entry and carry it through
single-item acquire, single-to-batch delegation, and sequential-fallback
lineage.
- Cap the raised execution-slot wait by that live request deadline on
both single-item and batch acquire paths.
- Keep the per-backend default concurrency at one and preserve all
tighter time budgets.
- Add queued single-item, batch, request-deadline, carried-deadline
lineage, and router-watchdog lifecycle regressions at the same owner
layer that currently fails.
- Preserve the explicit short-timeout fail-open path.

## Testing

- [x] Unit tests pass (`uv run pytest tests/test_kompress_failsafe.py
tests/test_kompress_request_nonblocking.py
tests/test_content_router_single_item_deadline.py
tests/test_transforms/test_kompress_deadline.py -v`)
- [x] Linting passes (`uv run ruff check
headroom/transforms/kompress_compressor.py
tests/test_kompress_failsafe.py
tests/test_content_router_single_item_deadline.py
tests/test_transforms/test_kompress_deadline.py`)
- [x] Formatting passes (`uv run ruff format
headroom/transforms/kompress_compressor.py
tests/test_kompress_failsafe.py
tests/test_content_router_single_item_deadline.py
tests/test_transforms/test_kompress_deadline.py --check`)
- [x] New regression tests prove the saturation fix
- [ ] Manual testing performed

### Test Output

```text
uv run pytest tests/test_kompress_failsafe.py tests/test_kompress_request_nonblocking.py tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py -v
37 passed in 4.22s

uv run ruff check headroom/transforms/kompress_compressor.py tests/test_kompress_failsafe.py tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py
All checks passed!

uv run ruff format headroom/transforms/kompress_compressor.py tests/test_kompress_failsafe.py tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py --check
4 files already formatted
```

## Real Behavior Proof

- Environment: worktree Python environment from `uv sync --extra dev`,
focused pytest with real Python threads and `threading.BoundedSemaphore`
- Exact command / steps: hold the sole execution slot with the
environment override unset, start queued single-item and batch
compression workers, wait until each worker proves it reached a blocked
acquire on the shared execution semaphore, release the slot, rerun the
explicit 1 ms timeout preservation case, then set
`HEADROOM_COMPRESSION_DEADLINE_MS=10` and repeat the held-slot
single-item and batch acquires plus a router single-cache-miss run whose
Kompress load sleeps past the request deadline.
- Observed result: The queued single-item and batch workers each proved
a real blocked acquire before release, then acquired after release and
compressed, while `HEADROOM_KOMPRESS_EXECUTION_TIMEOUT_MS=1` still
passed through promptly, the 10 ms request deadline capped the raised
default wait so both held-slot paths failed open before 200 ms without
reaching model inference, the single-to-batch and sequential-fallback
lineage regressions proved later branches inherit the original request
start instead of resetting it, and the router lifecycle proof showed the
carried deadline now allows slow Kompress load to start but still
expires before model inference after the outer request has already
failed open.
- Not tested: live ONNX proxy savings under sustained concurrent load

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

## Additional Notes

`CHANGELOG.md` stays unchanged because the release pipeline generates
changelog entries from conventional commits. The fail-open path from
headroomlabs-ai#1430 stays intact; this change stops it from firing spuriously under
ordinary queueing.

Author: rodboev

headroom/transforms/kompress_compressor.py

@@ -96,11 +96,12 @@ def _add_kompress_must_keep_words(
 KOMPRESS_COREML_CACHE_DIR_ENV = "HEADROOM_KOMPRESS_COREML_CACHE_DIR"
 KOMPRESS_MAX_CONCURRENT_ENV = "HEADROOM_KOMPRESS_MAX_CONCURRENT"
 KOMPRESS_EXECUTION_SEMAPHORE_WAIT_MS_ENV = "HEADROOM_KOMPRESS_EXECUTION_TIMEOUT_MS"
-KOMPRESS_EXECUTION_SEMAPHORE_WAIT_MS_DEFAULT = 25
+KOMPRESS_EXECUTION_SEMAPHORE_WAIT_MS_DEFAULT = 3000
 KOMPRESS_BATCH_SIZE_ENV = "HEADROOM_KOMPRESS_BATCH_SIZE"
 KOMPRESS_ACQUIRE_TIMEOUT_ENV = "HEADROOM_KOMPRESS_ACQUIRE_TIMEOUT_SECONDS"
 KOMPRESS_TIME_BUDGET_ENV = "HEADROOM_KOMPRESS_TIME_BUDGET_SECONDS"
 KOMPRESS_CANARY_THRESHOLD_ENV = "HEADROOM_KOMPRESS_CANARY_SECONDS"
+KOMPRESS_REQUEST_DEADLINE_ENV = "HEADROOM_COMPRESSION_DEADLINE_MS"
 
 # Both defaults sit well under the proxy's 30s compression-stage timeout so a
 # slow model gives up (passthrough) before the request is abandoned. A thread
@@ -175,6 +176,13 @@ def _execution_wait_budget_seconds() -> float:
     return parsed / 1000.0
 
 
+def _request_deadline_seconds() -> float:
+    try:
+        return max(0.0, float(os.environ.get(KOMPRESS_REQUEST_DEADLINE_ENV, "20000")) / 1000.0)
+    except ValueError:
+        return 20.0
+
+
 def _acquire_execution_slot(
     backend: str,
     device_type: str,
@@ -1199,6 +1207,7 @@ def compress(
         *,
         allow_download: bool = True,
         ccr_original: str | None = None,
+        _deadline_started_at: float | None = None,
     ) -> KompressResult:
         """Compress content using Kompress model.
 
@@ -1223,6 +1232,7 @@ def compress(
         Returns:
             KompressResult with compressed text.
         """
+        t_deadline = time.perf_counter() if _deadline_started_at is None else _deadline_started_at
         words = content.split()
         n_words = len(words)
 
@@ -1238,13 +1248,7 @@ def compress(
         # Cached per instance: operator config, read once -- not per compress() call.
         deadline_s = getattr(self, "_deadline_s", None)
         if deadline_s is None:
-            try:
-                deadline_s = max(
-                    0.0,
-                    float(os.environ.get("HEADROOM_COMPRESSION_DEADLINE_MS", "20000")) / 1000.0,
-                )
-            except ValueError:
-                deadline_s = 20.0
+            deadline_s = _request_deadline_seconds()
             self._deadline_s = deadline_s
 
         try:
@@ -1263,6 +1267,7 @@ def compress(
                     target_ratio=[target_ratio],
                     batch_size=_batch_size(),
                     ccr_originals=[ccr_original],
+                    _deadline_started_at=t_deadline,
                 )
                 if batch_result:
                     return batch_result[0]
@@ -1271,7 +1276,6 @@ def compress(
             kept_ids: set[int] = set()
             inference_ms = 0.0
             chunk_count = 0
-            t_deadline = time.perf_counter()
 
             acquire_timeout = _acquire_timeout_seconds()
             budget = _time_budget_seconds()
@@ -1327,12 +1331,29 @@ def compress(
                     input_ids = input_ids.to(device)
                     attention_mask = attention_mask.to(device)
 
+                request_remaining: float | None = None
+                if deadline_s:
+                    request_remaining = deadline_s - (time.perf_counter() - t_deadline)
+                    if request_remaining <= 0:
+                        kept_ids.update(range(chunk_start, n_words))
+                        logger.warning(
+                            "Kompress hit %.1fs deadline before acquire after %d/%d words "
+                            "(%d chunks done); kept remainder verbatim to free the request "
+                            "thread (#1171)",
+                            deadline_s,
+                            chunk_start,
+                            n_words,
+                            chunk_count,
+                        )
+                        break
+
                 acquire_bounds = [
                     bound
                     for bound in (
                         _execution_wait_budget_seconds(),
                         acquire_timeout,
                         remaining,
+                        request_remaining,
                     )
                     if bound is not None
                 ]
@@ -1471,6 +1492,7 @@ def compress_batch(
         batch_size: int = 32,
         *,
         ccr_originals: list[str | None] | None = None,
+        _deadline_started_at: float | None = None,
     ) -> list[KompressResult]:
         """Compress multiple texts. Uses batched inference on GPU, sequential on CPU.
 
@@ -1531,6 +1553,7 @@ def compress_batch(
         n = len(contents)
         if n == 0:
             return []
+        t_deadline = time.perf_counter() if _deadline_started_at is None else _deadline_started_at
 
         # Normalize target_ratio to a per-text list
         if isinstance(target_ratio, list):
@@ -1572,6 +1595,7 @@ def compress_batch(
                     question=question,
                     target_ratio=r,
                     ccr_original=ccr_source,
+                    _deadline_started_at=t_deadline,
                 )
                 for content, r, ccr_source in zip(contents, ratios, ccr_sources, strict=True)
             ]
@@ -1608,6 +1632,10 @@ def compress_batch(
         device_type = _model_device_type(model, backend)
         kept_ids_per_text: dict[int, set[int]] = {i: set() for i in range(n) if results[i] is None}
         inference_ms = 0.0
+        deadline_s = getattr(self, "_deadline_s", None)
+        if deadline_s is None:
+            deadline_s = _request_deadline_seconds()
+            self._deadline_s = deadline_s
 
         acquire_timeout = _acquire_timeout_seconds()
         budget = _time_budget_seconds()
@@ -1637,6 +1665,9 @@ def _bail_remaining(reason: str, batch_start: int) -> None:
                 if remaining <= 0:
                     _bail_remaining("time budget exhausted", batch_start)
                     break
+            if deadline_s and (deadline_s - (time.perf_counter() - t_deadline)) <= 0:
+                _bail_remaining("request deadline exhausted", batch_start)
+                break
 
             batch = chunk_queue[batch_start : batch_start + batch_size]
             batch_word_lists = [c[2] for c in batch]
@@ -1660,13 +1691,21 @@ def _bail_remaining(reason: str, batch_start: int) -> None:
                     input_ids = input_ids.to(device)
                     attention_mask = attention_mask.to(device)
 
+                request_remaining: float | None = None
+                if deadline_s:
+                    request_remaining = deadline_s - (time.perf_counter() - t_deadline)
+                    if request_remaining <= 0:
+                        _bail_remaining("request deadline exhausted", batch_start)
+                        break
+
                 # Single forward pass for all chunks in this batch.
                 acquire_bounds = [
                     bound
                     for bound in (
                         _execution_wait_budget_seconds(),
                         acquire_timeout,
                         remaining,
+                        request_remaining,
                     )
                     if bound is not None
                 ]

tests/test_content_router_single_item_deadline.py

@@ -2,6 +2,7 @@
 
 import time
 
+import headroom.transforms.kompress_compressor as kc
 from headroom.transforms.content_detector import ContentType
 from headroom.transforms.content_router import (
     CompressionStrategy,
@@ -10,6 +11,7 @@
     RouterCompressionResult,
     RoutingDecision,
 )
+from headroom.transforms.kompress_compressor import KompressCompressor, KompressConfig
 
 
 class _Tokenizer:
@@ -48,7 +50,7 @@ def _messages() -> list[dict[str, str]]:
         {"role": "assistant", "content": "frozen prefix content remains unchanged"},
         {
             "role": "assistant",
-            "content": "pending cache miss content takes the inline compression branch",
+            "content": "pending cache miss content takes the inline compression branch today",
         },
     ]
 
@@ -112,3 +114,71 @@ def test_single_cache_miss_preserves_disabled_deadline(monkeypatch):
     )
 
     assert result.messages[1]["content"] == "compressed output"
+
+
+def test_single_cache_miss_deadline_starts_before_kompress_load(monkeypatch, caplog):
+    router = _router()
+
+    class _Encoding(dict):
+        def __init__(self, rows: list[list[str]]):
+            super().__init__(
+                input_ids=[[0] * len(row) for row in rows],
+                attention_mask=[[1] * len(row) for row in rows],
+            )
+            self._rows = rows
+
+        def word_ids(self, batch_index: int = 0):
+            return list(range(len(self._rows[batch_index])))
+
+    class _Tokenizer:
+        def count_text(self, content: str) -> int:
+            return len(content.split())
+
+        def __call__(self, words, **_kwargs):
+            rows = words if words and isinstance(words[0], list) else [words]
+            return _Encoding(rows)
+
+    class _Model:
+        def __init__(self):
+            self.calls = 0
+
+        def get_keep_mask(self, input_ids, attention_mask):
+            self.calls += 1
+            return [[i % 2 == 0 for i in range(len(row))] for row in input_ids]
+
+    model = _Model()
+    compressor = KompressCompressor(config=KompressConfig(enable_ccr=False))
+    monkeypatch.setattr(compressor, "_should_batch_single_content", lambda *a, **k: False)
+    load_state = {"calls": 0}
+
+    def _slow_load(*_args, **_kwargs):
+        load_state["calls"] += 1
+        time.sleep(0.05)
+        return model, _Tokenizer(), "onnx"
+
+    monkeypatch.setattr(kc, "_load_kompress", _slow_load)
+    monkeypatch.setattr(
+        router,
+        "compress",
+        lambda content, *, context="", bias=1.0: _compression_result(
+            content,
+            compressor.compress(content).compressed,
+        ),
+    )
+    monkeypatch.setenv("HEADROOM_COMPRESSION_DEADLINE_MS", "10")
+
+    started = time.perf_counter()
+    result = router.apply(
+        _messages(),
+        _Tokenizer(),
+        frozen_message_count=1,
+        min_tokens_to_compress=1,
+    )
+    elapsed = time.perf_counter() - started
+    time.sleep(0.1)
+
+    assert elapsed < 0.12
+    assert result.messages[1]["content"] == _messages()[1]["content"]
+    assert "failing open via PASSTHROUGH" in caplog.text
+    assert load_state["calls"] == 1
+    assert model.calls == 0