mirror of
https://github.com/headroomlabs-ai/headroom.git
synced 2026-08-27 14:17:10 -04:00
fix(proxy): restore Anthropic compression on token mode (issue #327)
Three bugs combined to drive end-to-end compression on the Anthropic backend to ~0% in token mode (the default). User report #327 saw a ~9× drop in dashboard savings from one day to the next on Claude Code traffic; the dashboard headline was technically correct but the underlying compression genuinely was not running. After this change the same Claude Code-shape multi-turn conversation goes from 14987 → 14371 tokens at the request boundary on turn 1 and only recompresses the freshest tool_result on subsequent turns, with the prior turns frozen byte-identical to preserve the upstream prefix cache. Bug 1 — IntelligentContextManager inner ContentRouter has no observer PR #302 (commitcf979958, 2026-04-28) wired CompressionObserver onto the outer ContentRouter in proxy/server.py and onto SmartCrusher. The inner ContentRouter constructed lazily inside IntelligentContextManager._get_content_router (added Jan 18, 2026 in57b2de5alongside the COMPRESS_FIRST strategy) was missed. That inner router handles the bulk of Claude Code's tool_result-block compression, so per-strategy counters surfaced by PR #314 in v0.15.0 showed compressions_by_strategy={"text": 6} while summary.compression.total_tokens_removed=1.3M — math-impossible. Fix: add observer= parameter to IntelligentContextManager.__init__, forward it to the inner ContentRouter at intelligent_context.py:525, and pass observer=self.metrics from proxy/server.py. Bug 2 — TTL deferral marks every fresh tool_result as stable should_defer_compression in compression_cache.py returned True on first-sight (added 2026-04-07 in commit22dad13with the intent of batching first-time compressions near the 5-min cache TTL boundary to trade many small busts for one). The token-mode walker at anthropic.py:766-787 walks every message past frozen_message_count, calls should_defer_compression on each fresh tool_result, gets True, and advances ttl_frozen += 1 — every iteration. Result: frozen_message_count grows to len(messages), the pipeline freezes the entire request, and nothing reaches a real compressor. The defer-first-sight rationale assumes recurring content within TTL. Real Claude Code traffic produces unique content per turn, so "defer until next sight" defers forever. Compressing fresh content on first sight does not bust any prefix cache because Anthropic has not cached that byte position yet — it's a cache write either way. Fix: should_defer_compression returns False on first-sight (record the timestamp; compress now). Subsequent sightings within TTL still defer (batch window preserved for genuinely repeating content). Updated tests in test_compression_cache.py to assert the corrected semantics and verify _first_seen is recorded on first call. Bug 3 — cross-tokenizer comparison in token-mode inflation guard anthropic.py:634 sets original_tokens = tokenizer.count_messages(...) using the proxy-side EstimatingTokenCounter. The token-mode branch at line 816 set optimized_tokens = result.tokens_after from pipeline, which uses the provider-side AnthropicProvider tiktoken estimator. The two tokenizers disagree by ~25% on the same payload. The inflation guard at line 901 (if optimized_tokens > original_tokens: revert to originals) treats those two numbers as comparable. After a real 12% compression the provider-tokenizer figure was still higher than the proxy-tokenizer baseline, so the guard fired, optimized_messages was reset to the original input, transforms_applied was emptied, and tokens_saved went to 0. The dashboard showed no compression even when the pipeline successfully compressed. Fix: recount optimized_tokens with the proxy tokenizer right after the pipeline returns, so the guard compares apples-to-apples. The recount cost is a few ms on a 50K-token request and is dwarfed by upstream call latency. Verification * 80 targeted tests across test_compression_cache, test_compression_observability, test_proxy_anthropic_cache_stability, test_proxy_intelligent_context pass. * make ci-precheck clean. * End-to-end real-API run against api.anthropic.com via local proxy: - Turn 1 fresh: 14987 → 14371 (4.1%) on a 3-tool-round payload; smart_crusher and diff strategies fired with non-zero savings. - Turn 2 (turn 1 history + 1 new tool_result): 23161 → 21928 (5.3%); only the new tool_result compressed; older turns marked router:protected:user_message; Anthropic returned cache_creation_input_tokens > 0 confirming the prefix was not busted. Two new regression tests in test_compression_observability lock down the inner ContentRouter observer wiring so a future copy of Bug 1 fails the suite the day it lands.
This commit is contained in:
parent
2a0582dee1
commit
44944fb3fe
6 changed files with 120 additions and 12 deletions
22
headroom/cache/compression_cache.py
vendored
22
headroom/cache/compression_cache.py
vendored
|
|
@ -143,20 +143,30 @@ class CompressionCache:
|
|||
) -> bool:
|
||||
"""Whether to defer compressing this content to avoid mid-TTL busts.
|
||||
|
||||
Returns True if the content was first seen recently enough that
|
||||
compressing it now would bust the cached prefix with no TTL benefit.
|
||||
Returns False near the TTL boundary (within batch_window of expiry),
|
||||
meaning we should compress now and accept one bust.
|
||||
Returns True if we have evidence this content has been re-sent
|
||||
within the cache TTL window — recompressing it now would bust an
|
||||
existing prefix-cache entry without TTL-amortizing the bust over
|
||||
future turns. Returns False otherwise:
|
||||
|
||||
- **First sight** of the content. Compress now: there is no
|
||||
prefix-cache entry to preserve yet (this byte range was not in
|
||||
a prior request), so compression carries no bust cost. Issue
|
||||
#327: a previous version returned True here, which marked the
|
||||
freshest tool_result on every turn as "stable" and effectively
|
||||
disabled compression for typical Claude Code workloads where
|
||||
each tool_result is unique-per-turn.
|
||||
- **Near the TTL boundary**: compress now and amortize the bust
|
||||
across future turns (batched recompression).
|
||||
"""
|
||||
now = time.time()
|
||||
first_seen = self._first_seen.get(content_hash)
|
||||
if first_seen is None:
|
||||
self._first_seen[content_hash] = now
|
||||
return True # First time seeing this — defer
|
||||
return False # First time — compress now (no cache entry to preserve)
|
||||
age = now - first_seen
|
||||
if age >= ttl_seconds - batch_window:
|
||||
return False # Near TTL boundary — compress now (batch window)
|
||||
return True # Still within TTL — defer to preserve cache
|
||||
return True # Seen recently within TTL — defer to preserve cache
|
||||
|
||||
def get_stats(self) -> dict:
|
||||
"""Return cache statistics."""
|
||||
|
|
|
|||
|
|
@ -810,10 +810,22 @@ class AnthropicHandlerMixin:
|
|||
optimized_messages = result.messages
|
||||
transforms_applied = result.transforms_applied
|
||||
pipeline_timing = result.timing
|
||||
# Keep original_tokens as the REAL original (pre-Zone-1-swap)
|
||||
# so tokens_saved captures both Zone 1 + Zone 2 savings.
|
||||
# original_tokens was set at line ~2183 from uncompressed messages.
|
||||
optimized_tokens = result.tokens_after
|
||||
# Issue #327 / Bug 3: pipeline.apply uses the provider-
|
||||
# side tokenizer (AnthropicProvider tiktoken estimator),
|
||||
# which counts ~25% higher than the proxy-side
|
||||
# EstimatingTokenCounter used to set `original_tokens`
|
||||
# at line 634. Reusing `result.tokens_after` here
|
||||
# produced an apples-vs-oranges comparison against
|
||||
# `original_tokens` in the inflation guard below
|
||||
# (line ~901): even after a real 12% compression the
|
||||
# provider-tokenizer figure was higher than the proxy-
|
||||
# tokenizer baseline, triggering a spurious revert.
|
||||
# Recount optimized_messages with the proxy tokenizer
|
||||
# so original_tokens vs optimized_tokens is self-
|
||||
# consistent. The recount cost (~ms on a 50K-token
|
||||
# request) is paid once per request and is dwarfed by
|
||||
# the upstream call latency.
|
||||
optimized_tokens = tokenizer.count_messages(optimized_messages)
|
||||
elif not is_cache_mode(self.config.mode):
|
||||
async with stage_timer.measure("compression_first_stage"):
|
||||
result = await asyncio.wait_for(
|
||||
|
|
|
|||
|
|
@ -269,6 +269,7 @@ class HeadroomProxy(
|
|||
compress_threshold=0.10 if config.intelligent_context_compress_first else 0.0,
|
||||
),
|
||||
toin=toin,
|
||||
observer=self.metrics,
|
||||
)
|
||||
self._context_manager_status = "intelligent"
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -154,6 +154,7 @@ class IntelligentContextManager(Transform):
|
|||
config: IntelligentContextConfig | None = None,
|
||||
toin: ToolIntelligenceNetwork | None = None,
|
||||
summarize_fn: SummarizeFn | None = None,
|
||||
observer: Any = None,
|
||||
):
|
||||
"""
|
||||
Initialize intelligent context manager.
|
||||
|
|
@ -164,12 +165,21 @@ class IntelligentContextManager(Transform):
|
|||
summarize_fn: Optional callback for summarization.
|
||||
If provided and summarization_enabled=True, enables SUMMARIZE strategy.
|
||||
Signature: (messages: list[dict], context: str) -> str
|
||||
observer: Optional `CompressionObserver` (see
|
||||
`headroom.transforms.observability`). Forwarded to the
|
||||
lazy-loaded inner `ContentRouter` so per-strategy
|
||||
compression counters reflect the COMPRESS_FIRST path.
|
||||
Without it, compressions on `tool_result` content blocks
|
||||
are invisible to `compressions_by_strategy` /
|
||||
`tokens_saved_by_strategy` — the silent-regression class
|
||||
PR #302 was built to detect.
|
||||
"""
|
||||
from ..config import IntelligentContextConfig
|
||||
|
||||
self.config = config or IntelligentContextConfig()
|
||||
self.toin = toin
|
||||
self._summarize_fn = summarize_fn
|
||||
self._observer = observer
|
||||
|
||||
# Initialize scorer with TOIN if available
|
||||
self.scorer = MessageScorer(
|
||||
|
|
@ -522,7 +532,7 @@ class IntelligentContextManager(Transform):
|
|||
min_section_tokens=20,
|
||||
ccr_enabled=True,
|
||||
)
|
||||
self._content_router = ContentRouter(config=router_config)
|
||||
self._content_router = ContentRouter(config=router_config, observer=self._observer)
|
||||
except ImportError:
|
||||
logger.debug("ContentRouter not available for COMPRESS_FIRST")
|
||||
return self._content_router
|
||||
|
|
|
|||
|
|
@ -252,10 +252,25 @@ class TestCompressionCacheFrozenCount:
|
|||
assert hb not in cache._stable_hashes # msg[2] not included
|
||||
|
||||
def test_should_defer_compression_new_content(self, cache: CompressionCache) -> None:
|
||||
"""First-time content should be deferred."""
|
||||
"""First-time content should NOT be deferred — there is no
|
||||
prefix-cache entry to preserve, so compression carries no bust
|
||||
cost. Issue #327: prior behavior deferred first-sight, which
|
||||
marked every fresh tool_result as stable and disabled
|
||||
compression for typical Claude Code workloads.
|
||||
"""
|
||||
h = CompressionCache.content_hash("brand new content")
|
||||
assert cache.should_defer_compression(h, ttl_seconds=300, batch_window=30) is False
|
||||
# Subsequent sightings within TTL should defer (batch window).
|
||||
assert cache.should_defer_compression(h, ttl_seconds=300, batch_window=30) is True
|
||||
|
||||
def test_should_defer_compression_records_first_seen(self, cache: CompressionCache) -> None:
|
||||
"""First-sight call must record the timestamp so subsequent
|
||||
in-window calls can defer. Without this the deferral pathway
|
||||
for genuinely-repeated content stops working."""
|
||||
h = CompressionCache.content_hash("seen-twice content")
|
||||
cache.should_defer_compression(h) # first sight
|
||||
assert h in cache._first_seen
|
||||
|
||||
def test_should_defer_compression_near_ttl(self, cache: CompressionCache) -> None:
|
||||
"""Content near TTL boundary should NOT be deferred."""
|
||||
import time
|
||||
|
|
|
|||
|
|
@ -339,3 +339,63 @@ def test_router_with_prometheus_observer_increments_counters():
|
|||
"smart_crusher": (300 - 80) + (100 - 40), # 280
|
||||
"code_aware": (200 - 120), # 80
|
||||
}
|
||||
|
||||
|
||||
# ─── IntelligentContextManager wiring ──────────────────────────────────
|
||||
|
||||
|
||||
def test_intelligent_context_manager_forwards_observer_to_inner_router():
|
||||
"""COMPRESS_FIRST path must fire the observer.
|
||||
|
||||
Regression guard for the bug introduced in PR #302 (commit
|
||||
cf979958, 2026-04-28): the observer was wired onto the outer
|
||||
ContentRouter in `proxy/server.py` but NOT onto the inner
|
||||
ContentRouter inside `IntelligentContextManager._get_content_router`.
|
||||
On Anthropic/Claude Code traffic — where most compression happens
|
||||
inside `_apply_compress_first` walking `tool_result` blocks — that
|
||||
silently zero'd the per-strategy counters even when 1M+ tokens were
|
||||
being compressed (see issue #327).
|
||||
|
||||
The fix threads `observer=` through the IntelligentContextManager
|
||||
constructor into the inner router. This test asserts the wiring at
|
||||
the construction boundary; the observer fires when the inner router
|
||||
actually compresses content (covered indirectly by the existing
|
||||
`test_content_router_records_observer_call_per_routing_decision`).
|
||||
"""
|
||||
from headroom.config import IntelligentContextConfig
|
||||
from headroom.transforms.intelligent_context import IntelligentContextManager
|
||||
|
||||
spy = SpyObserver()
|
||||
icm = IntelligentContextManager(
|
||||
config=IntelligentContextConfig(enabled=True),
|
||||
observer=spy,
|
||||
)
|
||||
|
||||
# Force the lazy router to materialize.
|
||||
inner_router = icm._get_content_router()
|
||||
|
||||
assert inner_router is not None, (
|
||||
"IntelligentContextManager could not construct its inner ContentRouter; "
|
||||
"fixture setup is broken"
|
||||
)
|
||||
assert inner_router._observer is spy, (
|
||||
"Inner ContentRouter is missing the observer reference. "
|
||||
"IntelligentContextManager must forward `observer=` to "
|
||||
"ContentRouter(...) at intelligent_context.py:_get_content_router."
|
||||
)
|
||||
|
||||
|
||||
def test_intelligent_context_manager_observer_defaults_to_none():
|
||||
"""Default constructor (no observer kwarg) leaves the inner router
|
||||
unobserved. Lock the default behavior so SDK callers that don't
|
||||
have a metrics object aren't forced to plumb one through."""
|
||||
from headroom.config import IntelligentContextConfig
|
||||
from headroom.transforms.intelligent_context import IntelligentContextManager
|
||||
|
||||
icm = IntelligentContextManager(
|
||||
config=IntelligentContextConfig(enabled=True),
|
||||
)
|
||||
assert icm._observer is None
|
||||
inner_router = icm._get_content_router()
|
||||
assert inner_router is not None
|
||||
assert inner_router._observer is None
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue