feat(router): adaptive compression with Read lifecycle and context-pressure scaling

Enable ReadLifecycle by default so stale/superseded Read outputs are
automatically replaced with compact CCR markers — these are provably safe
to compress (file was edited or re-read).

Replace static compression thresholds with adaptive parameters that scale
with conversation length and context pressure:

- protect_recent_reads_fraction: protects the most-recent 50% of messages
from Read exclusion. Old Reads beyond this window become compressible,
preventing the "28 excluded Read/Glob, 0 tokens saved" problem.

- min_ratio_relaxed / min_ratio_aggressive: compression acceptance
threshold interpolates linearly with context pressure (tokens / model
limit). Low pressure → 0.85 (picky), high pressure → 0.65 (accept
anything helpful). Eliminates the fixed 0.9 gate that was rejecting
20+ messages per request.

Also adds --no-read-lifecycle CLI flag, and fixes a missing
pytest.importorskip guard for sentence-transformers in memory tests.

headroom/cli/proxy.py

@@ -29,6 +29,12 @@ from .main import main
 )
 # Code-aware compression (ON by default if installed)
 @click.option("--no-code-aware", is_flag=True, help="Disable AST-based code compression")
+# Read lifecycle (ON by default: compresses stale/superseded Read outputs)
+@click.option(
+    "--no-read-lifecycle",
+    is_flag=True,
+    help="Disable Read lifecycle management (stale/superseded Read compression)",
+)
 # Intelligent Context Management (ON by default)
 @click.option(
     "--no-intelligent-context",
@@ -110,6 +116,7 @@ def proxy(
     llmlingua_device: str,
     llmlingua_rate: float,
     no_code_aware: bool,
+    no_read_lifecycle: bool,
     no_intelligent_context: bool,
     no_intelligent_scoring: bool,
     no_compress_first: bool,
@@ -162,6 +169,8 @@ def proxy(
         llmlingua_target_rate=llmlingua_rate,
         # Code-aware: ON by default (use --no-code-aware to disable)
         code_aware_enabled=not no_code_aware,
+        # Read lifecycle: ON by default (use --no-read-lifecycle to disable)
+        read_lifecycle=not no_read_lifecycle,
         # Intelligent Context: ON by default (use --no-intelligent-context to disable)
         intelligent_context=not no_intelligent_context,
         intelligent_context_scoring=not no_intelligent_scoring,

headroom/config.py

@@ -388,7 +388,7 @@ class ReadLifecycleConfig:
     outputs still bypass ContentRouter compression.
     """
 
-    enabled: bool = False  # Opt-in; default preserves backward-compatible behavior
+    enabled: bool = True  # On by default: stale/superseded Reads are provably safe to compress
     compress_stale: bool = True  # Replace Reads of files that were later edited
     compress_superseded: bool = True  # Replace Reads of files that were later re-Read
     min_size_bytes: int = 512  # Skip tiny Read outputs (not worth the overhead)

headroom/proxy/server.py

@@ -79,6 +79,7 @@ from headroom.config import (
     CacheAlignerConfig,
     CCRConfig,
     IntelligentContextConfig,
+    ReadLifecycleConfig,
     RollingWindowConfig,
     SmartCrusherConfig,
 )
@@ -252,6 +253,9 @@ class ProxyConfig:
     # Per-tool compression profiles (parsed from CLI/env)
     tool_profiles: dict[str, Any] | None = None
 
+    # Read lifecycle management (compress stale/superseded Read outputs)
+    read_lifecycle: bool = True  # ON by default: stale/superseded are provably safe
+
     # Smart content routing (routes each message to optimal compressor)
     smart_routing: bool = True  # Use ContentRouter for intelligent compression
 
@@ -1069,6 +1073,7 @@ class HeadroomProxy:
                 enable_llmlingua=config.llmlingua_enabled,
                 enable_code_aware=config.code_aware_enabled,
                 tool_profiles=config.tool_profiles,
+                read_lifecycle=ReadLifecycleConfig(enabled=config.read_lifecycle),
             )
             transforms = [
                 CacheAligner(CacheAlignerConfig(enabled=True)),

headroom/transforms/content_router.py

@@ -284,6 +284,19 @@ class ContentRouterConfig:
     protect_recent_code: int = 4  # Don't compress CODE in last N messages (0 = disabled)
     protect_analysis_context: bool = True  # Detect "analyze/review" intent, protect code
 
+    # Adaptive Read protection: fraction of total messages to protect from
+    # compression.  At 10 msgs, protects ~5 Reads.  At 100 msgs, protects ~10.
+    # Old Reads beyond this window become compressible even though they are
+    # in DEFAULT_EXCLUDE_TOOLS.  0.0 = always exclude all (old behavior).
+    protect_recent_reads_fraction: float = 0.5  # protect the most-recent 50% of messages
+
+    # Adaptive compression ratio: scales with context pressure.
+    # At low pressure (<30% full), use the relaxed threshold (reject marginal).
+    # At high pressure (>80% full), use the aggressive threshold (accept anything helpful).
+    # Linearly interpolates between the two.
+    min_ratio_relaxed: float = 0.85  # when context is mostly empty
+    min_ratio_aggressive: float = 0.65  # when context is nearly full
+
     # CCR (Compress-Cache-Retrieve) settings for SmartCrusher
     ccr_enabled: bool = True  # Enable CCR marker injection for reversible compression
     ccr_inject_marker: bool = True  # Add retrieval markers to compressed content
@@ -1208,22 +1221,78 @@ class ContentRouter(Transform):
             tool_id for tool_id, name in tool_name_map.items() if name in exclude_tools
         }
 
+        # --- Adaptive parameters based on context pressure ---
+        num_messages = len(messages)
+        model_limit = kwargs.get("model_limit", 0)
+
+        # Adaptive Read protection: protect a fraction of recent messages
+        if self.config.protect_recent_reads_fraction > 0:
+            # Scale: at 10 msgs protect 5, at 50 msgs protect 25, at 200 msgs protect 100
+            # But cap at a reasonable floor so very short convos still protect everything
+            read_protection_window = max(
+                4,  # always protect at least last 4 messages
+                int(num_messages * self.config.protect_recent_reads_fraction),
+            )
+        else:
+            read_protection_window = num_messages  # 0.0 = protect all (old behavior)
+
+        # Adaptive compression ratio: scale with context pressure
+        if model_limit > 0:
+            context_pressure = min(1.0, tokens_before / model_limit)
+        else:
+            context_pressure = 0.5  # default: moderate
+
+        # Linear interpolation between relaxed and aggressive thresholds
+        # pressure 0.0 → relaxed, pressure 1.0 → aggressive
+        min_ratio = (
+            self.config.min_ratio_relaxed
+            + (self.config.min_ratio_aggressive - self.config.min_ratio_relaxed) * context_pressure
+        )
+        # Clamp to [aggressive, relaxed] range
+        min_ratio = max(
+            self.config.min_ratio_aggressive,
+            min(self.config.min_ratio_relaxed, min_ratio),
+        )
+
+        if context_pressure > 0.3:
+            logger.debug(
+                "content_router adaptive: pressure=%.2f, min_ratio=%.2f, "
+                "read_protect_window=%d/%d msgs",
+                context_pressure,
+                min_ratio,
+                read_protection_window,
+                num_messages,
+            )
+
         transformed_messages: list[dict[str, Any]] = []
         transforms_applied: list[str] = []
         warnings: list[str] = []
 
+        # Routing reason counters for summary logging
+        route_counts: dict[str, int] = {
+            "excluded_tool": 0,
+            "user_msg": 0,
+            "small": 0,
+            "recent_code": 0,
+            "analysis_ctx": 0,
+            "ratio_too_high": 0,
+            "non_string": 0,
+            "content_blocks": 0,
+        }
+        compressed_details: list[str] = []  # e.g. ["code_aware:0.72", "llmlingua:0.65"]
+
         # Check for analysis intent in the most recent user message
         analysis_intent = False
         if self.config.protect_analysis_context:
             analysis_intent = self._detect_analysis_intent(messages)
 
-        num_messages = len(messages)
-
         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
 
+            messages_from_end = num_messages - i
+
             # Handle list content (Anthropic format with content blocks)
             if isinstance(content, list):
                 transformed_message = self._process_content_blocks(
@@ -1233,22 +1302,37 @@ class ContentRouter(Transform):
                     transforms_applied,
                     excluded_tool_ids,
                     tool_name_map=tool_name_map,
+                    route_counts=route_counts,
+                    compressed_details=compressed_details,
+                    min_ratio=min_ratio,
+                    read_protection_window=read_protection_window,
+                    messages_from_end=messages_from_end,
                 )
                 transformed_messages.append(transformed_message)
+                route_counts["content_blocks"] += 1
                 continue
 
             # Skip non-string content (other types)
             if not isinstance(content, str):
                 transformed_messages.append(message)
+                route_counts["non_string"] += 1
                 continue
 
             # Skip OpenAI-style tool messages for excluded tools
+            # BUT: allow compression of old excluded-tool outputs beyond the
+            # adaptive protection window (age-based decay).
             if role == "tool":
                 tool_call_id = message.get("tool_call_id", "")
                 if tool_call_id in excluded_tool_ids:
-                    transformed_messages.append(message)
-                    transforms_applied.append("router:excluded:tool")
-                    continue
+                    if messages_from_end <= read_protection_window:
+                        # Recent — protect as before
+                        transformed_messages.append(message)
+                        transforms_applied.append("router:excluded:tool")
+                        route_counts["excluded_tool"] += 1
+                        continue
+                    # Old excluded-tool output — fall through to compression
+                    # (the LLM is unlikely to need exact content from this far back,
+                    # and CCR provides retrieval if it does)
                 # 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
@@ -1257,11 +1341,13 @@ class ContentRouter(Transform):
             if self.config.skip_user_messages and role == "user":
                 transformed_messages.append(message)
                 transforms_applied.append("router:protected:user_message")
+                route_counts["user_msg"] += 1
                 continue
 
             if not content or len(content.split()) < 50:
                 # Skip small content
                 transformed_messages.append(message)
+                route_counts["small"] += 1
                 continue
 
             # Detect content type for protection decisions
@@ -1277,12 +1363,14 @@ class ContentRouter(Transform):
             ):
                 transformed_messages.append(message)
                 transforms_applied.append("router:protected:recent_code")
+                route_counts["recent_code"] += 1
                 continue
 
             # Protection 3: Don't compress CODE when analysis intent detected
             if analysis_intent and is_code:
                 transformed_messages.append(message)
                 transforms_applied.append("router:protected:analysis_context")
+                route_counts["analysis_ctx"] += 1
                 continue
 
             # Route and compress based on content detection
@@ -1292,18 +1380,49 @@ class ContentRouter(Transform):
                 msg_bias *= hook_biases[i]
             result = self.compress(content, context=context, bias=msg_bias)
 
-            if result.compression_ratio < 0.9:
+            if result.compression_ratio < min_ratio:
                 transformed_messages.append({**message, "content": result.compressed})
                 transforms_applied.append(
                     f"router:{result.strategy_used.value}:{result.compression_ratio:.2f}"
                 )
+                compressed_details.append(
+                    f"{result.strategy_used.value}:{result.compression_ratio:.2f}"
+                )
             else:
                 transformed_messages.append(message)
+                route_counts["ratio_too_high"] += 1
 
         tokens_after = sum(
             tokenizer.count_text(str(m.get("content", ""))) for m in transformed_messages
         )
 
+        # Log routing summary
+        parts = []
+        if compressed_details:
+            parts.append(f"{len(compressed_details)} compressed ({', '.join(compressed_details)})")
+        if route_counts["excluded_tool"]:
+            parts.append(f"{route_counts['excluded_tool']} excluded (Read/Glob)")
+        if route_counts["user_msg"]:
+            parts.append(f"{route_counts['user_msg']} skipped (user)")
+        if route_counts["small"]:
+            parts.append(f"{route_counts['small']} skipped (<50 words)")
+        if route_counts["recent_code"]:
+            parts.append(f"{route_counts['recent_code']} protected (recent code)")
+        if route_counts["analysis_ctx"]:
+            parts.append(f"{route_counts['analysis_ctx']} protected (analysis ctx)")
+        if route_counts["ratio_too_high"]:
+            parts.append(f"{route_counts['ratio_too_high']} unchanged (ratio>={min_ratio:.2f})")
+        if route_counts["content_blocks"]:
+            parts.append(f"{route_counts['content_blocks']} content-block msgs")
+        if route_counts["non_string"]:
+            parts.append(f"{route_counts['non_string']} non-string")
+        if parts:
+            logger.info(
+                "content_router: %d msgs — %s",
+                num_messages,
+                ", ".join(parts),
+            )
+
         all_transforms = lifecycle_transforms + transforms_applied
         return TransformResult(
             messages=transformed_messages,
@@ -1343,6 +1462,11 @@ class ContentRouter(Transform):
         transforms_applied: list[str],
         excluded_tool_ids: set[str],
         tool_name_map: dict[str, str] | None = None,
+        route_counts: dict[str, int] | None = None,
+        compressed_details: list[str] | None = None,
+        min_ratio: float = 0.85,
+        read_protection_window: int = 8,
+        messages_from_end: int = 0,
     ) -> dict[str, Any]:
         """Process content blocks (Anthropic format) for tool_result compression.
 
@@ -1356,6 +1480,11 @@ class ContentRouter(Transform):
             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.
+            route_counts: Optional routing reason counters to update.
+            compressed_details: Optional list to append compression details to.
+            min_ratio: Adaptive compression ratio threshold.
+            read_protection_window: Messages from end within which excluded tools are protected.
+            messages_from_end: How far this message is from the end of the conversation.
 
         Returns:
             Transformed message with compressed content blocks.
@@ -1375,9 +1504,14 @@ class ContentRouter(Transform):
                 # Check if tool is excluded from compression
                 tool_use_id = block.get("tool_use_id", "")
                 if tool_use_id in excluded_tool_ids:
-                    new_blocks.append(block)
-                    transforms_applied.append("router:excluded:tool")
-                    continue
+                    if messages_from_end <= read_protection_window:
+                        # Recent — protect as before
+                        new_blocks.append(block)
+                        transforms_applied.append("router:excluded:tool")
+                        if route_counts is not None:
+                            route_counts["excluded_tool"] += 1
+                        continue
+                    # Old excluded-tool output — fall through to compression
 
                 # Look up tool-specific compression bias
                 tool_name = (tool_name_map or {}).get(tool_use_id, "")
@@ -1389,13 +1523,23 @@ class ContentRouter(Transform):
                 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, bias=bias)
-                    if result.compression_ratio < 0.9:
+                    if result.compression_ratio < min_ratio:
                         new_blocks.append({**block, "content": result.compressed})
                         transforms_applied.append(
                             f"router:tool_result:{result.strategy_used.value}"
                         )
+                        if compressed_details is not None:
+                            compressed_details.append(
+                                f"tool:{result.strategy_used.value}:{result.compression_ratio:.2f}"
+                            )
                         any_compressed = True
                         continue
+                    else:
+                        if route_counts is not None:
+                            route_counts["ratio_too_high"] += 1
+                else:
+                    if route_counts is not None:
+                        route_counts["small"] += 1
 
             # Keep block unchanged
             new_blocks.append(block)

tests/test_memory/test_hierarchical.py

@@ -754,6 +754,7 @@ class TestLocalEmbedder:
     @pytest.fixture
     def embedder(self):
         """Create a local embedder for testing."""
+        pytest.importorskip("sentence_transformers", reason="sentence-transformers not installed")
         from headroom.memory.adapters.embedders import LocalEmbedder
 
         return LocalEmbedder()

tests/test_transforms/test_read_lifecycle.py

@@ -152,9 +152,9 @@ SMALL_CONTENT = "tiny"  # Below min_size_bytes
 class TestReadLifecycleDisabled:
     """Verify backward compatibility when disabled."""
 
-    def test_disabled_by_default(self):
-        """Default config: no changes to messages."""
-        config = ReadLifecycleConfig()
+    def test_disabled_when_explicitly_off(self):
+        """Explicitly disabled config: no changes to messages."""
+        config = ReadLifecycleConfig(enabled=False)
         assert config.enabled is False
 
         mgr = ReadLifecycleManager(config)
@@ -168,6 +168,11 @@ class TestReadLifecycleDisabled:
         assert result.reads_total == 0
         assert result.transforms_applied == []
 
+    def test_enabled_by_default(self):
+        """Default config has lifecycle enabled."""
+        config = ReadLifecycleConfig()
+        assert config.enabled is True
+
 
 class TestStaleDetection:
     """Read outputs become stale when the file is subsequently edited."""