mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-10 14:27:00 -04:00
## Description Two related changes to the compression seams, plus the review fixes for both. Supersedes #2661 and #2662, which are closed in favour of this branch — the fixes are inseparable from the code they fix, so reviewing them together is cheaper than landing two PRs and patching them afterwards. **1. A registered lossless provider now competes on the general path.** The `headroom.transforms.lossless_provider` seam was only ever consulted from `_lossless_compact_excluded`, gated on `DEFAULT_EXCLUDE_TOOLS` (`config.py:216` — `Read/Grep/Glob/Write/Edit/WebSearch/WebFetch`). Gateway traffic carries the caller's own tool names — LiteLLM's `headroom` guardrail (https://docs.litellm.ai/docs/proxy/headroom) posts requests containing tools like `search_docs` / `run_ci` / `fetch_rows` — so a registered provider was structurally unreachable for every gateway/sidecar deployment. The seam existed; nothing could get to it. **2. `POST /v1/compress` is marker-free by default.** A CCR marker is only useful to a caller that also injects the `headroom_retrieve` tool AND can reach `/v1/retrieve`. Neither holds here: tool injection lives in the provider request handlers (`handlers/anthropic.py:1894`), never in `handle_compress`; and every `/v1/retrieve*` route is `Depends(_require_loopback)` (`server.py:4422, 4470, 4749, 4781`) with no remote opt-in — `HEADROOM_COMPRESS_ALLOW_REMOTE` drops the loopback dependency on `/v1/compress` only. So a gateway forwards a `Retrieve more: hash=…` pointer the model cannot follow, and the proxy pays a CCR store write nobody reads. `config.mode="ccr"` opts back in. Closes # ## Type of Change - [ ] Bug fix (non-breaking change that fixes an issue) - [x] New feature (non-breaking change that adds functionality) - [x] Breaking change (fix or feature that would cause existing functionality to change) - [ ] Documentation update - [ ] Performance improvement - [ ] Code refactoring (no functional changes) ## Changes Made ### Seam — `content_router.py`, `lossless_provider.py` - `_lossless_first` (STAGE 0, every block on every path) consults `get_lossless_provider()` and keeps whichever output is smaller. **Strict no-op when no provider is registered**, which is the default; and because it is best-of rather than authoritative, a provider can never do worse than the built-in folds. - Malformed provider output can no longer escape. Every shape check runs inside the `try`: result must be `None`, or a 2-element tuple/list of two `str`. Anything else is ignored at debug level. (Previously the unpack sat outside the `try`, so a 3-tuple raised `ValueError` up through `TransformPipeline.apply`, which re-raises.) - Empty / whitespace-only candidates are rejected rather than silently replacing block content. - **Providers are never offered diff content.** Diff folding is subtractive with no inverse check and a reflowed hunk breaks `git apply` — the same reason the built-in `diff` fold is restricted at `content_router.py:2481`. - The third-party `kind` label is sanitised against `^[a-z0-9_]{1,32}$` before reaching `transforms_applied` and the per-strategy metric dicts, so a caller-controlled string cannot explode Prometheus label cardinality. `fullmatch`, not `match`: `$` also matches before a trailing newline, which would put a newline in a label. - `set_lossless_provider(provider, *, verifier=None)` — in lossless-only mode, where STAGE 0's output is final and there is no marker to recover from, a registered verifier must confirm the fold or the candidate is dropped. No verifier registered = today's behaviour. `provider=None` clears both. - The provider is invoked once per block, not twice (`_has_lossless_fold` probes `_lossless_first` and discards the result, then STAGE 0 recomputes). Bounded memo, wholesale clear on overflow, no lock — a race costs one redundant fold. The memo keys on the provider registration generation, so registering or clearing a provider after a block was already folded takes effect. - The seam docstring now records that the provider runs on the general path and inside the parallel compression pool, so it must be thread-safe as well as deterministic. ### Route — `handlers/openai.py`, `server.py` - `_derived_compress_pipeline(key, **overrides)` replaces the copy-pasted pipeline-derivation block; `_no_ccr_pipeline` (the new default) and `_lossy_inline_pipeline` both use it. - The default pipeline is **built at startup** and included in `_eager_preload_transforms`, so a fresh pod does not pay ContentRouter construction and compressor load on its first request, inside the compression-executor budget. - An unrecognised `config.mode` returns 400 naming the valid values instead of silently falling back to the default. - Claude-family model names resolve their context limit from the Anthropic provider. Real divergence: `bedrock/anthropic.claude-3-5-sonnet` is 200000 there and 128000 on the OpenAI provider. The tokenizer still comes from the OpenAI pipeline's provider — a separate, larger change, noted in a comment. - Documents why the derived router deliberately does **not** share the base router's compression cache: keys do not encode CCR-marker mode, so sharing would leak marker-laden entries into the marker-free path. ### Behavior change A `/v1/compress` caller that relied on default markers now gets none. The only in-tree caller that can resolve them is the TypeScript SDK (`sdk/typescript/src/client.ts:398 retrieve`, `:422 handleToolCall`); it needs `config: {"mode": "ccr"}` to keep today's behaviour, and landing that SDK default in the same release would leave only gateway callers — for whom markers were never resolvable — seeing a difference. No `HEADROOM_COMPRESS_DEFAULT_MODE` compat env deliberately: a flag nobody sets becomes permanent debt, and the wire-level `mode` already covers the one caller that needs 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 - [x] Manual testing performed ### Test Output ```text $ python -m pytest tests/test_lossless_first_dispatch.py tests/test_lossless_excluded_compaction.py \ tests/test_lossless_mode.py tests/test_lossless_then_lossy.py tests/test_lossless_diff_fold_guard.py \ tests/test_bash_search_lossless_fold.py tests/test_proxy_compress_endpoint.py \ tests/test_ccr_row_drop_store_bridge.py tests/test_gateway_sidecar_ports.py tests/test_compress_api.py \ tests/test_platform_stabilization_functional.py tests/test_proxy_eager_preload_bind.py \ tests/test_proxy_warmup.py tests/test_router_registry_smartcrusher.py -q 189 passed in 32.04s $ ruff check <all 9 changed files> All checks passed! $ ruff format --check <all 9 changed files> 9 files already formatted $ mypy headroom/transforms/content_router.py headroom/transforms/lossless_provider.py \ headroom/proxy/handlers/openai.py headroom/proxy/server.py Success: no issues found in 4 source files ``` New tests cover, one concern each: every malformed provider shape; empty and whitespace-only results; diff content never reaching a provider (call-recording); `kind` sanitisation including the trailing-newline case; the verifier accepting / rejecting / raising; clearing a provider clearing its verifier; single provider invocation per block; memo invalidation on re-registration; unknown and valid `mode` values; the default pipeline existing before any request; and Claude vs OpenAI context-limit resolution with `token_budget` precedence preserved. ## Real Behavior Proof - **Environment:** macOS arm64, Python 3.12.6, proxy built from `_proxy_config_from_env()` with the default `coding` savings profile; Kompress both disabled and offloaded to a remote `kompress-v2-base` endpoint; `HEADROOM_COMPRESSION_TIMEOUT_SECONDS=300`. - **Steps:** `POST /v1/compress` over `TestClient` with OpenAI-shaped payloads under non-excluded tool names (`run_ci`, `list_files`, `code_search`, `fetch_rows`) — a CI log with ANSI escapes and repeated lines, a 160-path listing, a 150-line grep dump, a 150-row JSON array; plus a second payload with a RAG user blob, a 200-row JSON tool result and a 300-line log. Ran with and without a provider registered via `set_lossless_provider`. - **Observed — seam reachability:** | Kompress | no provider registered | provider registered | |---|---|---| | off | 19,284 → 10,265 tokens (46.8%) | 19,284 → **7,879 (59.1%)** | | on (remote) | 19,284 → 9,366 tokens (51.4%) | 19,284 → **7,146 (62.9%)** | Before this change the right-hand column was identical to the left — the registered provider was never called on this payload. - **Observed — marker-free default costs nothing:** | Config | tokens | saved | |---|---|---| | markers on (previous default) | 37,791 → 24,415 | 35.4% | | markers off (new default) | 37,791 → 24,415 | **35.4% — identical** | | `mode="lossy_inline"` | 37,791 → 25,129 | 33.5% | | `--lossless` | 37,791 → 35,100 | 7.1% | - **Observed — memo staleness, before the fix:** registering a provider that folds a grep block to 5 bytes left the block at its 1496-byte built-in fold, and clearing a provider kept serving the provider's output. Both correct after keying on the registration generation. - **Observed — context limit:** `bedrock/anthropic.claude-3-5-sonnet` resolves 200000 via the Anthropic provider, 128000 via the OpenAI provider. - **Not tested:** `tests/test_transforms_content_router.py` was not run — it does not complete on this machine, wedging on its 5th test while that test passes in 5.7s alone. Verified pre-existing before this work: with the diff stashed, the clean tree stalled at the identical test, and it stalled the same way under `HF_HUB_OFFLINE=1 HEADROOM_OFFLINE=1`. The machine was also out of disk at the time, which may be the real cause rather than the suspected native-detector deadlock (#575) — worth a separate issue either way. CI should be the arbiter here. ## 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 - [x] I did **not** edit `CHANGELOG.md` ## Additional Notes Two pre-existing test files needed adjusting, both direct consequences rather than scope creep: - `test_platform_stabilization_functional.py::test_v1_compress_success_reports_actual_metrics` patches `openai_pipeline.apply`, which the default mode no longer routes through. **This test was already failing on the marker-free-default commit** before any of the fixes — my original test selection missed it. It now patches the pipeline the route actually uses. - `test_proxy_eager_preload_bind.py` substitutes fake pipelines to control exactly what the preload walks; the eager build injected the real derived router's statuses into an exact-equality assertion. Its shared helper now clears the derived cache, preserving each test's intent without weakening an assertion. Docs unchecked — follow-ups worth doing in the same release: document `config.mode` values in `docs/content/docs/litellm.mdx` and `wiki/proxy.md`; the TS SDK `mode:"ccr"` default; and an operational note that `COMPRESSION_TIMEOUT_SECONDS` defaults to 30 (`helpers.py:687`) while a remote ML endpoint makes one sequential call per unit — on a large payload it trips the executor timeout and the handler fails open, returning `compression_skipped: true` with `tokens_before: 0`, which reads as "nothing to save" rather than "we gave up". Those zeroed counters are misleading and worth a separate fix.
169 lines
5.8 KiB
Python
169 lines
5.8 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import time
|
|
from typing import Any
|
|
|
|
import pytest
|
|
|
|
pytest.importorskip("fastapi")
|
|
pytest.importorskip("headroom._core")
|
|
|
|
from fastapi.testclient import TestClient
|
|
|
|
from headroom.config import TransformResult
|
|
from headroom.proxy.server import ProxyConfig, create_app
|
|
|
|
|
|
def _proxy_config(**overrides: Any) -> ProxyConfig:
|
|
defaults: dict[str, Any] = {
|
|
"optimize": True,
|
|
"cache_enabled": False,
|
|
"rate_limit_enabled": False,
|
|
"cost_tracking_enabled": False,
|
|
"log_requests": False,
|
|
"ccr_inject_tool": False,
|
|
"ccr_handle_responses": False,
|
|
"ccr_context_tracking": False,
|
|
"image_optimize": False,
|
|
"disable_kompress": True,
|
|
"compression_max_workers": 1,
|
|
}
|
|
defaults.update(overrides)
|
|
return ProxyConfig(**defaults)
|
|
|
|
|
|
def test_proxy_health_surfaces_compression_runtime_metrics(monkeypatch) -> None:
|
|
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
|
|
app = create_app(_proxy_config(optimize=False))
|
|
|
|
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
|
live = client.get("/livez")
|
|
health = client.get("/health")
|
|
|
|
assert live.status_code == 200
|
|
assert live.json()["alive"] is True
|
|
assert health.status_code == 200
|
|
runtime = health.json()["runtime"]
|
|
assert runtime["compression_executor"]["max_workers"] == 1
|
|
assert runtime["compression_executor"]["queued"] == 0
|
|
assert runtime["compression_executor"]["queue_timeouts_total"] == 0
|
|
|
|
|
|
def test_v1_compress_success_reports_actual_metrics(monkeypatch) -> None:
|
|
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
|
|
app = create_app(_proxy_config())
|
|
proxy = app.state.proxy
|
|
request_messages = [{"role": "user", "content": "summarize this repeated payload"}]
|
|
compressed_messages = [{"role": "user", "content": "summary payload"}]
|
|
|
|
def fake_apply(**kwargs):
|
|
assert kwargs["messages"] == request_messages
|
|
assert kwargs["model"] == "gpt-4o"
|
|
return TransformResult(
|
|
messages=compressed_messages,
|
|
tokens_before=100,
|
|
tokens_after=40,
|
|
transforms_applied=["test:compress"],
|
|
markers_inserted=["marker-1"],
|
|
)
|
|
|
|
# The default /v1/compress mode runs a marker-free pipeline derived from
|
|
# `openai_pipeline`, not `openai_pipeline` itself, so patch the one the
|
|
# route actually uses. It is built eagerly at create_app() time.
|
|
monkeypatch.setattr(proxy._compress_pipeline_cache["no_ccr"], "apply", fake_apply)
|
|
|
|
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
|
response = client.post(
|
|
"/v1/compress",
|
|
json={"model": "gpt-4o", "messages": request_messages},
|
|
)
|
|
|
|
body = response.json()
|
|
assert response.status_code == 200
|
|
assert body["messages"] == compressed_messages
|
|
assert body["tokens_before"] == 100
|
|
assert body["tokens_after"] == 40
|
|
assert body["tokens_saved"] == 60
|
|
assert body["compression_ratio"] == 0.4
|
|
assert body["transforms_applied"] == ["test:compress"]
|
|
assert body["transforms_summary"] == {"test:compress": 1}
|
|
assert body["ccr_hashes"] == ["marker-1"]
|
|
|
|
|
|
def test_v1_compress_timeout_fails_open_quickly(monkeypatch) -> None:
|
|
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
|
|
app = create_app(_proxy_config())
|
|
proxy = app.state.proxy
|
|
request_messages = [{"role": "user", "content": "do not mutate me"}]
|
|
|
|
async def timeout_executor(fn, *, timeout): # noqa: ANN001
|
|
raise TimeoutError("compression deadline exceeded")
|
|
|
|
monkeypatch.setattr(proxy, "_run_compression_in_executor", timeout_executor)
|
|
|
|
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
|
started = time.perf_counter()
|
|
response = client.post(
|
|
"/v1/compress",
|
|
json={"model": "gpt-4o", "messages": request_messages},
|
|
)
|
|
elapsed = time.perf_counter() - started
|
|
|
|
body = response.json()
|
|
assert response.status_code == 200
|
|
assert elapsed < 0.5
|
|
assert body["messages"] == request_messages
|
|
assert body["tokens_saved"] == 0
|
|
assert body["compression_ratio"] == 1.0
|
|
assert body["transforms_applied"] == []
|
|
assert body["compression_skipped"] is True
|
|
assert body["skip_reason"] == "compression_timeout"
|
|
|
|
|
|
def test_v1_compress_real_json_tool_payload_reduces_tokens(monkeypatch) -> None:
|
|
monkeypatch.setenv("HEADROOM_SKIP_UPSTREAM_CHECK", "1")
|
|
app = create_app(
|
|
_proxy_config(
|
|
ccr_inject_marker=False,
|
|
min_tokens_to_crush=20,
|
|
max_items_after_crush=10,
|
|
)
|
|
)
|
|
items = [
|
|
{
|
|
"id": i,
|
|
"status": "ok",
|
|
"score": i % 5,
|
|
"message": "same repeated value " * 20,
|
|
}
|
|
for i in range(80)
|
|
]
|
|
request = {
|
|
"model": "gpt-4o",
|
|
"messages": [
|
|
{"role": "user", "content": "summarize rows"},
|
|
{
|
|
"role": "assistant",
|
|
"content": None,
|
|
"tool_calls": [
|
|
{
|
|
"id": "call-1",
|
|
"type": "function",
|
|
"function": {"name": "list_rows", "arguments": "{}"},
|
|
}
|
|
],
|
|
},
|
|
{"role": "tool", "tool_call_id": "call-1", "content": json.dumps(items)},
|
|
],
|
|
}
|
|
|
|
with TestClient(app, base_url="http://127.0.0.1", client=("127.0.0.1", 12345)) as client:
|
|
response = client.post("/v1/compress", json=request)
|
|
|
|
body = response.json()
|
|
assert response.status_code == 200, response.text
|
|
assert body["tokens_before"] > body["tokens_after"], body
|
|
assert body["tokens_saved"] > 0
|
|
assert body["compression_ratio"] < 1.0
|
|
assert body["transforms_applied"], body
|