Add universal compression module with ML-based content detection

- Add headroom.compression module with UniversalCompressor
- ML-based content detection using Magika (JSON, code, logs, text)
- Structure-preserving compression via handler protocol
- JSON handler: preserves keys, brackets, high-entropy values (UUIDs)
- Code handler: preserves imports, signatures, types (tree-sitter AST)
- Entropy-based preservation for identifiers and hashes
- CCR integration for reversible compression
- Comprehensive test suite with LLM eval tests
- Add docs/compression.md with full API documentation

README.md

@@ -106,6 +106,7 @@ See the full [LangChain Integration Guide](docs/langchain.md) for memory, retrie
 | Feature | Description | Docs |
 |---------|-------------|------|
 | **Memory** | Persistent memory across conversations (zero-latency inline extraction) | [Memory](docs/memory.md) |
+| **Universal Compression** | ML-based content detection + structure-preserving compression | [Compression](docs/compression.md) |
 | **SmartCrusher** | Compresses JSON tool outputs statistically | [Transforms](docs/transforms.md) |
 | **CacheAligner** | Stabilizes prefixes for provider caching | [Transforms](docs/transforms.md) |
 | **RollingWindow** | Manages context limits without breaking tools | [Transforms](docs/transforms.md) |
@@ -173,6 +174,7 @@ pip install "headroom-ai[all]"       # Everything
 | Guide | Description |
 |-------|-------------|
 | [Memory Guide](docs/memory.md) | Persistent memory for LLMs |
+| [Compression Guide](docs/compression.md) | Universal compression with ML detection |
 | [LangChain Integration](docs/langchain.md) | Full LangChain support |
 | [SDK Guide](docs/sdk.md) | Fine-grained control |
 | [Proxy Guide](docs/proxy.md) | Production deployment |

docs/README.md

@@ -21,6 +21,7 @@ Welcome to the Headroom documentation.
 
 | Topic | Description |
 |-------|-------------|
+| [Universal Compression](compression.md) | ML-based content detection + structure preservation |
 | [Transforms](transforms.md) | How compression works |
 | [CCR](ccr.md) | Reversible compression architecture |
 | [Configuration](configuration.md) | All configuration options |
@@ -48,10 +49,11 @@ Headroom is the Context Optimization Layer for LLM applications. It reduces your
 
 ### How It Works
 
-1. **SmartCrusher** — Compresses JSON tool outputs, keeping errors, anomalies, and relevant items
-2. **CacheAligner** — Stabilizes message prefixes so provider caching works
-3. **RollingWindow** — Manages context limits without breaking tool call pairs
-4. **CCR** — Caches original data so compression is reversible
+1. **Universal Compression** — ML-based content detection with structure-preserving compression
+2. **SmartCrusher** — Compresses JSON tool outputs, keeping errors, anomalies, and relevant items
+3. **CacheAligner** — Stabilizes message prefixes so provider caching works
+4. **RollingWindow** — Manages context limits without breaking tool call pairs
+5. **CCR** — Caches original data so compression is reversible
 
 ### Safety Guarantees
 

docs/compression.md

@@ -0,0 +1,407 @@
+# Universal Compression
+
+Headroom's Universal Compression module provides intelligent, automatic compression with ML-based content detection and structure preservation.
+
+## Overview
+
+Universal Compression combines several techniques:
+
+1. **ML-based Detection** - Automatically detects content type (JSON, code, logs, text) using Magika
+2. **Structure Preservation** - Keeps keys, signatures, and templates intact via structure masks
+3. **Intelligent Compression** - Compresses content while preserving meaning with LLMLingua
+4. **Reversible via CCR** - Stores originals for retrieval when LLM needs full context
+
+## Quick Start
+
+### One-Liner
+
+```python
+from headroom.compression import compress
+
+result = compress(content)
+print(result.compressed)
+print(f"Saved {result.savings_percentage:.0f}% tokens")
+```
+
+### With Configuration
+
+```python
+from headroom.compression import UniversalCompressor, UniversalCompressorConfig
+
+config = UniversalCompressorConfig(
+    compression_ratio_target=0.5,  # Keep 50% of content
+    use_entropy_preservation=True,  # Preserve UUIDs, hashes
+)
+
+compressor = UniversalCompressor(config=config)
+result = compressor.compress(content)
+```
+
+---
+
+## How It Works
+
+### Detection Flow
+
+```
+┌─────────────┐    ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
+│   Content   │───>│   Detect    │───>│   Extract   │───>│  Compress   │
+│   Input     │    │   Type      │    │   Structure │    │  Content    │
+└─────────────┘    └─────────────┘    └─────────────┘    └─────────────┘
+                         │                   │                   │
+                         ▼                   ▼                   ▼
+                   ┌─────────────┐    ┌─────────────┐    ┌─────────────┐
+                   │   Magika    │    │   Handler   │    │  LLMLingua  │
+                   │   (ML)      │    │   (JSON,    │    │  (optional) │
+                   │             │    │   Code...)  │    │             │
+                   └─────────────┘    └─────────────┘    └─────────────┘
+```
+
+### Structure Masks
+
+Structure masks identify what to preserve:
+
+| Content Type | What's Preserved | What's Compressed |
+|--------------|------------------|-------------------|
+| **JSON** | Keys, brackets, booleans, nulls, short values, UUIDs | Long string values, whitespace |
+| **Code** | Imports, function signatures, class definitions, types | Function bodies, comments |
+| **Logs** | Timestamps, log levels, error messages | Repeated patterns, verbose details |
+| **Text** | High-entropy tokens (IDs, hashes) | Low-information content |
+
+---
+
+## Configuration
+
+### UniversalCompressorConfig
+
+```python
+from headroom.compression import UniversalCompressorConfig
+
+config = UniversalCompressorConfig(
+    # Detection
+    use_magika=True,               # Use ML-based detection (requires magika)
+
+    # Compression
+    use_llmlingua=True,            # Use LLMLingua for compression
+    compression_ratio_target=0.3,  # Keep 30% of content (70% reduction)
+    min_content_length=100,        # Skip content shorter than this
+
+    # Structure preservation
+    use_entropy_preservation=True, # Preserve high-entropy tokens
+    entropy_threshold=0.85,        # Entropy threshold for preservation
+
+    # CCR
+    ccr_enabled=True,              # Store originals for retrieval
+)
+```
+
+### Configuration Options
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `use_magika` | `True` | Use ML-based content detection |
+| `use_llmlingua` | `True` | Use LLMLingua for compression |
+| `compression_ratio_target` | `0.3` | Target ratio (0.3 = keep 30%) |
+| `min_content_length` | `100` | Minimum chars to compress |
+| `use_entropy_preservation` | `True` | Preserve high-entropy tokens |
+| `entropy_threshold` | `0.85` | Entropy threshold (0.0-1.0) |
+| `ccr_enabled` | `True` | Enable CCR storage |
+
+---
+
+## Content Handlers
+
+### JSON Handler
+
+Preserves JSON structure while compressing values:
+
+```python
+from headroom.compression.handlers.json_handler import JSONStructureHandler
+
+handler = JSONStructureHandler(
+    preserve_short_values=True,     # Keep values < 20 chars
+    short_value_threshold=20,       # Threshold for "short"
+    preserve_high_entropy=True,     # Keep UUIDs, hashes
+    entropy_threshold=0.85,         # Entropy threshold
+    max_array_items_full=3,         # Keep first N array items full
+    max_number_digits=10,           # Preserve numbers up to N digits
+)
+```
+
+**What's Preserved:**
+- All keys (navigational - LLM sees schema)
+- Structural syntax (`{`, `}`, `[`, `]`, `:`, `,`)
+- Booleans and nulls (semantically important)
+- High-entropy strings (UUIDs, hashes - identifiers)
+- Short numbers (often IDs)
+
+**Example:**
+
+```python
+# Before
+{
+    "id": "usr_abc123",
+    "name": "Alice Johnson",
+    "bio": "A long description that goes on and on..."
+}
+
+# After (structure preserved, long values compressed)
+{
+    "id": "usr_abc123",
+    "name": "Alice Johnson",
+    "bio": "A long...[compressed]..."
+}
+```
+
+### Code Handler
+
+Preserves code structure using AST parsing (tree-sitter) or regex fallback:
+
+```python
+from headroom.compression.handlers.code_handler import CodeStructureHandler
+
+handler = CodeStructureHandler(
+    preserve_comments=False,        # Preserve comments as structural
+    use_tree_sitter=True,           # Use tree-sitter for parsing
+    default_language="python",      # Default when detection fails
+)
+```
+
+**What's Preserved:**
+- Import statements
+- Function/method signatures
+- Class definitions
+- Type annotations
+- Decorators
+
+**What's Compressed:**
+- Function bodies (implementations)
+- Comments (unless `preserve_comments=True`)
+
+**Example:**
+
+```python
+# Before
+def process_data(items: List[str]) -> Dict[str, int]:
+    """Process items and count occurrences."""
+    result = {}
+    for item in items:
+        item = item.strip().lower()
+        if item in result:
+            result[item] += 1
+        else:
+            result[item] = 1
+    return result
+
+# After (signature preserved, body compressed)
+def process_data(items: List[str]) -> Dict[str, int]:
+    """Process items and count occurrences."""
+    result = {}
+    for item in items:
+    ...[compressed]...
+```
+
+### Supported Languages
+
+| Language | Parser | Support Level |
+|----------|--------|---------------|
+| Python | tree-sitter | Full AST |
+| JavaScript | tree-sitter | Full AST |
+| TypeScript | tree-sitter | Full AST |
+| Go | tree-sitter | Full AST |
+| Rust | tree-sitter | Full AST |
+| Java | tree-sitter | Full AST |
+| C | tree-sitter | Full AST |
+| C++ | tree-sitter | Full AST |
+
+---
+
+## Compression Result
+
+```python
+from headroom.compression import compress
+
+result = compress(content)
+
+# Access result fields
+print(result.compressed)           # Compressed content
+print(result.original)             # Original content
+print(result.compression_ratio)    # e.g., 0.35 (35% of original size)
+print(result.tokens_before)        # Estimated tokens before
+print(result.tokens_after)         # Estimated tokens after
+print(result.tokens_saved)         # tokens_before - tokens_after
+print(result.savings_percentage)   # e.g., 65.0 (65% savings)
+
+# Detection info
+print(result.content_type)         # ContentType.JSON, CODE, etc.
+print(result.detection_confidence) # 0.0-1.0
+
+# Structure info
+print(result.handler_used)         # "json", "code", etc.
+print(result.preservation_ratio)   # Fraction preserved as structure
+
+# CCR info
+print(result.ccr_key)              # Key for retrieval (if CCR enabled)
+```
+
+---
+
+## Batch Compression
+
+For multiple contents, batch compression is more efficient:
+
+```python
+from headroom.compression import UniversalCompressor
+
+compressor = UniversalCompressor()
+
+contents = [
+    '{"users": [...]}',
+    'def hello(): pass',
+    'Plain text content',
+]
+
+results = compressor.compress_batch(contents)
+
+for result in results:
+    print(f"{result.content_type}: {result.savings_percentage:.0f}% saved")
+```
+
+---
+
+## Custom Handlers
+
+Register custom handlers for specific content types:
+
+```python
+from headroom.compression import UniversalCompressor
+from headroom.compression.detector import ContentType
+from headroom.compression.handlers.base import BaseStructureHandler, HandlerResult
+from headroom.compression.masks import StructureMask
+
+
+class LogStructureHandler(BaseStructureHandler):
+    """Custom handler for log content."""
+
+    def __init__(self):
+        super().__init__(name="log")
+
+    def can_handle(self, content: str) -> bool:
+        return "[INFO]" in content or "[ERROR]" in content
+
+    def _extract_mask(self, content, tokens, **kwargs):
+        # Mark timestamps and log levels as structural
+        mask = [False] * len(content)
+        # ... (custom logic)
+        return HandlerResult(
+            mask=StructureMask(tokens=tokens, mask=mask),
+            handler_name=self.name,
+            confidence=0.9,
+        )
+
+
+# Register the custom handler
+compressor = UniversalCompressor()
+compressor.register_handler(ContentType.TEXT, LogStructureHandler())
+```
+
+---
+
+## CCR Integration
+
+Universal Compression integrates with CCR (Compress-Cache-Retrieve) for reversible compression:
+
+```python
+from headroom.compression import UniversalCompressor, UniversalCompressorConfig
+
+config = UniversalCompressorConfig(ccr_enabled=True)
+compressor = UniversalCompressor(config=config)
+
+result = compressor.compress(large_content)
+
+# CCR key for retrieval
+if result.ccr_key:
+    print(f"Original stored with key: {result.ccr_key}")
+    # LLM can request original via CCR when needed
+```
+
+See [CCR Guide](ccr.md) for full CCR documentation.
+
+---
+
+## Performance
+
+| Content Type | Compression | Speed | Accuracy |
+|--------------|-------------|-------|----------|
+| JSON (large arrays) | 70-90% | ~1ms | Keys preserved |
+| Code (Python) | 50-70% | ~10ms | Signatures preserved |
+| Plain text | 60-80% | ~5ms | High-entropy preserved |
+
+**Overhead:** ~1-10ms per compression depending on content size and type.
+
+---
+
+## Installation
+
+```bash
+# Basic compression (fallback to simple compression)
+pip install headroom-ai
+
+# With ML detection (recommended)
+pip install "headroom-ai[magika]"
+
+# With LLMLingua compression
+pip install "headroom-ai[llmlingua]"
+
+# With AST-based code handling
+pip install "headroom-ai[code]"
+
+# Everything
+pip install "headroom-ai[all]"
+```
+
+---
+
+## Example: Full Pipeline
+
+```python
+from headroom.compression import UniversalCompressor, UniversalCompressorConfig
+
+# Configure for aggressive compression
+config = UniversalCompressorConfig(
+    compression_ratio_target=0.25,  # Keep 25%
+    use_magika=True,
+    use_llmlingua=True,
+    ccr_enabled=True,
+)
+
+compressor = UniversalCompressor(config=config)
+
+# Compress JSON API response
+json_content = """
+{
+    "users": [
+        {"id": "usr_123", "name": "Alice", "bio": "Software engineer..."},
+        {"id": "usr_456", "name": "Bob", "bio": "Product manager..."}
+    ],
+    "total": 2,
+    "page": 1
+}
+"""
+
+result = compressor.compress(json_content)
+
+print(f"Type: {result.content_type}")          # ContentType.JSON
+print(f"Handler: {result.handler_used}")        # json
+print(f"Saved: {result.savings_percentage:.0f}%")  # ~60%
+print(f"Structure: {result.preservation_ratio:.0%} preserved")  # ~40%
+print(f"CCR Key: {result.ccr_key}")             # For retrieval
+```
+
+---
+
+## See Also
+
+- [Transforms Reference](transforms.md) - Other compression transforms
+- [CCR Guide](ccr.md) - Reversible compression architecture
+- [Text Compression](text-compression.md) - Opt-in utilities for search/logs

headroom/compression/__init__.py

@@ -0,0 +1,42 @@
+"""Universal compression with ML-based content detection.
+
+This module provides intelligent, automatic compression that:
+1. Detects content type using ML (Magika)
+2. Preserves structure (keys, signatures, templates)
+3. Compresses content with LLMLingua
+4. Enables retrieval via CCR
+
+Quick Start:
+    # One-liner for simple use
+    from headroom.compression import compress
+    result = compress(content)
+
+    # Or with configuration
+    from headroom.compression import UniversalCompressor, UniversalCompressorConfig
+
+    config = UniversalCompressorConfig(compression_ratio_target=0.5)
+    compressor = UniversalCompressor(config=config)
+    result = compressor.compress(content)
+"""
+
+from headroom.compression.detector import ContentType, MagikaDetector
+from headroom.compression.masks import StructureMask
+from headroom.compression.universal import (
+    CompressionResult,
+    UniversalCompressor,
+    UniversalCompressorConfig,
+    compress,
+)
+
+__all__ = [
+    # Simple API
+    "compress",
+    # Full API
+    "UniversalCompressor",
+    "UniversalCompressorConfig",
+    "CompressionResult",
+    # Advanced
+    "MagikaDetector",
+    "ContentType",
+    "StructureMask",
+]

headroom/compression/detector.py

