Add adaptive compression sizing and fix cost tracking bugs

- Fix LiteLLM pricing for claude-opus-4-6 by adding provider prefix
fallback (_resolve_litellm_model) when model name is unrecognized
- Fix cost savings double-counting: remove cache_read/cache_write
tokens from original_cost calculation at all 3 call sites, which
was inflating reported savings ~425x
- Add information saturation engine (adaptive_sizer.py) using Kneedle
algorithm on unique bigram coverage curves to statistically determine
optimal compression K instead of hardcoded thresholds
- Add CompressionProfile with per-tool bias multipliers (conservative,
moderate, aggressive) configurable via --tool-profile CLI flag and
HEADROOM_TOOL_PROFILES env var
- Integrate adaptive K into SmartCrusher, SearchCompressor, and
LogCompressor, threading bias through ContentRouter
- Remove Grep and Bash from DEFAULT_EXCLUDE_TOOLS so their outputs
are now compressed (only Read and Glob excluded)

headroom/config.py

@@ -348,24 +348,60 @@ class AnchorConfig:
 
 
 # Default tools to exclude from compression (local file/code tools)
-# These tools return precise content (line numbers, paths, code) where
-# exact fidelity matters. Web tools are NOT excluded - they benefit from
-# compression and can use CCR for retrieval if needed.
+# Read: Returns exact file content needed for Edit tool's old_string matching.
+#   Compressing would break the edit workflow.
+# Glob: Returns compact file path lists used for navigation. Low token count,
+#   not worth compressing.
+# Grep/Bash are NOT excluded - their outputs (search results, build logs,
+# test output) are ideal compression targets for SearchCompressor/LogCompressor,
+# and CCR provides safe retrieval if the LLM needs more detail.
 DEFAULT_EXCLUDE_TOOLS: frozenset[str] = frozenset(
     {
         "Read",
         "Glob",
-        "Grep",
-        "Bash",
         # Lowercase variants for case-insensitive matching
         "read",
         "glob",
-        "grep",
-        "bash",
     }
 )
 
 
+@dataclass
+class CompressionProfile:
+    """Per-tool compression bias applied to statistically-determined K.
+
+    Instead of hardcoding max_items=15, the adaptive sizer computes the optimal K
+    via information saturation (Kneedle on unique bigram coverage). This profile
+    applies a bias multiplier: >1 keeps more items (conservative), <1 keeps fewer
+    (aggressive).
+    """
+
+    bias: float = 1.0  # 0.7=aggressive, 1.0=moderate, 1.5=conservative
+    min_k: int = 3  # Never keep fewer than this
+    max_k: int | None = None  # Cap (None = no cap, let statistics decide)
+
+
+# Named presets for convenience
+PROFILE_PRESETS: dict[str, CompressionProfile] = {
+    "conservative": CompressionProfile(bias=1.5, min_k=5),
+    "moderate": CompressionProfile(bias=1.0, min_k=3),
+    "aggressive": CompressionProfile(bias=0.7, min_k=3),
+}
+
+# Default per-tool profiles: tools not listed here use moderate (bias=1.0)
+DEFAULT_TOOL_PROFILES: dict[str, CompressionProfile] = {
+    # Search results: keep more matches for accuracy
+    "Grep": PROFILE_PRESETS["conservative"],
+    "grep": PROFILE_PRESETS["conservative"],
+    # Logs/output: balanced compression
+    "Bash": PROFILE_PRESETS["moderate"],
+    "bash": PROFILE_PRESETS["moderate"],
+    # Web pages are verbose, compress aggressively
+    "WebFetch": PROFILE_PRESETS["aggressive"],
+    "webfetch": PROFILE_PRESETS["aggressive"],
+}
+
+
 @dataclass
 class SmartCrusherConfig:
     """Configuration for smart statistical crusher (DEFAULT).

headroom/proxy/server.py

@@ -247,6 +247,9 @@ class ProxyConfig:
     # Code-aware compression (ON by default if installed)
     code_aware_enabled: bool = True  # Enable AST-based code compression
 
+    # Per-tool compression profiles (parsed from CLI/env)
+    tool_profiles: dict[str, Any] | None = None
+
     # Smart content routing (routes each message to optimal compressor)
     smart_routing: bool = True  # Use ContentRouter for intelligent compression
 
@@ -536,6 +539,39 @@ class CostTracker:
         self._total_savings_usd: float = 0
         self._last_prune_time: datetime = datetime.now()
 
+    @staticmethod
+    def _resolve_litellm_model(model: str) -> str:
+        """Resolve model name to one LiteLLM recognizes, adding provider prefix if needed."""
+        if not LITELLM_AVAILABLE:
+            return model
+
+        # Try as-is first
+        try:
+            litellm.cost_per_token(model=model, prompt_tokens=1, completion_tokens=0)
+            return model
+        except Exception:
+            pass
+
+        # Try with provider prefix
+        prefixes = {
+            "claude-": "anthropic/",
+            "gpt-": "openai/",
+            "o1-": "openai/",
+            "o3-": "openai/",
+            "o4-": "openai/",
+            "gemini-": "google/",
+        }
+        for pattern, prefix in prefixes.items():
+            if model.startswith(pattern):
+                prefixed = f"{prefix}{model}"
+                try:
+                    litellm.cost_per_token(model=prefixed, prompt_tokens=1, completion_tokens=0)
+                    return prefixed
+                except Exception:
+                    break
+
+        return model
+
     def estimate_cost(
         self,
         model: str,
@@ -558,6 +594,9 @@ class CostTracker:
             return None
 
         try:
+            # Resolve model name (adds provider prefix if needed, e.g. claude-opus-4-6 → anthropic/claude-opus-4-6)
+            resolved_model = self._resolve_litellm_model(model)
+
             # cost_per_token returns (total_input_cost, total_output_cost) for the given token counts
             # Despite the name, it returns total cost not per-token cost
 
@@ -570,14 +609,14 @@ class CostTracker:
 
             # Get cost for regular (non-cached) input tokens
             input_cost, _ = litellm.cost_per_token(
-                model=model,
+                model=resolved_model,
                 prompt_tokens=regular_input,
                 completion_tokens=0,
             )
 
             # Get cost for output tokens
             _, output_cost = litellm.cost_per_token(
-                model=model,
+                model=resolved_model,
                 prompt_tokens=0,
                 completion_tokens=output_tokens,
             )
@@ -585,7 +624,7 @@ class CostTracker:
             # Get model info for cache pricing
             model_info: dict[str, Any] = {}
             try:
-                model_info = dict(litellm.get_model_info(model))
+                model_info = dict(litellm.get_model_info(resolved_model))
             except Exception:
                 pass
 
@@ -598,7 +637,7 @@ class CostTracker:
                 else:
                     # Fallback: most providers charge ~10% of input price for cache reads
                     cache_read_full_cost, _ = litellm.cost_per_token(
-                        model=model,
+                        model=resolved_model,
                         prompt_tokens=cache_read_tokens,
                         completion_tokens=0,
                     )
@@ -613,7 +652,7 @@ class CostTracker:
                 else:
                     # Fallback: most providers charge ~125% of input price for cache writes
                     cache_write_full_cost, _ = litellm.cost_per_token(
-                        model=model,
+                        model=resolved_model,
                         prompt_tokens=cache_write_tokens,
                         completion_tokens=0,
                     )
@@ -999,6 +1038,7 @@ class HeadroomProxy:
             router_config = ContentRouterConfig(
                 enable_llmlingua=config.llmlingua_enabled,
                 enable_code_aware=config.code_aware_enabled,
+                tool_profiles=config.tool_profiles,
             )
             transforms = [
                 CacheAligner(CacheAlignerConfig(enabled=True)),
@@ -2023,12 +2063,13 @@ class HeadroomProxy:
                         cache_read_tokens=cache_read_tokens,
                         cache_write_tokens=cache_write_tokens,
                     )
+                    # original_cost: what it would have cost without compression
+                    # Use only original_tokens at regular input rate — no cache params,
+                    # since caching is orthogonal to compression savings
                     original_cost = self.cost_tracker.estimate_cost(
                         model,
                         original_tokens,
                         output_tokens,
-                        cache_read_tokens=cache_read_tokens,
-                        cache_write_tokens=cache_write_tokens,
                     )
                     if cost_usd and original_cost:
                         savings_usd = original_cost - cost_usd
@@ -4128,11 +4169,12 @@ class HeadroomProxy:
                         output_tokens,
                         cache_read_tokens=cache_read_tokens,
                     )
+                    # original_cost: what it would have cost without compression
+                    # No cache params — caching is orthogonal to compression savings
                     original_cost = self.cost_tracker.estimate_cost(
                         model,
                         original_tokens,
                         output_tokens,
-                        cache_read_tokens=cache_read_tokens,
                     )
                     if cost_usd and original_cost:
                         savings_usd = original_cost - cost_usd
@@ -5183,11 +5225,12 @@ class HeadroomProxy:
                         output_tokens,
                         cache_read_tokens=cache_read_tokens,
                     )
+                    # original_cost: what it would have cost without compression
+                    # No cache params — caching is orthogonal to compression savings
                     original_cost = self.cost_tracker.estimate_cost(
                         model,
                         original_tokens,
                         output_tokens,
-                        cache_read_tokens=cache_read_tokens,
                     )
                     if cost_usd and original_cost:
                         savings_usd = original_cost - cost_usd
@@ -6627,6 +6670,45 @@ def _get_env_str(name: str, default: str) -> str:
     return os.environ.get(name, default)
 
 
+def _parse_tool_profiles(cli_profiles: list[str]) -> dict[str, Any]:
+    """Parse tool profiles from CLI args and HEADROOM_TOOL_PROFILES env var.
+
+    Format: ToolName:level (e.g., Grep:conservative, Bash:moderate)
+    Env var format: comma-separated (e.g., "Grep:conservative,Bash:moderate")
+
+    Returns:
+        Dict mapping tool names to CompressionProfile instances.
+    """
+    from headroom.config import PROFILE_PRESETS, CompressionProfile
+
+    profiles: dict[str, CompressionProfile] = {}
+    raw_entries: list[str] = list(cli_profiles)
+
+    # Also check env var
+    env_val = os.environ.get("HEADROOM_TOOL_PROFILES", "")
+    if env_val:
+        raw_entries.extend(e.strip() for e in env_val.split(",") if e.strip())
+
+    for entry in raw_entries:
+        if ":" not in entry:
+            logger.warning("Invalid tool profile format (expected ToolName:level): %s", entry)
+            continue
+        tool_name, level = entry.split(":", 1)
+        tool_name = tool_name.strip()
+        level = level.strip().lower()
+
+        if level in PROFILE_PRESETS:
+            profiles[tool_name] = PROFILE_PRESETS[level]
+        else:
+            logger.warning(
+                "Unknown profile level '%s' for tool '%s'. Use: conservative, moderate, aggressive",
+                level,
+                tool_name,
+            )
+
+    return profiles
+
+
 if __name__ == "__main__":
     parser = argparse.ArgumentParser(description="Headroom Proxy Server")
 
@@ -6690,6 +6772,13 @@ if __name__ == "__main__":
     parser.add_argument("--no-optimize", action="store_true", help="Disable optimization")
     parser.add_argument("--min-tokens", type=int, default=500, help="Min tokens to crush")
     parser.add_argument("--max-items", type=int, default=50, help="Max items after crush")
+    parser.add_argument(
+        "--tool-profile",
+        action="append",
+        default=[],
+        help="Per-tool compression profile: ToolName:level (e.g., Grep:conservative, Bash:moderate, WebFetch:aggressive). "
+        "Can be specified multiple times. Also settable via HEADROOM_TOOL_PROFILES env var.",
+    )
 
     # Caching
     parser.add_argument("--no-cache", action="store_true", help="Disable caching")
@@ -6783,6 +6872,9 @@ if __name__ == "__main__":
     if hasattr(args, "openrouter_api_key") and args.openrouter_api_key:
         os.environ["OPENROUTER_API_KEY"] = args.openrouter_api_key
 
+    # Parse per-tool compression profiles from CLI and env var
+    tool_profiles = _parse_tool_profiles(args.tool_profile)
+
     config = ProxyConfig(
         host=_get_env_str("HEADROOM_HOST", args.host),
         port=_get_env_int("HEADROOM_PORT", args.port),
@@ -6814,6 +6906,7 @@ if __name__ == "__main__":
         max_connections=_get_env_int("HEADROOM_MAX_CONNECTIONS", args.max_connections),
         max_keepalive_connections=_get_env_int("HEADROOM_MAX_KEEPALIVE", args.max_keepalive),
         http2=not args.no_http2 and _get_env_bool("HEADROOM_HTTP2", True),
+        tool_profiles=tool_profiles if tool_profiles else None,
     )
 
     # Get worker and concurrency settings

headroom/transforms/adaptive_sizer.py

@@ -0,0 +1,289 @@
+"""Adaptive compression sizing via information saturation detection.
+
+Instead of hardcoded max_items/max_matches, this module statistically determines
+how many items to keep by finding the "knee point" — where adding more items
+stops providing meaningful new information.
+
+Algorithm: Track unique bigrams as items are added in importance order. Build a
+cumulative coverage curve. Find the knee (Kneedle algorithm) where marginal
+information gain drops sharply. That's the optimal K.
+
+Per-tool profiles apply a bias multiplier on the statistically-determined K:
+- conservative (bias=1.5): keep 50% more than mathematically needed
+- moderate (bias=1.0): trust the statistics
+- aggressive (bias=0.7): compress harder
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import zlib
+from collections.abc import Sequence
+
+logger = logging.getLogger(__name__)
+
+
+def compute_optimal_k(
+    items: Sequence[str],
+    bias: float = 1.0,
+    min_k: int = 3,
+    max_k: int | None = None,
+) -> int:
+    """Compute the optimal number of items to keep using information saturation.
+
+    Three-tier decision system:
+      Tier 1 (fast path): trivial cases, near-duplicate detection
+      Tier 2 (standard):  Kneedle on unique bigram coverage curve
+      Tier 3 (validation): zlib compression ratio sanity check
+
+    Args:
+        items: Sequence of string representations of items (in importance order).
+        bias: Multiplier on the knee point. >1 = keep more, <1 = keep fewer.
+        min_k: Never return fewer than this.
+        max_k: Never return more than this (None = no cap).
+
+    Returns:
+        Optimal number of items to keep.
+    """
+    n = len(items)
+    effective_max = max_k if max_k is not None else n
+
+    # Tier 1: Fast path
+    if n <= 8:
+        return n
+
+    # Check for near-total redundancy
+    unique_count = count_unique_simhash(items)
+    if unique_count <= 3:
+        k = max(min_k, unique_count)
+        return min(k, effective_max)
+
+    # Tier 2: Kneedle on unique bigram coverage
+    curve = compute_unique_bigram_curve(items)
+    knee = find_knee(curve)
+
+    if knee is None:
+        # No clear knee — content is uniformly diverse, keep ~30% as heuristic
+        knee = max(min_k, int(n * 0.3))
+
+    # Apply bias multiplier
+    k = max(min_k, int(knee * bias))
+    k = min(k, effective_max)
+
+    # Tier 3: Validate with zlib compression ratio
+    k = _validate_with_zlib(items, k, effective_max)
+
+    k = max(min_k, min(k, effective_max))
+
+    logger.debug(
+        "adaptive_sizer: n=%d unique=%d knee=%s bias=%.1f → k=%d",
+        n,
+        unique_count,
+        knee,
+        bias,
+        k,
+    )
+    return k
+
+
+def find_knee(curve: list[int]) -> int | None:
+    """Find the knee point in a monotonically increasing curve.
+
+    Uses the Kneedle algorithm: normalize to [0,1], compute the difference
+    from the y=x diagonal, return the index of maximum difference.
+
+    Args:
+        curve: List of cumulative values (e.g., unique bigram counts).
+
+    Returns:
+        Index of the knee point, or None if no clear knee exists.
+    """
+    n = len(curve)
+    if n < 3:
+        return None
+
+    # Normalize x and y to [0, 1]
+    x_min, x_max = 0, n - 1
+    y_min, y_max = curve[0], curve[-1]
+
+    if y_max == y_min:
+        # Flat curve — all items are identical
+        return 1
+
+    x_range = x_max - x_min
+    y_range = y_max - y_min
+
+    # Compute difference from the diagonal (y = x in normalized space)
+    max_diff = -1.0
+    knee_idx = None
+
+    for i in range(n):
+        x_norm = (i - x_min) / x_range
+        y_norm = (curve[i] - y_min) / y_range
+        diff = y_norm - x_norm  # For concave curves, knee is where this is maximized
+        if diff > max_diff:
+            max_diff = diff
+            knee_idx = i
+
+    # Require a meaningful deviation from diagonal
+    if max_diff < 0.05:
+        return None
+
+    # Knee is at knee_idx, but we want to include items up to and including the knee
+    # Add 1 because we're converting from 0-indexed to count
+    return knee_idx + 1 if knee_idx is not None else None
+
+
+def compute_unique_bigram_curve(items: Sequence[str]) -> list[int]:
+    """Build cumulative unique bigram coverage curve.
+
+    For each item (in order), extracts word-level bigrams, adds them to a
+    running set, and records the total unique count.
+
+    Args:
+        items: Sequence of string items in importance order.
+
+    Returns:
+        List where curve[k] = number of unique bigrams after seeing items[0:k+1].
+    """
+    seen_bigrams: set[tuple[str, str]] = set()
+    curve: list[int] = []
+
+    for item in items:
+        words = item.lower().split()
+        if len(words) < 2:
+            # For single-word items, use the word itself as a "unigram bigram"
+            seen_bigrams.add((words[0] if words else "", ""))
+        else:
+            for j in range(len(words) - 1):
+                seen_bigrams.add((words[j], words[j + 1]))
+        curve.append(len(seen_bigrams))
+
+    return curve
+
+
+def _simhash(text: str) -> int:
+    """Compute a 64-bit SimHash fingerprint for a text string.
+
+    Uses character 4-grams hashed to 64-bit values, then aggregates
+    via weighted bit voting.
+
+    Args:
+        text: Input text.
+
+    Returns:
+        64-bit integer fingerprint.
+    """
+    v = [0] * 64
+    text_lower = text.lower()
+
+    # Character 4-grams
+    for i in range(max(1, len(text_lower) - 3)):
+        gram = text_lower[i : i + 4]
+        h = int(hashlib.md5(gram.encode(), usedforsecurity=False).hexdigest()[:16], 16)
+        for j in range(64):
+            if h & (1 << j):
+                v[j] += 1
+            else:
+                v[j] -= 1
+
+    fingerprint = 0
+    for j in range(64):
+        if v[j] > 0:
+            fingerprint |= 1 << j
+    return fingerprint
+
+
+def _hamming_distance(a: int, b: int) -> int:
+    """Count differing bits between two 64-bit integers."""
+    return bin(a ^ b).count("1")
+
+
+def count_unique_simhash(items: Sequence[str], threshold: int = 3) -> int:
+    """Count items with distinct content using SimHash.
+
+    Groups items by SimHash fingerprint similarity (Hamming distance <= threshold).
+    Returns the number of distinct groups.
+
+    Args:
+        items: Sequence of string items.
+        threshold: Max Hamming distance to consider items as duplicates.
+
+    Returns:
+        Number of unique content groups.
+    """
+    if not items:
+        return 0
+
+    # Compute fingerprints
+    fingerprints = [_simhash(item) for item in items]
+
+    # Greedy clustering: assign each item to the first matching cluster
+    clusters: list[int] = []  # Representative fingerprint per cluster
+    for fp in fingerprints:
+        matched = False
+        for rep in clusters:
+            if _hamming_distance(fp, rep) <= threshold:
+                matched = True
+                break
+        if not matched:
+            clusters.append(fp)
+
+    return len(clusters)
+
+
+def _validate_with_zlib(
+    items: Sequence[str],
+    k: int,
+    max_k: int,
+    tolerance: float = 0.15,
+) -> int:
+    """Validate K using zlib compression ratio comparison.
+
+    If the compression ratio of the selected subset differs significantly
+    from the full set, increase K.
+
+    Args:
+        items: All items.
+        k: Currently proposed K.
+        max_k: Maximum allowed K.
+        tolerance: Max allowed ratio difference (default 15%).
+
+    Returns:
+        Adjusted K (may be increased if validation fails).
+    """
+    if k >= len(items) or k >= max_k:
+        return k
+
+    full_text = "\n".join(items).encode()
+    subset_text = "\n".join(items[:k]).encode()
+
+    # Skip validation for very small content (zlib overhead dominates)
+    if len(full_text) < 200:
+        return k
+
+    full_compressed = len(zlib.compress(full_text, level=1))
+    subset_compressed = len(zlib.compress(subset_text, level=1))
+
+    full_ratio = full_compressed / len(full_text) if full_text else 1.0
+    subset_ratio = subset_compressed / len(subset_text) if subset_text else 1.0
+
+    # If subset compresses much better than full, it's missing diverse content
+    # A lower ratio means more redundancy. If subset ratio is much lower,
+    # it means the subset is more redundant than the full set — we're missing info.
+    ratio_diff = abs(full_ratio - subset_ratio)
+
+    if ratio_diff > tolerance:
+        # Increase K by 20% to capture more diversity
+        adjusted_k = min(int(k * 1.2), max_k)
+        logger.debug(
+            "zlib validation: ratio_diff=%.3f > %.3f, adjusting k=%d → %d",
+            ratio_diff,
+            tolerance,
+            k,
+            adjusted_k,
+        )
+        return adjusted_k
+
+    return k

headroom/transforms/content_router.py

@@ -292,6 +292,10 @@ class ContentRouterConfig:
     # Set to None to use DEFAULT_EXCLUDE_TOOLS, or provide custom set
     exclude_tools: set[str] | None = None
 
+    # Per-tool compression profiles (tool_name → CompressionProfile)
+    # Set to None to use DEFAULT_TOOL_PROFILES from config
+    tool_profiles: dict[str, Any] | None = None
+
 
 # Patterns for detecting mixed content
 _CODE_FENCE_PATTERN = re.compile(r"^```(\w*)\s*$", re.MULTILINE)
@@ -588,6 +592,7 @@ class ContentRouter(Transform):
         content: str,
         context: str = "",
         question: str | None = None,
+        bias: float = 1.0,
     ) -> RouterCompressionResult:
         """Compress content using optimal strategy based on content detection.
 
@@ -596,6 +601,7 @@ class ContentRouter(Transform):
             context: Optional context for relevance-aware compression.
             question: Optional question for QA-aware compression. When provided,
                 tokens relevant to answering this question are preserved.
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
 
         Returns:
             RouterCompressionResult with compressed content and routing metadata.
@@ -612,9 +618,9 @@ class ContentRouter(Transform):
         strategy = self._determine_strategy(content)
 
         if strategy == CompressionStrategy.MIXED:
-            return self._compress_mixed(content, context, question)
+            return self._compress_mixed(content, context, question, bias=bias)
         else:
-            return self._compress_pure(content, strategy, context, question)
+            return self._compress_pure(content, strategy, context, question, bias=bias)
 
     def _determine_strategy(self, content: str) -> CompressionStrategy:
         """Determine the compression strategy from content analysis.
