From e9000863fc5ee0a0e1d56b7b49afe36cc6111ada Mon Sep 17 00:00:00 2001 From: Rod Boev Date: Mon, 13 Jul 2026 23:50:49 -0400 Subject: [PATCH] fix(kompress): fail-open wall-clock guard on single-cache-miss compression (#2114) ## Description The single-cache-miss branch in `ContentRouter` ran compression inline on the request path without its own wall-clock guard, so a cooperative stall waited for the full call even when `HEADROOM_COMPRESSION_DEADLINE_MS` was meant to fail open. This change adds a branch-level watchdog that returns `PASSTHROUGH` after the deadline, scoped only to the one-pending-task path and not the native GIL-hold root cause. Closes #2046 ## 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 - Added `_compression_deadline_seconds()` and a watchdog around the single-cache-miss inline compression branch in `ContentRouter`. - Returned the original content with `PASSTHROUGH` and logged a fail-open warning after the configured deadline, while preserving under-deadline and deadline-disabled behavior. - Added focused regressions for timeout, under-deadline output, and disabled-deadline behavior, then kept the wider deadline suite green. - Raised the locked production floors for `click` and `pillow` to clear the current `pip-audit` findings that now fail external PR merge snapshots. ## Testing - [x] Unit tests pass (`uv run pytest tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py -q`) - [x] Linting passes (`uv run ruff check headroom/transforms/content_router.py tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py`) - [ ] Type checking passes (`uv run mypy headroom`) - [x] New tests added for new functionality - [x] Manual testing performed ### Test Output ```text uv run pytest tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py -q ============================= test session starts ============================= platform win32 -- Python 3.12.13, pytest-9.0.3, pluggy-1.6.0 rootdir: D:\Repos\headroom-pr-2046-compression-freeze configfile: pyproject.toml plugins: anyio-4.12.1, langsmith-0.9.3, asyncio-1.3.0, cov-7.0.0 asyncio: mode=Mode.AUTO, debug=False, asyncio_default_fixture_loop_scope=None, asyncio_default_test_loop_scope=function collected 5 items tests\test_content_router_single_item_deadline.py ... [ 60%] tests\test_transforms\test_kompress_deadline.py .. [100%] ============================== 5 passed in 0.42s ============================== uv run ruff check headroom/transforms/content_router.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/content_router.py tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py --check 3 files already formatted ``` ## Real Behavior Proof - Environment: Windows, Python 3.12.13, router-level harness with a cooperative slow-compression stub - Exact command / steps: `uv run pytest tests/test_content_router_single_item_deadline.py tests/test_transforms/test_kompress_deadline.py -q`, which forces one frozen prefix and one cache miss, then sleeps past a 10 ms deadline - Observed result: the guarded branch returns the original content through `PASSTHROUGH` at the deadline, while under-deadline and deadline-disabled behavior stay unchanged - Not tested: native GIL-holding freeze ## 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 - [x] New and existing unit tests pass locally with my changes - [ ] I have updated the CHANGELOG.md if applicable ## Screenshots (if applicable) Not applicable. ## Additional Notes - Security CI currently flags `click 8.3.1` and `pillow 12.2.0`, so this PR carries the narrow floor bump to `click>=8.3.3` and `pillow>=12.3.0` as a supply-chain unblock for the same final merge snapshot. - `CHANGELOG.md` remains untouched because Headroom generates release notes from conventional commits. - This PR is a Python-side mitigation for the single-cache-miss branch only; the native GIL-hold root cause remains a separate owner. --------- Co-authored-by: JerrettDavis --- headroom/transforms/content_router.py | 54 ++++++++- pyproject.toml | 4 +- ...est_content_router_single_item_deadline.py | 114 ++++++++++++++++++ uv.lock | 4 +- 4 files changed, 170 insertions(+), 6 deletions(-) create mode 100644 tests/test_content_router_single_item_deadline.py diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 5e7f26b3e..e9e07912c 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -78,6 +78,16 @@ _detect_panic_warned = False _detect_native_unhealthy = False # circuit breaker: native detect hung once (#575) +def _compression_deadline_seconds() -> float: + try: + return max( + 0.0, + float(os.environ.get("HEADROOM_COMPRESSION_DEADLINE_MS", "20000")) / 1000.0, + ) + except ValueError: + return 20.0 + + def _router_debug_dumps(value: Any) -> str: return json.dumps(value, ensure_ascii=False, default=str, separators=(",", ":")) @@ -3723,8 +3733,48 @@ class ContentRouter(Transform): task_results = [] for _, task_content, task_ctx, task_bias, _, _ in pending_tasks: t0 = time.perf_counter() - r = self.compress(task_content, context=task_ctx, bias=task_bias) - task_results.append((r, (time.perf_counter() - t0) * 1000)) + deadline_s = _compression_deadline_seconds() if len(pending_tasks) == 1 else 0.0 + if deadline_s: + box: dict[str, Any] = {} + + def _run( + _box: dict[str, Any] = box, + _content: str = task_content, + _context: str = task_ctx, + _bias: float = task_bias, + ) -> None: + try: + _box["result"] = self.compress( + _content, context=_context, bias=_bias + ) + except BaseException as exc: # noqa: BLE001 + _box["error"] = exc + + # ponytail: daemon watchdog cannot stop native GIL holds; native layer owns that fix. + worker = threading.Thread( + target=_run, name="headroom-single-compress-watchdog", daemon=True + ) + worker.start() + worker.join(deadline_s) + if worker.is_alive(): + logger.warning( + "ContentRouter single-cache-miss compression exceeded %.1fs; " + "failing open via PASSTHROUGH", + deadline_s, + ) + r = RouterCompressionResult( + compressed=task_content, + original=task_content, + strategy_used=CompressionStrategy.PASSTHROUGH, + ) + elif "error" in box: + raise box["error"] + else: + r = box["result"] + else: + r = self.compress(task_content, context=task_ctx, bias=task_bias) + compress_ms = (time.perf_counter() - t0) * 1000 + task_results.append((r, compress_ms)) else: # Parallel compression via thread pool with ThreadPoolExecutor(max_workers=max_workers) as executor: diff --git a/pyproject.toml b/pyproject.toml index b48a5cdb8..86ef1a514 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -54,7 +54,7 @@ dependencies = [ # ImportError-guarded. Marking it 3.14-optional lets headroom install on Python 3.14 # (core compression + the Anthropic proxy path never import litellm). See GH #956. "litellm>=1.86.2,<2.0; python_version < '3.14'", # model registry, pricing, providers (lazy) - "click>=8.1.0", # CLI framework + "click>=8.3.3", # CLI framework "rich>=13.0.0", # Rich terminal output "opentelemetry-api>=1.24.0", # Safe no-op OTEL API for instrumentation "ast-grep-cli>=0.30.0", # AST-aware code slicing (CodeCompressor); binary wheel @@ -165,7 +165,7 @@ relevance = [ # `headroom/image/compressor.py` adapts both API shapes at runtime via # a try/except cascade. See issue #372 for context. image = [ - "pillow>=10.0.0", + "pillow>=12.3.0", "sentencepiece>=0.1.99", # Required by SigLIP tokenizer (SiglipTokenizer) # Python 3.6–3.12: keep the proven ORT-bundled package directly. # ~15 MB ONNX models auto-downloaded on first use. diff --git a/tests/test_content_router_single_item_deadline.py b/tests/test_content_router_single_item_deadline.py new file mode 100644 index 000000000..258468f83 --- /dev/null +++ b/tests/test_content_router_single_item_deadline.py @@ -0,0 +1,114 @@ +from __future__ import annotations + +import time + +from headroom.transforms.content_detector import ContentType +from headroom.transforms.content_router import ( + CompressionStrategy, + ContentRouter, + ContentRouterConfig, + RouterCompressionResult, + RoutingDecision, +) + + +class _Tokenizer: + def count_text(self, content: str) -> int: + return len(content.split()) + + +def _compression_result(content: str, compressed: str) -> RouterCompressionResult: + return RouterCompressionResult( + compressed=compressed, + original=content, + strategy_used=CompressionStrategy.TEXT, + routing_log=[ + RoutingDecision( + content_type=ContentType.PLAIN_TEXT, + strategy=CompressionStrategy.TEXT, + original_tokens=len(content.split()), + compressed_tokens=len(compressed.split()), + ) + ], + ) + + +def _router() -> ContentRouter: + return ContentRouter( + ContentRouterConfig( + protect_recent_code=0, + protect_analysis_context=False, + skip_user_messages=False, + ) + ) + + +def _messages() -> list[dict[str, str]]: + return [ + {"role": "assistant", "content": "frozen prefix content remains unchanged"}, + { + "role": "assistant", + "content": "pending cache miss content takes the inline compression branch", + }, + ] + + +def test_single_cache_miss_fails_open_at_deadline(monkeypatch, caplog): + router = _router() + + def slow_compress(content, *, context="", bias=1.0): + time.sleep(0.2) + return _compression_result(content, "compressed output") + + monkeypatch.setattr(router, "compress", slow_compress) + 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, + ) + + assert time.perf_counter() - started < 0.12 + assert result.messages[1]["content"] == _messages()[1]["content"] + assert "failing open via PASSTHROUGH" in caplog.text + + +def test_single_cache_miss_preserves_under_deadline_output(monkeypatch): + router = _router() + monkeypatch.setattr( + router, + "compress", + lambda content, *, context="", bias=1.0: _compression_result(content, "compressed output"), + ) + monkeypatch.setenv("HEADROOM_COMPRESSION_DEADLINE_MS", "1000") + + result = router.apply( + _messages(), + _Tokenizer(), + frozen_message_count=1, + min_tokens_to_compress=1, + ) + + assert result.messages[1]["content"] == "compressed output" + + +def test_single_cache_miss_preserves_disabled_deadline(monkeypatch): + router = _router() + monkeypatch.setattr( + router, + "compress", + lambda content, *, context="", bias=1.0: _compression_result(content, "compressed output"), + ) + monkeypatch.setenv("HEADROOM_COMPRESSION_DEADLINE_MS", "0") + + result = router.apply( + _messages(), + _Tokenizer(), + frozen_message_count=1, + min_tokens_to_compress=1, + ) + + assert result.messages[1]["content"] == "compressed output" diff --git a/uv.lock b/uv.lock index 6a72283ed..c5fa3fbd3 100644 --- a/uv.lock +++ b/uv.lock @@ -1737,7 +1737,7 @@ requires-dist = [ { name = "ast-grep-cli", specifier = ">=0.30.0" }, { name = "boto3", marker = "extra == 'bedrock'", specifier = ">=1.41.0" }, { name = "botocore", extras = ["crt"], marker = "extra == 'bedrock'", specifier = ">=1.41.0" }, - { name = "click", specifier = ">=8.1.0" }, + { name = "click", specifier = ">=8.3.3" }, { name = "datasets", marker = "extra == 'evals'", specifier = ">=2.14.0" }, { name = "datasets", marker = "extra == 'voice-train'", specifier = ">=2.14.0" }, { name = "fastapi", marker = "extra == 'dev'", specifier = ">=0.100.0" }, @@ -1785,7 +1785,7 @@ requires-dist = [ { name = "opentelemetry-sdk", marker = "extra == 'dev'", specifier = ">=1.24.0" }, { name = "opentelemetry-sdk", marker = "extra == 'otel'", specifier = ">=1.24.0" }, { name = "orjson", marker = "platform_python_implementation != 'PyPy' and extra == 'proxy'", specifier = ">=3.9.14" }, - { name = "pillow", marker = "extra == 'image'", specifier = ">=10.0.0" }, + { name = "pillow", marker = "extra == 'image'", specifier = ">=12.3.0" }, { name = "pre-commit", marker = "extra == 'dev'", specifier = ">=3.0.0" }, { name = "pydantic", specifier = ">=2.0.0" }, { name = "pytest", marker = "extra == 'dev'", specifier = ">=7.0.0" },