@@ -0,0 +1,424 @@
+"""ML-based content type detection using Google's Magika.
+
+Magika is a deep learning model for content type detection that:
+- Runs locally (~5ms latency)
+- Supports 100+ content types
+- Has 99%+ accuracy on supported types
+- Requires no configuration
+
+This replaces rule-based detection with learned detection.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from enum import Enum
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from magika import Magika
+    from magika.types import MagikaResult
+
+logger = logging.getLogger(__name__)
+
+# Lazy-loaded Magika instance (singleton)
+_magika_instance: Magika | None = None
+
+
+class ContentType(Enum):
+    """High-level content categories for compression routing."""
+
+    JSON = "json"
+    CODE = "code"
+    LOG = "log"
+    MARKDOWN = "markdown"
+    TEXT = "text"
+    UNKNOWN = "unknown"
+
+
+@dataclass
+class DetectionResult:
+    """Result of ML-based content detection."""
+
+    content_type: ContentType
+    confidence: float  # 0.0 to 1.0
+    raw_label: str  # Original Magika label
+    language: str | None = None  # For code: python, javascript, etc.
+    metadata: dict = field(default_factory=dict)
+
+
+# Map Magika labels to our content types
+# This is the ONLY place where we map labels - no hardcoding elsewhere
+_CODE_LABELS = frozenset(
+    {
+        "python",
+        "javascript",
+        "typescript",
+        "go",
+        "rust",
+        "java",
+        "c",
+        "cpp",
+        "csharp",
+        "ruby",
+        "php",
+        "swift",
+        "kotlin",
+        "scala",
+        "shell",
+        "bash",
+        "powershell",
+        "sql",
+        "r",
+        "perl",
+        "lua",
+        "haskell",
+        "elixir",
+        "erlang",
+        "clojure",
+        "ocaml",
+        "fsharp",
+        "dart",
+        "julia",
+        "zig",
+        "nim",
+        "crystal",
+        "v",
+        "solidity",
+        "move",
+        "cairo",
+        "vyper",
+    }
+)
+
+_STRUCTURED_LABELS = frozenset(
+    {
+        "json",
+        "jsonl",
+        "yaml",
+        "toml",
+        "xml",
+        "html",
+        "csv",
+        "tsv",
+        "ini",
+        "properties",
+    }
+)
+
+_LOG_LABELS = frozenset(
+    {
+        "log",
+        "syslog",
+    }
+)
+
+_MARKDOWN_LABELS = frozenset(
+    {
+        "markdown",
+        "rst",
+        "asciidoc",
+        "org",
+    }
+)
+
+
+def _get_magika() -> Magika:
+    """Get or create the singleton Magika instance.
+
+    Lazy-loads on first use to avoid import cost if not needed.
+    """
+    global _magika_instance
+    if _magika_instance is None:
+        try:
+            from magika import Magika
+
+            _magika_instance = Magika()
+            logger.debug("Magika model loaded successfully")
+        except ImportError as e:
+            raise ImportError(
+                "Magika is required for ML-based content detection. "
+                "Install with: pip install magika"
+            ) from e
+    return _magika_instance
+
+
+def _magika_available() -> bool:
+    """Check if Magika is available without loading it."""
+    try:
+        import magika  # noqa: F401
+
+        return True
+    except ImportError:
+        return False
+
+
+class MagikaDetector:
+    """ML-based content type detector using Google's Magika.
+
+    This detector uses a deep learning model to identify content types
+    without relying on file extensions or brittle regex patterns.
+
+    Example:
+        detector = MagikaDetector()
+        result = detector.detect('def hello(): print("hi")')
+        # result.content_type == ContentType.CODE
+        # result.language == "python"
+    """
+
+    def __init__(self, min_confidence: float = 0.5):
+        """Initialize the detector.
+
+        Args:
+            min_confidence: Minimum confidence threshold. Below this,
+                returns ContentType.UNKNOWN.
+        """
+        self.min_confidence = min_confidence
+        self._magika: Magika | None = None
+
+    def _ensure_magika(self) -> Magika:
+        """Ensure Magika is loaded."""
+        if self._magika is None:
+            self._magika = _get_magika()
+        return self._magika
+
+    def detect(self, content: str) -> DetectionResult:
+        """Detect content type using ML.
+
+        Args:
+            content: The content to analyze.
+
+        Returns:
+            DetectionResult with type, confidence, and metadata.
+
+        Example:
+            >>> detector = MagikaDetector()
+            >>> result = detector.detect('{"users": [{"id": 1}]}')
+            >>> result.content_type
+            ContentType.JSON
+        """
+        if not content or not content.strip():
+            return DetectionResult(
+                content_type=ContentType.UNKNOWN,
+                confidence=0.0,
+                raw_label="empty",
+            )
+
+        # Get Magika prediction
+        magika = self._ensure_magika()
+        result: MagikaResult = magika.identify_bytes(content.encode("utf-8"))
+
+        raw_label = result.output.ct_label
+        confidence = result.output.score
+
+        # Map to our content type
+        content_type, language = self._map_label(raw_label)
+
+        # Apply confidence threshold
+        if confidence < self.min_confidence:
+            content_type = ContentType.UNKNOWN
+
+        return DetectionResult(
+            content_type=content_type,
+            confidence=confidence,
+            raw_label=raw_label,
+            language=language,
+            metadata={
+                "magika_group": result.output.group,
+                "magika_mime": result.output.mime_type,
+            },
+        )
+
+    def detect_batch(self, contents: list[str]) -> list[DetectionResult]:
+        """Detect content types for multiple contents.
+
+        More efficient than calling detect() in a loop.
+
+        Args:
+            contents: List of content strings to analyze.
+
+        Returns:
+            List of DetectionResults in same order as input.
+        """
+        if not contents:
+            return []
+
+        magika = self._ensure_magika()
+        results = []
+
+        # Convert to bytes for Magika
+        byte_contents = [c.encode("utf-8") for c in contents]
+
+        # Batch detection
+        magika_results = magika.identify_bytes_batch(byte_contents)
+
+        for content, magika_result in zip(contents, magika_results):
+            if not content or not content.strip():
+                results.append(
+                    DetectionResult(
+                        content_type=ContentType.UNKNOWN,
+                        confidence=0.0,
+                        raw_label="empty",
+                    )
+                )
+                continue
+
+            raw_label = magika_result.output.ct_label
+            confidence = magika_result.output.score
+            content_type, language = self._map_label(raw_label)
+
+            if confidence < self.min_confidence:
+                content_type = ContentType.UNKNOWN
+
+            results.append(
+                DetectionResult(
+                    content_type=content_type,
+                    confidence=confidence,
+                    raw_label=raw_label,
+                    language=language,
+                    metadata={
+                        "magika_group": magika_result.output.group,
+                        "magika_mime": magika_result.output.mime_type,
+                    },
+                )
+            )
+
+        return results
+
+    def _map_label(self, label: str) -> tuple[ContentType, str | None]:
+        """Map Magika label to our ContentType.
+
+        Args:
+            label: Raw Magika label (e.g., "python", "json").
+
+        Returns:
+            Tuple of (ContentType, optional language).
+        """
+        label_lower = label.lower()
+
+        # Check code languages
+        if label_lower in _CODE_LABELS:
+            return ContentType.CODE, label_lower
+
+        # Check structured data
+        if label_lower in _STRUCTURED_LABELS:
+            # JSON gets its own type for specialized handling
+            if label_lower in ("json", "jsonl"):
+                return ContentType.JSON, None
+            # Other structured data treated as JSON-like
+            return ContentType.JSON, None
+
+        # Check logs
+        if label_lower in _LOG_LABELS:
+            return ContentType.LOG, None
+
+        # Check markdown/docs
+        if label_lower in _MARKDOWN_LABELS:
+            return ContentType.MARKDOWN, None
+
+        # Text types
+        if label_lower in ("txt", "text", "ascii", "utf8", "empty"):
+            return ContentType.TEXT, None
+
+        # Default: treat as text
+        return ContentType.TEXT, None
+
+    @staticmethod
+    def is_available() -> bool:
+        """Check if Magika is available."""
+        return _magika_available()
+
+
+class FallbackDetector:
+    """Simple fallback detector when Magika is not available.
+
+    Uses basic heuristics - not as accurate but requires no dependencies.
+    """
+
+    def __init__(self, min_confidence: float = 0.5):
+        """Initialize the fallback detector."""
+        self.min_confidence = min_confidence
+
+    def detect(self, content: str) -> DetectionResult:
+        """Detect content type using simple heuristics.
+
+        Args:
+            content: The content to analyze.
+
+        Returns:
+            DetectionResult with type and confidence.
+        """
+        if not content or not content.strip():
+            return DetectionResult(
+                content_type=ContentType.UNKNOWN,
+                confidence=0.0,
+                raw_label="empty",
+            )
+
+        stripped = content.strip()
+
+        # JSON detection (simple but effective)
+        if stripped.startswith(("{", "[")):
+            try:
+                import json
+
+                json.loads(stripped)
+                return DetectionResult(
+                    content_type=ContentType.JSON,
+                    confidence=1.0,
+                    raw_label="json",
+                )
+            except (json.JSONDecodeError, ValueError):
+                pass
+
+        # Code detection (look for common patterns)
+        code_indicators = [
+            "def ",
+            "class ",
+            "function ",
+            "import ",
+            "const ",
+            "let ",
+            "var ",
+            "func ",
+            "fn ",
+            "pub ",
+            "package ",
+        ]
+        if any(indicator in content for indicator in code_indicators):
+            return DetectionResult(
+                content_type=ContentType.CODE,
+                confidence=0.7,
+                raw_label="code",
+            )
+
+        # Log detection
+        log_indicators = ["ERROR", "WARN", "INFO", "DEBUG", "FATAL"]
+        if any(indicator in content for indicator in log_indicators):
+            return DetectionResult(
+                content_type=ContentType.LOG,
+                confidence=0.6,
+                raw_label="log",
+            )
+
+        # Default to text
+        return DetectionResult(
+            content_type=ContentType.TEXT,
+            confidence=0.5,
+            raw_label="text",
+        )
+
+
+def get_detector(prefer_magika: bool = True) -> MagikaDetector | FallbackDetector:
+    """Get the best available detector.
+
+    Args:
+        prefer_magika: If True, use Magika if available.
+
+    Returns:
+        MagikaDetector if available and preferred, else FallbackDetector.
+    """
+    if prefer_magika and MagikaDetector.is_available():
+        return MagikaDetector()
+    return FallbackDetector()

headroom/compression/handlers/__init__.py

@@ -0,0 +1,22 @@
+"""Structure handlers for different content types.
+
+Each handler knows how to extract structural information from a specific
+content type and create a StructureMask marking what should be preserved.
+
+Handlers don't compress - they only identify structure. The actual
+compression is done by LLMLingua on the non-structural parts.
+"""
+
+from headroom.compression.handlers.base import (
+    HandlerResult,
+    StructureHandler,
+)
+from headroom.compression.handlers.code_handler import CodeStructureHandler
+from headroom.compression.handlers.json_handler import JSONStructureHandler
+
+__all__ = [
+    "StructureHandler",
+    "HandlerResult",
+    "JSONStructureHandler",
+    "CodeStructureHandler",
+]

headroom/compression/handlers/base.py

@@ -0,0 +1,219 @@
+"""Base class and protocol for structure handlers.
+
+Structure handlers extract structural information from content and create
+masks identifying what should be preserved during compression.
+
+The handler protocol is simple:
+1. get_mask(content) -> StructureMask
+2. can_handle(content) -> bool (optional)
+
+Handlers are content-type specific but domain-agnostic. A JSONStructureHandler
+preserves JSON keys whether it's user data, search results, or config files.
+"""
+
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from dataclasses import dataclass, field
+from typing import Any, Protocol, runtime_checkable
+
+from headroom.compression.masks import StructureMask
+
+
+@dataclass
+class HandlerResult:
+    """Result from a structure handler.
+
+    Contains the mask plus metadata about what was detected.
+    """
+
+    mask: StructureMask
+    handler_name: str
+    confidence: float = 1.0  # How confident the handler is in its detection
+    metadata: dict = field(default_factory=dict)
+
+    @property
+    def preservation_ratio(self) -> float:
+        """Fraction of content marked for preservation."""
+        return self.mask.preservation_ratio
+
+
+@runtime_checkable
+class StructureHandler(Protocol):
+    """Protocol for structure handlers.
+
+    Any class implementing get_mask() can be used as a handler.
+    """
+
+    @property
+    def name(self) -> str:
+        """Handler name for logging and metadata."""
+        ...
+
+    def get_mask(
+        self,
+        content: str,
+        tokens: list[str] | None = None,
+        **kwargs: Any,
+    ) -> HandlerResult:
+        """Extract structure mask from content.
+
+        Args:
+            content: The content to analyze.
+            tokens: Pre-tokenized content (optional). If not provided,
+                handler should tokenize internally.
+            **kwargs: Handler-specific options.
+
+        Returns:
+            HandlerResult with mask and metadata.
+        """
+        ...
+
+    def can_handle(self, content: str) -> bool:
+        """Check if this handler can process the content.
+
+        Default implementation returns True. Override for handlers
+        that need to verify content format before processing.
+
+        Args:
+            content: The content to check.
+
+        Returns:
+            True if handler can process this content.
+        """
+        ...
+
+
+class BaseStructureHandler(ABC):
+    """Base implementation for structure handlers.
+
+    Provides common functionality and enforces the handler interface.
+    Subclasses must implement _extract_mask().
+    """
+
+    def __init__(self, name: str | None = None):
+        """Initialize the handler.
+
+        Args:
+            name: Optional handler name. Defaults to class name.
+        """
+        self._name = name or self.__class__.__name__
+
+    @property
+    def name(self) -> str:
+        """Handler name."""
+        return self._name
+
+    def get_mask(
+        self,
+        content: str,
+        tokens: list[str] | None = None,
+        **kwargs: Any,
+    ) -> HandlerResult:
+        """Extract structure mask from content.
+
+        This is the main entry point. It handles common logic like
+        empty content and delegates to _extract_mask() for the
+        content-specific logic.
+
+        Args:
+            content: The content to analyze.
+            tokens: Pre-tokenized content (optional).
+            **kwargs: Handler-specific options.
+
+        Returns:
+            HandlerResult with mask and metadata.
+        """
+        # Handle empty content
+        if not content or not content.strip():
+            tokens = tokens or []
+            return HandlerResult(
+                mask=StructureMask.empty(tokens),
+                handler_name=self.name,
+                confidence=0.0,
+                metadata={"empty": True},
+            )
+
+        # Tokenize if not provided
+        if tokens is None:
+            tokens = self._tokenize(content)
+
+        # Delegate to subclass
+        return self._extract_mask(content, tokens, **kwargs)
+
+    def can_handle(self, content: str) -> bool:
+        """Check if this handler can process the content.
+
+        Default implementation returns True. Override for handlers
+        that need to verify content format.
+
+        Args:
+            content: The content to check.
+
+        Returns:
+            True if handler can process this content.
+        """
+        return True
+
+    @abstractmethod
+    def _extract_mask(
+        self,
+        content: str,
+        tokens: list[str],
+        **kwargs: Any,
+    ) -> HandlerResult:
+        """Extract structure mask from content.
+
+        Subclasses implement this to provide content-specific logic.
+
+        Args:
+            content: The content to analyze (non-empty, stripped).
+            tokens: Tokenized content.
+            **kwargs: Handler-specific options.
+
+        Returns:
+            HandlerResult with mask and metadata.
+        """
+        ...
+
+    def _tokenize(self, content: str) -> list[str]:
+        """Default tokenization - character-level.
+
+        Subclasses may override for more sophisticated tokenization.
+        For mask purposes, character-level is often sufficient and
+        aligns well with LLMLingua's token-level compression.
+
+        Args:
+            content: Content to tokenize.
+
+        Returns:
+            List of tokens (characters by default).
+        """
+        # Simple character-level tokenization
+        # This aligns well with structure detection (we mark ranges)
+        return list(content)
+
+
+class NoOpHandler(BaseStructureHandler):
+    """Handler that marks everything as compressible.
+
+    Used as a fallback when no structure is detected.
+    """
+
+    def __init__(self) -> None:
+        """Initialize the no-op handler."""
+        super().__init__(name="noop")
+
+    def _extract_mask(
+        self,
+        content: str,
+        tokens: list[str],
+        **kwargs: Any,
+    ) -> HandlerResult:
+        """Return mask with everything compressible."""
+        return HandlerResult(
+            mask=StructureMask.empty(tokens),
+            handler_name=self.name,
+            confidence=1.0,
+            metadata={"reason": "no structure detected"},
+        )

headroom/compression/handlers/code_handler.py

@@ -0,0 +1,506 @@
+"""Code structure handler using AST parsing.
+
+Extracts structural elements from source code:
+- Import statements
+- Function/method signatures
+- Class definitions
+- Type annotations
+- Decorators
+
+Function bodies are marked as compressible while preserving signatures.
+This enables the LLM to see all available functions/methods while body
+implementations are compressed.
+
+Uses tree-sitter for parsing when available, falls back to regex patterns.
+"""
+
+from __future__ import annotations
+
+import logging
+import re
+import threading
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from headroom.compression.handlers.base import BaseStructureHandler, HandlerResult
+from headroom.compression.masks import StructureMask
+
+logger = logging.getLogger(__name__)
+
+# Lazy-loaded tree-sitter
+_tree_sitter_available: bool | None = None
+_tree_sitter_parsers: dict[str, Any] = {}
+_tree_sitter_lock = threading.Lock()
+
+
+def _check_tree_sitter() -> bool:
+    """Check if tree-sitter is available."""
+    global _tree_sitter_available
+    if _tree_sitter_available is None:
+        try:
+            import tree_sitter_language_pack  # noqa: F401
+
+            _tree_sitter_available = True
+        except ImportError:
+            _tree_sitter_available = False
+    return _tree_sitter_available
+
+
+def _get_parser(language: str) -> Any:
+    """Get tree-sitter parser for language."""
+    global _tree_sitter_parsers
+
+    if not _check_tree_sitter():
+        raise ImportError("tree-sitter-language-pack not installed")
+
+    with _tree_sitter_lock:
+        if language not in _tree_sitter_parsers:
+            from tree_sitter_language_pack import get_parser
+
+            _tree_sitter_parsers[language] = get_parser(language)
+
+        return _tree_sitter_parsers[language]
+
+
+class CodeLanguage(Enum):
+    """Supported programming languages."""
+
+    PYTHON = "python"
+    JAVASCRIPT = "javascript"
+    TYPESCRIPT = "typescript"
+    GO = "go"
+    RUST = "rust"
+    JAVA = "java"
+    C = "c"
+    CPP = "cpp"
+
+
+@dataclass
+class CodeSpan:
+    """A span of code with its structural role."""
+
+    start: int
+    end: int
+    role: str  # "import", "signature", "body", "decorator", etc.
+    is_structural: bool
+
+
+# Language-specific AST node types that are structural
+_STRUCTURAL_NODE_TYPES: dict[str, set[str]] = {
+    "python": {
+        "import_statement",
+        "import_from_statement",
+        "function_definition",  # Just the signature part
+        "class_definition",
+        "decorated_definition",
+        "type_alias_statement",
+    },
+    "javascript": {
+        "import_statement",
+        "export_statement",
+        "function_declaration",
+        "class_declaration",
+        "method_definition",
+        "arrow_function",  # Signature only
+    },
+    "typescript": {
+        "import_statement",
+        "export_statement",
+        "function_declaration",
+        "class_declaration",
+        "method_definition",
+        "interface_declaration",
+        "type_alias_declaration",
+    },
+    "go": {
+        "import_declaration",
+        "function_declaration",
+        "method_declaration",
+        "type_declaration",
+        "interface_type",
+    },
+    "rust": {
+        "use_declaration",
+        "function_item",
+        "impl_item",
+        "struct_item",
+        "enum_item",
+        "trait_item",
+    },
+    "java": {
+        "import_declaration",
+        "class_declaration",
+        "method_declaration",
+        "interface_declaration",
+        "annotation",
+    },
+}
+
+# Regex patterns for fallback detection
+_SIGNATURE_PATTERNS: dict[str, list[re.Pattern[str]]] = {
+    "python": [
+        re.compile(r"^\s*(async\s+)?def\s+\w+\s*\([^)]*\)\s*(->\s*[^:]+)?:", re.MULTILINE),
+        re.compile(r"^\s*class\s+\w+(\([^)]*\))?:", re.MULTILINE),
+        re.compile(r"^\s*@\w+(\([^)]*\))?\s*$", re.MULTILINE),
+    ],
+    "javascript": [
+        re.compile(r"^\s*(async\s+)?function\s+\w+\s*\([^)]*\)", re.MULTILINE),
+        re.compile(r"^\s*class\s+\w+(\s+extends\s+\w+)?", re.MULTILINE),
+        re.compile(r"^\s*(const|let|var)\s+\w+\s*=\s*(async\s+)?\([^)]*\)\s*=>", re.MULTILINE),
+    ],
+    "typescript": [
+        re.compile(r"^\s*(async\s+)?function\s+\w+\s*(<[^>]+>)?\s*\([^)]*\)", re.MULTILINE),
+        re.compile(r"^\s*class\s+\w+(<[^>]+>)?(\s+extends\s+\w+)?", re.MULTILINE),
+        re.compile(r"^\s*interface\s+\w+(<[^>]+>)?", re.MULTILINE),
+        re.compile(r"^\s*type\s+\w+(<[^>]+>)?\s*=", re.MULTILINE),
+    ],
+    "go": [
+        re.compile(r"^\s*func\s+(\([^)]+\)\s+)?\w+\s*\([^)]*\)", re.MULTILINE),
+        re.compile(r"^\s*type\s+\w+\s+(struct|interface)", re.MULTILINE),
+    ],
+    "rust": [
+        re.compile(r"^\s*(pub\s+)?(async\s+)?fn\s+\w+\s*(<[^>]+>)?\s*\([^)]*\)", re.MULTILINE),
+        re.compile(r"^\s*(pub\s+)?struct\s+\w+", re.MULTILINE),
+        re.compile(r"^\s*(pub\s+)?enum\s+\w+", re.MULTILINE),
+        re.compile(r"^\s*(pub\s+)?trait\s+\w+", re.MULTILINE),
+        re.compile(r"^\s*impl(<[^>]+>)?\s+\w+", re.MULTILINE),
+    ],
+    "java": [
+        re.compile(
+            r"^\s*(public|private|protected)?\s*(static\s+)?\w+\s+\w+\s*\([^)]*\)", re.MULTILINE
+        ),
+        re.compile(r"^\s*(public\s+)?(class|interface|enum)\s+\w+", re.MULTILINE),
+        re.compile(r"^\s*@\w+(\([^)]*\))?\s*$", re.MULTILINE),
+    ],
+}
+
+# Import patterns for fallback
+_IMPORT_PATTERNS: dict[str, re.Pattern[str]] = {
+    "python": re.compile(r"^\s*(import\s+\w+|from\s+\w+\s+import)", re.MULTILINE),
+    "javascript": re.compile(r"^\s*(import\s+.*from|require\s*\()", re.MULTILINE),
+    "typescript": re.compile(r"^\s*(import\s+.*from|require\s*\()", re.MULTILINE),
+    "go": re.compile(r'^\s*import\s+(\(|")', re.MULTILINE),
+    "rust": re.compile(r"^\s*use\s+\w+", re.MULTILINE),
+    "java": re.compile(r"^\s*import\s+[\w.]+;", re.MULTILINE),
+}
+
+
+class CodeStructureHandler(BaseStructureHandler):
+    """Handler for source code.
+
+    Preserves:
+    - Import/use statements
+    - Function/method signatures (not bodies)
+    - Class/struct/interface definitions
+    - Type declarations
+    - Decorators/annotations
+
+    Marks as compressible:
+    - Function/method bodies
+    - Comments (optionally preserved)
+    - Whitespace
+
+    Example:
+        >>> handler = CodeStructureHandler()
+        >>> code = '''
+        ... def hello(name: str) -> str:
+        ...     message = f"Hello, {name}!"
+        ...     return message
+        ... '''
+        >>> result = handler.get_mask(code, language="python")
+        >>> # Signature "def hello(name: str) -> str:" preserved
+        >>> # Body content compressed
+    """
+
+    def __init__(
+        self,
+        preserve_comments: bool = False,
+        use_tree_sitter: bool = True,
+        default_language: str = "python",
+    ):
+        """Initialize the code handler.
+
+        Args:
+            preserve_comments: Whether to preserve comments as structural.
+            use_tree_sitter: Whether to use tree-sitter for parsing.
+                Falls back to regex if False or unavailable.
+            default_language: Default language when detection fails.
+        """
+        super().__init__(name="code")
+        self.preserve_comments = preserve_comments
+        self.use_tree_sitter = use_tree_sitter
+        self.default_language = default_language
+
+    def can_handle(self, content: str) -> bool:
+        """Check if content looks like source code."""
+        # Quick heuristic checks
+        code_indicators = [
+            "def ",
+            "class ",
+            "function ",
+            "import ",
+            "const ",
+            "let ",
+            "var ",
+            "func ",
+            "fn ",
+            "pub ",
+            "package ",
+            "struct ",
+            "interface ",
+        ]
+        return any(indicator in content for indicator in code_indicators)
+
+    def _extract_mask(
+        self,
+        content: str,
+        tokens: list[str],
+        language: str | None = None,
+        **kwargs: Any,
+    ) -> HandlerResult:
+        """Extract structure mask from code.
+
+        Args:
+            content: Source code content.
+            tokens: Character-level tokens.
+            language: Programming language (auto-detected if None).
+            **kwargs: Additional options.
+
+        Returns:
+            HandlerResult with mask marking structural elements.
+        """
+        # Detect language if not provided
+        if language is None:
+            language = self._detect_language(content)
+
+        # Try tree-sitter first
+        if self.use_tree_sitter and _check_tree_sitter():
+            try:
+                return self._extract_with_tree_sitter(content, tokens, language)
+            except Exception as e:
+                logger.debug("Tree-sitter parsing failed, using fallback: %s", e)
+
+        # Fallback to regex
+        return self._extract_with_regex(content, tokens, language)
+
+    def _extract_with_tree_sitter(
+        self,
+        content: str,
+        tokens: list[str],
+        language: str,
+    ) -> HandlerResult:
+        """Extract structure using tree-sitter AST.
+
+        Args:
+            content: Source code.
+            tokens: Character tokens.
+            language: Language name.
+
+        Returns:
+            HandlerResult with mask.
+        """
+        parser = _get_parser(language)
+        tree = parser.parse(content.encode("utf-8"))
+
+        # Collect structural spans
+        spans: list[CodeSpan] = []
+
+        def visit_node(node: Any, depth: int = 0) -> None:
+            """Visit AST node and collect structural spans."""
+            node_type = node.type
+            structural_types = _STRUCTURAL_NODE_TYPES.get(language, set())
+
+            # Check if this is a structural node type
+            if node_type in structural_types:
+                # For functions, only the signature is structural
+                if "function" in node_type or "method" in node_type:
+                    # Find the body node and exclude it
+                    body_node = None
+                    for child in node.children:
+                        if child.type in ("block", "statement_block", "compound_statement"):
+                            body_node = child
+                            break
+
+                    if body_node:
+                        # Signature is from start to body start
+                        spans.append(
+                            CodeSpan(
+                                start=node.start_byte,
+                                end=body_node.start_byte,
+                                role="signature",
+                                is_structural=True,
+                            )
+                        )
+                        # Body is compressible
+                        spans.append(
+                            CodeSpan(
+                                start=body_node.start_byte,
+                                end=body_node.end_byte,
+                                role="body",
+                                is_structural=False,
+                            )
+                        )
+                    else:
+                        # No body found, preserve whole thing
+                        spans.append(
+                            CodeSpan(
+                                start=node.start_byte,
+                                end=node.end_byte,
+                                role=node_type,
+                                is_structural=True,
+                            )
+                        )
+                else:
+                    # Non-function structural nodes
+                    spans.append(
+                        CodeSpan(
+                            start=node.start_byte,
+                            end=node.end_byte,
+                            role=node_type,
+                            is_structural=True,
+                        )
+                    )
+            elif node_type == "comment" and self.preserve_comments:
+                spans.append(
+                    CodeSpan(
+                        start=node.start_byte,
+                        end=node.end_byte,
+                        role="comment",
+                        is_structural=True,
+                    )
+                )
+
+            # Recurse into children
+            for child in node.children:
+                visit_node(child, depth + 1)
+
+        visit_node(tree.root_node)
+
+        # Build mask from spans
+        mask = self._spans_to_mask(spans, len(content))
+
+        return HandlerResult(
+            mask=StructureMask(tokens=tokens, mask=mask),
+            handler_name=self.name,
+            confidence=0.95,
+            metadata={
+                "language": language,
+                "parser": "tree-sitter",
+                "structural_spans": len([s for s in spans if s.is_structural]),
+            },
+        )
+
+    def _extract_with_regex(
+        self,
+        content: str,
+        tokens: list[str],
+        language: str,
+    ) -> HandlerResult:
+        """Extract structure using regex patterns (fallback).
+
+        Args:
+            content: Source code.
+            tokens: Character tokens.
+            language: Language name.
+
+        Returns:
+            HandlerResult with mask.
+        """
+        spans: list[CodeSpan] = []
+
+        # Match imports
+        import_pattern = _IMPORT_PATTERNS.get(language)
+        if import_pattern:
+            for match in import_pattern.finditer(content):
+                # Find end of import line
+                end = content.find("\n", match.end())
+                if end == -1:
+                    end = len(content)
+                spans.append(
+                    CodeSpan(
+                        start=match.start(),
+                        end=end,
+                        role="import",
+                        is_structural=True,
+                    )
+                )
+
+        # Match signatures
+        signature_patterns = _SIGNATURE_PATTERNS.get(language, [])
+        for pattern in signature_patterns:
+            for match in pattern.finditer(content):
+                spans.append(
+                    CodeSpan(
+                        start=match.start(),
+                        end=match.end(),
+                        role="signature",
+                        is_structural=True,
+                    )
+                )
+
+        # Build mask from spans
+        mask = self._spans_to_mask(spans, len(content))
+
+        return HandlerResult(
+            mask=StructureMask(tokens=tokens, mask=mask),
+            handler_name=self.name,
+            confidence=0.7,  # Lower confidence for regex
+            metadata={
+                "language": language,
+                "parser": "regex",
+                "structural_spans": len(spans),
+            },
+        )
+
+    def _spans_to_mask(self, spans: list[CodeSpan], length: int) -> list[bool]:
+        """Convert spans to character-level mask.
+
+        Args:
+            spans: List of code spans.
+            length: Total content length.
+
+        Returns:
+            Boolean mask aligned to characters.
+        """
+        mask = [False] * length
+
+        for span in spans:
+            if span.is_structural:
+                for i in range(span.start, min(span.end, length)):
+                    mask[i] = True
+
+        return mask
+
+    def _detect_language(self, content: str) -> str:
+        """Detect programming language from content.
+
+        Args:
+            content: Source code content.
+
+        Returns:
+            Language name (lowercase).
+        """
+        # Check for language-specific markers
+        markers = {
+            "python": ["def ", "import ", "from ", "class ", "async def"],
+            "javascript": ["function ", "const ", "let ", "var ", "=>"],
+            "typescript": ["interface ", "type ", ": string", ": number"],
+            "go": ["func ", "package ", "import (", "type "],
+            "rust": ["fn ", "let mut", "impl ", "pub fn", "use "],
+            "java": ["public class", "private ", "protected ", "void "],
+        }
+
+        scores: dict[str, int] = {}
+        for lang, patterns in markers.items():
+            scores[lang] = sum(1 for p in patterns if p in content)
+
+        if not scores or max(scores.values()) == 0:
+            return self.default_language
+
+        return max(scores, key=lambda k: scores[k])
+
+
+def is_tree_sitter_available() -> bool:
+    """Check if tree-sitter is available."""
+    return _check_tree_sitter()

headroom/compression/handlers/json_handler.py

@@ -0,0 +1,413 @@
+"""JSON structure handler.
+
+Extracts structural elements from JSON content:
+- Keys (navigational - tells LLM what fields exist)
+- Brackets and colons (structural syntax)
+- Short values like booleans, nulls, small numbers
+
+Values (strings, long numbers, nested content) are marked as compressible.
+
+This enables the LLM to see the full schema while values are compressed.
+"""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from enum import Enum
+from typing import Any
+
+from headroom.compression.handlers.base import BaseStructureHandler, HandlerResult
+from headroom.compression.masks import EntropyScore, StructureMask
+
+
+class JSONTokenType(Enum):
+    """Types of JSON tokens for structure detection."""
+
+    KEY = "key"  # Object key (always structural)
+    STRING_VALUE = "string_value"  # String value (compressible)
+    NUMBER = "number"  # Numeric value (preserve if short)
+    BOOLEAN = "boolean"  # true/false (always structural)
+    NULL = "null"  # null (always structural)
+    BRACKET = "bracket"  # {, }, [, ] (always structural)
+    COLON = "colon"  # : (always structural)
+    COMMA = "comma"  # , (always structural)
+    WHITESPACE = "whitespace"  # spaces, newlines (compressible)
+
+
+@dataclass
+class JSONToken:
+    """A token in JSON content with its type and position."""
+
+    text: str
+    token_type: JSONTokenType
+    start: int
+    end: int
+
+    @property
+    def is_structural(self) -> bool:
+        """Whether this token should be preserved."""
+        return self.token_type in (
+            JSONTokenType.KEY,
+            JSONTokenType.BOOLEAN,
+            JSONTokenType.NULL,
+            JSONTokenType.BRACKET,
+            JSONTokenType.COLON,
+            JSONTokenType.COMMA,
+        )
+
+
+class JSONStructureHandler(BaseStructureHandler):
+    """Handler for JSON content.
+
+    Preserves:
+    - All keys (navigational - LLM sees what fields exist)
+    - Structural syntax ({, }, [, ], :, ,)
+    - Booleans and nulls (small, semantically important)
+    - High-entropy strings (UUIDs, hashes - identifiers)
+    - Short numbers (often IDs or important values)
+
+    Compresses:
+    - Long string values (descriptions, content)
+    - Whitespace
+    - Redundant array elements (after first few)
+
+    Example:
+        >>> handler = JSONStructureHandler()
+        >>> result = handler.get_mask('{"name": "Alice", "id": "usr_123"}')
+        >>> # Keys "name" and "id" preserved, values may be compressed
+    """
+
+    def __init__(
+        self,
+        preserve_short_values: bool = True,
+        short_value_threshold: int = 20,
+        preserve_high_entropy: bool = True,
+        entropy_threshold: float = 0.85,
+        max_array_items_full: int = 3,  # Keep first N items fully
+        max_number_digits: int = 10,  # Preserve numbers up to N digits
+    ):
+        """Initialize the JSON handler.
+
+        Args:
+            preserve_short_values: Preserve short string values.
+            short_value_threshold: Max length for "short" values.
+            preserve_high_entropy: Preserve high-entropy strings (UUIDs, etc.).
+            entropy_threshold: Entropy threshold for preservation.
+            max_array_items_full: Number of array items to keep in full.
+            max_number_digits: Max digits for numbers to preserve (often IDs).
+        """
+        super().__init__(name="json")
+        self.preserve_short_values = preserve_short_values
+        self.short_value_threshold = short_value_threshold
+        self.preserve_high_entropy = preserve_high_entropy
+        self.entropy_threshold = entropy_threshold
+        self.max_array_items_full = max_array_items_full
+        self.max_number_digits = max_number_digits
+
+    def can_handle(self, content: str) -> bool:
+        """Check if content is valid JSON."""
+        stripped = content.strip()
+        if not stripped.startswith(("{", "[")):
+            return False
+        try:
+            json.loads(stripped)
+            return True
+        except (json.JSONDecodeError, ValueError):
+            return False
+
+    def _extract_mask(
+        self,
+        content: str,
+        tokens: list[str],
+        **kwargs: Any,
+    ) -> HandlerResult:
+        """Extract structure mask from JSON content.
+
+        Args:
+            content: JSON content.
+            tokens: Character-level tokens.
+            **kwargs: Additional options.
+
+        Returns:
+            HandlerResult with mask marking structural elements.
+        """
+        # Tokenize JSON to identify structure
+        json_tokens = self._tokenize_json(content)
+
+        # Build character-level mask
+        mask = [False] * len(content)
+
+        # Track array depth for selective preservation
+        array_depth = 0
+        array_item_counts: dict[int, int] = {}  # depth -> count
+
+        for token in json_tokens:
+            # Track array items
+            if token.token_type == JSONTokenType.BRACKET:
+                if token.text == "[":
+                    array_depth += 1
+                    array_item_counts[array_depth] = 0
+                elif token.text == "]":
+                    if array_depth in array_item_counts:
+                        del array_item_counts[array_depth]
+                    array_depth = max(0, array_depth - 1)
+
+            # Count array items at commas
+            if token.token_type == JSONTokenType.COMMA and array_depth > 0:
+                array_item_counts[array_depth] = array_item_counts.get(array_depth, 0) + 1
+
+            # Determine if this token should be preserved
+            preserve = self._should_preserve_token(
+                token,
+                array_depth,
+                array_item_counts.get(array_depth, 0),
+            )
+
+            # Mark in mask
+            if preserve:
+                for i in range(token.start, min(token.end, len(mask))):
+                    mask[i] = True
+
+        # Convert to character tokens if needed
+        char_tokens = list(content) if tokens == list(content) else tokens
+
+        return HandlerResult(
+            mask=StructureMask(tokens=char_tokens, mask=mask),
+            handler_name=self.name,
+            confidence=1.0,
+            metadata={
+                "token_count": len(json_tokens),
+                "key_count": sum(1 for t in json_tokens if t.token_type == JSONTokenType.KEY),
+            },
+        )
+
+    def _should_preserve_token(
+        self,
+        token: JSONToken,
+        array_depth: int,
+        array_item_index: int,
+    ) -> bool:
+        """Determine if a token should be preserved.
+
+        Args:
+            token: The JSON token.
+            array_depth: Current array nesting depth.
+            array_item_index: Index of current item in array.
+
+        Returns:
+            True if token should be preserved.
+        """
+        # Always preserve structural tokens
+        if token.is_structural:
+            return True
+
+        # Whitespace is never preserved
+        if token.token_type == JSONTokenType.WHITESPACE:
+            return False
+
+        # Numbers: preserve short ones (often IDs)
+        if token.token_type == JSONTokenType.NUMBER:
+            return len(token.text) <= self.max_number_digits
+
+        # String values: selective preservation
+        if token.token_type == JSONTokenType.STRING_VALUE:
+            # Check if we're past the max array items threshold
+            if array_depth > 0 and array_item_index >= self.max_array_items_full:
+                # In deep array, be more aggressive
+                return False
+
+            # Preserve short values
+            if self.preserve_short_values and len(token.text) <= self.short_value_threshold:
+                return True
+
+            # Preserve high-entropy values (UUIDs, hashes)
+            if self.preserve_high_entropy:
+                # Strip quotes for entropy calculation
+                value = token.text.strip('"')
+                score = EntropyScore.compute(value, self.entropy_threshold)
+                if score.should_preserve:
+                    return True
+
+            return False
+
+        return False
+
+    def _tokenize_json(self, content: str) -> list[JSONToken]:
+        """Tokenize JSON content into typed tokens.
+
+        This is a simple tokenizer that identifies JSON structure.
+        It's not a full parser - just enough to identify keys vs values.
+
+        Args:
+            content: JSON content.
+
+        Returns:
+            List of JSONToken objects.
+        """
+        tokens: list[JSONToken] = []
+        i = 0
+        n = len(content)
+
+        # Track if we're expecting a key (after { or ,)
+        expect_key = False
+        brace_stack: list[str] = []
+
+        while i < n:
+            char = content[i]
+
+            # Whitespace
+            if char in " \t\n\r":
+                start = i
+                while i < n and content[i] in " \t\n\r":
+                    i += 1
+                tokens.append(JSONToken(content[start:i], JSONTokenType.WHITESPACE, start, i))
+                continue
+
+            # Brackets
+            if char in "{}[]":
+                tokens.append(JSONToken(char, JSONTokenType.BRACKET, i, i + 1))
+                if char == "{":
+                    brace_stack.append("{")
+                    expect_key = True
+                elif char == "}":
+                    if brace_stack and brace_stack[-1] == "{":
+                        brace_stack.pop()
+                    expect_key = False
+                elif char == "[":
+                    brace_stack.append("[")
+                    expect_key = False
+                elif char == "]":
+                    if brace_stack and brace_stack[-1] == "[":
+                        brace_stack.pop()
+                i += 1
+                continue
+
+            # Colon
+            if char == ":":
+                tokens.append(JSONToken(char, JSONTokenType.COLON, i, i + 1))
+                expect_key = False
+                i += 1
+                continue
+
+            # Comma
+            if char == ",":
+                tokens.append(JSONToken(char, JSONTokenType.COMMA, i, i + 1))
+                # After comma in object, expect key
+                if brace_stack and brace_stack[-1] == "{":
+                    expect_key = True
+                i += 1
+                continue
+
+            # String (key or value)
+            if char == '"':
+                start = i
+                i += 1
+                while i < n and content[i] != '"':
+                    if content[i] == "\\":
+                        i += 2  # Skip escaped character
+                    else:
+                        i += 1
+                i += 1  # Include closing quote
+
+                text = content[start:i]
+
+                # Determine if this is a key or value
+                # Look ahead for colon (skipping whitespace)
+                j = i
+                while j < n and content[j] in " \t\n\r":
+                    j += 1
+
+                is_key = j < n and content[j] == ":" and expect_key
+
+                if is_key:
+                    tokens.append(JSONToken(text, JSONTokenType.KEY, start, i))
+                    expect_key = False
+                else:
+                    tokens.append(JSONToken(text, JSONTokenType.STRING_VALUE, start, i))
+
+                continue
+
+            # Number
+            if char in "-0123456789":
+                start = i
+                # Match JSON number pattern
+                if char == "-":
+                    i += 1
+                while i < n and content[i] in "0123456789":
+                    i += 1
+                if i < n and content[i] == ".":
+                    i += 1
+                    while i < n and content[i] in "0123456789":
+                        i += 1
+                if i < n and content[i] in "eE":
+                    i += 1
+                    if i < n and content[i] in "+-":
+                        i += 1
+                    while i < n and content[i] in "0123456789":
+                        i += 1
+
+                tokens.append(JSONToken(content[start:i], JSONTokenType.NUMBER, start, i))
+                continue
+
+            # Boolean or null
+            if content[i : i + 4] == "true":
+                tokens.append(JSONToken("true", JSONTokenType.BOOLEAN, i, i + 4))
+                i += 4
+                continue
+            if content[i : i + 5] == "false":
+                tokens.append(JSONToken("false", JSONTokenType.BOOLEAN, i, i + 5))
+                i += 5
+                continue
+            if content[i : i + 4] == "null":
+                tokens.append(JSONToken("null", JSONTokenType.NULL, i, i + 4))
+                i += 4
+                continue
+
+            # Unknown character - skip
+            i += 1
+
+        return tokens
+
+
+def extract_json_schema(content: str) -> dict[str, Any]:
+    """Extract the schema (keys only) from JSON content.
+
+    Useful for understanding the structure without the values.
+
+    Args:
+        content: JSON content.
+
+    Returns:
+        Schema dictionary with keys and types (no values).
+
+    Example:
+        >>> extract_json_schema('{"name": "Alice", "age": 30}')
+        {'name': 'string', 'age': 'number'}
+    """
+
+    def _extract(obj: Any) -> Any:
+        if isinstance(obj, dict):
+            return {k: _extract(v) for k, v in obj.items()}
+        elif isinstance(obj, list):
+            if obj:
+                return [_extract(obj[0])]  # Schema of first item
+            return []
+        elif isinstance(obj, str):
+            return "string"
+        elif isinstance(obj, bool):
+            return "boolean"
+        elif isinstance(obj, int):
+            return "integer"
+        elif isinstance(obj, float):
+            return "number"
+        elif obj is None:
+            return "null"
+        else:
+            return "unknown"
+
+    try:
+        parsed = json.loads(content)
+        return _extract(parsed)
+    except (json.JSONDecodeError, ValueError):
+        return {}

headroom/compression/masks.py

@@ -0,0 +1,345 @@
+"""Structure mask system for compression.
+
+A StructureMask identifies which parts of content are "structural" (should be
+preserved) vs "compressible" (can be compressed by LLMLingua).
+
+This separates the concerns of:
+1. Structure detection (handlers) - What tokens are navigational?
+2. Content compression (LLMLingua) - What tokens can be removed?
+
+The mask is content-agnostic - it's just a boolean array aligned to tokens.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable, Sequence
+from dataclasses import dataclass, field
+
+
+@dataclass
+class StructureMask:
+    """A mask identifying structural vs compressible tokens.
+
+    The mask is aligned to a token sequence. True means "preserve this token"
+    (it's structural/navigational), False means "compressible" (LLMLingua can
+    potentially remove it).
+
+    Attributes:
+        tokens: The tokenized content (list of strings or token IDs).
+        mask: Boolean array, True = preserve, False = compressible.
+        metadata: Optional handler-specific metadata.
+    """
+
+    tokens: Sequence[str | int]
+    mask: list[bool]
+    metadata: dict = field(default_factory=dict)
+
+    def __post_init__(self) -> None:
+        """Validate mask alignment."""
+        if len(self.tokens) != len(self.mask):
+            raise ValueError(
+                f"Mask length ({len(self.mask)}) must match tokens length ({len(self.tokens)})"
+            )
+
+    @property
+    def preservation_ratio(self) -> float:
+        """Fraction of tokens marked for preservation."""
+        if not self.mask:
+            return 0.0
+        return sum(self.mask) / len(self.mask)
+
+    @property
+    def structural_count(self) -> int:
+        """Number of structural (preserved) tokens."""
+        return sum(self.mask)
+
+    @property
+    def compressible_count(self) -> int:
+        """Number of compressible tokens."""
+        return len(self.mask) - sum(self.mask)
+
+    def get_structural_tokens(self) -> list[str | int]:
+        """Get list of tokens marked as structural."""
+        return [t for t, m in zip(self.tokens, self.mask) if m]
+
+    def get_compressible_tokens(self) -> list[str | int]:
+        """Get list of tokens marked as compressible."""
+        return [t for t, m in zip(self.tokens, self.mask) if not m]
+
+    @classmethod
+    def empty(cls, tokens: Sequence[str | int]) -> StructureMask:
+        """Create a mask with no structural tokens (all compressible)."""
+        return cls(tokens=tokens, mask=[False] * len(tokens))
+
+    @classmethod
+    def full(cls, tokens: Sequence[str | int]) -> StructureMask:
+        """Create a mask preserving all tokens (nothing compressible)."""
+        return cls(tokens=tokens, mask=[True] * len(tokens))
+
+    def union(self, other: StructureMask) -> StructureMask:
+        """Combine masks - preserve if EITHER mask says preserve.
+
+        Useful when combining multiple structure detection strategies.
+
+        Args:
+            other: Another mask to combine with.
+
+        Returns:
+            New mask with union of preserved tokens.
+
+        Raises:
+            ValueError: If masks have different lengths.
+        """
+        if len(self.mask) != len(other.mask):
+            raise ValueError("Cannot union masks of different lengths")
+
+        return StructureMask(
+            tokens=self.tokens,
+            mask=[a or b for a, b in zip(self.mask, other.mask)],
+            metadata={"source": "union", **self.metadata, **other.metadata},
+        )
+
+    def intersection(self, other: StructureMask) -> StructureMask:
+        """Combine masks - preserve only if BOTH masks say preserve.
+
+        Useful for being more aggressive with compression.
+
+        Args:
+            other: Another mask to combine with.
+
+        Returns:
+            New mask with intersection of preserved tokens.
+
+        Raises:
+            ValueError: If masks have different lengths.
+        """
+        if len(self.mask) != len(other.mask):
+            raise ValueError("Cannot intersect masks of different lengths")
+
+        return StructureMask(
+            tokens=self.tokens,
+            mask=[a and b for a, b in zip(self.mask, other.mask)],
+            metadata={"source": "intersection", **self.metadata, **other.metadata},
+        )
+
+
+@dataclass
+class MaskSpan:
+    """A contiguous span in the mask.
+
+    Useful for applying different compression strategies to different
+    parts of the content.
+    """
+
+    start: int
+    end: int
+    is_structural: bool
+    label: str = ""  # Optional label (e.g., "key", "value", "signature")
+
+    @property
+    def length(self) -> int:
+        """Length of the span."""
+        return self.end - self.start
+
+
+def mask_to_spans(mask: StructureMask) -> list[MaskSpan]:
+    """Convert a mask to a list of contiguous spans.
+
+    This is useful for processing structural and compressible regions
+    separately.
+
+    Args:
+        mask: The structure mask.
+
+    Returns:
+        List of MaskSpan objects representing contiguous regions.
+
+    Example:
+        >>> tokens = ["def", " ", "foo", "(", ")", ":", " ", "pass"]
+        >>> mask = StructureMask(tokens, [True, True, True, True, True, True, False, False])
+        >>> spans = mask_to_spans(mask)
+        >>> [(s.start, s.end, s.is_structural) for s in spans]
+        [(0, 6, True), (6, 8, False)]
+    """
+    if not mask.mask:
+        return []
+
+    spans = []
+    current_start = 0
+    current_structural = mask.mask[0]
+
+    for i, is_structural in enumerate(mask.mask[1:], start=1):
+        if is_structural != current_structural:
+            spans.append(
+                MaskSpan(
+                    start=current_start,
+                    end=i,
+                    is_structural=current_structural,
+                )
+            )
+            current_start = i
+            current_structural = is_structural
+
+    # Don't forget the last span
+    spans.append(
+        MaskSpan(
+            start=current_start,
+            end=len(mask.mask),
+            is_structural=current_structural,
+        )
+    )
+
+    return spans
+
+
+def apply_mask_to_text(
+    text: str,
+    mask: StructureMask,
+    compress_fn: Callable[[str], str],
+    tokenizer_decode: Callable[[Sequence[str | int]], str] | None = None,
+) -> str:
+    """Apply compression to non-structural regions of text.
+
+    This is the core function that enables structure-preserving compression.
+    Structural regions are kept verbatim, non-structural regions are
+    passed to the compression function.
+
+    Args:
+        text: Original text.
+        mask: Structure mask aligned to tokens.
+        compress_fn: Function to compress text (e.g., LLMLingua).
+        tokenizer_decode: Optional function to decode tokens to text.
+            If not provided, assumes tokens are strings and joins them.
+
+    Returns:
+        Text with non-structural regions compressed.
+    """
+    spans = mask_to_spans(mask)
+    result_parts = []
+
+    if tokenizer_decode is None:
+        # Default: assume tokens are strings
+        def tokenizer_decode(tokens: Sequence[str | int]) -> str:
+            return "".join(str(t) for t in tokens)
+
+    for span in spans:
+        span_tokens = mask.tokens[span.start : span.end]
+        span_text = tokenizer_decode(span_tokens)
+
+        if span.is_structural:
+            # Keep structural regions verbatim
+            result_parts.append(span_text)
+        else:
+            # Compress non-structural regions
+            compressed = compress_fn(span_text)
+            result_parts.append(compressed)
+
+    return "".join(result_parts)
+
+
+@dataclass
+class EntropyScore:
+    """Entropy-based preservation signal.
+
+    High entropy content (UUIDs, hashes, random strings) should generally
+    be preserved because:
+    1. They're information-dense (can't be reconstructed)
+    2. They're often identifiers (semantically important)
+    3. LLMLingua may mangle them
+
+    This is a self-signal - no external classifier needed.
+    """
+
+    value: float  # 0.0 to 1.0, normalized entropy
+    should_preserve: bool  # True if entropy above threshold
+
+    @classmethod
+    def compute(cls, text: str, threshold: float = 0.85) -> EntropyScore:
+        """Compute entropy score for text.
+
+        Args:
+            text: Text to analyze.
+            threshold: Entropy threshold for preservation (0.0-1.0).
+                Higher = more selective.
+
+        Returns:
+            EntropyScore with value and preservation recommendation.
+        """
+        if not text:
+            return cls(value=0.0, should_preserve=False)
+
+        # Calculate character entropy
+        import math
+        from collections import Counter
+
+        # Count character frequencies
+        counter = Counter(text)
+        total = len(text)
+
+        # Calculate Shannon entropy
+        entropy = 0.0
+        for count in counter.values():
+            if count > 0:
+                p = count / total
+                entropy -= p * math.log2(p)
+
+        # Normalize to 0-1 range
+        # Maximum possible entropy for this alphabet size
+        max_entropy = math.log2(len(counter)) if len(counter) > 1 else 1.0
+        normalized = entropy / max_entropy if max_entropy > 0 else 0.0
+
+        return cls(
+            value=normalized,
+            should_preserve=normalized >= threshold,
+        )
+
+
+def compute_entropy_mask(
+    tokens: Sequence[str],
+    threshold: float = 0.85,
+    min_token_length: int = 8,
+) -> StructureMask:
+    """Create a mask preserving high-entropy tokens.
+
+    This is a self-signal that doesn't require content classification.
+    High-entropy tokens (UUIDs, hashes, etc.) are marked for preservation.
+
+    Args:
+        tokens: List of string tokens.
+        threshold: Entropy threshold (0.0-1.0). Higher = more selective.
+        min_token_length: Only check tokens this long or longer.
+            Short tokens rarely have meaningful entropy.
+
+    Returns:
+        StructureMask with high-entropy tokens marked for preservation.
+
+    Example:
+        >>> tokens = ["user", ":", " ", "8f14e45f-ceea-4123-8f14-e45fceea4123"]
+        >>> mask = compute_entropy_mask(tokens)
+        >>> mask.mask
+        [False, False, False, True]  # UUID preserved
+    """
+    mask = []
+
+    for token in tokens:
+        if isinstance(token, int):
+            # Token ID, can't compute entropy
+            mask.append(False)
+            continue
+
+        token_str = str(token)
+
+        # Skip short tokens
+        if len(token_str) < min_token_length:
+            mask.append(False)
+            continue
+
+        # Compute entropy
+        score = EntropyScore.compute(token_str, threshold)
+        mask.append(score.should_preserve)
+
+    return StructureMask(
+        tokens=tokens,
+        mask=mask,
+        metadata={"source": "entropy", "threshold": threshold},
+    )