@@ -668,6 +674,7 @@ class ContentRouter(Transform):
         content: str,
         context: str,
         question: str | None = None,
+        bias: float = 1.0,
     ) -> RouterCompressionResult:
         """Compress mixed content by splitting and routing sections.
 
@@ -675,6 +682,7 @@ class ContentRouter(Transform):
             content: Mixed content to compress.
             context: User context for relevance.
             question: Optional question for QA-aware compression.
+            bias: Compression bias multiplier.
 
         Returns:
             RouterCompressionResult with reassembled content.
@@ -698,7 +706,12 @@ class ContentRouter(Transform):
             # Compress section
             original_tokens = len(section.content.split())
             compressed_content, compressed_tokens = self._apply_strategy_to_content(
-                section.content, strategy, context, section.language, question
+                section.content,
+                strategy,
+                context,
+                section.language,
+                question,
+                bias=bias,
             )
 
             # Preserve code fence markers
@@ -730,6 +743,7 @@ class ContentRouter(Transform):
         strategy: CompressionStrategy,
         context: str,
         question: str | None = None,
+        bias: float = 1.0,
     ) -> RouterCompressionResult:
         """Compress pure (non-mixed) content.
 
@@ -738,6 +752,7 @@ class ContentRouter(Transform):
             strategy: Selected strategy.
             context: User context.
             question: Optional question for QA-aware compression.
