diff --git a/headroom/client.py b/headroom/client.py index 84ee88924..c71133399 100644 --- a/headroom/client.py +++ b/headroom/client.py @@ -2,6 +2,7 @@ from __future__ import annotations +import logging from collections.abc import Iterator from datetime import datetime, timezone from typing import Any @@ -34,6 +35,8 @@ from .utils import ( generate_request_id, ) +logger = logging.getLogger(__name__) + class ChatCompletions: """Wrapper for chat.completions API (OpenAI-style).""" @@ -747,7 +750,7 @@ class HeadroomClient: "content": response.content, } except Exception: - pass + logger.debug("Failed to extract response content for semantic cache", exc_info=True) return None def _simulate( diff --git a/headroom/config.py b/headroom/config.py index db9363570..2dd07f8d2 100644 --- a/headroom/config.py +++ b/headroom/config.py @@ -215,14 +215,12 @@ DEFAULT_EXCLUDE_TOOLS: frozenset[str] = frozenset( "Grep", "Write", "Edit", - "Bash", # Lowercase variants for case-insensitive matching "read", "glob", "grep", "write", "edit", - "bash", } ) diff --git a/headroom/transforms/content_router.py b/headroom/transforms/content_router.py index 05b89cca6..3fb319deb 100644 --- a/headroom/transforms/content_router.py +++ b/headroom/transforms/content_router.py @@ -257,7 +257,7 @@ def _netcost_message_tokens(message: dict[str, Any], tokenizer: Tokenizer) -> in class CompressionCache: - """Two-tier compression cache with TTL. + """Two-tier compression cache with TTL. Thread-safe. Tier 1 (skip set): content hashes that won't compress — instant skip, near-zero memory (just ints in a set). @@ -271,9 +271,17 @@ class CompressionCache: Uses in-process dict for ultra-fast lookups (~100ns). Could be backed by memcached/Redis for multi-process deployments. + + Thread safety: a ``threading.Lock`` guards all read-modify-write + operations. The ``apply()`` path runs compression inside a + ``ThreadPoolExecutor``; without the lock concurrent cache misses for + the same content would produce duplicate compression work (correct but + wasteful) and metrics counters would drift. """ def __init__(self, ttl_seconds: int = 1800): + import threading + # Tier 2: compressed results {hash: (text, ratio, strategy, timestamp)} self._results: dict[int, tuple[str, float, str, float]] = {} # Tier 1: hashes of content that won't compress {hash: timestamp} @@ -286,80 +294,91 @@ class CompressionCache: self._evictions = 0 self._total_lookup_ns = 0 self._lookup_count = 0 + self._lock = threading.Lock() def get(self, key: int) -> tuple[str, float, str] | None: - """Get cached compression result. + """Get cached compression result. Thread-safe. Returns (compressed_text, ratio, strategy) or None if not found/expired. Use is_skipped() first to check if content is known non-compressible. """ t0 = time.perf_counter_ns() - entry = self._results.get(key) - if entry is not None: - compressed, ratio, strategy, created_at = entry - if (time.time() - created_at) < self._ttl_seconds: - self._hits += 1 - self._total_lookup_ns += time.perf_counter_ns() - t0 - self._lookup_count += 1 - return (compressed, ratio, strategy) - else: - del self._results[key] - self._evictions += 1 - self._misses += 1 - self._total_lookup_ns += time.perf_counter_ns() - t0 - self._lookup_count += 1 - return None + with self._lock: + entry = self._results.get(key) + if entry is not None: + compressed, ratio, strategy, created_at = entry + if (time.monotonic() - created_at) < self._ttl_seconds: + self._hits += 1 + self._total_lookup_ns += time.perf_counter_ns() - t0 + self._lookup_count += 1 + return (compressed, ratio, strategy) + else: + del self._results[key] + self._evictions += 1 + self._misses += 1 + self._total_lookup_ns += time.perf_counter_ns() - t0 + self._lookup_count += 1 + return None def is_skipped(self, key: int) -> bool: - """Check if content is known non-compressible (Tier 1).""" - ts = self._skip.get(key) - if ts is not None: - if (time.time() - ts) < self._ttl_seconds: - self._skip_hits += 1 - return True - else: - del self._skip[key] - self._evictions += 1 - return False + """Check if content is known non-compressible (Tier 1). Thread-safe.""" + with self._lock: + ts = self._skip.get(key) + if ts is not None: + if (time.monotonic() - ts) < self._ttl_seconds: + self._skip_hits += 1 + return True + else: + del self._skip[key] + self._evictions += 1 + return False def put(self, key: int, compressed: str, ratio: float, strategy: str) -> None: - """Store a compressed result (Tier 2).""" - self._results[key] = (compressed, ratio, strategy, time.time()) + """Store a compressed result (Tier 2). Thread-safe.""" + with self._lock: + self._results[key] = (compressed, ratio, strategy, time.monotonic()) def mark_skip(self, key: int) -> None: - """Mark content as non-compressible (Tier 1).""" - self._skip[key] = time.time() + """Mark content as non-compressible (Tier 1). Thread-safe.""" + with self._lock: + self._skip[key] = time.monotonic() def move_to_skip(self, key: int) -> None: - """Move a result to skip set (threshold tightened, no longer qualifies).""" - self._results.pop(key, None) - self._skip[key] = time.time() + """Move a result to skip set (threshold tightened, no longer qualifies). + Thread-safe.""" + with self._lock: + self._results.pop(key, None) + self._skip[key] = time.monotonic() @property def size(self) -> int: - return len(self._results) + with self._lock: + return len(self._results) @property def skip_size(self) -> int: - return len(self._skip) + with self._lock: + return len(self._skip) @property def stats(self) -> dict[str, int | float]: - avg_ns = self._total_lookup_ns / self._lookup_count if self._lookup_count else 0 - return { - "cache_hits": self._hits, - "cache_skip_hits": self._skip_hits, - "cache_misses": self._misses, - "cache_evictions": self._evictions, - "cache_size": len(self._results), - "cache_skip_size": len(self._skip), - "cache_avg_lookup_ns": avg_ns, - } + with self._lock: + avg_ns = self._total_lookup_ns / self._lookup_count if self._lookup_count else 0 + return { + "cache_hits": self._hits, + "cache_skip_hits": self._skip_hits, + "cache_misses": self._misses, + "cache_evictions": self._evictions, + "cache_size": len(self._results), + "cache_skip_size": len(self._skip), + "cache_avg_lookup_ns": avg_ns, + } def clear(self) -> None: - """Clear all entries (e.g., on session end).""" - self._results.clear() - self._skip.clear() + """Clear all entries (e.g., on session end). Thread-safe.""" + with self._lock: + self._results.clear() + self._skip.clear() class CompressionStrategy(Enum): @@ -1338,21 +1357,11 @@ class ContentRouter(Transform): result.compressed, len(result.compressed.split()), ) - smart_crusher_fallback = False - if result.compressed == content: - strategy_chain.append(CompressionStrategy.KOMPRESS.value) - fallback_compressed, fallback_tokens = self._try_ml_compressor( - content, context, question - ) - if fallback_tokens < compressed_tokens: - compressed = fallback_compressed - compressed_tokens = fallback_tokens - actual_strategy = CompressionStrategy.KOMPRESS - compressor_name = "KompressCompressor" - decision_reason = "smart_crusher_fallback_kompress_after_no_savings" - smart_crusher_fallback = True - if not smart_crusher_fallback: - decision_reason = "smart_crusher" + decision_reason = "smart_crusher" + # Fallback to Kompress (and possibly Log) is + # handled by the unified post-strategy block below + # — no inline fallback here to avoid duplicate + # Kompress invocations. elif strategy == CompressionStrategy.SEARCH: if self.config.enable_search_compressor: @@ -1435,10 +1444,19 @@ class ContentRouter(Transform): } fallback_no_savings = compressed == content or compressed_tokens >= original_tokens if fallback_eligible_strategy and fallback_no_savings: - strategy_chain.append(CompressionStrategy.KOMPRESS.value) - fallback_compressed, fallback_tokens = self._try_ml_compressor( - content, context, question - ) + # Skip if Kompress was already tried by an inline fallback + # (e.g. CODE_AWARE's code-compressor-unavailable path at + # line 1249). Prevents a duplicate strategy_chain entry + # and a wasted second _try_ml_compressor call. + already_tried_kompress = CompressionStrategy.KOMPRESS.value in strategy_chain + if not already_tried_kompress: + strategy_chain.append(CompressionStrategy.KOMPRESS.value) + fallback_compressed, fallback_tokens = self._try_ml_compressor( + content, context, question + ) + else: + fallback_compressed = compressed + fallback_tokens = compressed_tokens if fallback_tokens < compressed_tokens: compressed = fallback_compressed compressed_tokens = fallback_tokens @@ -2800,74 +2818,26 @@ class ContentRouter(Transform): route_counts["already_compressed"] += 1 continue - # Two-tier compression cache - content_key = hash(tool_content) - - # Tier 1: skip set — instant rejection - if self._cache.is_skipped(content_key): - new_blocks.append(block) - if route_counts is not None: - route_counts["ratio_too_high"] += 1 - route_counts.setdefault("cache_hit", 0) - route_counts["cache_hit"] += 1 - continue - - # Tier 2: result cache — reuse compressed output - cached = self._cache.get(content_key) - if cached is not None: - cached_compressed, cached_ratio, cached_strategy = cached - if cached_ratio < min_ratio: - new_blocks.append({**block, "content": cached_compressed}) - transforms_applied.append(f"router:tool_result:{cached_strategy}") - if compressed_details is not None: - compressed_details.append( - f"tool:{cached_strategy}:{cached_ratio:.2f}" - ) - any_compressed = True - else: - # Threshold tightened — move to skip - self._cache.move_to_skip(content_key) - new_blocks.append(block) - if route_counts is not None: - route_counts["ratio_too_high"] += 1 - if route_counts is not None: - route_counts.setdefault("cache_hit", 0) - route_counts["cache_hit"] += 1 - continue - - # Cache miss — run full compression - if route_counts is not None: - route_counts.setdefault("cache_miss", 0) - route_counts["cache_miss"] += 1 - t0 = time.perf_counter() - result = self.compress(tool_content, context=context, bias=bias) - compress_ms = (time.perf_counter() - t0) * 1000 - if compressor_timing is not None: - key = f"compressor:{result.strategy_used.value}" - compressor_timing[key] = compressor_timing.get(key, 0.0) + compress_ms - if result.compression_ratio < min_ratio: - # Compressed — store in result cache - self._cache.put( - content_key, - result.compressed, - result.compression_ratio, - result.strategy_used.value, - ) - new_blocks.append({**block, "content": result.compressed}) - transforms_applied.append( - f"router:tool_result:{result.strategy_used.value}" - ) - if compressed_details is not None: - compressed_details.append( - f"tool:{result.strategy_used.value}:{result.compression_ratio:.2f}" - ) + # Two-tier compression cache → shared helper + compressed_content, was_compressed = self._compress_block_content( + content=tool_content, + content_key=hash(tool_content), + context=context, + bias=bias, + min_ratio=min_ratio, + compressor_timing=compressor_timing, + transforms_applied=transforms_applied, + route_counts=route_counts, + compressed_details=compressed_details, + strategy_label="tool_result", + details_prefix="tool", + ) + if compressed_content is not None: + new_blocks.append({**block, "content": compressed_content}) any_compressed = True - continue else: - # Didn't compress — add to skip set - self._cache.mark_skip(content_key) - if route_counts is not None: - route_counts["ratio_too_high"] += 1 + new_blocks.append(block) + continue else: if route_counts is not None: route_counts["small"] += 1 @@ -2891,68 +2861,26 @@ class ContentRouter(Transform): route_counts["already_compressed"] += 1 continue - content_key = hash(text_content) - - # Tier 1: skip set - if self._cache.is_skipped(content_key): - new_blocks.append(block) - if route_counts is not None: - route_counts["ratio_too_high"] += 1 - route_counts.setdefault("cache_hit", 0) - route_counts["cache_hit"] += 1 - continue - - # Tier 2: result cache - cached = self._cache.get(content_key) - if cached is not None: - cached_compressed, cached_ratio, cached_strategy = cached - if cached_ratio < min_ratio: - new_blocks.append({**block, "text": cached_compressed}) - transforms_applied.append(f"router:text_block:{cached_strategy}") - if compressed_details is not None: - compressed_details.append( - f"text:{cached_strategy}:{cached_ratio:.2f}" - ) - any_compressed = True - else: - self._cache.move_to_skip(content_key) - new_blocks.append(block) - if route_counts is not None: - route_counts["ratio_too_high"] += 1 - if route_counts is not None: - route_counts.setdefault("cache_hit", 0) - route_counts["cache_hit"] += 1 - continue - - # Cache miss — full compression - if route_counts is not None: - route_counts.setdefault("cache_miss", 0) - route_counts["cache_miss"] += 1 - t0 = time.perf_counter() - result = self.compress(text_content, context=context, bias=1.0) - compress_ms = (time.perf_counter() - t0) * 1000 - if compressor_timing is not None: - key = f"compressor:{result.strategy_used.value}" - compressor_timing[key] = compressor_timing.get(key, 0.0) + compress_ms - if result.compression_ratio < min_ratio: - self._cache.put( - content_key, - result.compressed, - result.compression_ratio, - result.strategy_used.value, - ) - new_blocks.append({**block, "text": result.compressed}) - transforms_applied.append(f"router:text_block:{result.strategy_used.value}") - if compressed_details is not None: - compressed_details.append( - f"text:{result.strategy_used.value}:{result.compression_ratio:.2f}" - ) + # Two-tier compression cache → shared helper + compressed_content, _was_compressed = self._compress_block_content( + content=text_content, + content_key=hash(text_content), + context=context, + bias=1.0, + min_ratio=min_ratio, + compressor_timing=compressor_timing, + transforms_applied=transforms_applied, + route_counts=route_counts, + compressed_details=compressed_details, + strategy_label="text_block", + details_prefix="text", + ) + if compressed_content is not None: + new_blocks.append({**block, "text": compressed_content}) any_compressed = True - continue else: - self._cache.mark_skip(content_key) - if route_counts is not None: - route_counts["ratio_too_high"] += 1 + new_blocks.append(block) + continue else: if route_counts is not None: route_counts["small"] += 1 @@ -2964,6 +2892,102 @@ class ContentRouter(Transform): return {**message, "content": new_blocks} return message + def _compress_block_content( + self, + content: str, + content_key: int, + context: str, + bias: float, + min_ratio: float, + compressor_timing: dict[str, float] | None, + transforms_applied: list[str], + route_counts: dict[str, int] | None, + compressed_details: list[str] | None, + strategy_label: str, + details_prefix: str, + ) -> tuple[str | None, bool]: + """Apply two-tier cache lookup + compression to a single content string. + + Encapsulates the shared cache→compress→store logic used by both + ``tool_result`` and ``text`` block paths in ``_process_content_blocks``. + Previously this logic was duplicated ~60 lines per path; centralising + it ensures both paths stay in sync (cache expiry, pinning, ratio gating). + + Args: + content: The string content to compress. + content_key: Pre-computed ``hash(content)`` for cache lookups. + context: User/query context for relevance-aware compression. + bias: Compression bias multiplier (tool-specific or 1.0). + min_ratio: Adaptive minimum compression ratio threshold. + compressor_timing: Optional dict to accumulate per-strategy timing. + transforms_applied: List mutated in-place with transform labels. + route_counts: Optional dict mutated in-place with route counters. + compressed_details: Optional list mutated with compression details. + strategy_label: Transform label prefix (e.g. ``"tool_result"``). + details_prefix: Compressed-details prefix (e.g. ``"tool"``). + + Returns: + Tuple of ``(compressed_content_or_None, was_compressed)``. + When ``compressed_content`` is ``None`` the caller should keep + the original block unchanged. When ``was_compressed`` is + ``True`` the caller should update the block with the returned + content and set ``any_compressed``. + """ + # Tier 1: skip set — instant rejection + if self._cache.is_skipped(content_key): + if route_counts is not None: + route_counts["ratio_too_high"] = route_counts.get("ratio_too_high", 0) + 1 + route_counts["cache_hit"] = route_counts.get("cache_hit", 0) + 1 + return None, False + + # Tier 2: result cache — reuse compressed output + cached = self._cache.get(content_key) + if cached is not None: + cached_compressed, cached_ratio, cached_strategy = cached + if route_counts is not None: + route_counts["cache_hit"] = route_counts.get("cache_hit", 0) + 1 + if cached_ratio < min_ratio: + transforms_applied.append(f"router:{strategy_label}:{cached_strategy}") + if compressed_details is not None: + compressed_details.append( + f"{details_prefix}:{cached_strategy}:{cached_ratio:.2f}" + ) + return cached_compressed, True + # Threshold tightened — move result to skip set + self._cache.move_to_skip(content_key) + if route_counts is not None: + route_counts["ratio_too_high"] = route_counts.get("ratio_too_high", 0) + 1 + return None, False + + # Cache miss — run full compression + if route_counts is not None: + route_counts["cache_miss"] = route_counts.get("cache_miss", 0) + 1 + t0 = time.perf_counter() + result = self.compress(content, context=context, bias=bias) + compress_ms = (time.perf_counter() - t0) * 1000 + if compressor_timing is not None: + key = f"compressor:{result.strategy_used.value}" + compressor_timing[key] = compressor_timing.get(key, 0.0) + compress_ms + if result.compression_ratio < min_ratio: + # Compressed — store in result cache + self._cache.put( + content_key, + result.compressed, + result.compression_ratio, + result.strategy_used.value, + ) + transforms_applied.append(f"router:{strategy_label}:{result.strategy_used.value}") + if compressed_details is not None: + compressed_details.append( + f"{details_prefix}:{result.strategy_used.value}:{result.compression_ratio:.2f}" + ) + return result.compressed, True + # Didn't compress enough — add to skip set + self._cache.mark_skip(content_key) + if route_counts is not None: + route_counts["ratio_too_high"] = route_counts.get("ratio_too_high", 0) + 1 + return None, False + def _detect_analysis_intent(self, messages: list[dict[str, Any]]) -> bool: """Detect if user wants to analyze/review code. diff --git a/tests/test_compression_cache.py b/tests/test_compression_cache.py index 66833fe8d..20cfc82cc 100644 --- a/tests/test_compression_cache.py +++ b/tests/test_compression_cache.py @@ -554,6 +554,94 @@ class TestCompressionCacheConcurrency: assert stats["entries"] == n_threads * per_thread_calls assert stats["tokens_saved"] > 0 + def test_concurrent_hits_misses_consistent(self) -> None: + """Under concurrent reads + writes, hits+misses must be bounded by + total lookups (hits ≤ entries, misses ≥ 0 at all moments).""" + import random + import threading + + cache = CompressionCache(max_entries=1_000_000) + n_threads = 16 + per_thread = 50 + + # Pre-populate so reads have something to hit + for i in range(per_thread): + h = CompressionCache.content_hash(f"hit-{i}") + cache.store_compressed(h, f"comp-{i}", tokens_saved=3) + + errors: list[Exception] = [] + barrier = threading.Barrier(n_threads) + + def worker(tid: int) -> None: + try: + barrier.wait() + for i in range(per_thread): + if random.random() < 0.6: + # Read path + _ = cache.get_compressed( + CompressionCache.content_hash( + f"hit-{random.randint(0, per_thread - 1)}" + ) + ) + else: + # Write path + h = CompressionCache.content_hash(f"write-{tid}-{i}") + cache.store_compressed(h, f"w-{tid}-{i}", tokens_saved=1) + except Exception as e: # pragma: no cover + errors.append(e) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], f"Concurrent reads+writes raised: {errors}" + stats = cache.get_stats() + # hits + misses should be non-negative (sanity) + assert stats["hits"] >= 0 + assert stats["misses"] >= 0 + assert stats["entries"] > 0 + + def test_concurrent_stable_hash_ops_no_race(self) -> None: + """Concurrent mark_stable_from_messages + compute_frozen_count must + not race — stable_hashes must remain self-consistent.""" + import threading + + cache = CompressionCache() + n_threads = 12 + per_thread = 30 + + # Each thread has its own content; produce tool_result messages + # and mark them stable, then verify frozen count. + errors: list[Exception] = [] + barrier = threading.Barrier(n_threads) + + def worker(tid: int) -> None: + try: + barrier.wait() + for i in range(per_thread): + content = f"stable-content-{tid}-{i}" + h = CompressionCache.content_hash(content) + # Also store to make it appear cached + cache.store_compressed(h, f"comp-{tid}-{i}", tokens_saved=2) + # Mark stable + cache.mark_stable(h) + except Exception as e: # pragma: no cover + errors.append(e) + + threads = [threading.Thread(target=worker, args=(t,)) for t in range(n_threads)] + for t in threads: + t.start() + for t in threads: + t.join() + + assert errors == [], f"Concurrent stable-hash ops raised: {errors}" + stats = cache.get_stats() + # All entries should be recorded; stable_hashes should match entries + # (every store_compressed was followed by mark_stable in our test) + assert stats["entries"] == n_threads * per_thread + def test_get_compression_cache_returns_same_instance_under_contention() -> None: """`HeadroomProxy._get_compression_cache(session_id)` must return the diff --git a/tests/test_transforms/test_content_router.py b/tests/test_transforms/test_content_router.py index f93cf097a..8dfbe9ae0 100644 --- a/tests/test_transforms/test_content_router.py +++ b/tests/test_transforms/test_content_router.py @@ -878,3 +878,438 @@ class TestExcludeTools: # OtherTool may or may not be compressed, but should be processed # (we just verify it wasn't excluded) assert "router:excluded:tool" in result.transforms_applied + + def test_bash_not_in_default_exclude_tools(self): + """Bash is NOT excluded by default — its outputs (build logs, test + output) are ideal compression targets. Regression test for PR #704. + + This test validates the DEFAULT_EXCLUDE_TOOLS frozenset directly + (pure config check — no Rust dependency). + """ + from headroom.config import DEFAULT_EXCLUDE_TOOLS + + assert "Bash" not in DEFAULT_EXCLUDE_TOOLS, ( + "Bash should NOT be in DEFAULT_EXCLUDE_TOOLS — " + "its outputs (build logs, test output) are ideal compression targets" + ) + assert "bash" not in DEFAULT_EXCLUDE_TOOLS, "'bash' should NOT be in DEFAULT_EXCLUDE_TOOLS" + + def test_bash_lowercase_not_in_exclude_tools(self): + """Lowercase 'bash' is also NOT in default exclude tools.""" + from headroom.config import DEFAULT_EXCLUDE_TOOLS + + assert "bash" not in DEFAULT_EXCLUDE_TOOLS + + def test_default_exclude_tools_membership(self): + """Verify all expected exclude tools and their lowercase variants.""" + from headroom.config import DEFAULT_EXCLUDE_TOOLS + + # Tools that SHOULD be excluded (fresh Read/Write/Edit/Glob/Grep outputs) + for tool in ("Read", "Glob", "Grep", "Write", "Edit"): + assert tool in DEFAULT_EXCLUDE_TOOLS, f"{tool} should be in DEFAULT_EXCLUDE_TOOLS" + assert tool.lower() in DEFAULT_EXCLUDE_TOOLS, ( + f"{tool.lower()} should be in DEFAULT_EXCLUDE_TOOLS" + ) + + # Tools that should NOT be excluded + for tool in ("Bash", "bash", "TodoWrite", "todo_write"): + assert tool not in DEFAULT_EXCLUDE_TOOLS, ( + f"{tool} should NOT be in DEFAULT_EXCLUDE_TOOLS" + ) + + +# ============================================================================= +# TestSmartCrusherFallback — PR #704 regression suite +# ============================================================================= + + +class TestSmartCrusherFallback: + """Verify SmartCrusher→Kompress→Log fallback chain. + + The post-strategy unified fallback block (added in PR #704) replaces + inline duplicate Kompress invocations. When SmartCrusher returns no + savings, the unified block tries Kompress, then Log (structurally + repetitive content), without double-invoking Kompress. + + Uses ``_apply_strategy_to_content`` + monkeypatched fallback + compressors to avoid network/ML-model downloads in test environments. + """ + + def test_smart_crusher_with_no_savings_triggers_kompress_fallback(self, router, monkeypatch): + """When SmartCrusher produces no savings (returns content unchanged), + the unified post-strategy block must fire Kompress fallback. + + Monkeypatches ``_get_smart_crusher`` to return a mock whose + ``crush()`` returns *content* unchanged — this simulates "ran + but produced no savings" without depending on the Rust + ``headroom._core`` extension or an LLM round-trip. + """ + from unittest.mock import MagicMock + + import headroom.transforms.content_router as crm + from headroom.transforms.smart_crusher import CrushResult + + content = "this is repetitive text. " * 300 + + # Mock SmartCrusher: ran successfully but returned content as-is + # (no savings), so the unified fallback block is entered. + mock_crush_result = CrushResult( + compressed=content, + original=content, + was_modified=False, + strategy="passthrough", + ) + mock_crusher = MagicMock() + mock_crusher.crush.return_value = mock_crush_result + monkeypatch.setattr( + crm.ContentRouter, + "_get_smart_crusher", + lambda self: mock_crusher, + ) + + # Patch _try_ml_compressor to simulate Kompress also returning + # unchanged (no savings), forcing the full chain to exercise + monkeypatch.setattr( + crm.ContentRouter, + "_try_ml_compressor", + lambda self, c, context="", question=None: ( + c, + len(c.split()), + ), + ) + + compressed, compressed_tokens, strategy_chain = router._apply_strategy_to_content( + content, + CompressionStrategy.SMART_CRUSHER, + context="", + ) + + # Strategy chain must include smart_crusher + assert "smart_crusher" in strategy_chain + # Kompress fallback should have been attempted + assert "kompress" in strategy_chain, ( + f"Expected kompress in chain {strategy_chain} — " + f"unified post-strategy block should have fired" + ) + + def test_smart_crusher_json_compresses_directly(self, router, monkeypatch): + """When SmartCrusher successfully compresses JSON, the chain is + just [smart_crusher] with no fallback entries. + + Uses a mock SmartCrusher to avoid depending on the Rust + ``headroom._core`` extension in test environments. + """ + import json + from unittest.mock import MagicMock + + import headroom.transforms.content_router as crm + from headroom.transforms.smart_crusher import CrushResult + + content = json.dumps([{"id": i, "name": f"item_{i}", "value": i * 10} for i in range(100)]) + + # Mock SmartCrusher: simulated compression (shorter output) + mock_compressed = json.dumps([{"id": i, "name": f"item_{i}"} for i in range(50)]) + mock_crush_result = CrushResult( + compressed=mock_compressed, + original=content, + was_modified=True, + strategy="smart_crusher", + ) + mock_crusher = MagicMock() + mock_crusher.crush.return_value = mock_crush_result + monkeypatch.setattr( + crm.ContentRouter, + "_get_smart_crusher", + lambda self: mock_crusher, + ) + + compressed, compressed_tokens, strategy_chain = router._apply_strategy_to_content( + content, + CompressionStrategy.SMART_CRUSHER, + context="", + ) + + # SmartCrusher should handle JSON directly + assert "smart_crusher" in strategy_chain + # With real savings, no fallback should be triggered + assert "kompress" not in strategy_chain + assert len(compressed.strip()) > 0 + + def test_post_strategy_block_no_duplicate_kompress(self, router, monkeypatch): + """The unified post-strategy block must NOT produce duplicate + 'kompress' entries in the strategy chain. + + Pre-PR #704: an inline duplicate Kompress fallback existed for + SmartCrusher that could fire alongside the post-strategy block, + causing 'kompress' to appear twice in the chain. + + Uses a mock SmartCrusher returning no savings so the fallback + block is entered deterministically, without depending on the + Rust ``headroom._core`` extension. + """ + from unittest.mock import MagicMock + + import headroom.transforms.content_router as crm + from headroom.transforms.smart_crusher import CrushResult + + repetitive = "line " * 300 + "\n" + + # Mock SmartCrusher: ran successfully but returned content as-is + # (no savings) — fallback block must fire. + mock_crush_result = CrushResult( + compressed=repetitive, + original=repetitive, + was_modified=False, + strategy="passthrough", + ) + mock_crusher = MagicMock() + mock_crusher.crush.return_value = mock_crush_result + monkeypatch.setattr( + crm.ContentRouter, + "_get_smart_crusher", + lambda self: mock_crusher, + ) + + # Monkeypatch Kompress to return unchanged (no savings), + # forcing the full fallback chain without network access + monkeypatch.setattr( + crm.ContentRouter, + "_try_ml_compressor", + lambda self, c, context="", question=None: ( + c, + len(c.split()), + ), + ) + + compressed, compressed_tokens, strategy_chain = router._apply_strategy_to_content( + repetitive, + CompressionStrategy.SMART_CRUSHER, + context="", + ) + + # The chain must include the requested strategy + assert "smart_crusher" in strategy_chain + + # No duplicate "kompress" entries — the key regression check + kompress_count = strategy_chain.count("kompress") + assert kompress_count <= 1, ( + f"Kompress appeared {kompress_count} times in chain; " + f"duplicate fallback suggests inline+post-strategy both fired: " + f"{strategy_chain}" + ) + + def test_code_aware_fallback_also_uses_unified_block(self, router, monkeypatch): + """CodeAware strategy also uses the unified fallback block. + Verify it doesn't double-invoke Kompress either.""" + import headroom.transforms.content_router as crm + + monkeypatch.setattr( + crm.ContentRouter, + "_try_ml_compressor", + lambda self, content, context="", question=None: ( + content, + len(content.split()), + ), + ) + + plain = "This is just plain text. " * 200 + + compressed, compressed_tokens, strategy_chain = router._apply_strategy_to_content( + plain, + CompressionStrategy.CODE_AWARE, + context="", + ) + + # CodeAware should be in the chain + assert "code_aware" in strategy_chain + + # No duplicate fallback entries + kompress_count = strategy_chain.count("kompress") + assert kompress_count <= 1, ( + f"Kompress appeared {kompress_count} times; " + f"duplicate fallback in CodeAware path: {strategy_chain}" + ) + + +# ============================================================================= +# TestCompressBlockContent — PR #704 shared-path regression +# ============================================================================= + + +class TestCompressBlockContent: + """Verify `_compress_block_content` shared path for tool_result and text blocks. + + Before PR #704, the two block paths had ~60 lines of duplicate cache + logic each. The shared helper ensures both paths stay in sync (cache + expiry, pinning, ratio gating). + + Tests target the two-tier ``CompressionCache`` (content_router-local, + line 191) and the ``_compress_block_content`` method directly, + avoiding the Rust content-detection extension by pre-populating + the cache and verifying cache-hit/skip behaviour. + """ + + @pytest.fixture + def router_with_cache(self): + """ContentRouter with all compressors enabled.""" + config = ContentRouterConfig( + enable_smart_crusher=True, + enable_kompress=True, + enable_log_compressor=True, + min_section_tokens=10, + ) + return ContentRouter(config) + + def test_skip_set_prevents_recompression(self, router_with_cache): + """Tier 1 (skip set): content_key in the skip set returns + (None, False) immediately — no compression attempted.""" + cache = router_with_cache._cache + key = hash("test-content-that-wont-compress") + + # Mark as skipped + cache.mark_skip(key) + assert cache.is_skipped(key) is True + + # _compress_block_content should return early on skip + compressed, was_compressed = router_with_cache._compress_block_content( + content="test-content-that-wont-compress", + content_key=key, + context="", + bias=1.0, + min_ratio=0.5, + compressor_timing=None, + transforms_applied=[], + route_counts=None, + compressed_details=None, + strategy_label="test", + details_prefix="test", + ) + + assert compressed is None + assert was_compressed is False + + def test_result_cache_hit_returns_cached(self, router_with_cache): + """Tier 2 (result cache): cached content is returned without + re-running compression.""" + cache = router_with_cache._cache + key = hash("cacheable-content") + original = "compressed-version-of-content" + + # Populate result cache + cache.put(key, original, ratio=0.3, strategy="kompress") + assert cache.get(key) == (original, 0.3, "kompress") + + # _compress_block_content should return cached result + compressed, was_compressed = router_with_cache._compress_block_content( + content="cacheable-content", + content_key=key, + context="", + bias=1.0, + min_ratio=0.5, + compressor_timing=None, + transforms_applied=[], + route_counts=None, + compressed_details=None, + strategy_label="test", + details_prefix="test", + ) + + assert compressed == original + assert was_compressed is True + + def test_result_cache_ratio_above_min_moves_to_skip(self, router_with_cache): + """When the cached ratio is ≥ min_ratio, the entry is moved from + Tier 2 to Tier 1 (skip set) — ratio threshold has tightened.""" + cache = router_with_cache._cache + key = hash("borderline-content") + + # Cached with ratio 0.8 (high — barely compressed) + cache.put(key, "slightly-compressed", ratio=0.8, strategy="text") + + # min_ratio=0.7 — cached ratio (0.8) ≥ threshold → move to skip + compressed, was_compressed = router_with_cache._compress_block_content( + content="borderline-content", + content_key=key, + context="", + bias=1.0, + min_ratio=0.7, + compressor_timing=None, + transforms_applied=[], + route_counts=None, + compressed_details=None, + strategy_label="test", + details_prefix="test", + ) + + assert compressed is None, "Should move to skip when ratio ≥ min_ratio" + assert was_compressed is False + assert cache.is_skipped(key), "Entry should now be in skip set" + assert cache.get(key) is None, "Entry should be removed from result cache" + + def test_compress_block_content_route_counts_mutated(self, router_with_cache): + """route_counts dict is mutated in-place with cache hit/miss info.""" + cache = router_with_cache._cache + key_skip = hash("skip-content") + key_hit = hash("hit-content") + + cache.mark_skip(key_skip) + cache.put(key_hit, "compressed", ratio=0.3, strategy="kompress") + + route_counts: dict[str, int] = {} + + # Skip hit + router_with_cache._compress_block_content( + content="skip-content", + content_key=key_skip, + context="", + bias=1.0, + min_ratio=0.5, + compressor_timing=None, + transforms_applied=[], + route_counts=route_counts, + compressed_details=None, + strategy_label="test", + details_prefix="test", + ) + assert route_counts.get("ratio_too_high", 0) >= 1 + assert route_counts.get("cache_hit", 0) >= 1 + + # Cache hit + router_with_cache._compress_block_content( + content="hit-content", + content_key=key_hit, + context="", + bias=1.0, + min_ratio=0.5, + compressor_timing=None, + transforms_applied=[], + route_counts=route_counts, + compressed_details=None, + strategy_label="test", + details_prefix="test", + ) + + def test_compress_block_content_transforms_applied_mutated(self, router_with_cache): + """transforms_applied list is mutated with strategy info on cache hit.""" + cache = router_with_cache._cache + key = hash("transform-test-content") + cache.put(key, "short", ratio=0.25, strategy="kompress") + + transforms_applied: list[str] = [] + router_with_cache._compress_block_content( + content="transform-test-content", + content_key=key, + context="", + bias=1.0, + min_ratio=0.5, + compressor_timing=None, + transforms_applied=transforms_applied, + route_counts=None, + compressed_details=None, + strategy_label="tool_result", + details_prefix="tool", + ) + + assert any("router:tool_result" in t for t in transforms_applied), ( + f"Expected router:tool_result:* in transforms, got: {transforms_applied}" + ) diff --git a/tests/test_transforms_content_router.py b/tests/test_transforms_content_router.py index 01c3732c6..0d16bb29e 100644 --- a/tests/test_transforms_content_router.py +++ b/tests/test_transforms_content_router.py @@ -25,7 +25,7 @@ def test_compression_cache_handles_hits_skips_evictions_and_clear( monkeypatch: pytest.MonkeyPatch, ) -> None: times = iter([100.0, 100.0, 100.0, 100.0, 100.0, 100.0, 112.0, 112.0]) - monkeypatch.setattr(content_router_module.time, "time", lambda: next(times)) + monkeypatch.setattr(content_router_module.time, "monotonic", lambda: next(times)) monkeypatch.setattr(content_router_module.time, "perf_counter_ns", lambda: 50) cache = CompressionCache(ttl_seconds=10)