headroom/compression/universal.py

@@ -0,0 +1,466 @@
+"""Universal compressor with ML-based detection and structure preservation.
+
+This is the main entry point for compression. It:
+1. Detects content type using Magika (ML)
+2. Extracts structure using appropriate handler
+3. Compresses non-structural content with LLMLingua
+4. Optionally stores original in CCR for retrieval
+
+Usage:
+    compressor = UniversalCompressor()
+    result = compressor.compress(content)
+
+    # Result contains:
+    # - compressed: The compressed content
+    # - compression_ratio: original_tokens / compressed_tokens
+    # - content_type: Detected content type
+    # - preservation_ratio: Fraction of content preserved as structure
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from dataclasses import dataclass, field
+from typing import Any
+
+from headroom.compression.detector import (
+    ContentType,
+    DetectionResult,
+    FallbackDetector,
+    get_detector,
+)
+from headroom.compression.handlers.base import (
+    NoOpHandler,
+    StructureHandler,
+)
+from headroom.compression.handlers.code_handler import CodeStructureHandler
+from headroom.compression.handlers.json_handler import JSONStructureHandler
+from headroom.compression.masks import (
+    StructureMask,
+    compute_entropy_mask,
+    mask_to_spans,
+)
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class UniversalCompressorConfig:
+    """Configuration for UniversalCompressor.
+
+    Attributes:
+        use_magika: Use ML-based detection (requires magika package).
+        use_llmlingua: Use LLMLingua for content compression.
+        use_entropy_preservation: Preserve high-entropy tokens (UUIDs, etc.).
+        entropy_threshold: Threshold for entropy-based preservation.
+        min_content_length: Minimum content length to compress.
+        compression_ratio_target: Target compression ratio (0.0-1.0).
+        ccr_enabled: Store originals in CCR for retrieval.
+    """
+
+    use_magika: bool = True
+    use_llmlingua: bool = True
+    use_entropy_preservation: bool = True
+    entropy_threshold: float = 0.85
+    min_content_length: int = 100
+    compression_ratio_target: float = 0.3  # Target 70% reduction
+    ccr_enabled: bool = True
+
+
+@dataclass
+class CompressionResult:
+    """Result from compression.
+
+    Attributes:
+        compressed: The compressed content.
+        original: The original content (for reference).
+        compression_ratio: compressed_length / original_length.
+        tokens_before: Estimated token count before compression.
+        tokens_after: Estimated token count after compression.
+        content_type: Detected content type.
+        detection_confidence: Confidence of content type detection.
+        handler_used: Name of structure handler used.
+        preservation_ratio: Fraction of content marked as structural.
+        ccr_key: CCR storage key (if CCR enabled).
+        metadata: Additional metadata.
+    """
+
+    compressed: str
+    original: str
+    compression_ratio: float
+    tokens_before: int
+    tokens_after: int
+    content_type: ContentType
+    detection_confidence: float
+    handler_used: str
+    preservation_ratio: float
+    ccr_key: str | None = None
+    metadata: dict = field(default_factory=dict)
+
+    @property
+    def tokens_saved(self) -> int:
+        """Number of tokens saved."""
+        return max(0, self.tokens_before - self.tokens_after)
+
+    @property
+    def savings_percentage(self) -> float:
+        """Percentage of tokens saved."""
+        if self.tokens_before == 0:
+            return 0.0
+        return (self.tokens_saved / self.tokens_before) * 100
+
+
+class UniversalCompressor:
+    """Universal compressor with ML detection and structure preservation.
+
+    This compressor automatically:
+    1. Detects content type (JSON, code, logs, text) using ML
+    2. Extracts structure (keys, signatures, templates)
+    3. Preserves structure while compressing content
+    4. Stores original for CCR retrieval
+
+    Example:
+        >>> compressor = UniversalCompressor()
+        >>> result = compressor.compress('{"users": [{"id": 1, "name": "Alice"}]}')
+        >>> print(result.content_type)  # ContentType.JSON
+        >>> print(result.compressed)     # Structure preserved, values compressed
+    """
+
+    def __init__(
+        self,
+        config: UniversalCompressorConfig | None = None,
+        handlers: dict[ContentType, StructureHandler] | None = None,
+        compress_fn: Callable[[str], str] | None = None,
+    ):
+        """Initialize the compressor.
+
+        Args:
+            config: Compression configuration.
+            handlers: Custom handlers for content types.
+            compress_fn: Custom compression function. If None, uses
+                LLMLingua when available, else simple truncation.
+        """
+        self.config = config or UniversalCompressorConfig()
+
+        # Initialize detector
+        if self.config.use_magika:
+            self._detector = get_detector(prefer_magika=True)
+        else:
+            self._detector = FallbackDetector()
+
+        # Initialize handlers
+        self._handlers: dict[ContentType, StructureHandler] = handlers or {
+            ContentType.JSON: JSONStructureHandler(),
+            ContentType.CODE: CodeStructureHandler(),
+        }
+        self._noop_handler = NoOpHandler()
+
+        # Initialize compression function
+        self._compress_fn = compress_fn or self._get_default_compress_fn()
+
+        # CCR store (lazy initialized)
+        self._ccr_store: Any | None = None
+
+    def _get_default_compress_fn(self) -> Callable[[str], str]:
+        """Get default compression function.
+
+        Returns LLMLingua wrapper if available, else simple truncation.
+        """
+        if self.config.use_llmlingua:
+            try:
+                return self._llmlingua_compress
+            except ImportError:
+                logger.info("LLMLingua not available, using simple compression")
+
+        return self._simple_compress
+
+    def _llmlingua_compress(self, text: str) -> str:
+        """Compress using LLMLingua.
+
+        Args:
+            text: Text to compress.
+
+        Returns:
+            Compressed text.
+        """
+        try:
+            from headroom.transforms.llmlingua_compressor import compress_with_llmlingua
+
+            result = compress_with_llmlingua(
+                text,
+                target_ratio=self.config.compression_ratio_target,
+            )
+            return result.compressed
+        except ImportError:
+            return self._simple_compress(text)
+        except Exception as e:
+            logger.warning("LLMLingua compression failed: %s", e)
+            return self._simple_compress(text)
+
+    def _simple_compress(self, text: str) -> str:
+        """Simple compression fallback (truncation with indicator).
+
+        Args:
+            text: Text to compress.
+
+        Returns:
+            Truncated text with indicator.
+        """
+        target_len = int(len(text) * self.config.compression_ratio_target)
+        if len(text) <= target_len:
+            return text
+
+        # Keep first and last portions
+        keep_start = target_len * 2 // 3
+        keep_end = target_len // 3
+
+        return text[:keep_start] + "\n...[compressed]...\n" + text[-keep_end:]
+
+    def compress(
+        self,
+        content: str,
+        content_type: ContentType | None = None,
+        **kwargs: Any,
+    ) -> CompressionResult:
+        """Compress content with structure preservation.
+
+        Args:
+            content: Content to compress.
+            content_type: Override content type detection.
+            **kwargs: Handler-specific options.
+
+        Returns:
+            CompressionResult with compressed content and metadata.
+        """
+        # Handle empty/short content
+        if not content or len(content) < self.config.min_content_length:
+            return CompressionResult(
+                compressed=content,
+                original=content,
+                compression_ratio=1.0,
+                tokens_before=self._estimate_tokens(content),
+                tokens_after=self._estimate_tokens(content),
+                content_type=ContentType.UNKNOWN,
+                detection_confidence=0.0,
+                handler_used="none",
+                preservation_ratio=1.0,
+                metadata={"skipped": "content too short"},
+            )
+
+        # Detect content type
+        if content_type is None:
+            detection = self._detector.detect(content)
+        else:
+            detection = DetectionResult(
+                content_type=content_type,
+                confidence=1.0,
+                raw_label="override",
+            )
+
+        # Get handler for content type
+        handler = self._handlers.get(detection.content_type, self._noop_handler)
+
+        # Tokenize content (character-level for masks)
+        tokens = list(content)
+
+        # Get structure mask from handler
+        handler_result = handler.get_mask(content, tokens, **kwargs)
+        structure_mask = handler_result.mask
+
+        # Optionally add entropy-based preservation
+        if self.config.use_entropy_preservation:
+            entropy_mask = compute_entropy_mask(
+                tokens,
+                threshold=self.config.entropy_threshold,
+            )
+            # Union: preserve if either mask says preserve
+            structure_mask = structure_mask.union(entropy_mask)
+
+        # Apply compression to non-structural parts
+        compressed = self._compress_with_mask(content, structure_mask)
+
+        # Estimate tokens
+        tokens_before = self._estimate_tokens(content)
+        tokens_after = self._estimate_tokens(compressed)
+
+        # Store in CCR if enabled
+        ccr_key = None
+        if self.config.ccr_enabled:
+            ccr_key = self._store_in_ccr(content, compressed)
+
+        return CompressionResult(
+            compressed=compressed,
+            original=content,
+            compression_ratio=len(compressed) / len(content) if content else 1.0,
+            tokens_before=tokens_before,
+            tokens_after=tokens_after,
+            content_type=detection.content_type,
+            detection_confidence=detection.confidence,
+            handler_used=handler_result.handler_name,
+            preservation_ratio=structure_mask.preservation_ratio,
+            ccr_key=ccr_key,
+            metadata={
+                "detection": {
+                    "raw_label": detection.raw_label,
+                    "language": detection.language,
+                },
+                "handler": handler_result.metadata,
+            },
+        )
+
+    def _compress_with_mask(self, content: str, mask: StructureMask) -> str:
+        """Apply compression respecting structure mask.
+
+        Args:
+            content: Original content.
+            mask: Structure mask.
+
+        Returns:
+            Compressed content with structure preserved.
+        """
+        spans = mask_to_spans(mask)
+        result_parts: list[str] = []
+
+        for span in spans:
+            span_content = content[span.start : span.end]
+
+            if span.is_structural:
+                # Preserve structural content
+                result_parts.append(span_content)
+            else:
+                # Compress non-structural content
+                if len(span_content) > 50:  # Only compress if substantial
+                    compressed = self._compress_fn(span_content)
+                    result_parts.append(compressed)
+                else:
+                    result_parts.append(span_content)
+
+        return "".join(result_parts)
+
+    def _estimate_tokens(self, text: str) -> int:
+        """Estimate token count.
+
+        Uses simple heuristic: ~4 characters per token.
+
+        Args:
+            text: Text to estimate.
+
+        Returns:
+            Estimated token count.
+        """
+        if not text:
+            return 0
+        # Simple estimation: ~4 chars per token on average
+        return len(text) // 4
+
+    def _store_in_ccr(self, original: str, compressed: str) -> str | None:
+        """Store original in CCR for retrieval.
+
+        Args:
+            original: Original content.
+            compressed: Compressed content.
+
+        Returns:
+            CCR key if stored, None otherwise.
+        """
+        try:
+            if self._ccr_store is None:
+                from headroom.cache.compression_store import CompressionStore
+
+                self._ccr_store = CompressionStore()
+
+            key = self._ccr_store.store(
+                original_content=original,
+                compressed_content=compressed,
+                original_tokens=self._estimate_tokens(original),
+                compressed_tokens=self._estimate_tokens(compressed),
+            )
+            return key
+        except ImportError:
+            logger.debug("CCR store not available")
+            return None
+        except Exception as e:
+            logger.warning("Failed to store in CCR: %s", e)
+            return None
+
+    def compress_batch(
+        self,
+        contents: list[str],
+        **kwargs: Any,
+    ) -> list[CompressionResult]:
+        """Compress multiple contents.
+
+        More efficient than calling compress() in a loop for
+        ML detection.
+
+        Args:
+            contents: List of contents to compress.
+            **kwargs: Handler-specific options.
+
+        Returns:
+            List of CompressionResults.
+        """
+        if not contents:
+            return []
+
+        # Batch detection
+        if hasattr(self._detector, "detect_batch"):
+            detections = self._detector.detect_batch(contents)
+        else:
+            detections = [self._detector.detect(c) for c in contents]
+
+        # Compress each with detected type
+        results = []
+        for content, detection in zip(contents, detections):
+            result = self.compress(
+                content,
+                content_type=detection.content_type,
+                **kwargs,
+            )
+            results.append(result)
+
+        return results
+
+    def get_handler(self, content_type: ContentType) -> StructureHandler:
+        """Get handler for content type.
+
+        Args:
+            content_type: Content type.
+
+        Returns:
+            Handler for the content type.
+        """
+        return self._handlers.get(content_type, self._noop_handler)
+
+    def register_handler(
+        self,
+        content_type: ContentType,
+        handler: StructureHandler,
+    ) -> None:
+        """Register a custom handler for a content type.
+
+        Args:
+            content_type: Content type to handle.
+            handler: Handler instance.
+        """
+        self._handlers[content_type] = handler
+
+
+def compress(content: str, **kwargs: Any) -> CompressionResult:
+    """Convenience function for one-off compression.
+
+    Args:
+        content: Content to compress.
+        **kwargs: Passed to UniversalCompressor.compress().
+
+    Returns:
+        CompressionResult.
+
+    Example:
+        >>> from headroom.compression import compress
+        >>> result = compress('{"users": [{"id": 1}, {"id": 2}]}')
+        >>> print(result.compressed)
+    """
+    compressor = UniversalCompressor()
+    return compressor.compress(content, **kwargs)

tests/test_compression/__init__.py

@@ -0,0 +1 @@
+"""Tests for the universal compression module."""

tests/test_compression/test_json_handler.py

@@ -0,0 +1,337 @@
+"""Tests for JSON structure handler."""
+
+import json
+
+import pytest
+
+from headroom.compression.handlers.json_handler import (
+    JSONStructureHandler,
+    JSONTokenType,
+    extract_json_schema,
+)
+
+
+class TestJSONStructureHandler:
+    """Tests for JSONStructureHandler."""
+
+    @pytest.fixture
+    def handler(self):
+        """Create handler instance."""
+        return JSONStructureHandler()
+
+    def test_can_handle_json_object(self, handler):
+        """Test detection of JSON objects."""
+        assert handler.can_handle('{"key": "value"}') is True
+
+    def test_can_handle_json_array(self, handler):
+        """Test detection of JSON arrays."""
+        assert handler.can_handle('[{"id": 1}, {"id": 2}]') is True
+
+    def test_cannot_handle_invalid_json(self, handler):
+        """Test rejection of invalid JSON."""
+        assert handler.can_handle("not json") is False
+        assert handler.can_handle('{"unclosed": ') is False
+
+    def test_cannot_handle_plain_text(self, handler):
+        """Test rejection of plain text."""
+        assert handler.can_handle("Hello, world!") is False
+
+    def test_preserves_keys(self, handler):
+        """Test that JSON keys are marked as structural."""
+        content = '{"name": "Alice", "age": 30}'
+        result = handler.get_mask(content)
+
+        # Find the key positions in the mask
+        # "name" should be preserved (with quotes)
+        name_start = content.index('"name"')
+        name_end = name_start + len('"name"')
+
+        for i in range(name_start, name_end):
+            assert result.mask.mask[i] is True, f"Key char at {i} should be preserved"
+
+    def test_preserves_brackets(self, handler):
+        """Test that brackets are preserved."""
+        content = '{"items": [1, 2, 3]}'
+        result = handler.get_mask(content)
+
+        # Find bracket positions
+        for i, char in enumerate(content):
+            if char in "{}[]":
+                assert result.mask.mask[i] is True, f"Bracket at {i} should be preserved"
+
+    def test_preserves_booleans(self, handler):
+        """Test that boolean values are preserved."""
+        content = '{"active": true, "deleted": false}'
+        result = handler.get_mask(content)
+
+        # Find boolean positions
+        true_start = content.index("true")
+        false_start = content.index("false")
+
+        for i in range(true_start, true_start + 4):
+            assert result.mask.mask[i] is True
+        for i in range(false_start, false_start + 5):
+            assert result.mask.mask[i] is True
+
+    def test_preserves_null(self, handler):
+        """Test that null values are preserved."""
+        content = '{"value": null}'
+        result = handler.get_mask(content)
+
+        null_start = content.index("null")
+        for i in range(null_start, null_start + 4):
+            assert result.mask.mask[i] is True
+
+    def test_preserves_short_strings(self, handler):
+        """Test that short string values are preserved."""
+        handler = JSONStructureHandler(
+            preserve_short_values=True,
+            short_value_threshold=10,
+        )
+        content = '{"status": "ok"}'
+        result = handler.get_mask(content)
+
+        # "ok" is short, should be preserved
+        ok_start = content.index('"ok"')
+        for i in range(ok_start, ok_start + 4):
+            assert result.mask.mask[i] is True
+
+    def test_compresses_long_strings(self, handler):
+        """Test that long string values are marked compressible."""
+        handler = JSONStructureHandler(
+            preserve_short_values=True,
+            short_value_threshold=5,
+        )
+        # Use a much longer string to ensure compression kicks in
+        long_value = "x" * 200  # Repetitive content with low entropy
+        content = f'{{"description": "{long_value}"}}'
+        result = handler.get_mask(content)
+
+        # The long repetitive value should have many compressible characters
+        desc_start = content.index('"x')
+        # Not all characters in the long string should be preserved
+        preserved = sum(result.mask.mask[desc_start:])
+        # At minimum, some characters should be compressible (not preserved)
+        total_after_desc = len(content) - desc_start
+        assert preserved < total_after_desc, (
+            "Long repetitive strings should be partially compressible"
+        )
+
+    def test_preserves_high_entropy_values(self):
+        """Test that high-entropy values (UUIDs) are preserved."""
+        handler = JSONStructureHandler(
+            preserve_high_entropy=True,
+            entropy_threshold=0.8,
+        )
+        content = '{"id": "8f14e45f-ceea-4123-8f14-e45fceea4123"}'
+        result = handler.get_mask(content)
+
+        # UUID should be mostly preserved due to high entropy
+        uuid_start = content.index('"8f14e45f')
+        preserved = sum(result.mask.mask[uuid_start:])
+        # Should preserve a significant portion
+        assert preserved > 10
+
+    def test_nested_object(self, handler):
+        """Test handling of nested objects."""
+        content = '{"user": {"name": "Alice", "email": "alice@example.com"}}'
+        result = handler.get_mask(content)
+
+        # Both "user" and "name" keys should be preserved
+        assert result.mask.mask[content.index('"user"')] is True
+        assert result.mask.mask[content.index('"name"')] is True
+
+    def test_array_of_objects(self, handler):
+        """Test handling of array of objects."""
+        content = '[{"id": 1}, {"id": 2}]'
+        result = handler.get_mask(content)
+
+        # Both "id" keys should be preserved
+        first_id = content.index('"id"')
+        second_id = content.index('"id"', first_id + 1)
+
+        assert result.mask.mask[first_id] is True
+        assert result.mask.mask[second_id] is True
+
+    def test_metadata_contains_key_count(self, handler):
+        """Test that metadata includes key count."""
+        content = '{"a": 1, "b": 2, "c": 3}'
+        result = handler.get_mask(content)
+
+        assert "key_count" in result.metadata
+        assert result.metadata["key_count"] == 3
+
+    def test_empty_json_object(self, handler):
+        """Test handling of empty object."""
+        content = "{}"
+        result = handler.get_mask(content)
+
+        # Should preserve the brackets
+        assert result.mask.mask[0] is True  # {
+        assert result.mask.mask[1] is True  # }
+
+    def test_empty_json_array(self, handler):
+        """Test handling of empty array."""
+        content = "[]"
+        result = handler.get_mask(content)
+
+        assert result.mask.mask[0] is True  # [
+        assert result.mask.mask[1] is True  # ]
+
+    def test_whitespace_not_preserved(self, handler):
+        """Test that whitespace is not preserved."""
+        content = '{\n  "key": "value"\n}'
+        result = handler.get_mask(content)
+
+        # Newlines and spaces should not be preserved
+        for i, char in enumerate(content):
+            if char in " \n\t":
+                assert result.mask.mask[i] is False
+
+    def test_handler_name(self, handler):
+        """Test handler name is correct."""
+        assert handler.name == "json"
+
+
+class TestJSONTokenization:
+    """Tests for JSON tokenization."""
+
+    @pytest.fixture
+    def handler(self):
+        """Create handler instance."""
+        return JSONStructureHandler()
+
+    def test_tokenizes_simple_object(self, handler):
+        """Test tokenization of simple object."""
+        content = '{"key": "value"}'
+        tokens = handler._tokenize_json(content)
+
+        # Should have: {, "key", :, "value", }
+        types = [t.token_type for t in tokens]
+        assert JSONTokenType.BRACKET in types
+        assert JSONTokenType.KEY in types
+        assert JSONTokenType.COLON in types
+        assert JSONTokenType.STRING_VALUE in types
+
+    def test_identifies_keys_vs_values(self, handler):
+        """Test that keys and values are correctly identified."""
+        content = '{"name": "Alice"}'
+        tokens = handler._tokenize_json(content)
+
+        key_tokens = [t for t in tokens if t.token_type == JSONTokenType.KEY]
+        value_tokens = [t for t in tokens if t.token_type == JSONTokenType.STRING_VALUE]
+
+        assert len(key_tokens) == 1
+        assert key_tokens[0].text == '"name"'
+        assert len(value_tokens) == 1
+        assert value_tokens[0].text == '"Alice"'
+
+    def test_tokenizes_numbers(self, handler):
+        """Test tokenization of numbers."""
+        content = '{"int": 42, "float": 3.14, "negative": -10, "exp": 1e5}'
+        tokens = handler._tokenize_json(content)
+
+        number_tokens = [t for t in tokens if t.token_type == JSONTokenType.NUMBER]
+        numbers = [t.text for t in number_tokens]
+
+        assert "42" in numbers
+        assert "3.14" in numbers
+        assert "-10" in numbers
+        assert "1e5" in numbers
+
+    def test_tokenizes_booleans_and_null(self, handler):
+        """Test tokenization of booleans and null."""
+        content = '{"a": true, "b": false, "c": null}'
+        tokens = handler._tokenize_json(content)
+
+        bool_tokens = [t for t in tokens if t.token_type == JSONTokenType.BOOLEAN]
+        null_tokens = [t for t in tokens if t.token_type == JSONTokenType.NULL]
+
+        assert len(bool_tokens) == 2
+        assert len(null_tokens) == 1
+
+
+class TestExtractJSONSchema:
+    """Tests for extract_json_schema function."""
+
+    def test_simple_object_schema(self):
+        """Test schema extraction from simple object."""
+        content = '{"name": "Alice", "age": 30}'
+        schema = extract_json_schema(content)
+
+        assert schema == {"name": "string", "age": "integer"}
+
+    def test_nested_object_schema(self):
+        """Test schema extraction from nested object."""
+        content = '{"user": {"name": "Alice", "active": true}}'
+        schema = extract_json_schema(content)
+
+        assert schema == {"user": {"name": "string", "active": "boolean"}}
+
+    def test_array_schema(self):
+        """Test schema extraction from array."""
+        content = '[{"id": 1}, {"id": 2}]'
+        schema = extract_json_schema(content)
+
+        assert schema == [{"id": "integer"}]
+
+    def test_invalid_json_returns_empty(self):
+        """Test that invalid JSON returns empty schema."""
+        content = "not json"
+        schema = extract_json_schema(content)
+
+        assert schema == {}
+
+    def test_mixed_types(self):
+        """Test schema with mixed types."""
+        content = '{"str": "hello", "num": 1.5, "bool": true, "null": null}'
+        schema = extract_json_schema(content)
+
+        assert schema == {
+            "str": "string",
+            "num": "number",
+            "bool": "boolean",
+            "null": "null",
+        }
+
+
+class TestJSONStructurePreservation:
+    """Integration tests for JSON structure preservation."""
+
+    def test_all_keys_visible_after_compression(self):
+        """Test that all keys remain visible in compressed output."""
+        handler = JSONStructureHandler()
+        content = json.dumps(
+            {
+                "users": [
+                    {"id": 1, "name": "Alice", "email": "alice@example.com"},
+                    {"id": 2, "name": "Bob", "email": "bob@example.com"},
+                ],
+                "total": 2,
+                "page": 1,
+            }
+        )
+
+        result = handler.get_mask(content)
+
+        # All keys should be preserved
+        for key in ["users", "id", "name", "email", "total", "page"]:
+            key_str = f'"{key}"'
+            key_start = content.find(key_str)
+            if key_start != -1:
+                # At least the first character of the key should be preserved
+                assert result.mask.mask[key_start] is True, f"Key {key} should be preserved"
+
+    def test_large_array_handling(self):
+        """Test handling of large arrays."""
+        handler = JSONStructureHandler(max_array_items_full=3)
+
+        # Create array with 100 items
+        items = [{"id": i, "value": f"item_{i}_" + "x" * 50} for i in range(100)]
+        content = json.dumps(items)
+
+        result = handler.get_mask(content)
+
+        # Should have reasonable preservation ratio
+        # Not everything should be preserved for large arrays
+        assert 0.1 < result.mask.preservation_ratio < 0.9

tests/test_compression/test_llm_eval.py

@@ -0,0 +1,644 @@
+"""Real-world LLM evaluation tests for compression efficacy.
+
+These tests use actual LLM calls to validate that:
+1. Compressed content is still understandable
+2. LLM can identify what data exists (for CCR retrieval)
+3. Structure preservation enables meaningful reasoning
+
+Run with: pytest tests/test_compression/test_llm_eval.py -v -s
+
+Requires OPENAI_API_KEY environment variable.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+
+import pytest
+
+from headroom.compression.detector import ContentType
+from headroom.compression.universal import (
+    UniversalCompressor,
+    UniversalCompressorConfig,
+)
+
+# Skip all tests if no API key
+pytestmark = pytest.mark.skipif(
+    not os.getenv("OPENAI_API_KEY"),
+    reason="OPENAI_API_KEY not set - skipping LLM eval tests",
+)
+
+
+# =============================================================================
+# Test Fixtures
+# =============================================================================
+
+PRODUCT_CATALOG = json.dumps(
+    {
+        "catalog": {
+            "products": [
+                {
+                    "id": "prod_001",
+                    "sku": "LAPTOP-PRO-15",
+                    "name": "ProBook Laptop 15-inch",
+                    "category": "electronics",
+                    "price": 1299.99,
+                    "currency": "USD",
+                    "description": "High-performance laptop with 16GB RAM, 512GB SSD, Intel i7 processor. "
+                    "Perfect for professionals and power users who need reliable computing power "
+                    "for demanding tasks like video editing, software development, and data analysis. "
+                    "Features include backlit keyboard, fingerprint reader, and Thunderbolt 4 ports.",
+                    "specs": {
+                        "processor": "Intel Core i7-1260P",
+                        "ram": "16GB DDR5",
+                        "storage": "512GB NVMe SSD",
+                        "display": "15.6-inch FHD IPS",
+                        "battery": "72Wh",
+                        "weight": "1.8kg",
+                    },
+                    "stock": 45,
+                    "rating": 4.7,
+                    "reviews_count": 234,
+                },
+                {
+                    "id": "prod_002",
+                    "sku": "HEADPHONES-NC-100",
+                    "name": "NoiseCanceller Pro Headphones",
+                    "category": "audio",
+                    "price": 349.99,
+                    "currency": "USD",
+                    "description": "Premium wireless headphones with industry-leading active noise cancellation. "
+                    "Immerse yourself in crystal-clear audio with 30-hour battery life and quick charge "
+                    "capability. Comfortable memory foam ear cushions make these perfect for long listening "
+                    "sessions, flights, or focused work environments.",
+                    "specs": {
+                        "driver_size": "40mm",
+                        "frequency_response": "20Hz-20kHz",
+                        "battery_life": "30 hours",
+                        "bluetooth": "5.2",
+                        "weight": "250g",
+                    },
+                    "stock": 128,
+                    "rating": 4.8,
+                    "reviews_count": 567,
+                },
+                {
+                    "id": "prod_003",
+                    "sku": "MONITOR-4K-27",
+                    "name": "UltraView 4K Monitor 27-inch",
+                    "category": "electronics",
+                    "price": 599.99,
+                    "currency": "USD",
+                    "description": "Professional-grade 4K monitor with exceptional color accuracy for creative "
+                    "professionals. Features HDR400 support, USB-C connectivity with 65W power delivery, "
+                    "and an ergonomic stand with height, tilt, and swivel adjustments.",
+                    "specs": {
+                        "resolution": "3840x2160",
+                        "panel_type": "IPS",
+                        "refresh_rate": "60Hz",
+                        "response_time": "5ms",
+                        "color_gamut": "99% sRGB",
+                    },
+                    "stock": 72,
+                    "rating": 4.5,
+                    "reviews_count": 189,
+                },
+            ],
+            "total_products": 3,
+            "last_updated": "2024-06-20T15:30:00Z",
+        },
+        "metadata": {
+            "api_version": "v2",
+            "request_id": "req_abc123xyz789",
+        },
+    },
+    indent=2,
+)
+
+CODE_FILE = '''"""User authentication service with JWT tokens."""
+
+from datetime import datetime, timedelta
+from typing import Optional
+import jwt
+from pydantic import BaseModel
+
+SECRET_KEY = "your-secret-key-here"
+ALGORITHM = "HS256"
+ACCESS_TOKEN_EXPIRE_MINUTES = 30
+
+
+class TokenData(BaseModel):
+    """Data stored in JWT token."""
+    username: Optional[str] = None
+    scopes: list[str] = []
+
+
+class User(BaseModel):
+    """User model."""
+    username: str
+    email: str
+    full_name: Optional[str] = None
+    disabled: bool = False
+
+
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None) -> str:
+    """Create a new JWT access token.
+
+    Args:
+        data: Payload data to encode in the token.
+        expires_delta: Custom expiration time.
+
+    Returns:
+        Encoded JWT token string.
+    """
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
+    to_encode.update({"exp": expire})
+    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
+    return encoded_jwt
+
+
+def verify_token(token: str) -> Optional[TokenData]:
+    """Verify and decode a JWT token.
+
+    Args:
+        token: The JWT token to verify.
+
+    Returns:
+        TokenData if valid, None otherwise.
+    """
+    try:
+        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
+        username: str = payload.get("sub")
+        if username is None:
+            return None
+        scopes = payload.get("scopes", [])
+        return TokenData(username=username, scopes=scopes)
+    except jwt.JWTError:
+        return None
+
+
+def authenticate_user(username: str, password: str) -> Optional[User]:
+    """Authenticate a user by username and password.
+
+    Args:
+        username: The username to authenticate.
+        password: The password to verify.
+
+    Returns:
+        User object if authenticated, None otherwise.
+    """
+    # In production, this would check against a database
+    # This is a placeholder implementation
+    if username == "admin" and password == "secret":
+        return User(
+            username="admin",
+            email="admin@example.com",
+            full_name="Admin User",
+            disabled=False,
+        )
+    return None
+
+
+class RateLimiter:
+    """Simple rate limiter for API endpoints."""
+
+    def __init__(self, max_requests: int = 100, window_seconds: int = 60):
+        self.max_requests = max_requests
+        self.window_seconds = window_seconds
+        self._requests: dict[str, list[datetime]] = {}
+
+    def is_allowed(self, client_id: str) -> bool:
+        """Check if a request from client_id is allowed."""
+        now = datetime.utcnow()
+        cutoff = now - timedelta(seconds=self.window_seconds)
+
+        if client_id not in self._requests:
+            self._requests[client_id] = []
+
+        # Clean old requests
+        self._requests[client_id] = [
+            t for t in self._requests[client_id] if t > cutoff
+        ]
+
+        if len(self._requests[client_id]) >= self.max_requests:
+            return False
+
+        self._requests[client_id].append(now)
+        return True
+'''
+
+
+@dataclass
+class LLMEvalResult:
+    """Result from an LLM evaluation."""
+
+    test_name: str
+    passed: bool
+    expected: str
+    actual: str
+    tokens_original: int
+    tokens_compressed: int
+    compression_ratio: float
+    details: str = ""
+
+    def __str__(self) -> str:
+        status = "✓ PASS" if self.passed else "✗ FAIL"
+        return (
+            f"{status}: {self.test_name}\n"
+            f"  Compression: {self.tokens_original} → {self.tokens_compressed} "
+            f"({self.compression_ratio:.1%})\n"
+            f"  Expected: {self.expected}\n"
+            f"  Actual: {self.actual}\n"
+            f"  {self.details}"
+        )
+
+
+def call_openai(prompt: str, system: str = "You are a helpful assistant.") -> str:
+    """Call OpenAI API with given prompt.
+
+    Args:
+        prompt: User prompt.
+        system: System prompt.
+
+    Returns:
+        Model response text.
+    """
+    try:
+        from openai import OpenAI
+
+        client = OpenAI()
+        response = client.chat.completions.create(
+            model="gpt-4o-mini",  # Cost-effective for evals
+            messages=[
+                {"role": "system", "content": system},
+                {"role": "user", "content": prompt},
+            ],
+            max_tokens=500,
+            temperature=0,  # Deterministic for evals
+        )
+        return response.choices[0].message.content or ""
+    except Exception as e:
+        pytest.skip(f"OpenAI API error: {e}")
+        return ""
+
+
+# =============================================================================
+# LLM Evaluation Tests
+# =============================================================================
+
+
+class TestJSONDiscoverability:
+    """Test that LLM can discover structure in compressed JSON."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            use_llmlingua=False,
+            ccr_enabled=False,
+        )
+        return UniversalCompressor(config=config)
+
+    def test_llm_can_list_product_fields(self, compressor):
+        """Test that LLM can identify available fields from compressed JSON."""
+        result = compressor.compress(PRODUCT_CATALOG)
+
+        prompt = f"""Here is a product catalog (may be compressed):
+
+{result.compressed}
+
+List ALL the field names/keys that are available for each product.
+Format your answer as a comma-separated list of field names only."""
+
+        response = call_openai(prompt)
+
+        # Check that key fields are mentioned
+        expected_fields = [
+            "id",
+            "sku",
+            "name",
+            "category",
+            "price",
+            "description",
+            "specs",
+            "stock",
+            "rating",
+        ]
+        found_fields = [f for f in expected_fields if f.lower() in response.lower()]
+
+        eval_result = LLMEvalResult(
+            test_name="JSON Field Discoverability",
+            passed=len(found_fields) >= 7,  # At least 7 of 9 fields
+            expected=", ".join(expected_fields),
+            actual=response[:200],
+            tokens_original=result.tokens_before,
+            tokens_compressed=result.tokens_after,
+            compression_ratio=result.compression_ratio,
+            details=f"Found {len(found_fields)}/9 fields: {found_fields}",
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, f"LLM could not discover enough fields: {found_fields}"
+
+    def test_llm_can_answer_specific_question(self, compressor):
+        """Test that LLM can answer questions about compressed data."""
+        result = compressor.compress(PRODUCT_CATALOG)
+
+        prompt = f"""Here is a product catalog (may be compressed):
+
+{result.compressed}
+
+What is the price of the laptop? Just answer with the number."""
+
+        response = call_openai(prompt)
+
+        # The price should be visible (1299.99)
+        passed = "1299" in response or "1,299" in response
+
+        eval_result = LLMEvalResult(
+            test_name="JSON Specific Query",
+            passed=passed,
+            expected="1299.99",
+            actual=response[:100],
+            tokens_original=result.tokens_before,
+            tokens_compressed=result.tokens_after,
+            compression_ratio=result.compression_ratio,
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, "LLM could not find laptop price"
+
+    def test_llm_knows_what_to_retrieve(self, compressor):
+        """Test that LLM can identify what additional info might be needed."""
+        result = compressor.compress(PRODUCT_CATALOG)
+
+        prompt = f"""Here is a product catalog (may be compressed):
+
+{result.compressed}
+
+I want to write a detailed product comparison. Looking at the compressed data,
+which specific product fields or details would you need me to retrieve in full
+to write a good comparison? List the field names."""
+
+        response = call_openai(prompt)
+
+        # LLM should identify description and specs as needing full retrieval
+        wants_description = "description" in response.lower()
+        wants_specs = "spec" in response.lower()
+
+        passed = wants_description or wants_specs
+
+        eval_result = LLMEvalResult(
+            test_name="CCR Retrieval Identification",
+            passed=passed,
+            expected="description, specs (compressed fields)",
+            actual=response[:200],
+            tokens_original=result.tokens_before,
+            tokens_compressed=result.tokens_after,
+            compression_ratio=result.compression_ratio,
+            details=f"Identified description: {wants_description}, specs: {wants_specs}",
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, "LLM could not identify what to retrieve"
+
+
+class TestCodeUnderstanding:
+    """Test that LLM can understand compressed code."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            use_llmlingua=False,
+            ccr_enabled=False,
+        )
+        return UniversalCompressor(config=config)
+
+    def test_llm_can_list_functions(self, compressor):
+        """Test that LLM can identify functions from compressed code."""
+        result = compressor.compress(CODE_FILE)
+
+        prompt = f"""Here is a Python file (may be compressed):
+
+{result.compressed}
+
+List all the function names defined in this file.
+Format: one function name per line."""
+
+        response = call_openai(prompt)
+
+        expected_functions = [
+            "create_access_token",
+            "verify_token",
+            "authenticate_user",
+        ]
+        found = [f for f in expected_functions if f in response]
+
+        eval_result = LLMEvalResult(
+            test_name="Code Function Discovery",
+            passed=len(found) >= 2,
+            expected=", ".join(expected_functions),
+            actual=response[:200],
+            tokens_original=result.tokens_before,
+            tokens_compressed=result.tokens_after,
+            compression_ratio=result.compression_ratio,
+            details=f"Found {len(found)}/3 functions: {found}",
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, "LLM could not find enough functions"
+
+    def test_llm_can_describe_function_purpose(self, compressor):
+        """Test that LLM can describe what a function does from signature."""
+        result = compressor.compress(CODE_FILE)
+
+        prompt = f"""Here is a Python file (may be compressed):
+
+{result.compressed}
+
+What does the `create_access_token` function do?
+Answer in one sentence based on the function signature and any visible docstring."""
+
+        response = call_openai(prompt)
+
+        # Should mention JWT, token, or access in description
+        keywords = ["jwt", "token", "access", "create"]
+        found_keywords = [k for k in keywords if k.lower() in response.lower()]
+
+        passed = len(found_keywords) >= 2
+
+        eval_result = LLMEvalResult(
+            test_name="Code Function Understanding",
+            passed=passed,
+            expected="Creates a JWT access token",
+            actual=response[:200],
+            tokens_original=result.tokens_before,
+            tokens_compressed=result.tokens_after,
+            compression_ratio=result.compression_ratio,
+            details=f"Keywords found: {found_keywords}",
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, "LLM could not understand function purpose"
+
+    def test_llm_can_identify_classes(self, compressor):
+        """Test that LLM can identify classes from compressed code."""
+        result = compressor.compress(CODE_FILE)
+
+        prompt = f"""Here is a Python file (may be compressed):
+
+{result.compressed}
+
+List all class names defined in this file."""
+
+        response = call_openai(prompt)
+
+        expected_classes = ["TokenData", "User", "RateLimiter"]
+        found = [c for c in expected_classes if c in response]
+
+        eval_result = LLMEvalResult(
+            test_name="Code Class Discovery",
+            passed=len(found) >= 2,
+            expected=", ".join(expected_classes),
+            actual=response[:200],
+            tokens_original=result.tokens_before,
+            tokens_compressed=result.tokens_after,
+            compression_ratio=result.compression_ratio,
+            details=f"Found {len(found)}/3 classes: {found}",
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, "LLM could not find enough classes"
+
+
+class TestMultiContentAgent:
+    """Test multi-content scenario simulating an agent."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            use_llmlingua=False,
+            ccr_enabled=False,
+        )
+        return UniversalCompressor(config=config)
+
+    def test_agent_mixed_content_understanding(self, compressor):
+        """Test that LLM can work with mixed compressed content."""
+        # Compress both
+        json_result = compressor.compress(PRODUCT_CATALOG)
+        code_result = compressor.compress(CODE_FILE)
+
+        prompt = f"""You are an agent with access to two data sources.
+
+## Data Source 1: Product Catalog (JSON)
+{json_result.compressed}
+
+## Data Source 2: Authentication Code (Python)
+{code_result.compressed}
+
+Based on the available data, answer these questions:
+1. What is the most expensive product?
+2. What function would I use to create a login token?
+3. What product categories are available?
+
+Answer each question briefly."""
+
+        response = call_openai(prompt)
+
+        # Check answers
+        checks = {
+            "expensive_product": any(x in response.lower() for x in ["laptop", "probook", "1299"]),
+            "token_function": "create_access_token" in response,
+            "categories": any(x in response.lower() for x in ["electronics", "audio"]),
+        }
+
+        passed = sum(checks.values()) >= 2
+
+        total_original = json_result.tokens_before + code_result.tokens_before
+        total_compressed = json_result.tokens_after + code_result.tokens_after
+
+        eval_result = LLMEvalResult(
+            test_name="Multi-Content Agent Understanding",
+            passed=passed,
+            expected="Laptop ($1299), create_access_token, electronics/audio",
+            actual=response[:300],
+            tokens_original=total_original,
+            tokens_compressed=total_compressed,
+            compression_ratio=total_compressed / total_original,
+            details=f"Checks: {checks}",
+        )
+
+        print(f"\n{eval_result}")
+        assert eval_result.passed, "Agent could not understand mixed content"
+
+
+class TestCompressionEfficacy:
+    """Test overall compression efficacy with real metrics."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            use_llmlingua=False,
+            ccr_enabled=False,
+        )
+        return UniversalCompressor(config=config)
+
+    def test_compression_summary(self, compressor):
+        """Generate summary of compression efficacy."""
+        test_cases = [
+            ("Product Catalog (JSON)", PRODUCT_CATALOG, ContentType.JSON),
+            ("Auth Service (Python)", CODE_FILE, ContentType.CODE),
+        ]
+
+        print("\n" + "=" * 70)
+        print("COMPRESSION EFFICACY SUMMARY (with LLM Validation)")
+        print("=" * 70)
+
+        all_passed = True
+
+        for name, content, expected_type in test_cases:
+            result = compressor.compress(content)
+
+            # Test LLM can extract basic info
+            if expected_type == ContentType.JSON:
+                prompt = f"What are the top-level keys in this JSON?\n\n{result.compressed}"
+                test_query = "JSON keys"
+            else:
+                prompt = f"What functions are defined in this code?\n\n{result.compressed}"
+                test_query = "Function names"
+
+            response = call_openai(prompt)
+
+            # Basic validation
+            llm_understood = len(response) > 20 and "error" not in response.lower()
+
+            status = "✓" if llm_understood else "✗"
+            all_passed = all_passed and llm_understood
+
+            print(f"\n{name}:")
+            print(f"  Type: {result.content_type.name}")
+            print(
+                f"  Tokens: {result.tokens_before} → {result.tokens_after} ({result.compression_ratio:.1%})"
+            )
+            print(f"  Savings: {result.tokens_before - result.tokens_after} tokens")
+            print(f"  LLM Test ({test_query}): {status}")
+            print(f"  LLM Response: {response[:100]}...")
+
+        print("\n" + "=" * 70)
+        print(f"Overall: {'✓ ALL TESTS PASSED' if all_passed else '✗ SOME TESTS FAILED'}")
+        print("=" * 70)
+
+        assert all_passed, "Some LLM validation tests failed"

tests/test_compression/test_masks.py

@@ -0,0 +1,272 @@
+"""Tests for structure mask system."""
+
+import pytest
+
+from headroom.compression.masks import (
+    EntropyScore,
+    MaskSpan,
+    StructureMask,
+    apply_mask_to_text,
+    compute_entropy_mask,
+    mask_to_spans,
+)
+
+
+class TestStructureMask:
+    """Tests for StructureMask class."""
+
+    def test_create_mask(self):
+        """Test basic mask creation."""
+        tokens = ["a", "b", "c", "d"]
+        mask = [True, False, False, True]
+
+        sm = StructureMask(tokens=tokens, mask=mask)
+
+        assert len(sm.tokens) == 4
+        assert len(sm.mask) == 4
+        assert sm.structural_count == 2
+        assert sm.compressible_count == 2
+
+    def test_mask_length_mismatch_raises(self):
+        """Test that mismatched lengths raise ValueError."""
+        tokens = ["a", "b", "c"]
+        mask = [True, False]  # Wrong length
+
+        with pytest.raises(ValueError, match="must match"):
+            StructureMask(tokens=tokens, mask=mask)
+
+    def test_preservation_ratio(self):
+        """Test preservation ratio calculation."""
+        tokens = list("abcdefghij")  # 10 tokens
+        mask = [True, True, False, False, False, False, False, False, False, False]
+
+        sm = StructureMask(tokens=tokens, mask=mask)
+
+        assert sm.preservation_ratio == 0.2  # 2/10
+
+    def test_empty_mask(self):
+        """Test creating empty mask (all compressible)."""
+        tokens = list("hello")
+        sm = StructureMask.empty(tokens)
+
+        assert all(not m for m in sm.mask)
+        assert sm.preservation_ratio == 0.0
+
+    def test_full_mask(self):
+        """Test creating full mask (all preserved)."""
+        tokens = list("hello")
+        sm = StructureMask.full(tokens)
+
+        assert all(m for m in sm.mask)
+        assert sm.preservation_ratio == 1.0
+
+    def test_get_structural_tokens(self):
+        """Test extracting structural tokens."""
+        tokens = ["def", " ", "foo", "(", ")", ":"]
+        mask = [True, False, True, True, True, True]
+
+        sm = StructureMask(tokens=tokens, mask=mask)
+        structural = sm.get_structural_tokens()
+
+        assert structural == ["def", "foo", "(", ")", ":"]
+
+    def test_get_compressible_tokens(self):
+        """Test extracting compressible tokens."""
+        tokens = ["def", " ", "foo", "(", ")", ":"]
+        mask = [True, False, True, True, True, True]
+
+        sm = StructureMask(tokens=tokens, mask=mask)
+        compressible = sm.get_compressible_tokens()
+
+        assert compressible == [" "]
+
+    def test_union_masks(self):
+        """Test union of two masks."""
+        tokens = list("abcd")
+        mask1 = StructureMask(tokens=tokens, mask=[True, False, False, False])
+        mask2 = StructureMask(tokens=tokens, mask=[False, False, True, False])
+
+        result = mask1.union(mask2)
+
+        assert result.mask == [True, False, True, False]
+
+    def test_union_different_lengths_raises(self):
+        """Test that union of different length masks raises."""
+        mask1 = StructureMask(tokens=["a", "b"], mask=[True, False])
+        mask2 = StructureMask(tokens=["a", "b", "c"], mask=[True, False, True])
+
+        with pytest.raises(ValueError, match="different lengths"):
+            mask1.union(mask2)
+
+    def test_intersection_masks(self):
+        """Test intersection of two masks."""
+        tokens = list("abcd")
+        mask1 = StructureMask(tokens=tokens, mask=[True, True, False, False])
+        mask2 = StructureMask(tokens=tokens, mask=[True, False, True, False])
+
+        result = mask1.intersection(mask2)
+
+        assert result.mask == [True, False, False, False]
+
+
+class TestMaskToSpans:
+    """Tests for mask_to_spans function."""
+
+    def test_simple_spans(self):
+        """Test converting mask to spans."""
+        tokens = list("abcdef")
+        mask = StructureMask(
+            tokens=tokens,
+            mask=[True, True, True, False, False, False],
+        )
+
+        spans = mask_to_spans(mask)
+
+        assert len(spans) == 2
+        assert spans[0] == MaskSpan(start=0, end=3, is_structural=True)
+        assert spans[1] == MaskSpan(start=3, end=6, is_structural=False)
+
+    def test_alternating_spans(self):
+        """Test mask with alternating regions."""
+        tokens = list("abcdef")
+        mask = StructureMask(
+            tokens=tokens,
+            mask=[True, False, True, False, True, False],
+        )
+
+        spans = mask_to_spans(mask)
+
+        assert len(spans) == 6  # Each token is its own span
+
+    def test_empty_mask(self):
+        """Test empty mask produces no spans."""
+        mask = StructureMask(tokens=[], mask=[])
+        spans = mask_to_spans(mask)
+
+        assert spans == []
+
+    def test_span_length(self):
+        """Test span length property."""
+        span = MaskSpan(start=5, end=15, is_structural=True)
+        assert span.length == 10
+
+
+class TestEntropyScore:
+    """Tests for entropy-based preservation."""
+
+    def test_high_entropy_uuid(self):
+        """Test that UUIDs have high entropy."""
+        uuid = "8f14e45f-ceea-4123-8f14-e45fceea4123"
+        score = EntropyScore.compute(uuid, threshold=0.8)
+
+        assert score.value > 0.8
+        assert score.should_preserve is True
+
+    def test_low_entropy_repeated(self):
+        """Test that repeated text has low entropy."""
+        text = "aaaaaaaaaaaaaaaa"
+        score = EntropyScore.compute(text, threshold=0.5)
+
+        assert score.value < 0.3
+        assert score.should_preserve is False
+
+    def test_normal_text_entropy(self):
+        """Test normal text entropy."""
+        text = "The quick brown fox"
+        score = EntropyScore.compute(text, threshold=0.85)
+
+        # Normal diverse text has high entropy (no repetition)
+        assert 0.5 < score.value <= 1.0
+
+    def test_empty_text(self):
+        """Test empty text."""
+        score = EntropyScore.compute("", threshold=0.5)
+
+        assert score.value == 0.0
+        assert score.should_preserve is False
+
+    def test_custom_threshold(self):
+        """Test custom threshold."""
+        text = "abc123xyz"  # Moderate entropy
+
+        high_threshold = EntropyScore.compute(text, threshold=0.95)
+        low_threshold = EntropyScore.compute(text, threshold=0.5)
+
+        # Same value, different preservation decisions
+        assert high_threshold.value == low_threshold.value
+        assert (
+            high_threshold.should_preserve != low_threshold.should_preserve
+            or high_threshold.value >= 0.95
+            or high_threshold.value < 0.5
+        )
+
+
+class TestComputeEntropyMask:
+    """Tests for compute_entropy_mask function."""
+
+    def test_preserves_uuids(self):
+        """Test that UUIDs are preserved."""
+        tokens = ["user", ":", " ", "8f14e45f-ceea-4123-8f14-e45fceea4123"]
+        mask = compute_entropy_mask(tokens, threshold=0.8)
+
+        # Only the UUID token should be preserved
+        assert mask.mask[0] is False  # "user"
+        assert mask.mask[1] is False  # ":"
+        assert mask.mask[2] is False  # " "
+        assert mask.mask[3] is True  # UUID
+
+    def test_short_tokens_not_checked(self):
+        """Test that short tokens are not checked for entropy."""
+        tokens = ["ab", "cd", "ef"]
+        mask = compute_entropy_mask(tokens, min_token_length=10)
+
+        # All tokens too short to check
+        assert all(not m for m in mask.mask)
+
+    def test_metadata_contains_threshold(self):
+        """Test that metadata contains threshold."""
+        tokens = ["test"]
+        mask = compute_entropy_mask(tokens, threshold=0.9)
+
+        assert mask.metadata["source"] == "entropy"
+        assert mask.metadata["threshold"] == 0.9
+
+
+class TestApplyMaskToText:
+    """Tests for apply_mask_to_text function."""
+
+    def test_preserves_structural(self):
+        """Test that structural regions are preserved."""
+        text = "def foo(): pass"
+        tokens = list(text)
+        mask = StructureMask(
+            tokens=tokens,
+            # Preserve "def foo():" (first 10 chars)
+            mask=[True] * 10 + [False] * 5,
+        )
+
+        def mock_compress(s: str) -> str:
+            return "[C]"
+
+        result = apply_mask_to_text(text, mask, mock_compress)
+
+        assert result.startswith("def foo():")
+        assert "[C]" in result
+
+    def test_compresses_non_structural(self):
+        """Test that non-structural regions are compressed."""
+        text = "aaa bbb ccc"
+        tokens = list(text)
+        mask = StructureMask(
+            tokens=tokens,
+            mask=[True, True, True, False, False, False, False, True, True, True, True],
+        )
+
+        def mock_compress(s: str) -> str:
+            return "X"
+
+        result = apply_mask_to_text(text, mask, mock_compress)
+
+        # "aaa" preserved, " bbb " compressed to "X", "ccc" preserved
+        assert "aaa" in result
+        assert "ccc" in result

tests/test_compression/test_universal.py

@@ -0,0 +1,340 @@
+"""Tests for UniversalCompressor."""
+
+import json
+
+import pytest
+
+from headroom.compression.detector import ContentType
+from headroom.compression.handlers.base import NoOpHandler
+from headroom.compression.universal import (
+    CompressionResult,
+    UniversalCompressor,
+    UniversalCompressorConfig,
+    compress,
+)
+
+
+class TestUniversalCompressorConfig:
+    """Tests for UniversalCompressorConfig."""
+
+    def test_default_config(self):
+        """Test default configuration values."""
+        config = UniversalCompressorConfig()
+
+        assert config.use_magika is True
+        assert config.use_llmlingua is True
+        assert config.use_entropy_preservation is True
+        assert config.entropy_threshold == 0.85
+        assert config.min_content_length == 100
+        assert config.compression_ratio_target == 0.3
+
+    def test_custom_config(self):
+        """Test custom configuration."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            compression_ratio_target=0.5,
+        )
+
+        assert config.use_magika is False
+        assert config.compression_ratio_target == 0.5
+
+
+class TestCompressionResult:
+    """Tests for CompressionResult."""
+
+    def test_tokens_saved(self):
+        """Test tokens_saved calculation."""
+        result = CompressionResult(
+            compressed="short",
+            original="much longer original content",
+            compression_ratio=0.5,
+            tokens_before=100,
+            tokens_after=50,
+            content_type=ContentType.TEXT,
+            detection_confidence=0.9,
+            handler_used="test",
+            preservation_ratio=0.5,
+        )
+
+        assert result.tokens_saved == 50
+
+    def test_savings_percentage(self):
+        """Test savings_percentage calculation."""
+        result = CompressionResult(
+            compressed="short",
+            original="longer",
+            compression_ratio=0.5,
+            tokens_before=100,
+            tokens_after=25,
+            content_type=ContentType.TEXT,
+            detection_confidence=0.9,
+            handler_used="test",
+            preservation_ratio=0.5,
+        )
+
+        assert result.savings_percentage == 75.0
+
+    def test_zero_tokens_before(self):
+        """Test handling of zero tokens_before."""
+        result = CompressionResult(
+            compressed="",
+            original="",
+            compression_ratio=1.0,
+            tokens_before=0,
+            tokens_after=0,
+            content_type=ContentType.UNKNOWN,
+            detection_confidence=0.0,
+            handler_used="none",
+            preservation_ratio=1.0,
+        )
+
+        assert result.savings_percentage == 0.0
+
+
+class TestUniversalCompressor:
+    """Tests for UniversalCompressor."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor with fallback detector (no Magika required)."""
+        config = UniversalCompressorConfig(
+            use_magika=False,  # Use fallback detector
+            use_llmlingua=False,  # Use simple compression
+            ccr_enabled=False,  # Skip CCR
+        )
+        return UniversalCompressor(config=config)
+
+    def test_compress_short_content_unchanged(self, compressor):
+        """Test that short content is not compressed."""
+        content = "short"
+        result = compressor.compress(content)
+
+        assert result.compressed == content
+        assert result.compression_ratio == 1.0
+        assert "skipped" in result.metadata
+
+    def test_compress_empty_content(self, compressor):
+        """Test handling of empty content."""
+        result = compressor.compress("")
+
+        assert result.compressed == ""
+        assert result.content_type == ContentType.UNKNOWN
+
+    def test_compress_json_content(self, compressor):
+        """Test compression of JSON content."""
+        content = json.dumps(
+            {"users": [{"id": i, "name": f"User {i}", "bio": "x" * 100} for i in range(10)]}
+        )
+
+        result = compressor.compress(content)
+
+        assert result.content_type == ContentType.JSON
+        assert result.handler_used == "json"
+        # Compression should reduce size
+        assert len(result.compressed) < len(content)
+
+    def test_compress_code_content(self, compressor):
+        """Test compression of code content."""
+        content = (
+            '''
+def hello_world():
+    """Say hello to the world."""
+    message = "Hello, World!"
+    print(message)
+    return message
+
+def another_function():
+    """Another function with a long body."""
+    x = 1
+    y = 2
+    z = x + y
+    '''
+            + "result = z * " * 50
+            + """
+    return result
+"""
+        )
+        result = compressor.compress(content)
+
+        assert result.content_type == ContentType.CODE
+        assert result.handler_used == "code"
+
+    def test_compress_plain_text(self, compressor):
+        """Test compression of plain text."""
+        content = "This is plain text without any special structure. " * 20
+
+        result = compressor.compress(content)
+
+        assert result.content_type == ContentType.TEXT
+
+    def test_compress_with_override_type(self, compressor):
+        """Test compression with overridden content type."""
+        content = '{"key": "value"}' + " " * 100  # Pad to meet min length
+
+        result = compressor.compress(content, content_type=ContentType.TEXT)
+
+        # Should use TEXT even though it looks like JSON
+        assert result.content_type == ContentType.TEXT
+
+    def test_compression_result_has_metadata(self, compressor):
+        """Test that result includes metadata."""
+        content = json.dumps({"items": [{"id": i} for i in range(20)]})
+
+        result = compressor.compress(content)
+
+        assert "detection" in result.metadata
+        assert "handler" in result.metadata
+
+    def test_register_custom_handler(self, compressor):
+        """Test registering a custom handler."""
+        custom_handler = NoOpHandler()
+        compressor.register_handler(ContentType.JSON, custom_handler)
+
+        content = '{"key": "value"}' + " " * 100
+
+        result = compressor.compress(content)
+
+        # Should use our custom handler
+        assert result.handler_used == "noop"
+
+    def test_get_handler(self, compressor):
+        """Test getting handler for content type."""
+        json_handler = compressor.get_handler(ContentType.JSON)
+        assert json_handler is not None
+        assert json_handler.name == "json"
+
+        unknown_handler = compressor.get_handler(ContentType.UNKNOWN)
+        assert unknown_handler.name == "noop"
+
+
+class TestUniversalCompressorBatch:
+    """Tests for batch compression."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor with fallback detector."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            use_llmlingua=False,
+            ccr_enabled=False,
+        )
+        return UniversalCompressor(config=config)
+
+    def test_compress_batch_empty(self, compressor):
+        """Test batch compression with empty list."""
+        results = compressor.compress_batch([])
+        assert results == []
+
+    def test_compress_batch_mixed_content(self, compressor):
+        """Test batch compression with mixed content types."""
+        contents = [
+            json.dumps({"id": 1, "data": "x" * 100}),
+            "def foo(): pass\n" * 10,
+            "Plain text content " * 10,
+        ]
+
+        results = compressor.compress_batch(contents)
+
+        assert len(results) == 3
+        assert results[0].content_type == ContentType.JSON
+        assert results[1].content_type == ContentType.CODE
+        assert results[2].content_type == ContentType.TEXT
+
+
+class TestCompressFunction:
+    """Tests for the convenience compress function."""
+
+    def test_compress_function(self):
+        """Test one-off compression function."""
+        content = json.dumps({"items": [{"id": i} for i in range(20)]})
+
+        result = compress(content)
+
+        assert isinstance(result, CompressionResult)
+        assert result.content_type == ContentType.JSON
+
+
+class TestStructurePreservation:
+    """Integration tests for structure preservation."""
+
+    @pytest.fixture
+    def compressor(self):
+        """Create compressor."""
+        config = UniversalCompressorConfig(
+            use_magika=False,
+            use_llmlingua=False,
+            ccr_enabled=False,
+        )
+        return UniversalCompressor(config=config)
+
+    def test_json_keys_preserved(self, compressor):
+        """Test that JSON keys are visible after compression."""
+        data = {
+            "user_id": "12345",
+            "user_name": "Alice",
+            "user_email": "alice@example.com",
+            "user_bio": "A very long biography that goes on and on " * 10,
+        }
+        content = json.dumps(data)
+
+        result = compressor.compress(content)
+
+        # All keys should be visible in compressed output
+        for key in data.keys():
+            assert key in result.compressed, f"Key {key} should be in compressed output"
+
+    def test_code_signatures_preserved(self, compressor):
+        """Test that code signatures are visible after compression."""
+        content = (
+            '''
+def calculate_total(items, tax_rate=0.1):
+    """Calculate total with tax."""
+    subtotal = sum(item.price for item in items)
+    tax = subtotal * tax_rate
+    total = subtotal + tax
+    '''
+            + "# padding " * 50
+            + '''
+    return total
+
+class ShoppingCart:
+    """Shopping cart implementation."""
+
+    def __init__(self):
+        self.items = []
+        '''
+            + "# more padding " * 30
+            + '''
+
+    def add_item(self, item):
+        """Add item to cart."""
+        self.items.append(item)
+'''
+        )
+        result = compressor.compress(content)
+
+        # Function and class names should be visible
+        assert "calculate_total" in result.compressed
+        assert "ShoppingCart" in result.compressed
+        assert "add_item" in result.compressed
+
+    def test_compression_reduces_tokens(self, compressor):
+        """Test that compression actually reduces token count."""
+        # Large content that should be compressible
+        data = {
+            "results": [
+                {
+                    "id": i,
+                    "title": f"Result {i}",
+                    "description": f"This is a detailed description for result {i}. " * 5,
+                }
+                for i in range(50)
+            ]
+        }
+        content = json.dumps(data)
+
+        result = compressor.compress(content)
+
+        # Should achieve some compression
+        assert result.tokens_after < result.tokens_before
+        assert result.compression_ratio < 1.0

uv.lock

@@ -382,7 +382,7 @@ wheels = [
 
 [[package]]
 name = "headroom-ai"
-version = "0.2.3"
+version = "0.2.4"
 source = { editable = "." }
 dependencies = [
     { name = "openai" },