+            bias: Compression bias multiplier.
 
         Returns:
             RouterCompressionResult.
@@ -745,7 +760,7 @@ class ContentRouter(Transform):
         original_tokens = len(content.split())
 
         compressed, compressed_tokens = self._apply_strategy_to_content(
-            content, strategy, context, question=question
+            content, strategy, context, question=question, bias=bias
         )
 
         return RouterCompressionResult(
@@ -769,6 +784,7 @@ class ContentRouter(Transform):
         context: str,
         language: str | None = None,
         question: str | None = None,
+        bias: float = 1.0,
     ) -> tuple[str, int]:
         """Apply a compression strategy to content.
 
@@ -778,6 +794,7 @@ class ContentRouter(Transform):
             context: User context.
             language: Language hint for code.
             question: Optional question for QA-aware compression.
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
 
         Returns:
             Tuple of (compressed_content, compressed_token_count).
@@ -804,14 +821,14 @@ class ContentRouter(Transform):
                 if self.config.enable_smart_crusher:
                     crusher = self._get_smart_crusher()
                     if crusher:
-                        result = crusher.crush(content, query=context)
+                        result = crusher.crush(content, query=context, bias=bias)
                         return result.compressed, len(result.compressed.split())
 
             elif strategy == CompressionStrategy.SEARCH:
                 if self.config.enable_search_compressor:
                     compressor = self._get_search_compressor()
                     if compressor:
-                        result = compressor.compress(content, context=context)
+                        result = compressor.compress(content, context=context, bias=bias)
                         compressed, compressed_tokens = (
                             result.compressed,
                             len(result.compressed.split()),
@@ -821,7 +838,7 @@ class ContentRouter(Transform):
                 if self.config.enable_log_compressor:
                     compressor = self._get_log_compressor()
                     if compressor:
-                        result = compressor.compress(content)
+                        result = compressor.compress(content, bias=bias)
                         compressed, compressed_tokens = (
                             result.compressed,
                             result.compressed_line_count,
@@ -1184,11 +1201,17 @@ class ContentRouter(Transform):
         for i, message in enumerate(messages):
             role = message.get("role", "")
             content = message.get("content", "")
+            bias = 1.0  # Default bias, may be overridden for tool messages
 
             # Handle list content (Anthropic format with content blocks)
             if isinstance(content, list):
                 transformed_message = self._process_content_blocks(
-                    message, content, context, transforms_applied, excluded_tool_ids
+                    message,
+                    content,
+                    context,
+                    transforms_applied,
+                    excluded_tool_ids,
+                    tool_name_map=tool_name_map,
                 )
                 transformed_messages.append(transformed_message)
                 continue
@@ -1205,6 +1228,9 @@ class ContentRouter(Transform):
                     transformed_messages.append(message)
                     transforms_applied.append("router:excluded:tool")
                     continue
+                # Look up tool-specific compression bias for OpenAI tool messages
+                tool_name = tool_name_map.get(tool_call_id, "")
+                bias = self._get_tool_bias(tool_name) if tool_name else 1.0
 
             # Protection 1: Never compress user messages
             if self.config.skip_user_messages and role == "user":
@@ -1239,7 +1265,9 @@ class ContentRouter(Transform):
                 continue
 
             # Route and compress based on content detection
-            result = self.compress(content, context=context)
+            # Use tool-specific bias for tool messages, default 1.0 for others
+            msg_bias = bias if role == "tool" else 1.0
+            result = self.compress(content, context=context, bias=msg_bias)
 
             if result.compression_ratio < 0.9:
                 transformed_messages.append({**message, "content": result.compressed})
@@ -1261,6 +1289,27 @@ class ContentRouter(Transform):
             warnings=warnings,
         )
 
+    def _get_tool_bias(self, tool_name: str) -> float:
+        """Look up compression bias for a tool name.
+
+        Checks user-configured profiles first, then DEFAULT_TOOL_PROFILES.
+        Returns 1.0 (moderate) if no profile is configured.
+        """
+        from ..config import DEFAULT_TOOL_PROFILES
+
+        # Check user-configured profiles
+        if self.config.tool_profiles:
+            profile = self.config.tool_profiles.get(tool_name)
+            if profile:
+                return float(profile.bias)
+
+        # Check default profiles
+        profile = DEFAULT_TOOL_PROFILES.get(tool_name)
+        if profile:
+            return profile.bias
+
+        return 1.0  # Default: moderate
+
     def _process_content_blocks(
         self,
         message: dict[str, Any],
@@ -1268,6 +1317,7 @@ class ContentRouter(Transform):
         context: str,
         transforms_applied: list[str],
         excluded_tool_ids: set[str],
+        tool_name_map: dict[str, str] | None = None,
     ) -> dict[str, Any]:
         """Process content blocks (Anthropic format) for tool_result compression.
 
