"""Tests for ContentRouter - intelligent content-based compression routing. Comprehensive tests covering: - ContentRouterConfig: Configuration validation and defaults - ContentRouter: Core routing functionality - Strategy detection: Code, JSON, search, logs, text - Mixed content handling: Split, route, reassemble - Transform interface: apply(), should_apply() methods """ import json import logging from types import SimpleNamespace import pytest from headroom.transforms.content_detector import ContentType from headroom.transforms.content_router import ( CompressionStrategy, ContentRouter, ContentRouterConfig, RouterCompressionResult, RoutingDecision, ) from headroom.transforms.lossless_compaction import search_unheading # ============================================================================= # Test Fixtures # ============================================================================= @pytest.fixture def default_config(): """Default ContentRouterConfig for testing.""" return ContentRouterConfig( min_section_tokens=10, # Low threshold for tests ) @pytest.fixture def router(default_config): """ContentRouter instance with default config.""" return ContentRouter(default_config) @pytest.fixture def tokenizer(): """Get a tokenizer for Transform interface tests.""" from headroom.providers import OpenAIProvider from headroom.tokenizer import Tokenizer provider = OpenAIProvider() token_counter = provider.get_token_counter("gpt-4o") return Tokenizer(token_counter, "gpt-4o") # ============================================================================= # Test Data Generators # ============================================================================= def generate_python_code(n_functions: int = 5) -> str: """Generate Python code for testing.""" lines = [ '"""Module with functions."""', "import os", "from typing import Any", "", ] for i in range(n_functions): lines.extend( [ f"def function_{i}(arg: Any) -> str:", f' """Process argument {i}."""', " return str(arg)", "", ] ) return "\n".join(lines) def generate_json_data(n_items: int = 20) -> str: """Generate JSON content for testing.""" import json items = [ {"id": i, "name": f"Item {i}", "value": i * 10, "active": i % 2 == 0} for i in range(n_items) ] return json.dumps(items, indent=2) def generate_search_results(n_results: int = 10) -> str: """Generate grep/search-like results for testing.""" lines = [] for i in range(n_results): lines.append(f"src/module{i}.py:42: def process_data(input: str) -> str:") lines.append(f"src/module{i}.py:43: return transform(input)") return "\n".join(lines) def generate_log_output(n_lines: int = 30) -> str: """Generate build/test log output for testing.""" lines = [ "Running tests...", "=== Test Suite: Unit Tests ===", ] for i in range(n_lines): if i % 10 == 0: lines.append(f"PASS tests/test_module{i}.py::test_function") elif i % 15 == 0: lines.append(f"FAIL tests/test_module{i}.py::test_failing") else: lines.append(f" Running test_{i}... ok") lines.append("=== Summary ===") lines.append(f"Tests: {n_lines}, Passed: {n_lines - 2}, Failed: 2") return "\n".join(lines) def generate_mixed_content() -> str: """Generate content with mixed types (markdown with code).""" return """# Documentation This is a README file with code examples. ## Python Example ```python def example(): return "hello" ``` ## JSON Configuration ```json {"key": "value", "number": 42} ``` ## Usage Run the following command: ```bash python main.py --verbose ``` That's all! """ def test_force_kompress_routes_anthropic_tool_result_to_targeted_kompress( router, tokenizer, monkeypatch ): captured: dict[str, object] = {} class FakeKompress: def is_ready(self) -> bool: return True def ensure_background_load(self) -> None: pass def compress(self, content, **kwargs): captured.update(kwargs) # Real Kompress appends a CCR retrieval marker when CCR is enabled, # keeping the lossy result recoverable. Include one so the router's # reversibility gate (tool ground truth must stay recoverable, #1307) # accepts the compression instead of reverting to verbatim. compressed = " ".join(content.split()[:20]) + " Retrieve more: hash=deadbeef" return SimpleNamespace( compressed=compressed, compressed_tokens=len(compressed.split()), ) monkeypatch.setattr(router, "_get_kompress", lambda: FakeKompress()) tool_content = " ".join( f'{{"file":"src/module_{i}.py","line":{i},"text":"repeated search payload"}}' for i in range(160) ) messages = [ { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_search_1", "content": tool_content, } ], } ] result = router.apply( messages, tokenizer, force_kompress=True, target_ratio=0.10, compress_user_messages=True, min_tokens_to_compress=10, read_protection_window=0, ) assert result.messages[0]["content"][0]["content"] != tool_content assert result.transforms_applied == ["router:tool_result:kompress"] assert captured["target_ratio"] == 0.10 def test_anthropic_tool_result_lossy_without_marker_stays_verbatim(router, tokenizer, monkeypatch): """Reversibility gate (#1307): a lossy Kompress result on a tool_result block with no CCR retrieval marker is unrecoverable, so the router must keep the original verbatim rather than hand the agent a fabricated summary. This is the block-path counterpart to the string/`role=="tool"` guard.""" class FakeKompress: def is_ready(self) -> bool: return True def ensure_background_load(self) -> None: pass def compress(self, content, **kwargs): # Lossy summary with NO retrieval marker → unrecoverable. compressed = " ".join(content.split()[:20]) return SimpleNamespace( compressed=compressed, compressed_tokens=len(compressed.split()), ) monkeypatch.setattr(router, "_get_kompress", lambda: FakeKompress()) tool_content = " ".join( f'{{"file":"src/module_{i}.py","line":{i},"text":"repeated search payload"}}' for i in range(160) ) messages = [ { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_search_1", "content": tool_content, } ], } ] result = router.apply( messages, tokenizer, force_kompress=True, target_ratio=0.10, compress_user_messages=True, min_tokens_to_compress=10, read_protection_window=0, ) # Unrecoverable lossy compression is rejected → original kept verbatim. assert result.messages[0]["content"][0]["content"] == tool_content # ============================================================================= # TestContentRouterConfig # ============================================================================= class TestContentRouterConfig: """Tests for ContentRouterConfig dataclass.""" def test_default_values(self): """Default config values are sensible.""" config = ContentRouterConfig() assert config.enable_code_aware is False # Disabled by default; use code graph MCP instead assert config.enable_kompress is True assert config.enable_smart_crusher is True assert config.enable_search_compressor is True assert config.enable_log_compressor is True assert config.min_section_tokens == 20 assert config.fallback_strategy == CompressionStrategy.KOMPRESS def test_custom_values(self): """Custom config values are applied.""" config = ContentRouterConfig( min_section_tokens=50, enable_code_aware=False, fallback_strategy=CompressionStrategy.TEXT, ) assert config.min_section_tokens == 50 assert config.enable_code_aware is False assert config.fallback_strategy == CompressionStrategy.TEXT def test_all_strategies_in_enum(self): """All expected strategies are in the enum.""" expected = [ "CODE_AWARE", "SMART_CRUSHER", "SEARCH", "LOG", "TEXT", "MIXED", "PASSTHROUGH", ] actual = [s.name for s in CompressionStrategy] for strategy in expected: assert strategy in actual, f"Missing strategy: {strategy}" # ============================================================================= # TestRouterCompressionResult # ============================================================================= class TestRouterCompressionResult: """Tests for RouterCompressionResult dataclass.""" def test_tokens_saved_from_routing_log(self): """tokens_saved property calculates correctly from routing log.""" result = RouterCompressionResult( compressed="short", original="long content here", strategy_used=CompressionStrategy.CODE_AWARE, routing_log=[ RoutingDecision( content_type=ContentType.SOURCE_CODE, strategy=CompressionStrategy.CODE_AWARE, confidence=0.9, original_tokens=100, compressed_tokens=30, ) ], sections_processed=1, ) assert result.tokens_saved == 70 def test_tokens_saved_no_negative(self): """tokens_saved never returns negative.""" result = RouterCompressionResult( compressed="expanded", original="short", strategy_used=CompressionStrategy.PASSTHROUGH, routing_log=[ RoutingDecision( content_type=ContentType.PLAIN_TEXT, strategy=CompressionStrategy.PASSTHROUGH, confidence=1.0, original_tokens=10, compressed_tokens=20, # Expanded ) ], sections_processed=1, ) # Should be 0 not negative assert result.tokens_saved == 0 def test_savings_percentage(self): """savings_percentage property calculates correctly.""" result = RouterCompressionResult( compressed="short", original="long content", strategy_used=CompressionStrategy.TEXT, routing_log=[ RoutingDecision( content_type=ContentType.PLAIN_TEXT, strategy=CompressionStrategy.TEXT, confidence=0.8, original_tokens=100, compressed_tokens=25, ) ], sections_processed=1, ) assert result.savings_percentage == 75.0 def test_empty_routing_log(self): """Handles empty routing log gracefully.""" result = RouterCompressionResult( compressed="content", original="content", strategy_used=CompressionStrategy.PASSTHROUGH, routing_log=[], sections_processed=0, ) assert result.total_original_tokens == 0 assert result.total_compressed_tokens == 0 assert result.savings_percentage == 0.0 # ============================================================================= # TestStrategyDetection # ============================================================================= class TestStrategyDetection: """Tests for content type and strategy detection.""" def test_detect_python_code(self, router): """Python code is detected.""" code = generate_python_code(5) strategy = router._determine_strategy(code) # Should be either CODE_AWARE or fallback assert strategy in CompressionStrategy def test_detect_json_content(self, router): """JSON content is detected.""" json_data = generate_json_data(20) strategy = router._determine_strategy(json_data) assert strategy in CompressionStrategy def test_detect_search_results(self, router): """Search/grep results are detected.""" search_results = generate_search_results(10) strategy = router._determine_strategy(search_results) assert strategy in CompressionStrategy def test_detect_log_output(self, router): """Build/test logs are detected.""" logs = generate_log_output(30) strategy = router._determine_strategy(logs) assert strategy in CompressionStrategy def test_detect_plain_text(self, router): """Plain text detection.""" text = "This is just plain text without any special formatting." strategy = router._determine_strategy(text) assert strategy in CompressionStrategy # ============================================================================= # TestContentRouter # ============================================================================= class TestContentRouter: """Tests for ContentRouter core functionality.""" def test_init_with_default_config(self): """Router initializes with default config.""" router = ContentRouter() assert router.config is not None assert router.config.enable_code_aware is False # Disabled by default def test_init_with_custom_config(self, default_config): """Router initializes with custom config.""" router = ContentRouter(default_config) assert router.config == default_config def test_compress_empty_content(self, router): """Empty content returns passthrough.""" result = router.compress("") assert result.compressed == "" assert result.strategy_used == CompressionStrategy.PASSTHROUGH def test_compress_small_content(self, router): """Small content returns same content.""" result = router.compress("small") assert result.compressed == "small" # Small content might use TEXT or PASSTHROUGH strategy assert result.strategy_used in ( CompressionStrategy.PASSTHROUGH, CompressionStrategy.TEXT, ) def test_compress_returns_result(self, router): """compress() returns RouterCompressionResult.""" content = generate_python_code(10) result = router.compress(content) assert isinstance(result, RouterCompressionResult) assert result.original == content assert result.strategy_used is not None def test_compress_diff_accepts_none_context_with_debug(self, router, caplog): """None context is normalized before debug logging and compressor dispatch.""" class FakeDiffCompressor: def compress(self, content, context): assert context == "" return SimpleNamespace(compressed="diff summary") diff = "diff --git a/file.py b/file.py\n@@ -1 +1 @@\n-old\n+new\n" router._diff_compressor = FakeDiffCompressor() caplog.set_level(logging.DEBUG, logger="headroom.transforms.content_router") result = router.compress(diff, context=None) assert result.compressed == "diff summary" assert result.strategy_used == CompressionStrategy.DIFF def test_name_property(self, router): """Router has correct name.""" assert router.name == "content_router" # ============================================================================= # TestTransformInterface # ============================================================================= class TestTransformInterface: """Tests for Transform interface (apply, should_apply).""" def test_should_apply_returns_bool(self, default_config, tokenizer): """should_apply returns a boolean.""" router = ContentRouter(default_config) messages = [{"role": "user", "content": "small"}] result = router.should_apply(messages, tokenizer) assert isinstance(result, bool) def test_should_apply_returns_true_for_large_content(self, default_config, tokenizer): """should_apply returns True for large content.""" router = ContentRouter(default_config) content = generate_python_code(20) messages = [{"role": "tool", "tool_call_id": "call_1", "content": content}] assert router.should_apply(messages, tokenizer) def test_apply_returns_transform_result(self, default_config, tokenizer): """apply() returns proper TransformResult.""" router = ContentRouter(default_config) content = generate_python_code(10) messages = [{"role": "tool", "tool_call_id": "call_1", "content": content}] result = router.apply(messages, tokenizer) assert result is not None assert result.tokens_before > 0 assert len(result.messages) == 1 def test_apply_passes_through_non_tool_messages(self, default_config, tokenizer): """apply() passes through non-tool messages unchanged.""" router = ContentRouter(default_config) messages = [ {"role": "user", "content": "Hello"}, {"role": "assistant", "content": "Hi there!"}, ] result = router.apply(messages, tokenizer) assert result.messages[0]["content"] == "Hello" assert result.messages[1]["content"] == "Hi there!" # ============================================================================= # TestCompressorDisabling # ============================================================================= class TestCompressorDisabling: """Tests for disabling specific compressors. Note: These tests verify the config is accepted, not that the router actually respects the disable flags (which may not be fully implemented). """ def test_config_accepts_disable_code_compression(self): """Config accepts enable_code_aware=False.""" config = ContentRouterConfig( enable_code_aware=False, min_section_tokens=10, ) router = ContentRouter(config) code = generate_python_code(10) # Should not crash result = router.compress(code) assert result is not None def test_config_accepts_disable_search_compression(self): """Config accepts enable_search_compressor=False.""" config = ContentRouterConfig( enable_search_compressor=False, min_section_tokens=10, ) router = ContentRouter(config) search_results = generate_search_results(10) # Should not crash result = router.compress(search_results) assert result is not None def test_config_accepts_disable_log_compression(self): """Config accepts enable_log_compressor=False.""" config = ContentRouterConfig( enable_log_compressor=False, min_section_tokens=10, ) router = ContentRouter(config) logs = generate_log_output(30) # Should not crash result = router.compress(logs) assert result is not None # ============================================================================= # TestEdgeCases # ============================================================================= class TestEdgeCases: """Edge case tests for ContentRouter.""" def test_whitespace_only_content(self, router): """Whitespace-only content is handled gracefully.""" result = router.compress(" \n\t\n ") assert result.strategy_used == CompressionStrategy.PASSTHROUGH def test_unicode_content(self, router): """Unicode content is handled correctly.""" content = "This has unicode: \u4e2d\u6587 \u65e5\u672c\u8a9e " * 50 result = router.compress(content) assert result is not None def test_very_long_content(self, router): """Very long content is handled.""" content = generate_python_code(100) result = router.compress(content) assert result is not None # ============================================================================= # TestRoutingLog # ============================================================================= class TestRoutingLog: """Tests for routing log functionality.""" def test_routing_log_populated(self, router): """Routing log is populated with decisions.""" content = generate_python_code(10) result = router.compress(content) # Routing log should be a list assert isinstance(result.routing_log, list) def test_routing_log_entries_have_strategy(self, router): """Routing log entries contain strategy.""" content = generate_python_code(10) result = router.compress(content) for entry in result.routing_log: assert hasattr(entry, "strategy") assert entry.strategy in CompressionStrategy # ============================================================================= # TestSummary # ============================================================================= class TestSummary: """Tests for result summary generation.""" def test_summary_property(self, router): """Summary property exists and is callable or returns string.""" content = generate_python_code(10) result = router.compress(content) # Check summary property exists assert hasattr(result, "summary") # Get summary (call if callable) summary = result.summary if callable(summary): summary = summary() # Should be a string assert summary is not None # ============================================================================= # TestExcludeTools # ============================================================================= class TestExcludeTools: """Tests for exclude_tools feature - bypassing compression for specific tools.""" @pytest.fixture def tokenizer(self): """Get a tokenizer for tests.""" from headroom.providers import OpenAIProvider from headroom.tokenizer import Tokenizer provider = OpenAIProvider() token_counter = provider.get_token_counter("gpt-4o") return Tokenizer(token_counter, "gpt-4o") def test_default_exclude_tools_uses_defaults(self, tokenizer): """Default config excludes DEFAULT_EXCLUDE_TOOLS (Read, Glob, etc).""" config = ContentRouterConfig(min_section_tokens=10) router = ContentRouter(config) # Create message with tool call from "Read" tool (should be excluded) messages = [ { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_read_1", "type": "function", "function": {"name": "Read", "arguments": "{}"}, } ], }, { "role": "tool", "tool_call_id": "call_read_1", "content": generate_python_code(20), # Large content that would normally compress }, ] result = router.apply(messages, tokenizer) # Content should be unchanged (passed through, not compressed) assert result.messages[1]["content"] == messages[1]["content"] # Check transform was marked as excluded assert "router:excluded:tool" in result.transforms_applied def test_custom_exclude_tools(self, tokenizer): """Custom exclude_tools set is respected.""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"MyCustomTool"}, # Only exclude this tool ) router = ContentRouter(config) # Create message with MyCustomTool (should be excluded) messages = [ { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_custom_1", "type": "function", "function": {"name": "MyCustomTool", "arguments": "{}"}, } ], }, { "role": "tool", "tool_call_id": "call_custom_1", "content": generate_json_data(50), }, ] result = router.apply(messages, tokenizer) # Excluded from *lossy* compression, but JSON still gets a data-lossless # minify (same object, fewer tokens). Assert recovery, not byte-identity. assert json.loads(result.messages[1]["content"]) == json.loads(messages[1]["content"]) assert "router:excluded:lossless_json" in result.transforms_applied def test_glob_exclude_tools(self, tokenizer): """Glob patterns in exclude_tools match by prefix (issue #870).""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"mcp__*"}, # One pattern excludes every MCP tool ) router = ContentRouter(config) messages = [ { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_mcp_1", "type": "function", "function": { "name": "mcp__build123d__measure", "arguments": "{}", }, } ], }, { "role": "tool", "tool_call_id": "call_mcp_1", "content": generate_json_data(50), }, ] result = router.apply(messages, tokenizer) # The MCP tool result matched the glob → excluded from lossy compression. # Its JSON still gets a data-lossless minify; assert recovery. assert json.loads(result.messages[1]["content"]) == json.loads(messages[1]["content"]) assert "router:excluded:lossless_json" in result.transforms_applied def test_anthropic_mcp_alias_exclude_tools(self, tokenizer): """Single-underscore MCP names from custom agents honor documented MCP globs.""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"mcp__*"}, ) router = ContentRouter(config) messages = [ { "role": "assistant", "content": [ { "type": "tool_use", "id": "toolu_mcp_1", "name": "mcp_CursorTaskRegistry_cursor_list_tasks", "input": {"project": "headroom"}, } ], }, { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_mcp_1", "content": generate_json_data(50), } ], }, ] result = router.apply(messages, tokenizer) tool_result_block = result.messages[1]["content"][0] assert json.loads(tool_result_block["content"]) == json.loads( messages[1]["content"][0]["content"] ) assert "router:excluded:lossless_json" in result.transforms_applied def test_anthropic_mcp_bare_tool_alias_exclude_tools(self, tokenizer): """Bare tool exclusions match custom-agent MCP wrappers (#1822).""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"headroom_retrieve"}, ) router = ContentRouter(config) messages = [ { "role": "assistant", "content": [ { "type": "tool_use", "id": "toolu_retrieve_1", "name": "mcp_HeadroomZai_headroom_retrieve", "input": {"key": "abc123"}, } ], }, { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_retrieve_1", "content": generate_json_data(50), } ], }, ] result = router.apply(messages, tokenizer) tool_result_block = result.messages[1]["content"][0] assert json.loads(tool_result_block["content"]) == json.loads( messages[1]["content"][0]["content"] ) assert "router:excluded:lossless_json" in result.transforms_applied def test_is_tool_excluded_helper(self): """is_tool_excluded: exact (case-insensitive) and glob matching.""" from headroom.config import is_tool_excluded # Glob entry covers a whole MCP server; unrelated tools are untouched. assert is_tool_excluded("mcp__build123d__measure", {"mcp__*"}) assert is_tool_excluded("mcp_CursorTaskRegistry_cursor_list_tasks", {"mcp__*"}) assert not is_tool_excluded("Bash", {"mcp__*"}) # Plain entries keep exact, case-insensitive membership. assert is_tool_excluded("Read", {"read"}) assert is_tool_excluded("MCP__X", {"mcp__*"}) # MCP wrapper aliases can still be excluded by their bare tool name. assert is_tool_excluded("mcp_HeadroomZai_headroom_retrieve", {"headroom_retrieve"}) assert is_tool_excluded("mcp__Headroom__headroom_retrieve", {"headroom_retrieve"}) # Empty set never excludes. assert not is_tool_excluded("Read", set()) def test_non_excluded_tools_are_compressed(self, tokenizer): """Tools not in exclude_tools set are still compressed.""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"Read"}, # Only exclude Read, not OtherTool ) router = ContentRouter(config) original_content = generate_json_data(100) # Large JSON array messages = [ { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_other_1", "type": "function", "function": {"name": "OtherTool", "arguments": "{}"}, } ], }, { "role": "tool", "tool_call_id": "call_other_1", "content": original_content, }, ] result = router.apply(messages, tokenizer) # Content should be compressed (different from original) # Note: Compression may or may not change the content depending on strategy # But it should NOT have the excluded marker assert "router:excluded:tool" not in result.transforms_applied def test_empty_exclude_tools_compresses_all(self, tokenizer): """Empty exclude_tools set means no tools are excluded.""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools=set(), # Empty set - exclude nothing ) router = ContentRouter(config) messages = [ { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_read_1", "type": "function", "function": {"name": "Read", "arguments": "{}"}, } ], }, { "role": "tool", "tool_call_id": "call_read_1", "content": generate_python_code(20), }, ] result = router.apply(messages, tokenizer) # Should NOT be excluded (empty set means compress everything) assert "router:excluded:tool" not in result.transforms_applied def test_anthropic_format_tool_result_exclusion(self, tokenizer): """Anthropic format tool_result blocks are also excluded.""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"Glob"}, ) router = ContentRouter(config) # Anthropic format with tool_use and tool_result in content blocks messages = [ { "role": "assistant", "content": [ { "type": "tool_use", "id": "toolu_glob_1", "name": "Glob", "input": {"pattern": "*.py"}, } ], }, { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_glob_1", "content": generate_search_results(50), } ], }, ] result = router.apply(messages, tokenizer) # Find the tool_result block and verify content unchanged user_msg = result.messages[1] tool_result_block = next( (b for b in user_msg["content"] if b.get("type") == "tool_result"), None ) assert tool_result_block is not None # Excluded from lossy compression; search results get a byte-lossless # heading fold. Verify byte-exact recovery (Anthropic block format). original = messages[1]["content"][0]["content"] assert search_unheading(tool_result_block["content"]) == original assert "router:excluded:lossless_search" in result.transforms_applied def test_anthropic_tool_result_runtime_window_allows_old_excluded_tools(self, tokenizer): """Agent profiles can shrink the protected window for Claude tool results.""" config = ContentRouterConfig( min_section_tokens=10, min_chars_for_block_compression=10, exclude_tools={"Glob"}, ) router = ContentRouter(config) old_tool_content = generate_search_results(80) messages = [ { "role": "assistant", "content": [ { "type": "tool_use", "id": "toolu_glob_old", "name": "Glob", "input": {"pattern": "*.py"}, } ], }, { "role": "user", "content": [ { "type": "tool_result", "tool_use_id": "toolu_glob_old", "content": old_tool_content, } ], }, {"role": "assistant", "content": "ack"}, {"role": "user", "content": "continue"}, {"role": "assistant", "content": "ack"}, ] result = router.apply(messages, tokenizer, read_protection_window=2) assert "router:excluded:tool" not in result.transforms_applied def test_mixed_excluded_and_non_excluded_tools(self, tokenizer): """Multiple tools in same conversation - only excluded ones pass through.""" config = ContentRouterConfig( min_section_tokens=10, exclude_tools={"Read"}, # Only exclude Read ) router = ContentRouter(config) read_content = generate_python_code(20) other_content = generate_json_data(100) messages = [ { "role": "assistant", "content": None, "tool_calls": [ { "id": "call_read_1", "type": "function", "function": {"name": "Read", "arguments": "{}"}, }, { "id": "call_other_1", "type": "function", "function": {"name": "OtherTool", "arguments": "{}"}, }, ], }, { "role": "tool", "tool_call_id": "call_read_1", "content": read_content, }, { "role": "tool", "tool_call_id": "call_other_1", "content": other_content, }, ] result = router.apply(messages, tokenizer) # Read tool content should be unchanged (excluded) read_result = next(m for m in result.messages if m.get("tool_call_id") == "call_read_1") assert read_result["content"] == read_content # 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}" )