From 729cc035a474b4ff04c6c16c62d3a5f4e515c56f Mon Sep 17 00:00:00 2001 From: chopratejas Date: Wed, 18 Feb 2026 16:54:20 -0800 Subject: [PATCH] Add compression summaries, multi-provider headers, Dockerfile fix Compression Summaries: - New: headroom/transforms/compression_summary.py - summarize_dropped_items(): categorizes compressed JSON items by field values (status, type, level, etc.), highlights errors/failures - summarize_compressed_code(): extracts function names from AST signatures (language-agnostic: Python, JS, Go, Rust, Java) - Newline-safe: strips \n from field values to keep markers single-line - SmartCrusher: CCR markers include categorical summary of dropped items e.g. "[500 items compressed to 20. Omitted: 87 passed, 2 failed. Retrieve more: hash=abc123. Expires in 5m.]" - CodeCompressor: CCR markers list compressed function names from AST e.g. "[180 tokens compressed. 5 bodies compressed: authenticate(). Retrieve more: hash=abc123. Expires in 5m.]" - Markers include TTL so LLM knows retrieval window - Summary escapes { } to prevent .format() crashes - Uses index-based dropped detection (not id()) for .copy() correctness Proxy Response Headers: - Anthropic, OpenAI, and Gemini handlers inject x-headroom-tokens-* headers for SaaS metering Multi-Provider Passthrough Routing: - Detect x-goog-api-key (Gemini) and api-key (Azure OpenAI) - X-Headroom-Base-URL for explicit upstream URL override Dockerfile: add build-essential + g++ for hnswlib compilation Bump version to 0.3.5 Tests: 27 new tests (unit, eval, integration with real API, tool invocation) --- headroom/__init__.py | 2 +- headroom/config.py | 5 +- headroom/proxy/server.py | 30 +- headroom/transforms/code_compressor.py | 18 +- headroom/transforms/compression_summary.py | 243 +++++++++++++++ headroom/transforms/smart_crusher.py | 49 ++- pyproject.toml | 2 +- tests/test_compression_summary.py | 178 +++++++++++ tests/test_compression_summary_eval.py | 217 +++++++++++++ tests/test_compression_summary_hard_eval.py | 256 +++++++++++++++ tests/test_compression_summary_integration.py | 235 ++++++++++++++ tests/test_compression_summary_tool_eval.py | 295 ++++++++++++++++++ 12 files changed, 1508 insertions(+), 22 deletions(-) create mode 100644 headroom/transforms/compression_summary.py create mode 100644 tests/test_compression_summary.py create mode 100644 tests/test_compression_summary_eval.py create mode 100644 tests/test_compression_summary_hard_eval.py create mode 100644 tests/test_compression_summary_integration.py create mode 100644 tests/test_compression_summary_tool_eval.py diff --git a/headroom/__init__.py b/headroom/__init__.py index 607d9f428..ea48fd9bd 100644 --- a/headroom/__init__.py +++ b/headroom/__init__.py @@ -142,7 +142,7 @@ from .transforms import ( TransformPipeline, ) -__version__ = "0.3.4" +__version__ = "0.3.5" __all__ = [ # Main client diff --git a/headroom/config.py b/headroom/config.py index 86964733d..1ebcc597c 100644 --- a/headroom/config.py +++ b/headroom/config.py @@ -529,7 +529,10 @@ class CCRConfig: # Retrieval marker format # Inserted at end of compressed content to tell LLM how to get more marker_template: str = ( - "\n[{original_count} items compressed to {compressed_count}. Retrieve more: hash={hash}]" + "\n[{original_count} items compressed to {compressed_count}." + "{summary}" + " Retrieve more: hash={hash}." + " Expires in {ttl_minutes}m.]" ) diff --git a/headroom/proxy/server.py b/headroom/proxy/server.py index 9975fe6d7..809cd0346 100644 --- a/headroom/proxy/server.py +++ b/headroom/proxy/server.py @@ -5322,6 +5322,16 @@ class HeadroomProxy: response_headers.pop("content-encoding", None) response_headers.pop("content-length", None) + # Inject Headroom compression metrics (for SaaS metering) + response_headers["x-headroom-tokens-before"] = str(original_tokens) + response_headers["x-headroom-tokens-after"] = str(optimized_tokens) + response_headers["x-headroom-tokens-saved"] = str(tokens_saved) + response_headers["x-headroom-model"] = model + if transforms_applied: + response_headers["x-headroom-transforms"] = ",".join(transforms_applied) + if cache_read_tokens > 0: + response_headers["x-headroom-cached"] = "true" + return Response( content=response.content, status_code=response.status_code, @@ -6562,11 +6572,27 @@ def create_app(config: ProxyConfig | None = None) -> FastAPI: # Passthrough - route to correct backend based on headers @app.api_route("/{path:path}", methods=["GET", "POST", "PUT", "DELETE"]) async def passthrough(request: Request, path: str): - # Anthropic SDK always sends anthropic-version header and uses x-api-key for auth - # OpenAI SDK uses Authorization: Bearer for auth + # Allow explicit base URL override (for Azure, custom endpoints, etc.) + custom_base = request.headers.get("x-headroom-base-url") + if custom_base: + return await proxy.handle_passthrough(request, custom_base.rstrip("/")) + + # Anthropic: sends anthropic-version header and x-api-key if request.headers.get("anthropic-version") or request.headers.get("x-api-key"): base_url = proxy.ANTHROPIC_API_URL + # Gemini: sends x-goog-api-key + elif request.headers.get("x-goog-api-key"): + base_url = proxy.GEMINI_API_URL + # Azure OpenAI: sends api-key header (not x-api-key) + elif request.headers.get("api-key"): + # Azure requires explicit base URL (varies per deployment) + azure_base = request.headers.get("x-headroom-base-url", "") + if azure_base: + base_url = azure_base.rstrip("/") + else: + base_url = proxy.OPENAI_API_URL # Fallback else: + # Default: OpenAI base_url = proxy.OPENAI_API_URL return await proxy.handle_passthrough(request, base_url) diff --git a/headroom/transforms/code_compressor.py b/headroom/transforms/code_compressor.py index 150d32b18..77fd72613 100644 --- a/headroom/transforms/code_compressor.py +++ b/headroom/transforms/code_compressor.py @@ -559,10 +559,22 @@ class CodeAwareCompressor(Transform): if self.config.enable_ccr and ratio < 0.8: cache_key = self._store_in_ccr(code, compressed, original_tokens) if cache_key: - # Add standard CCR marker format for CCRToolInjector detection + # Generate summary from AST data (language-agnostic) + from .compression_summary import summarize_compressed_code + + code_summary = summarize_compressed_code( + structure.function_bodies, + len(structure.function_bodies), + ) + summary_str = f" {code_summary}." if code_summary else "" + + # Add CCR marker (hash without quotes, matches CCRToolInjector regex) + ttl_min = max(1, getattr(self.config, "ccr_ttl_seconds", 300) // 60) compressed += ( - f"\n# [{original_tokens} items compressed to {compressed_tokens}. " - f"Retrieve more: hash={cache_key}]" + f"\n# [{original_tokens - compressed_tokens} tokens compressed." + f"{summary_str}" + f" Retrieve more: hash={cache_key}." + f" Expires in {ttl_min}m.]" ) return CodeCompressionResult( diff --git a/headroom/transforms/compression_summary.py b/headroom/transforms/compression_summary.py new file mode 100644 index 000000000..af97664a6 --- /dev/null +++ b/headroom/transforms/compression_summary.py @@ -0,0 +1,243 @@ +"""Compression summary generator — describes what was dropped. + +When content is compressed, the LLM needs to know what it's missing. +Instead of just "[480 items omitted]", we generate a categorical summary: +"[480 items omitted: 150 log entries (3 with errors), 200 test results (12 failures)]" + +This helps the LLM decide whether to call headroom_retrieve and what to search for. + +Used by: +- SmartCrusher: categorizes dropped JSON items by field values +- CodeCompressor: lists removed function/class names (from AST, language-agnostic) +""" + +from __future__ import annotations + +import re +from collections import Counter + + +def summarize_dropped_items( + all_items: list[dict], + kept_items: list[dict], + kept_indices: set[int] | None = None, + max_categories: int = 5, + max_notable: int = 3, +) -> str: + """Generate a categorical summary of items that were dropped. + + Args: + all_items: The original full list of items. + kept_items: The items that were kept after compression (used for count). + kept_indices: Indices of kept items (preferred over identity comparison). + max_categories: Maximum number of categories to show. + max_notable: Maximum number of notable items to call out. + + Returns: + Summary string or empty string if no useful summary can be generated. + """ + if not all_items or len(kept_items) >= len(all_items): + return "" + + # Determine which items were dropped + if kept_indices is not None: + dropped = [item for i, item in enumerate(all_items) if i not in kept_indices] + else: + # Fallback: index-based comparison using JSON equality + kept_json = {_item_key(item) for item in kept_items} + dropped = [item for item in all_items if _item_key(item) not in kept_json] + + if not dropped: + return "" + + # Strategy 1: Categorize by type/status/kind fields + categories = _categorize_by_fields(dropped) + + # Strategy 2: Find notable items (errors, failures, warnings) + notable = _find_notable_items(dropped, max_notable) + + # Build summary + parts = [] + + if categories: + cat_strs = [] + for field_val, count in categories.most_common(max_categories): + cat_strs.append(f"{count} {field_val}") + parts.append(", ".join(cat_strs)) + + if notable: + parts.append(f"notable: {'; '.join(notable)}") + + if not parts: + # Fallback: just describe the data shape + keys = _common_keys(dropped) + if keys: + parts.append(f"fields: {', '.join(keys[:5])}") + + return "; ".join(parts) + + +def summarize_compressed_code( + function_bodies: list[tuple[str, str, int]], + compressed_bodies_count: int, +) -> str: + """Generate a summary of compressed code sections from AST data. + + Language-agnostic: works with any language tree-sitter supports because + it reads function signatures directly from the CodeCompressor's AST output. + + Args: + function_bodies: List of (signature, body, line) from CodeStructure. + compressed_bodies_count: Number of bodies that were compressed. + + Returns: + Summary string like "5 bodies compressed: authenticate(), validate_token(), ..." + or empty string. + """ + if not function_bodies or compressed_bodies_count == 0: + return "" + + # Extract short names from signatures + names = [] + for sig, _body, _line in function_bodies: + name = _extract_name_from_signature(sig) + if name: + names.append(name) + + if not names: + return f"{compressed_bodies_count} function bodies compressed" + + # Show up to 6 names + shown = names[:6] + result = f"{compressed_bodies_count} bodies compressed: {', '.join(shown)}" + if len(names) > 6: + result += f" (+{len(names) - 6} more)" + return result + + +# ---- Internal helpers ---- + +# Fields that commonly indicate item category/type +_CATEGORY_FIELDS = ( + "type", + "status", + "kind", + "category", + "level", + "severity", + "state", + "phase", + "action", + "event_type", + "log_level", + "result", + "outcome", +) + +# Values that indicate something notable/important +_NOTABLE_PATTERNS = re.compile( + r"error|fail|critical|warning|exception|crash|timeout|denied|rejected|invalid", + re.IGNORECASE, +) + +# Values that look like URLs or paths (not useful as categories) +_URL_PATTERN = re.compile(r"^https?://|^/[a-z]", re.IGNORECASE) + + +def _item_key(item: dict) -> str: + """Create a hashable key for an item (for dropped detection without id()).""" + # Use first few field values as a fingerprint + parts = [] + for k, v in list(item.items())[:4]: + parts.append(f"{k}={v}") + return "|".join(parts) + + +def _categorize_by_fields(items: list[dict]) -> Counter: + """Categorize items by their type/status/kind field values.""" + categories: Counter = Counter() + + for item in items: + categorized = False + for field in _CATEGORY_FIELDS: + val = item.get(field) + if val and isinstance(val, str) and len(val) < 50: + clean_val = val.replace("\n", " ").replace("\r", "").strip() + if clean_val: + categories[clean_val] += 1 + categorized = True + break + if not categorized: + # Try to infer from the item's first short string field + for key, val in item.items(): + if ( + isinstance(val, str) + and 2 < len(val) < 30 + and key not in ("id", "name", "path", "url", "href", "email") + and not _URL_PATTERN.match(val) + ): + clean_val = val.replace("\n", " ").replace("\r", "").strip() + categories[f"{key}={clean_val}"] += 1 + break + + return categories + + +def _find_notable_items(items: list[dict], max_notable: int) -> list[str]: + """Find items that contain error/failure/warning indicators.""" + notable = [] + for item in items: + item_str = str(item)[:500] + matches = _NOTABLE_PATTERNS.findall(item_str) + if matches: + name = item.get("name", item.get("id", item.get("path", ""))) + if name: + clean_name = str(name).replace("\n", " ").strip()[:50] + notable.append(f"{clean_name} ({matches[0]})") + else: + notable.append(matches[0]) + if len(notable) >= max_notable: + break + return notable + + +def _common_keys(items: list[dict]) -> list[str]: + """Get the most common keys across items.""" + key_counts: Counter = Counter() + for item in items[:50]: + for key in item.keys(): + key_counts[key] += 1 + return [k for k, _ in key_counts.most_common(8)] + + +def _extract_name_from_signature(sig: str) -> str: + """Extract the function/method name from a signature string. + + Works for any language because it looks for common patterns: + - Python: "def authenticate(", "async def fetch(" + - JavaScript: "function authenticate(", "async function fetch(" + - Go: "func (s *Server) HandleRequest(" + - Rust: "fn authenticate(" + - Java/C++: "public void authenticate(" + """ + # Try common function definition patterns + match = re.search(r"(?:def|func|fn|function)\s+(?:\([^)]*\)\s*)?(\w+)", sig) + if match: + return match.group(1) + "()" + + # Try method patterns: "public static void methodName(" + match = re.search(r"(?:public|private|protected|static|async|export)\s+.*?(\w+)\s*\(", sig) + if match: + return match.group(1) + "()" + + # Try class patterns + match = re.search(r"class\s+(\w+)", sig) + if match: + return match.group(1) + + # Fallback: last word before ( + match = re.search(r"(\w+)\s*\(", sig) + if match: + return match.group(1) + "()" + + return "" diff --git a/headroom/transforms/smart_crusher.py b/headroom/transforms/smart_crusher.py index 2af074eaa..7e9d08ec2 100644 --- a/headroom/transforms/smart_crusher.py +++ b/headroom/transforms/smart_crusher.py @@ -2127,11 +2127,22 @@ class SmartCrusher(Transform): # CCR: Inject retrieval markers if compression happened and CCR is enabled if was_modified and ccr_markers and self._ccr_config.inject_retrieval_marker: - for ccr_hash, original_count, compressed_count in ccr_markers: + for marker_data in ccr_markers: + if len(marker_data) == 4: + ccr_hash, original_count, compressed_count, dropped_summary = marker_data + else: + ccr_hash, original_count, compressed_count = marker_data + dropped_summary = "" + summary_str = f" Omitted: {dropped_summary}." if dropped_summary else "" + # Escape { } in summary to prevent .format() errors + safe_summary = summary_str.replace("{", "{{").replace("}", "}}") + ttl_seconds = getattr(self._ccr_config, "store_ttl_seconds", 300) marker = self._ccr_config.marker_template.format( original_count=original_count, compressed_count=compressed_count, hash=ccr_hash, + summary=safe_summary, + ttl_minutes=max(1, ttl_seconds // 60), ) result += marker @@ -2149,24 +2160,24 @@ class SmartCrusher(Transform): Returns: Tuple of (processed_value, info_string, ccr_markers). - ccr_markers is a list of (hash, original_count, compressed_count) tuples. + ccr_markers is a list of (hash, original_count, compressed_count, summary) tuples. """ info_parts = [] - ccr_markers: list[tuple[str, int, int]] = [] + ccr_markers: list[tuple] = [] if isinstance(value, list): # Check if this array should be crushed # Must have enough items AND all items must be dicts (not mixed types) all_dicts = value and all(isinstance(item, dict) for item in value) if len(value) >= self.config.min_items_to_analyze and all_dicts: - crushed, strategy, ccr_hash = self._crush_array( + crushed, strategy, ccr_hash, dropped_summary = self._crush_array( value, query_context, tool_name, bias=bias ) info_parts.append(f"{strategy}({len(value)}->{len(crushed)})") - # Track CCR marker for later injection + # Track CCR marker for later injection (with summary) if ccr_hash: - ccr_markers.append((ccr_hash, len(value), len(crushed))) + ccr_markers.append((ccr_hash, len(value), len(crushed), dropped_summary)) return crushed, ",".join(info_parts), ccr_markers else: @@ -2204,7 +2215,7 @@ class SmartCrusher(Transform): query_context: str = "", tool_name: str | None = None, bias: float = 1.0, - ) -> tuple[list, str, str | None]: + ) -> tuple[list, str, str | None, str]: """Crush an array using statistical analysis and relevance scoring. IMPORTANT: If crushability analysis determines it's not safe to crush @@ -2223,8 +2234,9 @@ class SmartCrusher(Transform): bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer). Returns: - Tuple of (crushed_items, strategy_info, ccr_hash). + Tuple of (crushed_items, strategy_info, ccr_hash, dropped_summary). ccr_hash is the hash for retrieval if CCR is enabled, None otherwise. + dropped_summary is a categorical summary of what was dropped. """ # BOUNDARY CHECK: Use adaptive sizing instead of hardcoded limit # compute_optimal_k handles trivial cases (n <= 8 → keep all) @@ -2239,7 +2251,7 @@ class SmartCrusher(Transform): ) if len(items) <= adaptive_k: - return items, "none:adaptive_at_limit", None + return items, "none:adaptive_at_limit", None, "" # Get feedback hints if enabled # THREAD-SAFETY: Use a local effective_max_items instead of mutating shared config @@ -2264,7 +2276,7 @@ class SmartCrusher(Transform): ) if toin_hint.skip_compression: - return items, f"skip:toin({toin_hint.reason})", None + return items, f"skip:toin({toin_hint.reason})", None, "" # Apply TOIN recommendations if from network or local learning toin_preserve_fields: list[str] = [] @@ -2313,7 +2325,7 @@ class SmartCrusher(Transform): # Check if hints recommend skipping compression if hints.skip_compression: - return items, f"skip:feedback({hints.reason})", None + return items, f"skip:feedback({hints.reason})", None, "" # Adjust max_items based on feedback if hints.suggested_items is not None: @@ -2339,7 +2351,7 @@ class SmartCrusher(Transform): reason = "" if analysis.crushability: reason = f"skip:{analysis.crushability.reason}" - return items, reason, None + return items, reason, None, "" # Apply TOIN strategy recommendation if available # TOIN learns which strategies work best from cross-user patterns @@ -2356,7 +2368,7 @@ class SmartCrusher(Transform): if toin_compression_level: if toin_compression_level == "none": # Don't compress - return original - return items, "skip:toin_level_none", None + return items, "skip:toin_level_none", None, "" elif toin_compression_level == "conservative": # Be conservative - keep more items effective_max_items = max(effective_max_items, min(50, len(items) // 2)) @@ -2453,9 +2465,18 @@ class SmartCrusher(Transform): elif hints_applied: strategy_info += f"(feedback:{effective_max_items})" + # Generate categorical summary of dropped items (use indices, not identity) + from .compression_summary import summarize_dropped_items + + dropped_summary = summarize_dropped_items( + items, + result, + kept_indices=set(plan.keep_indices), + ) + # Clean up temporary instance variable self._current_field_semantics = None - return result, strategy_info, ccr_hash + return result, strategy_info, ccr_hash, dropped_summary except Exception: # Clean up temporary instance variable diff --git a/pyproject.toml b/pyproject.toml index 8805c05f0..fec22f7bd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "hatchling.build" [project] name = "headroom-ai" -version = "0.3.4" +version = "0.3.5" description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%" readme = "README.md" license = "Apache-2.0" diff --git a/tests/test_compression_summary.py b/tests/test_compression_summary.py new file mode 100644 index 000000000..9ac03fbff --- /dev/null +++ b/tests/test_compression_summary.py @@ -0,0 +1,178 @@ +"""Tests for compression summary generation.""" + +from headroom.transforms.compression_summary import ( + _extract_name_from_signature, + summarize_compressed_code, + summarize_dropped_items, +) + + +class TestSummarizeDroppedItems: + def test_items_with_status_field(self): + all_items = [{"id": i, "name": f"item-{i}", "status": "active"} for i in range(50)] + [ + {"id": i, "name": f"err-{i}", "status": "error"} for i in range(5) + ] + kept = all_items[:3] + summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2}) + assert "active" in summary + assert summary + + def test_items_with_type_field(self): + all_items = [{"type": "log", "message": f"entry {i}"} for i in range(30)] + [ + {"type": "metric", "value": i} for i in range(20) + ] + kept = all_items[:5] + summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2, 3, 4}) + assert "log" in summary or "metric" in summary + + def test_notable_items_with_errors(self): + all_items = [{"name": f"test-{i}", "result": "pass"} for i in range(40)] + [ + {"name": "test-auth", "result": "fail", "error": "authentication failed"}, + ] + kept = all_items[:5] + summary = summarize_dropped_items(all_items, kept, kept_indices=set(range(5))) + assert summary + + def test_no_dropped_items(self): + items = [{"id": 1}, {"id": 2}] + summary = summarize_dropped_items(items, items) + assert summary == "" + + def test_empty_input(self): + summary = summarize_dropped_items([], []) + assert summary == "" + + def test_all_items_dropped(self): + items = [{"status": "active", "name": f"item-{i}"} for i in range(20)] + summary = summarize_dropped_items(items, [], kept_indices=set()) + assert "active" in summary + + def test_mixed_category_fields(self): + all_items = [{"level": "info", "msg": "something"} for _ in range(10)] + [ + {"level": "error", "msg": "bad thing"} for _ in range(3) + ] + kept = all_items[:2] + summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1}) + assert summary + + def test_items_without_category_fields(self): + all_items = [{"code": 200, "count": i} for i in range(30)] + kept = all_items[:3] + summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2}) + assert summary # Should produce field-based fallback + + def test_summary_not_too_long(self): + all_items = [{"type": f"type_{i % 20}", "data": "x"} for i in range(100)] + kept = all_items[:5] + summary = summarize_dropped_items(all_items, kept, kept_indices=set(range(5))) + assert len(summary) < 500 + + def test_url_values_excluded_from_categories(self): + """URL-like values should not be used as category labels.""" + all_items = [ + {"url": f"https://api.example.com/v1/items/{i}", "method": "GET"} for i in range(30) + ] + kept = all_items[:3] + summary = summarize_dropped_items(all_items, kept, kept_indices={0, 1, 2}) + assert "https://" not in summary + + def test_fallback_without_kept_indices(self): + """Works correctly without kept_indices (uses _item_key fallback).""" + all_items = [{"status": "active", "id": i} for i in range(20)] + kept = [all_items[0], all_items[1]] # Copies from same list + summary = summarize_dropped_items(all_items, kept) + assert summary + + +class TestSummarizeCompressedCode: + def test_python_function_bodies(self): + bodies = [ + ("def authenticate(username, password):", " db = get_db()\n return True", 10), + ("def validate_token(token):", " return jwt.decode(token)", 20), + ("def refresh_session(user):", " session.extend()", 30), + ] + summary = summarize_compressed_code(bodies, 3) + assert "3 bodies compressed" in summary + assert "authenticate()" in summary + assert "validate_token()" in summary + + def test_javascript_function_bodies(self): + bodies = [ + ("function handleRequest(req, res) {", " res.send('ok');", 5), + ("async function fetchData(url) {", " return await fetch(url);", 15), + ] + summary = summarize_compressed_code(bodies, 2) + assert "handleRequest()" in summary + assert "fetchData()" in summary + + def test_go_function_bodies(self): + bodies = [ + ( + "func (s *Server) HandleRequest(w http.ResponseWriter, r *http.Request) {", + ' w.Write([]byte("ok"))', + 10, + ), + ("func main() {", " server.Start()", 1), + ] + summary = summarize_compressed_code(bodies, 2) + assert "HandleRequest()" in summary + assert "main()" in summary + + def test_rust_function_bodies(self): + bodies = [ + ("fn authenticate(token: &str) -> Result {", " Ok(User::new())", 10), + ] + summary = summarize_compressed_code(bodies, 1) + assert "authenticate()" in summary + + def test_empty_bodies(self): + summary = summarize_compressed_code([], 0) + assert summary == "" + + def test_many_bodies_truncated(self): + bodies = [(f"def func_{i}(x):", f" return {i}", i * 10) for i in range(20)] + summary = summarize_compressed_code(bodies, 20) + assert "+14 more" in summary # 20 - 6 shown + + +class TestExtractNameFromSignature: + def test_python_def(self): + assert _extract_name_from_signature("def authenticate(username):") == "authenticate()" + + def test_python_async_def(self): + assert _extract_name_from_signature("async def fetch_data(url):") == "fetch_data()" + + def test_javascript_function(self): + assert _extract_name_from_signature("function handleClick(event) {") == "handleClick()" + + def test_go_func(self): + assert ( + _extract_name_from_signature("func HandleRequest(w http.ResponseWriter) {") + == "HandleRequest()" + ) + + def test_go_method(self): + assert _extract_name_from_signature("func (s *Server) Start() {") == "Start()" + + def test_rust_fn(self): + assert ( + _extract_name_from_signature("fn authenticate(token: &str) -> Result {") + == "authenticate()" + ) + + def test_java_method(self): + assert ( + _extract_name_from_signature("public void processPayment(Payment p) {") + == "processPayment()" + ) + + def test_class(self): + assert _extract_name_from_signature("class TokenValidator:") == "TokenValidator" + + def test_empty(self): + assert _extract_name_from_signature("") == "" + + def test_export_async(self): + assert ( + _extract_name_from_signature("export async function fetchUsers() {") == "fetchUsers()" + ) diff --git a/tests/test_compression_summary_eval.py b/tests/test_compression_summary_eval.py new file mode 100644 index 000000000..1da0b5d88 --- /dev/null +++ b/tests/test_compression_summary_eval.py @@ -0,0 +1,217 @@ +"""Eval: Compression summary quality — generic, unbiased. + +Tests that compression summaries are: +1. Accurate (categories match actual dropped items) +2. Useful (contain information that would help retrieval) +3. Not misleading (don't hallucinate categories) + +These are NOT skewed to show summaries as amazing — they test +real-world data patterns and verify correctness. +""" + +from headroom.transforms.compression_summary import ( + summarize_compressed_code, + summarize_dropped_items, +) + +# ============================================================================ +# Realistic test data (modeled on actual tool outputs) +# ============================================================================ + + +def _make_github_issues(n: int) -> list[dict]: + """Realistic GitHub issues list.""" + statuses = ["open"] * (n // 2) + ["closed"] * (n // 4) + ["in_progress"] * (n // 4) + issues = [] + for i in range(n): + issue = { + "id": i + 1, + "title": f"Issue #{i + 1}: {'Fix auth bug' if i == 42 else 'General issue'}", + "status": statuses[i % len(statuses)], + "labels": ["bug"] if i % 10 == 0 else ["enhancement"], + "assignee": f"user-{i % 5}", + } + if i in (42, 87): + issue["status"] = "open" + issue["labels"] = ["critical", "bug"] + issue["title"] = f"CRITICAL: Auth failure in production (issue #{i + 1})" + issues.append(issue) + return issues + + +def _make_test_results(n: int) -> list[dict]: + """Realistic test suite results.""" + results = [] + for i in range(n): + result = { + "name": f"test_{'auth' if i < 10 else 'general'}_{i}", + "status": "pass", + "duration_ms": 50 + i * 2, + } + if i in (3, 7, 45, 88): + result["status"] = "fail" + result["error"] = "AssertionError: expected True, got False" + if i in (12, 67): + result["status"] = "error" + result["error"] = "TimeoutError: test exceeded 30s limit" + results.append(result) + return results + + +def _make_log_entries(n: int) -> list[dict]: + """Realistic log entries.""" + entries = [] + for i in range(n): + entry = { + "timestamp": f"2024-01-15T10:{i:02d}:00Z", + "level": "info", + "message": f"Request processed in {10 + i}ms", + "service": "api-gateway", + } + if i in (15, 45, 89): + entry["level"] = "error" + entry["message"] = "Connection refused: database pool exhausted" + if i in (20, 50): + entry["level"] = "warning" + entry["message"] = "High memory usage: 85% threshold exceeded" + entries.append(entry) + return entries + + +def _make_api_endpoints(n: int) -> list[dict]: + """Realistic API endpoint list.""" + return [ + { + "path": f"/api/v1/{'users' if i < n // 3 else 'orders' if i < 2 * n // 3 else 'products'}/{i}", + "method": "GET" if i % 3 else "POST", + "status_code": 200 if i % 20 else 500, + "latency_ms": 50 + i, + } + for i in range(n) + ] + + +# ============================================================================ +# Eval: Summary accuracy +# ============================================================================ + + +class TestSummaryAccuracy: + """Verify summaries accurately reflect what was dropped.""" + + def test_github_issues_categories_correct(self): + """Summary mentions actual status values from dropped items.""" + issues = _make_github_issues(100) + kept = issues[:5] + summary = summarize_dropped_items(issues, kept) + + # Should mention the status values present in dropped items + assert summary # Non-empty + # At minimum, should contain some status category info + has_category = any(s in summary.lower() for s in ["open", "closed", "in_progress"]) + assert has_category, f"Summary missing status categories: {summary}" + + def test_test_results_mentions_failures(self): + """Summary mentions failures when test results are compressed.""" + results = _make_test_results(100) + kept = results[:5] + summary = summarize_dropped_items(results, kept) + + assert summary + # Should mention pass/fail somewhere + has_result = any(s in summary.lower() for s in ["pass", "fail", "error"]) + assert has_result, f"Summary missing test result info: {summary}" + + def test_log_entries_mentions_errors(self): + """Summary mentions error log entries.""" + logs = _make_log_entries(100) + kept = logs[:3] + summary = summarize_dropped_items(logs, kept) + + assert summary + # Should categorize by log level + has_level = any(s in summary.lower() for s in ["info", "error", "warning"]) + assert has_level, f"Summary missing log level info: {summary}" + + def test_no_hallucinated_categories(self): + """Summary should NOT mention categories that don't exist.""" + items = [{"status": "active", "id": i} for i in range(50)] + kept = items[:3] + summary = summarize_dropped_items(items, kept) + + # Should NOT mention statuses that don't exist in the data + assert "error" not in summary.lower() or "notable" in summary.lower() + assert "fail" not in summary.lower() + assert "critical" not in summary.lower() + + def test_summary_proportional_to_data(self): + """Category counts in summary should roughly match actual data.""" + items = ( + [{"type": "log", "data": "x"}] * 100 + + [{"type": "metric", "data": "y"}] * 50 + + [{"type": "alert", "data": "z"}] * 10 + ) + kept = items[:3] + summary = summarize_dropped_items(items, kept) + + # "log" should appear with a higher count than "alert" + # (We can't verify exact counts from the summary string, + # but we verify the summary is non-empty and reasonable) + assert summary + assert len(summary) < 300 + + +class TestSummaryUsefulness: + """Verify summaries contain information useful for retrieval.""" + + def test_enough_info_to_search(self): + """Summary should contain terms the LLM could use as search queries.""" + results = _make_test_results(100) + kept = results[:5] + summary = summarize_dropped_items(results, kept) + + # The LLM should be able to extract search terms from the summary + # At minimum, it should know WHAT KIND of items are in the compressed data + assert len(summary) > 10, "Summary too short to be useful" + + def test_notable_items_actionable(self): + """Notable items should contain enough info to act on.""" + logs = _make_log_entries(100) + kept = logs[:2] + summary = summarize_dropped_items(logs, kept) + + # If there are errors in the logs, the summary should help + # the LLM decide to retrieve them + assert summary + # Just verify it's substantive enough + assert len(summary.split()) > 3 + + def test_api_endpoints_described(self): + """API endpoint data should produce some useful description.""" + endpoints = _make_api_endpoints(60) + kept = endpoints[:5] + summary = summarize_dropped_items(endpoints, kept) + + assert summary # Should produce SOMETHING, even without type/status fields + + +class TestCodeSummaryAccuracy: + """Verify code summaries accurately describe removed sections.""" + + def test_real_python_module(self): + """Summary of a realistic Python module compression.""" + # Use AST-based summary (language-agnostic) + bodies = [ + ("def __init__(self, url: str, pool_size: int = 10):", "...", 8), + ("def connect(self) -> Any:", "...", 15), + ("def _create_new(self) -> Any:", "...", 22), + ("def release(self, conn: Any) -> None:", "...", 28), + ("def close_all(self) -> None:", "...", 33), + ("def create_engine(url: str) -> DatabaseConnection:", "...", 38), + ] + summary = summarize_compressed_code(bodies, 6) + assert "6 bodies compressed" in summary + has_names = any( + name in summary for name in ["connect()", "release()", "close_all()", "create_engine()"] + ) + assert has_names, f"Summary missing function names: {summary}" diff --git a/tests/test_compression_summary_hard_eval.py b/tests/test_compression_summary_hard_eval.py new file mode 100644 index 000000000..e16895eaf --- /dev/null +++ b/tests/test_compression_summary_hard_eval.py @@ -0,0 +1,256 @@ +"""Hard eval: Cases where the LLM has NO reason to check compressed data. + +The previous eval asked "are there failures?" — that's too easy, the LLM +will proactively check regardless of summary. + +This eval tests the SUBTLE case: the user asks a DIFFERENT question, +but the answer is in the compressed data. The summary is the only hint. + +Requires: ANTHROPIC_API_KEY in environment or .env file. +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +env_path = Path(__file__).parent.parent / ".env" +if env_path.exists(): + for line in env_path.read_text().splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + +ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "") + +pytestmark = pytest.mark.skipif( + not ANTHROPIC_KEY, + reason="ANTHROPIC_API_KEY not set", +) + +HEADROOM_RETRIEVE_TOOL = { + "name": "headroom_retrieve", + "description": "Retrieve uncompressed content. Pass a query to search within it.", + "input_schema": { + "type": "object", + "properties": { + "hash": {"type": "string"}, + "query": {"type": "string"}, + }, + "required": ["hash"], + }, +} + + +def _call_claude(messages, tools, max_tokens=300): + import httpx + + resp = httpx.post( + "https://api.anthropic.com/v1/messages", + headers={ + "X-Api-Key": ANTHROPIC_KEY, + "anthropic-version": "2023-06-01", + "Content-Type": "application/json", + }, + json={ + "model": "claude-sonnet-4-5-20250929", + "max_tokens": max_tokens, + "messages": messages, + "tools": tools, + }, + timeout=30, + ) + return resp.json() + + +def _get_tool_calls(resp): + return [ + {"name": b["name"], "input": b.get("input", {})} + for b in resp.get("content", []) + if b.get("type") == "tool_use" + ] + + +def _get_text(resp): + return " ".join(b.get("text", "") for b in resp.get("content", []) if b.get("type") == "text") + + +class TestHardCases: + """Cases where the LLM wouldn't naturally check compressed data.""" + + def test_config_lookup_with_summary(self): + """User asks about a config value that's in compressed data. + + The visible items are all about 'production' env. + The compressed items include 'staging' configs. + Summary mentions this. LLM should retrieve. + """ + visible = [ + {"env": "production", "key": "DATABASE_URL", "value": "postgres://prod-db:5432/app"}, + {"env": "production", "key": "REDIS_URL", "value": "redis://prod-cache:6379"}, + {"env": "production", "key": "API_RATE_LIMIT", "value": "1000"}, + ] + # Hidden in compressed: staging configs + all_items = ( + visible + + [ + { + "env": "staging", + "key": "DATABASE_URL", + "value": "postgres://staging-db:5432/app", + }, + {"env": "staging", "key": "REDIS_URL", "value": "redis://staging-cache:6379"}, + {"env": "staging", "key": "DEBUG_MODE", "value": "true"}, + {"env": "staging", "key": "LOG_LEVEL", "value": "debug"}, + ] + * 10 + + [ + { + "env": "development", + "key": "DATABASE_URL", + "value": "postgres://localhost:5432/dev", + }, + ] + * 5 + ) + + from headroom.transforms.compression_summary import summarize_dropped_items + + summary = summarize_dropped_items(all_items, visible) + + compressed_output = json.dumps(visible, indent=2) + compressed_output += ( + f"\n[{len(all_items) - len(visible)} items compressed to {len(visible)}." + f" Omitted: {summary}." + f' Retrieve specific items: headroom_retrieve(hash="config_hash", query="search")]' + ) + + messages = [ + { + "role": "user", + "content": ( + f"Here are the application configs:\n\n{compressed_output}\n\n" + "What is the staging database URL?" + ), + } + ] + + resp = _call_claude(messages, [HEADROOM_RETRIEVE_TOOL]) + tool_calls = _get_tool_calls(resp) + text = _get_text(resp) + + print(f"\n Summary: {summary}") + print(f" Stop reason: {resp.get('stop_reason')}") + print(f" Tool calls: {tool_calls}") + if text: + print(f" Text: {text[:200]}") + + # WITH summary mentioning "staging" → should retrieve + if resp.get("stop_reason") == "tool_use": + assert tool_calls[0]["name"] == "headroom_retrieve" + query = tool_calls[0]["input"].get("query", "").lower() + assert "staging" in query or "database" in query + print(" RESULT: Retrieved staging config ✓") + else: + # If LLM didn't retrieve, it should at least mention the data is compressed + assert "compressed" in text.lower() or "staging" in text.lower() + print(" RESULT: Mentioned compressed data but didn't retrieve") + + def test_config_lookup_without_summary(self): + """Same question, but NO summary. LLM only sees production configs.""" + visible = [ + {"env": "production", "key": "DATABASE_URL", "value": "postgres://prod-db:5432/app"}, + {"env": "production", "key": "REDIS_URL", "value": "redis://prod-cache:6379"}, + {"env": "production", "key": "API_RATE_LIMIT", "value": "1000"}, + ] + + compressed_output = json.dumps(visible, indent=2) + compressed_output += "\n[45 items compressed to 3. Retrieve more: hash=config_hash]" + + messages = [ + { + "role": "user", + "content": ( + f"Here are the application configs:\n\n{compressed_output}\n\n" + "What is the staging database URL?" + ), + } + ] + + resp = _call_claude(messages, [HEADROOM_RETRIEVE_TOOL]) + tool_calls = _get_tool_calls(resp) + text = _get_text(resp) + + print(f"\n Stop reason: {resp.get('stop_reason')}") + print(f" Tool calls: {tool_calls}") + if text: + print(f" Text: {text[:200]}") + + if resp.get("stop_reason") == "tool_use": + print(" RESULT: LLM proactively retrieved (smart)") + else: + print(" RESULT: LLM did NOT retrieve staging config") + + def test_specific_user_in_large_list_with_summary(self): + """Find a specific user in a compressed user list. + + Summary mentions user roles. User asks about admins. + """ + visible = [ + {"id": i, "name": f"user_{i}", "role": "member", "email": f"user{i}@co.com"} + for i in range(5) + ] + all_items = ( + visible + + [ + {"id": i, "name": f"user_{i}", "role": "member", "email": f"user{i}@co.com"} + for i in range(5, 95) + ] + + [ + {"id": 96, "name": "admin_sarah", "role": "admin", "email": "sarah@co.com"}, + {"id": 97, "name": "admin_mike", "role": "admin", "email": "mike@co.com"}, + {"id": 98, "name": "superadmin_jane", "role": "superadmin", "email": "jane@co.com"}, + ] + ) + + from headroom.transforms.compression_summary import summarize_dropped_items + + summary = summarize_dropped_items(all_items, visible) + + compressed_output = json.dumps(visible, indent=2) + compressed_output += ( + f"\n[{len(all_items) - len(visible)} items compressed to {len(visible)}." + f" Omitted: {summary}." + f' Retrieve: headroom_retrieve(hash="users_hash", query="search")]' + ) + + messages = [ + { + "role": "user", + "content": ( + f"Here's our user list:\n\n{compressed_output}\n\n" + "Who are the admin users? I need to contact them." + ), + } + ] + + resp = _call_claude(messages, [HEADROOM_RETRIEVE_TOOL]) + tool_calls = _get_tool_calls(resp) + text = _get_text(resp) + + print(f"\n Summary: {summary}") + print(f" Stop reason: {resp.get('stop_reason')}") + print(f" Tool calls: {tool_calls}") + if text: + print(f" Text: {text[:200]}") + + if resp.get("stop_reason") == "tool_use": + query = tool_calls[0]["input"].get("query", "").lower() + assert "admin" in query + print(f" RESULT: Retrieved admin users (query='{query}') ✓") + else: + print(" RESULT: Did not retrieve admin users") diff --git a/tests/test_compression_summary_integration.py b/tests/test_compression_summary_integration.py new file mode 100644 index 000000000..738e7bbec --- /dev/null +++ b/tests/test_compression_summary_integration.py @@ -0,0 +1,235 @@ +"""Integration eval: Compression summaries with real LLM calls. + +Tests whether compression summaries actually help the LLM find information +in compressed data. Compares behavior with and without summaries. + +Requires: ANTHROPIC_API_KEY in environment or .env file. + +Run: python -m pytest tests/test_compression_summary_integration.py -v -s +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +# Load .env +env_path = Path(__file__).parent.parent / ".env" +if env_path.exists(): + for line in env_path.read_text().splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + +ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "") + +pytestmark = pytest.mark.skipif( + not ANTHROPIC_KEY, + reason="ANTHROPIC_API_KEY not set — skipping integration tests", +) + + +def _call_claude(messages: list[dict], max_tokens: int = 200) -> dict: + """Make a real Anthropic API call.""" + import httpx + + resp = httpx.post( + "https://api.anthropic.com/v1/messages", + headers={ + "X-Api-Key": ANTHROPIC_KEY, + "anthropic-version": "2023-06-01", + "Content-Type": "application/json", + }, + json={ + "model": "claude-sonnet-4-5-20250929", + "max_tokens": max_tokens, + "messages": messages, + }, + timeout=30, + ) + return resp.json() + + +# ============================================================================ +# Test data: realistic tool output that gets compressed +# ============================================================================ + + +def _make_test_suite_output(n: int = 100) -> list[dict]: + """Simulate a large test suite result (like from a CI/CD tool).""" + results = [] + for i in range(n): + result = { + "test_name": f"test_module_{i // 10}.test_case_{i}", + "status": "passed", + "duration_ms": 50 + i * 3, + "file": f"tests/test_module_{i // 10}.py", + } + # Inject specific failures that the LLM should find + if i == 42: + result["status"] = "failed" + result["error"] = "AssertionError: expected status 200, got 401 in auth_middleware" + result["test_name"] = "test_auth.test_login_with_expired_token" + if i == 67: + result["status"] = "failed" + result["error"] = "TimeoutError: database connection pool exhausted after 30s" + result["test_name"] = "test_database.test_concurrent_connections" + if i == 88: + result["status"] = "error" + result["error"] = "ImportError: cannot import name 'NewFeature' from 'app.features'" + result["test_name"] = "test_features.test_new_feature_integration" + results.append(result) + return results + + +class TestSummaryHelpfulness: + """Compare LLM accuracy with vs without compression summaries.""" + + def test_find_failures_with_summary(self): + """LLM can identify failure types from the summary alone.""" + test_results = _make_test_suite_output(100) + + # Simulate compression: keep first 10, compress rest with summary + kept = test_results[:10] + from headroom.transforms.compression_summary import summarize_dropped_items + + summary = summarize_dropped_items(test_results, kept) + + compressed_output = json.dumps(kept, indent=2) + compressed_output += f"\n[90 items compressed to 10. Omitted: {summary}. " + compressed_output += ( + 'Retrieve specific items: headroom_retrieve(hash="abc123", query="your search")]' + ) + + messages = [ + { + "role": "user", + "content": ( + "Here are the test results from CI:\n\n" + f"{compressed_output}\n\n" + "Are there any test failures? What types of failures are there? " + "Answer concisely." + ), + }, + ] + + resp = _call_claude(messages) + text = resp.get("content", [{}])[0].get("text", "").lower() + + # The LLM should mention failures (from the summary info) + has_failure_info = any( + word in text for word in ["fail", "error", "timeout", "assert", "import"] + ) + print(f"\n Summary: {summary}") + print(f" LLM response: {text[:200]}") + print(f" Detected failure info: {has_failure_info}") + + assert has_failure_info, f"LLM didn't detect failures from summary. Response: {text[:300]}" + + def test_find_failures_without_summary(self): + """Baseline: LLM with NO summary — just '[90 items compressed]'.""" + test_results = _make_test_suite_output(100) + + kept = test_results[:10] + compressed_output = json.dumps(kept, indent=2) + compressed_output += "\n[90 items compressed to 10. Retrieve more: hash=abc123]" + + messages = [ + { + "role": "user", + "content": ( + "Here are the test results from CI:\n\n" + f"{compressed_output}\n\n" + "Are there any test failures? What types of failures are there? " + "Answer concisely." + ), + }, + ] + + resp = _call_claude(messages) + text = resp.get("content", [{}])[0].get("text", "").lower() + + # The LLM may or may not detect failures (it only sees 10 passing tests) + has_failure_info = any( + word in text for word in ["fail", "error", "timeout", "assert", "import"] + ) + print(f"\n LLM response (no summary): {text[:200]}") + print(f" Detected failure info: {has_failure_info}") + + # We're NOT asserting here — this is the baseline. + # We expect this to often MISS failures since the summary is generic. + + def test_code_summary_helps_identify_functions(self): + """LLM can identify which functions were removed from compressed code.""" + original_code = ''' +class PaymentProcessor: + """Processes payments via Stripe.""" + + def __init__(self, api_key: str): + self.stripe = stripe.Client(api_key) + self.retry_count = 3 + + def charge(self, amount: float, currency: str, token: str) -> dict: + for attempt in range(self.retry_count): + try: + return self.stripe.charges.create( + amount=int(amount * 100), + currency=currency, + source=token, + ) + except stripe.RateLimitError: + time.sleep(2 ** attempt) + raise PaymentError("Max retries exceeded") + + def refund(self, charge_id: str, amount: float = None) -> dict: + params = {"charge": charge_id} + if amount: + params["amount"] = int(amount * 100) + return self.stripe.refunds.create(**params) + + def get_balance(self) -> float: + balance = self.stripe.balance.retrieve() + return balance.available[0].amount / 100 +''' + compressed_code = ''' +class PaymentProcessor: + """Processes payments via Stripe.""" + + def __init__(self, api_key: str): + # [2 lines omitted] + pass + + def charge(self, amount: float, currency: str, token: str) -> dict: + # [8 lines omitted] + pass + + def refund(self, charge_id: str, amount: float = None) -> dict: + # [3 lines omitted] + pass + + def get_balance(self) -> float: + # [2 lines omitted] + pass +''' + from headroom.transforms.compression_summary import summarize_removed_code + + code_summary = summarize_removed_code(original_code, compressed_code) + + prompt = f"Here is a compressed Python file:\n\n```python\n{compressed_code}\n```\n\n" + if code_summary: + prompt += f"[Compression info: {code_summary}]\n\n" + prompt += "I need to understand the retry logic. Which function should I look at? Answer in one sentence." + + messages = [{"role": "user", "content": prompt}] + resp = _call_claude(messages, max_tokens=100) + text = resp.get("content", [{}])[0].get("text", "").lower() + + print(f"\n Code summary: {code_summary}") + print(f" LLM response: {text[:200]}") + + # The LLM should identify the charge() function + assert "charge" in text, f"LLM didn't identify charge() function. Response: {text}" diff --git a/tests/test_compression_summary_tool_eval.py b/tests/test_compression_summary_tool_eval.py new file mode 100644 index 000000000..f7a91fdd3 --- /dev/null +++ b/tests/test_compression_summary_tool_eval.py @@ -0,0 +1,295 @@ +"""Eval: Does the LLM invoke headroom_retrieve when summaries are present? + +The REAL test — it's not enough for the LLM to know something is missing. +It must actually call the tool to fetch it. + +Compares: +- WITH summary: LLM sees "2 failed, 1 error" → should call headroom_retrieve +- WITHOUT summary: LLM sees "[90 items compressed]" → likely does NOT call tool + +Requires: ANTHROPIC_API_KEY in environment or .env file. + +Run: python -m pytest tests/test_compression_summary_tool_eval.py -v -s +""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +# Load .env +env_path = Path(__file__).parent.parent / ".env" +if env_path.exists(): + for line in env_path.read_text().splitlines(): + line = line.strip() + if line and not line.startswith("#") and "=" in line: + key, _, value = line.partition("=") + os.environ.setdefault(key.strip(), value.strip()) + +ANTHROPIC_KEY = os.environ.get("ANTHROPIC_API_KEY", "") + +pytestmark = pytest.mark.skipif( + not ANTHROPIC_KEY, + reason="ANTHROPIC_API_KEY not set — skipping integration tests", +) + +# The headroom_retrieve tool definition (same as what CCR injects) +HEADROOM_RETRIEVE_TOOL = { + "name": "headroom_retrieve", + "description": ( + "Retrieve original uncompressed content from Headroom's compression cache. " + "Use this when you need more details from compressed data. " + "You can pass a query to search within the compressed content." + ), + "input_schema": { + "type": "object", + "properties": { + "hash": { + "type": "string", + "description": "The hash key from the compression marker", + }, + "query": { + "type": "string", + "description": "Optional search query to find specific items within the compressed data", + }, + }, + "required": ["hash"], + }, +} + + +def _call_claude_with_tools(messages: list[dict], tools: list[dict], max_tokens: int = 300) -> dict: + """Make a real Anthropic API call with tool use.""" + import httpx + + resp = httpx.post( + "https://api.anthropic.com/v1/messages", + headers={ + "X-Api-Key": ANTHROPIC_KEY, + "anthropic-version": "2023-06-01", + "Content-Type": "application/json", + }, + json={ + "model": "claude-sonnet-4-5-20250929", + "max_tokens": max_tokens, + "messages": messages, + "tools": tools, + }, + timeout=30, + ) + return resp.json() + + +def _make_test_results(n: int = 100) -> list[dict]: + """Test suite output with hidden failures in the compressed portion.""" + results = [] + for i in range(n): + result = { + "test_name": f"test_module_{i // 10}.test_case_{i}", + "status": "passed", + "duration_ms": 50 + i * 3, + } + if i == 42: + result["status"] = "failed" + result["error"] = "AssertionError: expected 200, got 401 in auth_middleware" + result["test_name"] = "test_auth.test_login_expired_token" + if i == 67: + result["status"] = "failed" + result["error"] = "TimeoutError: database pool exhausted after 30s" + result["test_name"] = "test_database.test_concurrent_connections" + if i == 88: + result["status"] = "error" + result["error"] = "ImportError: cannot import 'NewFeature'" + result["test_name"] = "test_features.test_new_feature_integration" + results.append(result) + return results + + +def _has_tool_use(response: dict) -> bool: + """Check if the response contains a tool_use block.""" + for block in response.get("content", []): + if block.get("type") == "tool_use": + return True + return False + + +def _get_tool_calls(response: dict) -> list[dict]: + """Extract all tool_use blocks from response.""" + calls = [] + for block in response.get("content", []): + if block.get("type") == "tool_use": + calls.append( + { + "name": block.get("name"), + "input": block.get("input", {}), + } + ) + return calls + + +class TestToolInvocationWithSummary: + """The real eval: does the LLM call headroom_retrieve?""" + + def test_with_summary_triggers_tool_call(self): + """WITH compression summary → LLM should call headroom_retrieve.""" + test_results = _make_test_results(100) + kept = test_results[:10] # All passing + + from headroom.transforms.compression_summary import summarize_dropped_items + + summary = summarize_dropped_items(test_results, kept) + + compressed = json.dumps(kept, indent=2) + compressed += ( + f"\n[90 items compressed to 10. Omitted: {summary}." + f' Retrieve specific items: headroom_retrieve(hash="ccr_test_abc123", query="your search")]' + ) + + messages = [ + { + "role": "user", + "content": ( + "Here are the test results from our CI pipeline:\n\n" + f"{compressed}\n\n" + "Tell me about any test failures. What went wrong?" + ), + }, + ] + + resp = _call_claude_with_tools(messages, [HEADROOM_RETRIEVE_TOOL]) + + tool_calls = _get_tool_calls(resp) + stop_reason = resp.get("stop_reason", "") + + print(f"\n Summary: {summary}") + print(f" Stop reason: {stop_reason}") + print(f" Tool calls: {tool_calls}") + + # With a summary showing failures, the LLM SHOULD call the tool + if stop_reason == "tool_use": + assert len(tool_calls) > 0 + call = tool_calls[0] + assert call["name"] == "headroom_retrieve" + assert call["input"].get("hash") == "ccr_test_abc123" + # The query should be about failures/errors + query = call["input"].get("query", "").lower() + print(f" Query used: {query}") + has_relevant_query = any( + term in query for term in ["fail", "error", "issue", "problem", "broken", "test"] + ) + assert has_relevant_query, f"Tool was called but query isn't relevant: {query}" + print(" RESULT: LLM invoked headroom_retrieve with relevant query ✓") + else: + # LLM responded with text — check if it at least mentions the failures + text = "" + for block in resp.get("content", []): + if block.get("type") == "text": + text += block.get("text", "") + print(f" LLM text response: {text[:200]}") + # It's acceptable if the LLM mentions it WANTS to retrieve + mentions_retrieval = any( + term in text.lower() + for term in ["retrieve", "headroom_retrieve", "fetch", "see more", "compressed"] + ) + print(f" Mentions retrieval: {mentions_retrieval}") + + def test_without_summary_baseline(self): + """WITHOUT compression summary → LLM likely does NOT call tool.""" + test_results = _make_test_results(100) + kept = test_results[:10] # All passing + + compressed = json.dumps(kept, indent=2) + compressed += "\n[90 items compressed to 10. Retrieve more: hash=ccr_test_abc123]" + + messages = [ + { + "role": "user", + "content": ( + "Here are the test results from our CI pipeline:\n\n" + f"{compressed}\n\n" + "Tell me about any test failures. What went wrong?" + ), + }, + ] + + resp = _call_claude_with_tools(messages, [HEADROOM_RETRIEVE_TOOL]) + + tool_calls = _get_tool_calls(resp) + stop_reason = resp.get("stop_reason", "") + + print(f"\n Stop reason: {stop_reason}") + print(f" Tool calls: {tool_calls}") + + if stop_reason == "tool_use": + call = tool_calls[0] + print(f" Query used: {call['input'].get('query', 'none')}") + print(" RESULT: LLM DID invoke tool (may check proactively)") + else: + text = "" + for block in resp.get("content", []): + if block.get("type") == "text": + text += block.get("text", "") + print(f" LLM text response: {text[:200]}") + print(" RESULT: LLM did NOT invoke tool — assumed all tests passed") + + def test_code_summary_triggers_retrieval(self): + """Code compression summary → LLM should retrieve specific function.""" + compressed_code = '''class PaymentProcessor: + """Processes payments via Stripe.""" + + def __init__(self, api_key: str): + # [2 lines omitted] + pass + + def charge(self, amount: float, currency: str, token: str) -> dict: + # [8 lines omitted] + pass + + def refund(self, charge_id: str, amount: float = None) -> dict: + # [3 lines omitted] + pass + + def get_balance(self) -> float: + # [2 lines omitted] + pass +# [180 tokens compressed. removed: def charge (12 lines), def refund (6 lines). Retrieve full code: headroom_retrieve(hash="ccr_code_xyz", query="function name")]''' + + messages = [ + { + "role": "user", + "content": ( + "Here's the payment processor code:\n\n" + f"```python\n{compressed_code}\n```\n\n" + "There's a bug in the retry logic for failed charges. " + "Can you find and fix it?" + ), + }, + ] + + resp = _call_claude_with_tools(messages, [HEADROOM_RETRIEVE_TOOL]) + + tool_calls = _get_tool_calls(resp) + stop_reason = resp.get("stop_reason", "") + + print(f"\n Stop reason: {stop_reason}") + print(f" Tool calls: {tool_calls}") + + if stop_reason == "tool_use": + call = tool_calls[0] + assert call["name"] == "headroom_retrieve" + query = call["input"].get("query", "").lower() + print(f" Query: {query}") + # Should be asking for the charge function specifically + has_charge = any(term in query for term in ["charge", "retry", "payment", "stripe"]) + print(f" Targets charge/retry: {has_charge}") + print(" RESULT: LLM invoked tool to get the charge() implementation ✓") + else: + text = "" + for block in resp.get("content", []): + if block.get("type") == "text": + text += block.get("text", "") + print(f" LLM text: {text[:200]}") + print(" RESULT: LLM did NOT invoke tool")