@@ -1279,6 +1329,8 @@ class ContentRouter(Transform):
             content_blocks: List of content blocks.
             context: Context for compression.
             transforms_applied: List to append transform names to.
+            excluded_tool_ids: Tool IDs to skip compression for.
+            tool_name_map: Mapping from tool_call_id to tool_name for profile lookup.
 
         Returns:
             Transformed message with compressed content blocks.
@@ -1302,12 +1354,16 @@ class ContentRouter(Transform):
                     transforms_applied.append("router:excluded:tool")
                     continue
 
+                # Look up tool-specific compression bias
+                tool_name = (tool_name_map or {}).get(tool_use_id, "")
+                bias = self._get_tool_bias(tool_name) if tool_name else 1.0
+
                 tool_content = block.get("content", "")
 
                 # Only process string content
                 if isinstance(tool_content, str) and len(tool_content) > 500:
                     # Compress using content detection (will auto-detect JSON arrays, etc.)
-                    result = self.compress(tool_content, context=context)
+                    result = self.compress(tool_content, context=context, bias=bias)
                     if result.compression_ratio < 0.9:
                         new_blocks.append({**block, "content": result.compressed})
                         transforms_applied.append(

headroom/transforms/log_compressor.py

@@ -176,12 +176,13 @@ class LogCompressor:
         """
         self.config = config or LogCompressorConfig()
 
-    def compress(self, content: str, context: str = "") -> LogCompressionResult:
+    def compress(self, content: str, context: str = "", bias: float = 1.0) -> LogCompressionResult:
         """Compress log output.
 
         Args:
             content: Raw log output.
             context: User query context (unused for now).
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
 
         Returns:
             LogCompressionResult with compressed output and metadata.
@@ -204,8 +205,8 @@ class LogCompressor:
         # Parse and categorize lines
         log_lines = self._parse_lines(lines)
 
-        # Select important lines
-        selected = self._select_lines(log_lines)
+        # Select important lines (with adaptive sizing)
+        selected = self._select_lines(log_lines, bias=bias)
 
         # Format output with summaries
         compressed, stats = self._format_output(selected, log_lines)
@@ -319,10 +320,22 @@ class LogCompressor:
 
         return min(1.0, score)
 
-    def _select_lines(self, log_lines: list[LogLine]) -> list[LogLine]:
-        """Select important lines to keep."""
+    def _select_lines(self, log_lines: list[LogLine], bias: float = 1.0) -> list[LogLine]:
+        """Select important lines to keep using adaptive sizing."""
+        from .adaptive_sizer import compute_optimal_k
+
         selected: list[LogLine] = []
 
+        # Compute adaptive line budget from the FULL log output
+        # This determines "how many lines of value exist" across the entire log
+        all_line_strings = [line.content for line in log_lines]
+        adaptive_max = compute_optimal_k(
+            all_line_strings,
+            bias=bias,
+            min_k=10,
+            max_k=self.config.max_total_lines,
+        )
+
         # Group by category
         errors: list[LogLine] = []
         fails: list[LogLine] = []
@@ -381,11 +394,11 @@ class LogCompressor:
         # Sort by line number and dedupe
         selected = sorted(set(selected), key=lambda x: x.line_number)
 
-        # Limit total lines
-        if len(selected) > self.config.max_total_lines:
+        # Apply adaptive line limit
+        if len(selected) > adaptive_max:
             # Keep most important lines
             selected = sorted(selected, key=lambda x: x.score, reverse=True)
-            selected = selected[: self.config.max_total_lines]
+            selected = selected[:adaptive_max]
             selected = sorted(selected, key=lambda x: x.line_number)
 
         return selected

headroom/transforms/search_compressor.py

@@ -106,12 +106,14 @@ class SearchCompressor:
         self,
         content: str,
         context: str = "",
+        bias: float = 1.0,
     ) -> SearchCompressionResult:
         """Compress search results.
 
         Args:
             content: Raw grep/ripgrep output.
             context: User query context for relevance scoring.
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
 
         Returns:
             SearchCompressionResult with compressed output and metadata.
@@ -135,8 +137,8 @@ class SearchCompressor:
         # Score matches by relevance
         self._score_matches(file_matches, context)
 
-        # Select top matches per file
-        selected = self._select_matches(file_matches)
+        # Select top matches per file (with adaptive sizing)
+        selected = self._select_matches(file_matches, bias=bias)
 
         # Format compressed output
         compressed, summaries = self._format_output(selected, file_matches)
@@ -235,8 +237,11 @@ class SearchCompressor:
     def _select_matches(
         self,
         file_matches: dict[str, FileMatches],
+        bias: float = 1.0,
     ) -> dict[str, FileMatches]:
-        """Select top matches per file and globally."""
+        """Select top matches per file and globally using adaptive sizing."""
+        from .adaptive_sizer import compute_optimal_k
+
         selected: dict[str, FileMatches] = {}
 
         # Sort files by total match score (highest first)
@@ -249,9 +254,22 @@ class SearchCompressor:
         # Limit number of files
         sorted_files = sorted_files[: self.config.max_files]
 
+        # Compute adaptive total matches limit
+        all_match_strings = [
+            f"{file_path}:{m.line_number}:{m.content}"
+            for file_path, fm in sorted_files
+            for m in fm.matches
+        ]
+        adaptive_total = compute_optimal_k(
+            all_match_strings,
+            bias=bias,
+            min_k=5,
+            max_k=self.config.max_total_matches,
+        )
+
         total_selected = 0
         for file_path, fm in sorted_files:
-            if total_selected >= self.config.max_total_matches:
+            if total_selected >= adaptive_total:
                 break
 
             # Sort matches by score
@@ -261,7 +279,7 @@ class SearchCompressor:
             file_selected: list[SearchMatch] = []
             remaining_slots = min(
                 self.config.max_matches_per_file,
-                self.config.max_total_matches - total_selected,
+                adaptive_total - total_selected,
             )
 
             # Always include first and last if configured

headroom/transforms/smart_crusher.py

@@ -1440,7 +1440,7 @@ class SmartCrusher(Transform):
             CompressionStrategy.SMART_SAMPLE: AnchorDataPattern.GENERIC,
         }.get(strategy, AnchorDataPattern.GENERIC)
 
-    def crush(self, content: str, query: str = "") -> CrushResult:
+    def crush(self, content: str, query: str = "", bias: float = 1.0) -> CrushResult:
         """Crush content string directly (for use by ContentRouter).
 
         This is a simplified interface for compressing a single content string,
@@ -1449,12 +1449,13 @@ class SmartCrusher(Transform):
         Args:
             content: JSON string content to compress.
             query: Query context for relevance-based compression.
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
 
         Returns:
             CrushResult with compressed content and metadata.
         """
         compressed, was_modified, analysis_info = self._smart_crush_content(
-            content, query_context=query
+            content, query_context=query, bias=bias
         )
         return CrushResult(
             compressed=compressed,
@@ -2088,7 +2089,11 @@ class SmartCrusher(Transform):
         return " ".join(context_parts)
 
     def _smart_crush_content(
-        self, content: str, query_context: str = "", tool_name: str | None = None
+        self,
+        content: str,
+        query_context: str = "",
+        tool_name: str | None = None,
+        bias: float = 1.0,
     ) -> tuple[str, bool, str]:
         """
         Apply smart crushing to content.
@@ -2100,6 +2105,7 @@ class SmartCrusher(Transform):
             content: Content to crush (JSON or plain text).
             query_context: Context string from user messages for relevance scoring.
             tool_name: Name of the tool that produced this output.
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
 
         Returns:
             Tuple of (crushed_content, was_modified, analysis_info).
@@ -2113,7 +2119,7 @@ class SmartCrusher(Transform):
 
         # Recursively process and crush arrays
         crushed, info, ccr_markers = self._process_value(
-            parsed, query_context=query_context, tool_name=tool_name
+            parsed, query_context=query_context, tool_name=tool_name, bias=bias
         )
 
         result = safe_json_dumps(crushed, indent=None)
@@ -2132,7 +2138,12 @@ class SmartCrusher(Transform):
         return result, was_modified, info
 
     def _process_value(
-        self, value: Any, depth: int = 0, query_context: str = "", tool_name: str | None = None
+        self,
+        value: Any,
+        depth: int = 0,
+        query_context: str = "",
+        tool_name: str | None = None,
+        bias: float = 1.0,
     ) -> tuple[Any, str, list[tuple[str, int, int]]]:
         """Recursively process a value, crushing arrays where appropriate.
 
@@ -2148,7 +2159,9 @@ class SmartCrusher(Transform):
             # 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(value, query_context, tool_name)
+                crushed, strategy, ccr_hash = 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
@@ -2161,7 +2174,7 @@ class SmartCrusher(Transform):
                 processed = []
                 for item in value:
                     p_item, p_info, p_markers = self._process_value(
-                        item, depth + 1, query_context, tool_name
+                        item, depth + 1, query_context, tool_name, bias=bias
                     )
                     processed.append(p_item)
                     if p_info:
@@ -2174,7 +2187,7 @@ class SmartCrusher(Transform):
             processed_dict: dict[str, Any] = {}
             for k, v in value.items():
                 p_val, p_info, p_markers = self._process_value(
-                    v, depth + 1, query_context, tool_name
+                    v, depth + 1, query_context, tool_name, bias=bias
                 )
                 processed_dict[k] = p_val
                 if p_info:
@@ -2186,7 +2199,11 @@ class SmartCrusher(Transform):
             return value, "", []
 
     def _crush_array(
-        self, items: list[dict], query_context: str = "", tool_name: str | None = None
+        self,
+        items: list[dict],
+        query_context: str = "",
+        tool_name: str | None = None,
+        bias: float = 1.0,
     ) -> tuple[list, str, str | None]:
         """Crush an array using statistical analysis and relevance scoring.
 
@@ -2199,17 +2216,34 @@ class SmartCrusher(Transform):
         Feedback-aware: Uses learned patterns to adjust compression aggressiveness.
         High retrieval rate for a tool → compress less aggressively.
 
+        Args:
+            items: List of dict items to compress.
+            query_context: Context string from user messages for relevance scoring.
+            tool_name: Name of the tool that produced this output.
+            bias: Compression bias multiplier (>1 = keep more, <1 = keep fewer).
+
         Returns:
             Tuple of (crushed_items, strategy_info, ccr_hash).
             ccr_hash is the hash for retrieval if CCR is enabled, None otherwise.
         """
-        # BOUNDARY CHECK: If already at or below max_items, no compression needed
-        if len(items) <= self.config.max_items_after_crush:
-            return items, "none:at_limit", None
+        # BOUNDARY CHECK: Use adaptive sizing instead of hardcoded limit
+        # compute_optimal_k handles trivial cases (n <= 8 → keep all)
+        from .adaptive_sizer import compute_optimal_k
+
+        item_strings = [json.dumps(item, default=str) for item in items]
+        adaptive_k = compute_optimal_k(
+            item_strings,
+            bias=bias,
+            min_k=3,
+            max_k=self.config.max_items_after_crush if self.config.max_items_after_crush else None,
+        )
+
+        if len(items) <= adaptive_k:
+            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
-        effective_max_items = self.config.max_items_after_crush
+        effective_max_items = adaptive_k
         hints_applied = False
         toin_hint_applied = False