Add TOIN field-level learning and comprehensive integration tests

Features:
- Add field-level learning to TOIN from retrieved items
- CompressionStore now passes retrieved_items to TOIN for learning
- Add FieldSemantics class for tracking field usage patterns

Test improvements:
- Add TestCacheOptimizerInvocation to verify optimizer is actually invoked
- Add TestSemanticCacheIntegration to verify cache hit returns without API call
- Add TestSessionStatsTracking to verify session stats are tracked
- Add TestEndToEndTOINIntegration for full CCR cycle with TOIN
- Add critical field_semantics assertions to catch feedback loop bugs

Fixes:
- Remove unused imports and variables (ruff linting)

Bump version to 0.2.11

headroom/cache/compression_store.py

@@ -684,7 +684,10 @@ class CompressionStore:
             self._pending_feedback_events = []
 
             # Gather entry data while holding lock to avoid race conditions
-            event_data: list[tuple[RetrievalEvent, str | None, str | None, str | None]] = []
+            # Tuple: (event, tool_name, sig_hash, strategy, compressed_content)
+            event_data: list[
+                tuple[RetrievalEvent, str | None, str | None, str | None, str | None]
+            ] = []
             for event in events:
                 entry = self._store.get(event.hash)
                 if entry:
@@ -696,10 +699,11 @@ class CompressionStore:
                             entry.tool_name,
                             entry.tool_signature_hash,  # The correct hash!
                             entry.compression_strategy,
+                            entry.compressed_content,  # For TOIN field-level learning
                         )
                     )
                 else:
-                    event_data.append((event, None, None, None))
+                    event_data.append((event, None, None, None, None))
 
         # Process outside lock
         if event_data:
@@ -707,7 +711,7 @@ class CompressionStore:
             telemetry = get_telemetry_collector()
             toin = get_toin()
 
-            for event, _tool_name, sig_hash, strategy in event_data:
+            for event, _tool_name, sig_hash, strategy, compressed_content in event_data:
                 # Notify feedback system (pass strategy for success rate tracking)
                 feedback.record_retrieval(event, strategy=strategy)
 
@@ -729,6 +733,30 @@ class CompressionStore:
                     # Telemetry should never break the feedback loop
                     logger.debug("Telemetry record_retrieval failed", exc_info=True)
 
+                # Parse compressed content to extract items for TOIN field-level learning
+                retrieved_items: list[dict[str, Any]] | None = None
+                if compressed_content:
+                    try:
+                        parsed = json.loads(compressed_content)
+                        # Handle both direct arrays and wrapped arrays
+                        if isinstance(parsed, list):
+                            # Filter to dicts only (field learning needs dict items)
+                            retrieved_items = [
+                                item for item in parsed if isinstance(item, dict)
+                            ]
+                        elif isinstance(parsed, dict):
+                            # Check for common wrapper patterns: {"items": [...], "results": [...]}
+                            for key in ("items", "results", "data", "records"):
+                                if key in parsed and isinstance(parsed[key], list):
+                                    retrieved_items = [
+                                        item for item in parsed[key]
+                                        if isinstance(item, dict)
+                                    ]
+                                    break
+                    except (json.JSONDecodeError, TypeError):
+                        # Invalid JSON - skip field learning for this retrieval
+                        pass
+
                 # Notify TOIN for cross-user learning
                 try:
                     if sig_hash is not None:
@@ -738,6 +766,7 @@ class CompressionStore:
                             query=event.query,
                             query_fields=query_fields,
                             strategy=strategy,  # Pass strategy for success rate tracking
+                            retrieved_items=retrieved_items,  # For field-level learning
                         )
                 except Exception:
                     # TOIN should never break the feedback loop

headroom/providers/litellm.py

@@ -22,7 +22,6 @@ Requires: pip install litellm
 from __future__ import annotations
 
 import logging
-from collections.abc import Callable
 from typing import Any
 
 from headroom.tokenizers import EstimatingTokenCounter

headroom/telemetry/models.py

@@ -367,6 +367,308 @@ class ToolSignature:
         return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__})
 
 
+@dataclass
+class FieldSemantics:
+    """Learned semantics for a field based on retrieval patterns.
+
+    This is the evolution of TOIN - we learn WHAT fields mean
+    from HOW users retrieve them. No hardcoded patterns, no assumptions.
+
+    Learning process:
+    1. User retrieves items where field X has value Y
+    2. TOIN records: field_hash, value_hash, retrieval context
+    3. After N retrievals, TOIN infers: "This field behaves like an error indicator"
+    4. SmartCrusher uses this learned signal (O(1) lookup, zero latency)
+
+    Privacy: All field names and values are hashed (SHA256[:8]).
+    """
+
+    field_hash: str  # SHA256[:8] of field name
+
+    # Inferred semantic type (learned from retrieval patterns, NOT hardcoded)
+    # These are behavioral categories, not syntactic patterns:
+    # - "identifier": Users query by exact value (e.g., "show me item X")
+    # - "error_indicator": Users retrieve when value != most common value
+    # - "score": Users retrieve top-N by this field
+    # - "status": Low cardinality, specific values trigger retrieval
+    # - "temporal": Users query by time ranges
+    # - "content": Users do text search on this field
+    inferred_type: Literal[
+        "unknown",  # Not enough data yet
+        "identifier",  # High uniqueness, exact-match queries
+        "error_indicator",  # Retrieved when value != default
+        "score",  # Top-N / sorted queries
+        "status",  # Categorical, certain values matter
+        "temporal",  # Range queries
+        "content",  # Text search queries
+    ] = "unknown"
+
+    confidence: float = 0.0  # 0.0 = no data, 1.0 = high confidence
+
+    # Value patterns (all hashed for privacy)
+    # important_value_hashes: values that triggered retrieval
+    # default_value_hash: most common value (probably NOT important)
+    important_value_hashes: list[str] = field(default_factory=list)
+    default_value_hash: str | None = None
+    value_retrieval_frequency: dict[str, int] = field(default_factory=dict)  # value_hash -> count
+
+    # Value statistics (for inferring type)
+    total_unique_values_seen: int = 0
+    total_values_seen: int = 0
+    most_common_value_frequency: float = 0.0  # Fraction of items with most common value
+
+    # Query patterns (anonymized)
+    # Tracks HOW users query this field (equals, not-equals, greater-than, etc.)
+    query_operator_frequency: dict[str, int] = field(default_factory=dict)  # operator -> count
+
+    # Learning metadata
+    retrieval_count: int = 0
+    compression_count: int = 0  # How many times we've seen this field in compression
+    last_updated: float = 0.0
+
+    # Bounds for memory management
+    MAX_IMPORTANT_VALUES: int = 50
+    MAX_VALUE_FREQUENCY_ENTRIES: int = 100
+
+    def to_dict(self) -> dict[str, Any]:
+        """Convert to dictionary for serialization."""
+        return {
+            "field_hash": self.field_hash,
+            "inferred_type": self.inferred_type,
+            "confidence": self.confidence,
+            "important_value_hashes": self.important_value_hashes[:self.MAX_IMPORTANT_VALUES],
+            "default_value_hash": self.default_value_hash,
+            "value_retrieval_frequency": dict(
+                sorted(
+                    self.value_retrieval_frequency.items(),
+                    key=lambda x: x[1],
+                    reverse=True,
+                )[: self.MAX_VALUE_FREQUENCY_ENTRIES]
+            ),
+            "total_unique_values_seen": self.total_unique_values_seen,
+            "total_values_seen": self.total_values_seen,
+            "most_common_value_frequency": self.most_common_value_frequency,
+            "query_operator_frequency": self.query_operator_frequency,
+            "retrieval_count": self.retrieval_count,
+            "compression_count": self.compression_count,
+            "last_updated": self.last_updated,
+        }
+
+    @classmethod
+    def from_dict(cls, data: dict[str, Any]) -> FieldSemantics:
+        """Create from dictionary."""
+        # Filter to valid fields only
+        valid_fields = {
+            "field_hash",
+            "inferred_type",
+            "confidence",
+            "important_value_hashes",
+            "default_value_hash",
+            "value_retrieval_frequency",
+            "total_unique_values_seen",
+            "total_values_seen",
+            "most_common_value_frequency",
+            "query_operator_frequency",
+            "retrieval_count",
+            "compression_count",
+            "last_updated",
+        }
+        filtered = {k: v for k, v in data.items() if k in valid_fields}
+        return cls(**filtered)
+
+    def record_retrieval_value(self, value_hash: str, operator: str = "=") -> None:
+        """Record that a value was retrieved for this field.
+
+        Args:
+            value_hash: SHA256[:8] hash of the retrieved value.
+            operator: Query operator used ("=", "!=", ">", "<", "contains", etc.)
+        """
+        import time
+
+        self.retrieval_count += 1
+        self.last_updated = time.time()
+
+        # Track value frequency
+        self.value_retrieval_frequency[value_hash] = (
+            self.value_retrieval_frequency.get(value_hash, 0) + 1
+        )
+
+        # Bound the frequency dict
+        if len(self.value_retrieval_frequency) > self.MAX_VALUE_FREQUENCY_ENTRIES:
+            sorted_items = sorted(
+                self.value_retrieval_frequency.items(),
+                key=lambda x: x[1],
+                reverse=True,
+            )[: self.MAX_VALUE_FREQUENCY_ENTRIES]
+            self.value_retrieval_frequency = dict(sorted_items)
+
+        # Track important values (values that get retrieved)
+        if value_hash not in self.important_value_hashes:
+            self.important_value_hashes.append(value_hash)
+            if len(self.important_value_hashes) > self.MAX_IMPORTANT_VALUES:
+                # Keep most frequently retrieved values
+                self.important_value_hashes = sorted(
+                    self.important_value_hashes,
+                    key=lambda v: self.value_retrieval_frequency.get(v, 0),
+                    reverse=True,
+                )[: self.MAX_IMPORTANT_VALUES]
+
+        # Track query operators
+        self.query_operator_frequency[operator] = (
+            self.query_operator_frequency.get(operator, 0) + 1
+        )
+
+    def record_compression_stats(
+        self,
+        unique_values: int,
+        total_values: int,
+        most_common_value_hash: str | None,
+        most_common_frequency: float,
+    ) -> None:
+        """Record statistics from compression for type inference.
+
+        Args:
+            unique_values: Number of unique values seen for this field.
+            total_values: Total number of items with this field.
+            most_common_value_hash: Hash of the most common value.
+            most_common_frequency: Fraction of items with the most common value.
+        """
+        import time
+
+        self.compression_count += 1
+        self.last_updated = time.time()
+
+        # Update rolling statistics
+        n = self.compression_count
+        self.total_unique_values_seen = int(
+            (self.total_unique_values_seen * (n - 1) + unique_values) / n
+        )
+        self.total_values_seen = int(
+            (self.total_values_seen * (n - 1) + total_values) / n
+        )
+        self.most_common_value_frequency = (
+            self.most_common_value_frequency * (n - 1) + most_common_frequency
+        ) / n
+
+        # Track default value (most common)
+        if most_common_value_hash and most_common_frequency > 0.5:
+            self.default_value_hash = most_common_value_hash
+
+    def infer_type(self) -> None:
+        """Infer semantic type from accumulated statistics.
+
+        This is the learning algorithm - purely data-driven, no hardcoded patterns.
+        """
+        # Need minimum data to infer
+        min_retrievals = 3
+        min_compressions = 2
+
+        if self.retrieval_count < min_retrievals or self.compression_count < min_compressions:
+            self.inferred_type = "unknown"
+            self.confidence = 0.0
+            return
+
+        # Calculate metrics
+        uniqueness_ratio = (
+            self.total_unique_values_seen / max(1, self.total_values_seen)
+        )
+        has_dominant_default = self.most_common_value_frequency > 0.7
+        retrieval_diversity = len(self.value_retrieval_frequency) / max(1, self.retrieval_count)
+
+        # Check query operator patterns
+        total_ops = sum(self.query_operator_frequency.values())
+        equals_ratio = self.query_operator_frequency.get("=", 0) / max(1, total_ops)
+        range_ratio = (
+            self.query_operator_frequency.get(">", 0)
+            + self.query_operator_frequency.get("<", 0)
+            + self.query_operator_frequency.get(">=", 0)
+            + self.query_operator_frequency.get("<=", 0)
+        ) / max(1, total_ops)
+        contains_ratio = self.query_operator_frequency.get("contains", 0) / max(1, total_ops)
+
+        # Inference logic (data-driven, no field name patterns)
+        inferred = "unknown"
+        confidence = 0.0
+
+        # IDENTIFIER: High uniqueness + exact match queries
+        if uniqueness_ratio > 0.8 and equals_ratio > 0.7:
+            inferred = "identifier"
+            confidence = min(0.9, uniqueness_ratio * equals_ratio)
+
+        # ERROR_INDICATOR: Has dominant default + retrievals are for non-default values
+        elif has_dominant_default and self.default_value_hash:
+            # Check if retrieved values are different from default
+            default_retrieval_count = self.value_retrieval_frequency.get(
+                self.default_value_hash, 0
+            )
+            non_default_retrieval_ratio = 1 - (
+                default_retrieval_count / max(1, self.retrieval_count)
+            )
+            if non_default_retrieval_ratio > 0.7:
+                inferred = "error_indicator"
+                confidence = min(0.9, non_default_retrieval_ratio * self.most_common_value_frequency)
+
+        # STATUS: Low uniqueness + specific values retrieved
+        elif uniqueness_ratio < 0.2 and retrieval_diversity < 0.5:
+            inferred = "status"
+            confidence = min(0.85, (1 - uniqueness_ratio) * (1 - retrieval_diversity))
+
+        # SCORE: Range queries or sorted access patterns
+        elif range_ratio > 0.5:
+            inferred = "score"
+            confidence = min(0.85, range_ratio)
+
+        # TEMPORAL: Range queries + high uniqueness (likely timestamps)
+        elif range_ratio > 0.3 and uniqueness_ratio > 0.7:
+            inferred = "temporal"
+            confidence = min(0.8, range_ratio * uniqueness_ratio)
+
+        # CONTENT: Contains/text search queries
+        elif contains_ratio > 0.5:
+            inferred = "content"
+            confidence = min(0.85, contains_ratio)
+
+        # Apply minimum confidence threshold
+        if confidence < 0.3:
+            inferred = "unknown"
+            confidence = 0.0
+
+        self.inferred_type = inferred
+        self.confidence = confidence
+
+    def is_value_important(self, value_hash: str) -> bool:
+        """Check if a specific value is considered important.
+
+        A value is important if:
+        1. It's in the important_value_hashes list (has been retrieved)
+        2. It's NOT the default value (for error_indicator type)
+
+        Args:
+            value_hash: SHA256[:8] hash of the value to check.
+
+        Returns:
+            True if this value should be preserved during compression.
+        """
+        # If we don't have enough data, be conservative
+        if self.confidence < 0.3:
+            return False
+
+        # For error_indicator: non-default values are important
+        if self.inferred_type == "error_indicator":
+            if self.default_value_hash and value_hash != self.default_value_hash:
+                return True
+
+        # For any type: values that have been retrieved are important
+        if value_hash in self.important_value_hashes:
+            return True
+
+        # For status: check if this value has been retrieved
+        if self.inferred_type == "status":
+            return value_hash in self.value_retrieval_frequency
+
+        return False
+
+
 @dataclass
 class CompressionEvent:
     """Record of a single compression decision (anonymized).

headroom/telemetry/toin.py

@@ -52,7 +52,7 @@ from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Any, Literal
 
-from .models import ToolSignature
+from .models import FieldSemantics, ToolSignature
 
 logger = logging.getLogger(__name__)
 
@@ -116,6 +116,11 @@ class ToolPattern:
     skip_compression_recommended: bool = False
     preserve_fields: list[str] = field(default_factory=list)
 
+    # === Field-Level Semantics (TOIN Evolution) ===
+    # Learned semantic types for each field based on retrieval patterns
+    # This enables zero-latency signal detection without hardcoded patterns
+    field_semantics: dict[str, FieldSemantics] = field(default_factory=dict)
+
     # === Confidence ===
     sample_size: int = 0
     user_count: int = 0  # Number of unique users (anonymized)
@@ -163,6 +168,10 @@ class ToolPattern:
             "optimal_max_items": self.optimal_max_items,
             "skip_compression_recommended": self.skip_compression_recommended,
             "preserve_fields": self.preserve_fields,
+            # Field-level semantics (TOIN Evolution)
+            "field_semantics": {
+                k: v.to_dict() for k, v in self.field_semantics.items()
+            },
             "sample_size": self.sample_size,
             "user_count": self.user_count,
             "confidence": self.confidence,
@@ -226,6 +235,13 @@ class ToolPattern:
         if pattern.user_count > len(pattern._seen_instance_hashes):
             pattern._tracking_truncated = True
 
+        # Load field semantics (TOIN Evolution)
+        field_semantics_data = data.get("field_semantics", {})
+        if field_semantics_data:
+            pattern.field_semantics = {
+                k: FieldSemantics.from_dict(v) for k, v in field_semantics_data.items()
+            }
+
         return pattern
 
 
@@ -257,6 +273,11 @@ class CompressionHint:
     source: Literal["network", "local", "default"] = "default"
     based_on_samples: int = 0
 
+    # === TOIN Evolution: Learned Field Semantics ===
+    # These enable zero-latency signal detection in SmartCrusher.
+    # field_hash -> FieldSemantics (learned semantic type, important values, etc.)
+    field_semantics: dict[str, FieldSemantics] = field(default_factory=dict)
+
 
 @dataclass
 class TOINConfig:
@@ -375,12 +396,16 @@ class ToolIntelligenceNetwork:
         compressed_tokens: int,
         strategy: str,
         query_context: str | None = None,
+        items: list[dict[str, Any]] | None = None,
     ) -> None:
         """Record a compression event.
 
         Called after SmartCrusher compresses data. Updates the pattern
         for this tool type.
 
+        TOIN Evolution: When items are provided, we capture field statistics
+        for learning semantic types (uniqueness, default values, etc.).
+
         Args:
             tool_signature: Signature of the tool output structure.
             original_count: Original number of items.
@@ -389,6 +414,7 @@ class ToolIntelligenceNetwork:
             compressed_tokens: Compressed token count.
             strategy: Compression strategy used.
             query_context: Optional user query that triggered this tool call.
+            items: Optional list of items being compressed for field-level learning.
         """
         # HIGH FIX: Check enabled FIRST to avoid computing structure_hash if disabled
         # This saves CPU when TOIN is turned off
@@ -518,6 +544,11 @@ class ToolIntelligenceNetwork:
             if pattern.total_compressions % 10 == 0:
                 self._update_recommendations(pattern)
 
+            # === TOIN Evolution: Field Statistics for Semantic Learning ===
+            # Capture field-level statistics to learn default values and uniqueness
+            if items:
+                self._update_field_statistics(pattern, items)
+
             pattern.last_updated = time.time()
             pattern.confidence = self._calculate_confidence(pattern)
             self._dirty = True
@@ -525,6 +556,81 @@ class ToolIntelligenceNetwork:
         # Auto-save if needed (outside lock)
         self._maybe_auto_save()
 
+    def _update_field_statistics(
+        self,
+        pattern: ToolPattern,
+        items: list[dict[str, Any]],
+    ) -> None:
+        """Update field statistics from compression items.
+
+        Captures uniqueness, default values, and value distribution for
+        learning field semantic types.
+
+        Args:
+            pattern: ToolPattern to update.
+            items: Items being compressed.
+        """
+        if not items:
+            return
+
+        # Analyze field statistics (sample up to 100 items to limit CPU)
+        sample_items = items[:100] if len(items) > 100 else items
+
+        # Collect values for each field
+        field_values: dict[str, list[str]] = {}  # field_hash -> list of value_hashes
+
+        for item in sample_items:
+            if not isinstance(item, dict):
+                continue
+
+            for field_name, value in item.items():
+                field_hash = self._hash_field_name(field_name)
+                value_hash = self._hash_value(value)
+
+                if field_hash not in field_values:
+                    field_values[field_hash] = []
+                field_values[field_hash].append(value_hash)
+
+        # Update FieldSemantics with statistics
+        for field_hash, values in field_values.items():
+            if not values:
+                continue
+
+            # Get or create FieldSemantics
+            if field_hash not in pattern.field_semantics:
+                pattern.field_semantics[field_hash] = FieldSemantics(field_hash=field_hash)
+
+            field_sem = pattern.field_semantics[field_hash]
+
+            # Calculate statistics
+            unique_values = len(set(values))
+            total_values = len(values)
+
+            # Find most common value
+            from collections import Counter
+
+            value_counts = Counter(values)
+            most_common_value, most_common_count = value_counts.most_common(1)[0]
+            most_common_frequency = most_common_count / total_values if total_values > 0 else 0.0
+
+            # Record compression stats
+            field_sem.record_compression_stats(
+                unique_values=unique_values,
+                total_values=total_values,
+                most_common_value_hash=most_common_value,
+                most_common_frequency=most_common_frequency,
+            )
+
+        # Bound field_semantics to prevent unbounded growth (max 100 fields)
+        if len(pattern.field_semantics) > 100:
+            # Keep fields with highest activity (retrieval + compression count)
+            sorted_fields = sorted(
+                pattern.field_semantics.items(),
+                key=lambda x: x[1].retrieval_count + x[1].compression_count,
+                reverse=True,
+            )[:100]
+            pattern.field_semantics = dict(sorted_fields)
+
     def record_retrieval(
         self,
         tool_signature_hash: str,
@@ -532,18 +638,23 @@ class ToolIntelligenceNetwork:
         query: str | None = None,
         query_fields: list[str] | None = None,
         strategy: str | None = None,
+        retrieved_items: list[dict[str, Any]] | None = None,
     ) -> None:
         """Record a retrieval event.
 
         Called when LLM retrieves compressed content. This is the key
         feedback signal - it means compression was too aggressive.
 
+        TOIN Evolution: When retrieved_items are provided, we learn field
+        semantics from the values. This enables zero-latency signal detection.
+
         Args:
             tool_signature_hash: Hash of the tool signature.
             retrieval_type: "full" or "search".
             query: Optional search query (will be anonymized).
             query_fields: Fields mentioned in query (will be hashed).
             strategy: Compression strategy that was used (for success rate tracking).
+            retrieved_items: Optional list of retrieved items for field-level learning.
         """
         if not self._config.enabled:
             return
@@ -636,6 +747,49 @@ class ToolIntelligenceNetwork:
                             reverse=True,
                         )[: self._config.max_query_patterns]
 
+            # === TOIN Evolution: Field-Level Semantic Learning ===
+            # Learn from retrieved items to build zero-latency signal detection
+            if retrieved_items:
+                # Extract query operator from query string (for learning)
+                query_operator = self._extract_query_operator(query) if query else "="
+
+                for item in retrieved_items:
+                    if not isinstance(item, dict):
+                        continue
+
+                    for field_name, value in item.items():
+                        field_hash = self._hash_field_name(field_name)
+
+                        # Get or create FieldSemantics for this field
+                        if field_hash not in pattern.field_semantics:
+                            pattern.field_semantics[field_hash] = FieldSemantics(
+                                field_hash=field_hash
+                            )
+
+                        field_sem = pattern.field_semantics[field_hash]
+
+                        # Hash the value for privacy
+                        value_hash = self._hash_value(value)
+
+                        # Record this retrieval
+                        field_sem.record_retrieval_value(value_hash, query_operator)
+
+                # Periodically infer types (every 5 retrievals to save CPU)
+                if pattern.total_retrievals % 5 == 0:
+                    for field_sem in pattern.field_semantics.values():
+                        if field_sem.retrieval_count >= 3:  # Need minimum data
+                            field_sem.infer_type()
+
+                # Bound field_semantics to prevent unbounded growth (max 100 fields)
+                if len(pattern.field_semantics) > 100:
+                    # Keep fields with highest retrieval counts
+                    sorted_fields = sorted(
+                        pattern.field_semantics.items(),
+                        key=lambda x: x[1].retrieval_count,
+                        reverse=True,
+                    )[:100]
+                    pattern.field_semantics = dict(sorted_fields)
+
             # Update recommendations based on new retrieval data
             self._update_recommendations(pattern)
 
@@ -850,6 +1004,16 @@ class ToolIntelligenceNetwork:
                                 hint.reason += " (query uses fields from retrieval pattern)"
                                 break
 
+        # === TOIN Evolution: Include learned field semantics ===
+        # Copy field_semantics with sufficient confidence for SmartCrusher to use
+        # Only include fields with confidence >= 0.3 to reduce noise
+        if pattern.field_semantics:
+            hint.field_semantics = {
+                field_hash: field_sem
+                for field_hash, field_sem in pattern.field_semantics.items()
+                if field_sem.confidence >= 0.3 or field_sem.retrieval_count >= 3
+            }
+
         return hint
 
     def _find_best_strategy(self, pattern: ToolPattern) -> str | None:
@@ -948,6 +1112,62 @@ class ToolIntelligenceNetwork:
 
         return pattern
 
+    def _hash_value(self, value: Any) -> str:
+        """Hash a value for privacy-preserving storage.
+
+        Handles all types by converting to a canonical string representation.
+        """
+        if value is None:
+            canonical = "null"
+        elif isinstance(value, bool):
+            canonical = "true" if value else "false"
+        elif isinstance(value, (int, float)):
+            canonical = str(value)
+        elif isinstance(value, str):
+            canonical = value
+        elif isinstance(value, (list, dict)):
+            # For complex types, use JSON serialization
+            try:
+                canonical = json.dumps(value, sort_keys=True, default=str)
+            except (TypeError, ValueError):
+                canonical = str(value)
+        else:
+            canonical = str(value)
+
+        return hashlib.sha256(canonical.encode()).hexdigest()[:8]
+
+    def _extract_query_operator(self, query: str) -> str:
+        """Extract the dominant query operator from a search query.
+
+        Used for learning field semantic types from query patterns.
+
+        Returns:
+            Query operator: "=", "!=", ">", "<", ">=", "<=", "contains", or "="
+        """
+        if not query:
+            return "="
+
+        query_lower = query.lower()
+
+        # Check for inequality operators
+        if "!=" in query or " not " in query_lower or " ne " in query_lower:
+            return "!="
+        if ">=" in query or " gte " in query_lower:
+            return ">="
+        if "<=" in query or " lte " in query_lower:
+            return "<="
+        if ">" in query or " gt " in query_lower:
+            return ">"
+        if "<" in query or " lt " in query_lower:
+            return "<"
+
+        # Check for text search operators
+        if " like " in query_lower or " contains " in query_lower or "*" in query:
+            return "contains"
+
+        # Default to equality
+        return "="
+
     def get_stats(self) -> dict[str, Any]:
         """Get overall TOIN statistics."""
         with self._lock:

headroom/transforms/smart_crusher.py

@@ -57,6 +57,7 @@ from ..cache.compression_store import CompressionStore, get_compression_store
 from ..config import CCRConfig, RelevanceScorerConfig, TransformResult
 from ..relevance import RelevanceScorer, create_scorer
 from ..telemetry import TelemetryCollector, ToolSignature, get_telemetry_collector
+from ..telemetry.models import FieldSemantics
 from ..telemetry.toin import ToolIntelligenceNetwork, get_toin
 from ..tokenizer import Tokenizer
 from ..utils import (
@@ -616,6 +617,81 @@ def _detect_error_items_for_preservation(items: list[dict]) -> list[int]:
     return error_indices
 
 
+def _detect_items_by_learned_semantics(
+    items: list[dict],
+    field_semantics: dict[str, FieldSemantics],
+) -> list[int]:
+    """Detect items with important values based on learned field semantics.
+
+    This is the TOIN Evolution integration - uses learned field semantic types
+    to identify items that should be preserved during compression.
+
+    Key insight: Instead of hardcoded patterns, we learn from user behavior
+    which field values are actually important (e.g., error indicators, rare
+    status values, identifiers that get queried).
+
+    Args:
+        items: List of items to analyze.
+        field_semantics: Learned field semantics from TOIN (field_hash -> FieldSemantics).
+
+    Returns:
+        List of indices for items containing important values.
+    """
+    if not field_semantics or not items:
+        return []
+
+    important_indices: list[int] = []
+
+    # Build a quick lookup for field_hash -> FieldSemantics
+    # Pre-filter to fields with sufficient confidence
+    confident_semantics = {
+        fh: fs for fh, fs in field_semantics.items()
+        if fs.confidence >= 0.3 and fs.inferred_type != "unknown"
+    }
+
+    if not confident_semantics:
+        return []
+
+    for i, item in enumerate(items):
+        if not isinstance(item, dict):
+            continue
+
+        for field_name, value in item.items():
+            # Hash the field name to match TOIN's format
+            field_hash = hashlib.sha256(field_name.encode()).hexdigest()[:8]
+
+            if field_hash not in confident_semantics:
+                continue
+
+            field_sem = confident_semantics[field_hash]
+
+            # Hash the value to check importance
+            if value is None:
+                value_canonical = "null"
+            elif isinstance(value, bool):
+                value_canonical = "true" if value else "false"
+            elif isinstance(value, (int, float)):
+                value_canonical = str(value)
+            elif isinstance(value, str):
+                value_canonical = value
+            elif isinstance(value, (list, dict)):
+                try:
+                    value_canonical = json.dumps(value, sort_keys=True, default=str)
+                except (TypeError, ValueError):
+                    value_canonical = str(value)
+            else:
+                value_canonical = str(value)
+
+            value_hash = hashlib.sha256(value_canonical.encode()).hexdigest()[:8]
+
+            # Check if this value is important based on learned semantics
+            if field_sem.is_value_important(value_hash):
+                important_indices.append(i)
+                break  # Only need to mark item once
+
+    return important_indices
+
+
 @dataclass
 class CrushabilityAnalysis:
     """Analysis of whether an array is safe to crush.
@@ -1484,6 +1560,7 @@ class SmartCrusher(Transform):
         n: int,
         analysis: ArrayAnalysis | None = None,
         max_items: int | None = None,
+        field_semantics: dict[str, FieldSemantics] | None = None,
     ) -> set[int]:
         """Prioritize indices when we exceed max_items, ALWAYS keeping critical items.
 
@@ -1491,11 +1568,13 @@ class SmartCrusher(Transform):
         1. ALL error items (non-negotiable) - items with error keywords
         2. ALL structural outliers (non-negotiable) - items with rare fields/status values
         3. ALL numeric anomalies (non-negotiable) - e.g., unusual values like 999999
-        4. First 3 items (context)
-        5. Last 2 items (context)
-        6. Other important items by index order
+        4. ALL items with important values (learned) - TOIN field semantics
+        5. First 3 items (context)
+        6. Last 2 items (context)
+        7. Other important items by index order
 
-        Uses BOTH keyword detection (for preservation guarantee) AND statistical detection.
+        Uses BOTH keyword detection (for preservation guarantee) AND statistical detection,
+        PLUS learned field semantics from TOIN for zero-latency signal detection.
 
         HIGH FIX: Note that this function may return MORE items than effective_max
         when critical items (errors, outliers, anomalies) exceed the limit. This is
@@ -1508,6 +1587,7 @@ class SmartCrusher(Transform):
             n: Total number of items.
             analysis: Optional analysis results for anomaly detection.
             max_items: Thread-safe max items limit (defaults to config value).
+            field_semantics: Optional learned field semantics from TOIN.
 
         Returns:
             Set of indices to keep (may exceed max_items if critical items require it).
@@ -1518,6 +1598,9 @@ class SmartCrusher(Transform):
         if len(keep_indices) <= effective_max:
             return keep_indices
 
+        # Use provided field_semantics or fall back to instance variable (set by crush())
+        effective_field_semantics = field_semantics or getattr(self, "_current_field_semantics", None)
+
         # Identify error items using KEYWORD detection (preservation guarantee)
         # This ensures ALL error items are kept, regardless of frequency
         error_indices = set(_detect_error_items_for_preservation(items))
@@ -1540,22 +1623,31 @@ class SmartCrusher(Transform):
                                 if abs(val - stats.mean_val) > threshold:
                                     anomaly_indices.add(i)
 
+        # === TOIN Evolution: Identify items with important values (learned) ===
+        # Uses learned field semantics for zero-latency signal detection
+        learned_important_indices: set[int] = set()
+        if effective_field_semantics:
+            learned_important_indices = set(
+                _detect_items_by_learned_semantics(items, effective_field_semantics)
+            )
+
         # Start with all critical items (these are non-negotiable)
         # Error items are ALWAYS preserved (quality guarantee)
-        prioritized = error_indices | outlier_indices | anomaly_indices
+        prioritized = error_indices | outlier_indices | anomaly_indices | learned_important_indices
 
         # HIGH FIX: Log warning if critical items alone exceed the limit
         # This helps diagnose why compression may be less effective than expected
         critical_count = len(prioritized)
         if critical_count > effective_max:
             logger.warning(
-                "Critical items (%d) exceed max_items (%d): errors=%d outliers=%d anomalies=%d. "
+                "Critical items (%d) exceed max_items (%d): errors=%d outliers=%d anomalies=%d learned=%d. "
                 "Quality guarantee takes precedence - keeping all critical items.",
                 critical_count,
                 effective_max,
                 len(error_indices),
                 len(outlier_indices),
                 len(anomaly_indices),
+                len(learned_important_indices),
             )
 
         # Add first/last items if we have room
@@ -1929,6 +2021,11 @@ class SmartCrusher(Transform):
             if toin_hint.compression_level != "moderate":
                 toin_compression_level = toin_hint.compression_level
 
+        # === TOIN Evolution: Extract field semantics for signal detection ===
+        # Store temporarily on instance for use in _prioritize_indices
+        # This enables learned signal detection without changing all method signatures
+        self._current_field_semantics = toin_hint.field_semantics if toin_hint.field_semantics else None
+
         # Local feedback hints (if TOIN didn't apply)
         if not toin_hint_applied and self.config.use_feedback_hints and tool_name:
             feedback = self._get_feedback()
@@ -2059,6 +2156,7 @@ class SmartCrusher(Transform):
                     compressed_tokens=compressed_tokens,
                     strategy=analysis.recommended_strategy.value,
                     query_context=query_context,
+                    items=items,  # Pass items for field-level semantic learning
                 )
             except Exception:
                 # TOIN should never break compression
@@ -2075,9 +2173,13 @@ class SmartCrusher(Transform):
             elif hints_applied:
                 strategy_info += f"(feedback:{effective_max_items})"
 
+            # Clean up temporary instance variable
+            self._current_field_semantics = None
             return result, strategy_info, ccr_hash
 
         except Exception:
+            # Clean up temporary instance variable
+            self._current_field_semantics = None
             # Re-raise any exceptions (removed finally block since we no longer mutate config)
             raise
 

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "hatchling.build"
 
 [project]
 name = "headroom-ai"
-version = "0.2.10"
+version = "0.2.11"
 description = "The Context Optimization Layer for LLM Applications - Cut costs by 50-90%"
 readme = "README.md"
 license = "Apache-2.0"

tests/test_cache/test_client_integration.py

@@ -2,7 +2,8 @@
 
 import os
 import tempfile
-from unittest.mock import MagicMock
+from dataclasses import dataclass
+from unittest.mock import MagicMock, patch
 
 import pytest
 
@@ -10,6 +11,7 @@ from headroom import (
     AnthropicCacheOptimizer,
     HeadroomClient,
 )
+from headroom.cache.base import CacheMetrics, CacheResult
 
 
 @pytest.fixture
@@ -25,20 +27,30 @@ def temp_db():
 class MockTokenCounter:
     """Mock token counter for testing."""
 
-    def count_tokens(self, text: str) -> int:
+    def count_text(self, text: str) -> int:
+        """Count tokens in text (required by Tokenizer interface)."""
         return len(text) // 4
 
+    def count_tokens(self, text: str) -> int:
+        """Alias for count_text."""
+        return self.count_text(text)
+
+    def count_message(self, message: dict) -> int:
+        """Count tokens in a single message."""
+        content = message.get("content", "")
+        if isinstance(content, str):
+            return len(content) // 4
+        elif isinstance(content, list):
+            total = 0
+            for block in content:
+                if isinstance(block, dict):
+                    total += len(block.get("text", "")) // 4
+            return total
+        return 0
+
     def count_messages(self, messages: list) -> int:
-        total = 0
-        for msg in messages:
-            content = msg.get("content", "")
-            if isinstance(content, str):
-                total += len(content) // 4
-            elif isinstance(content, list):
-                for block in content:
-                    if isinstance(block, dict):
-                        total += len(block.get("text", "")) // 4
-        return total
+        """Count tokens in messages."""
+        return sum(self.count_message(msg) for msg in messages)
 
 
 class MockAnthropicProvider:
@@ -65,6 +77,42 @@ class MockOpenAIProvider:
         return 128000
 
 
+# Mock response classes for testing (avoid MagicMock in sqlite)
+
+
+@dataclass
+class MockTextBlock:
+    """Mock text block for Anthropic response."""
+
+    type: str = "text"
+    text: str = "Hello!"
+
+
+@dataclass
+class MockUsage:
+    """Mock usage for Anthropic response."""
+
+    input_tokens: int = 100
+    output_tokens: int = 20
+
+
+@dataclass
+class MockAnthropicResponse:
+    """Mock Anthropic API response."""
+
+    content: list = None
+    usage: MockUsage = None
+    model: str = "claude-sonnet-4-20250514"
+    id: str = "msg_123"
+    stop_reason: str = "end_turn"
+
+    def __post_init__(self):
+        if self.content is None:
+            self.content = [MockTextBlock()]
+        if self.usage is None:
+            self.usage = MockUsage()
+
+
 class TestHeadroomClientCacheIntegration:
     """Test HeadroomClient cache optimizer integration."""
 
@@ -219,3 +267,375 @@ class TestHeadroomClientCacheIntegration:
 
         assert client._config.cache_optimizer.enabled is True
         assert client._config.cache_optimizer.enable_semantic_cache is True
+
+
+class TestCacheOptimizerInvocation:
+    """Test that cache optimizer is actually INVOKED during chat completion.
+
+    These tests catch bugs where the optimizer is assigned but never called
+    in the production code path.
+    """
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_optimizer_optimize_is_called_during_chat(self, mock_save, temp_db):
+        """CRITICAL: Verify optimizer.optimize() is called during chat completion.
+
+        This test catches the gap where tests verify assignment but not invocation.
+        Note: Cache optimizer is only invoked in OPTIMIZE mode, not AUDIT mode (the default).
+        """
+        from headroom import HeadroomMode
+
+        # Use module-level mock classes to avoid sqlite issues with MagicMock
+        mock_client = MagicMock()
+        mock_client.messages.create.return_value = MockAnthropicResponse()
+
+        provider = MockAnthropicProvider()
+
+        # Create a spy optimizer to track calls
+        real_optimizer = AnthropicCacheOptimizer()
+        spy_optimize = MagicMock(
+            return_value=CacheResult(
+                messages=[{"role": "user", "content": "test"}],
+                metrics=CacheMetrics(
+                    cacheable_tokens=100,
+                    breakpoints_inserted=1,
+                    estimated_cache_hit=False,
+                    estimated_savings_percent=0.0,
+                ),
+                transforms_applied=["test_transform"],
+            )
+        )
+        real_optimizer.optimize = spy_optimize
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+            cache_optimizer=real_optimizer,
+        )
+
+        # Make a chat completion call in OPTIMIZE mode (cache optimizer only runs in OPTIMIZE mode)
+        messages = [
+            {"role": "user", "content": "Hello, how are you?"},
+        ]
+
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.OPTIMIZE,
+        )
+
+        # CRITICAL: Verify optimizer.optimize() was actually called
+        assert spy_optimize.called, (
+            "Cache optimizer.optimize() should be called during chat completion. "
+            "If this fails, the optimizer is assigned but never invoked."
+        )
+
+        # Verify it was called with the right arguments
+        call_args = spy_optimize.call_args
+        assert call_args is not None
+        optimized_messages, context = call_args[0]
+        assert len(optimized_messages) >= 1, "Should pass messages to optimizer"
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_optimizer_transforms_applied_in_response(self, mock_save, temp_db):
+        """Verify optimizer transforms are reported in the response metadata."""
+        from headroom import HeadroomMode
+
+        # Use module-level mock classes to avoid sqlite issues with MagicMock
+        mock_client = MagicMock()
+        mock_client.messages.create.return_value = MockAnthropicResponse()
+
+        provider = MockAnthropicProvider()
+
+        # Create optimizer that applies a transform
+        real_optimizer = AnthropicCacheOptimizer()
+        real_optimizer.optimize = MagicMock(
+            return_value=CacheResult(
+                messages=[{"role": "user", "content": "test"}],
+                metrics=CacheMetrics(
+                    cacheable_tokens=500,
+                    breakpoints_inserted=2,
+                    estimated_cache_hit=True,
+                    estimated_savings_percent=0.5,
+                ),
+                transforms_applied=["add_cache_control"],
+            )
+        )
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+            cache_optimizer=real_optimizer,
+        )
+
+        messages = [
+            {"role": "user", "content": "x" * 1000},  # Large message
+        ]
+
+        # Use OPTIMIZE mode so cache optimizer is invoked
+        result = client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.OPTIMIZE,
+        )
+
+        # Verify the response includes cache optimizer info
+        assert hasattr(result, "headroom"), "Response should have headroom metadata"
+        headroom_meta = result.headroom
+
+        # Check that cache optimizer was reported
+        assert headroom_meta.cache_optimizer_used is not None or \
+            any("cache_optimizer" in t for t in (headroom_meta.transforms_applied or [])), \
+            "Cache optimizer usage should be reported in metadata"
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_optimizer_not_called_in_audit_mode(self, mock_save, temp_db):
+        """Verify optimizer is NOT called in AUDIT mode (observe only)."""
+        from headroom import HeadroomMode
+
+        # Use module-level mock classes to avoid sqlite issues with MagicMock
+        mock_client = MagicMock()
+        mock_client.messages.create.return_value = MockAnthropicResponse()
+
+        provider = MockAnthropicProvider()
+
+        spy_optimize = MagicMock(
+            return_value=CacheResult(
+                messages=[{"role": "user", "content": "test"}],
+                metrics=CacheMetrics(),
+            )
+        )
+        real_optimizer = AnthropicCacheOptimizer()
+        real_optimizer.optimize = spy_optimize
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+            cache_optimizer=real_optimizer,
+        )
+
+        messages = [{"role": "user", "content": "Hello"}]
+
+        # Make call in AUDIT mode (observe only, no modifications)
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.AUDIT,
+        )
+
+        # Optimizer should NOT be called in AUDIT mode
+        assert not spy_optimize.called, (
+            "Cache optimizer should NOT be called in AUDIT mode"
+        )
+
+
+class TestSemanticCacheIntegration:
+    """Test semantic cache integration with HeadroomClient.
+
+    These tests verify the full production code path for semantic caching,
+    including that cache hits actually return cached responses without calling
+    the underlying API.
+    """
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_semantic_cache_hit_returns_cached_response_without_api_call(
+        self, mock_save, temp_db
+    ):
+        """CRITICAL: Verify semantic cache hit returns cached response without API call.
+
+        This test catches the gap where semantic cache is enabled but cached
+        responses are never actually returned (API is always called).
+        """
+        from headroom import HeadroomMode
+
+        # Mock OpenAI-style response (chat.completions.create uses OpenAI API style)
+        mock_client = MagicMock()
+        mock_openai_response = MagicMock()
+        mock_openai_response.choices = [MagicMock(message=MagicMock(content="4"))]
+        mock_openai_response.usage = MagicMock(
+            prompt_tokens=10, completion_tokens=5, total_tokens=15
+        )
+        mock_openai_response.model = "claude-sonnet-4-20250514"
+        mock_openai_response.id = "chatcmpl-123"
+        mock_client.chat.completions.create.return_value = mock_openai_response
+
+        provider = MockAnthropicProvider()
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+            enable_cache_optimizer=True,
+            enable_semantic_cache=True,
+        )
+
+        messages = [
+            {"role": "system", "content": "You are helpful."},
+            {"role": "user", "content": "What is 2+2?"},
+        ]
+
+        # First call - should call API and potentially cache
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.OPTIMIZE,
+        )
+
+        first_call_count = mock_client.chat.completions.create.call_count
+        assert first_call_count == 1, "First call should hit API"
+
+        # Manually store response in semantic cache for test
+        if client._semantic_cache_layer is not None:
+            from headroom.cache import OptimizationContext
+
+            context = OptimizationContext(
+                provider="anthropic",
+                model="claude-sonnet-4-20250514",
+                query="What is 2+2?",
+            )
+            client._semantic_cache_layer.store_response(
+                messages,
+                {"text": "4", "role": "assistant"},
+                context,
+            )
+
+        # Second call with same messages - should hit cache, NOT call API
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.OPTIMIZE,
+        )
+
+        second_call_count = mock_client.chat.completions.create.call_count
+
+        # If semantic cache is working, API should NOT be called again
+        assert second_call_count == 1, (
+            f"Semantic cache hit should NOT call API. "
+            f"Expected 1 API call, got {second_call_count}. "
+            "If this fails, cached responses are not being returned."
+        )
+
+
+class TestSessionStatsTracking:
+    """Test session statistics tracking in HeadroomClient.
+
+    These tests verify that session stats are actually updated during
+    chat completion calls.
+    """
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_session_stats_incremented_after_request(self, mock_save, temp_db):
+        """CRITICAL: Verify session stats are incremented after requests."""
+        from headroom import HeadroomMode
+
+        mock_client = MagicMock()
+        mock_client.messages.create.return_value = MockAnthropicResponse()
+
+        provider = MockAnthropicProvider()
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+        )
+
+        # Get initial stats
+        initial_stats = client.get_stats()
+        initial_requests = initial_stats["session"]["requests_total"]
+
+        # Make a request in AUDIT mode
+        messages = [{"role": "user", "content": "Hello"}]
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.AUDIT,
+        )
+
+        # Verify stats were updated
+        after_stats = client.get_stats()
+        after_requests = after_stats["session"]["requests_total"]
+
+        assert after_requests == initial_requests + 1, (
+            f"requests_total should increment. "
+            f"Before: {initial_requests}, After: {after_requests}"
+        )
+        assert after_stats["session"]["requests_audit"] >= 1, (
+            "requests_audit should be at least 1 after AUDIT mode request"
+        )
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_session_stats_tracks_optimize_mode(self, mock_save, temp_db):
+        """Verify session stats track OPTIMIZE mode requests separately."""
+        from headroom import HeadroomMode
+
+        mock_client = MagicMock()
+        mock_client.messages.create.return_value = MockAnthropicResponse()
+
+        provider = MockAnthropicProvider()
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+        )
+
+        messages = [{"role": "user", "content": "Hello"}]
+
+        # Make request in OPTIMIZE mode
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.OPTIMIZE,
+        )
+
+        stats = client.get_stats()
+
+        assert stats["session"]["requests_optimized"] >= 1, (
+            "requests_optimized should be at least 1 after OPTIMIZE mode request"
+        )
+
+    @patch("headroom.storage.sqlite.SQLiteStorage.save")
+    def test_session_stats_tracks_tokens_saved(self, mock_save, temp_db):
+        """Verify session stats track tokens saved."""
+        from headroom import HeadroomMode
+
+        mock_client = MagicMock()
+        mock_client.messages.create.return_value = MockAnthropicResponse()
+
+        provider = MockAnthropicProvider()
+
+        client = HeadroomClient(
+            original_client=mock_client,
+            provider=provider,
+            store_url=temp_db,
+        )
+
+        # Create a conversation that will trigger some optimization
+        messages = [
+            {"role": "system", "content": "You are helpful. " * 100},
+            {"role": "user", "content": "Hello"},
+        ]
+
+        client.chat.completions.create(
+            model="claude-sonnet-4-20250514",
+            messages=messages,
+            max_tokens=100,
+            headroom_mode=HeadroomMode.OPTIMIZE,
+        )
+
+        stats = client.get_stats()
+
+        # tokens_saved_total should be tracked (may be 0 if no compression)
+        assert "tokens_saved_total" in stats["session"], (
+            "Session stats should track tokens_saved_total"
+        )

tests/test_proxy_ccr.py

@@ -331,3 +331,322 @@ class TestCCREdgeCases:
         retrieved = json.loads(data["original_content"])
         assert retrieved[0]["text"] == "日本語テキスト"
         assert "🎉" in retrieved[1]["text"]
+
+
+class TestEndToEndTOINIntegration:
+    """End-to-end tests verifying the production path from proxy → TOIN.
+
+    These tests verify that:
+    1. SmartCrusher compresses tool outputs when called through the proxy pipeline
+    2. TOIN records compression events
+    3. Retrieval events update TOIN field semantics
+    4. The full feedback loop works
+
+    This catches bugs where components are wired correctly but don't communicate
+    (e.g., compression_store not passing retrieved_items to TOIN).
+    """
+
+    @pytest.fixture
+    def fresh_toin(self):
+        """Create a fresh TOIN instance."""
+        import tempfile
+        from pathlib import Path
+
+        from headroom.telemetry.toin import (
+            TOINConfig,
+            get_toin,
+            reset_toin,
+        )
+
+        reset_toin()
+        with tempfile.TemporaryDirectory() as tmpdir:
+            storage_path = str(Path(tmpdir) / "toin.json")
+            toin = get_toin(
+                TOINConfig(
+                    storage_path=storage_path,
+                    auto_save_interval=0,
+                )
+            )
+            yield toin
+            reset_toin()
+
+    @pytest.fixture
+    def client_with_optimization(self, fresh_toin):
+        """Create test client with optimization enabled."""
+        reset_compression_store()
+        config = ProxyConfig(
+            optimize=True,  # Enable optimization
+            smart_routing=False,  # Use legacy mode for simpler testing
+            cache_enabled=False,
+            rate_limit_enabled=False,
+            cost_tracking_enabled=False,
+        )
+        app = create_app(config)
+        with TestClient(app) as client:
+            yield client
+        reset_compression_store()
+
+    def test_pipeline_compresses_tool_output_and_records_toin(self, fresh_toin, client_with_optimization):
+        """CRITICAL: Verify SmartCrusher compression records events in TOIN.
+
+        This tests the production code path:
+        1. Tool output comes in through proxy
+        2. SmartCrusher compresses it
+        3. TOIN records the compression event
+        """
+        from headroom.config import CCRConfig, SmartCrusherConfig
+        from headroom.providers import AnthropicProvider
+        from headroom.telemetry import ToolSignature
+        from headroom.transforms import SmartCrusher, TransformPipeline
+
+        # Create tool output with 100 items that will trigger compression
+        # Key: score field with varying values signals sortable data
+        # Having repetitive category values helps trigger compression
+        items = [
+            {
+                "id": i,
+                "score": 1000 - i,  # Decreasing scores signal sorting
+                "category": f"cat_{i % 3}",  # Only 3 unique categories
+                "status": "active" if i % 2 == 0 else "inactive",  # Binary status
+            }
+            for i in range(100)
+        ]
+        tool_output = json.dumps(items)
+
+        # Create messages with tool_result containing our data
+        messages = [
+            {"role": "user", "content": "Search for items"},
+            {
+                "role": "assistant",
+                "content": [
+                    {
+                        "type": "tool_use",
+                        "id": "tool_123",
+                        "name": "search_api",
+                        "input": {"query": "test"},
+                    }
+                ],
+            },
+            {
+                "role": "user",
+                "content": [
+                    {
+                        "type": "tool_result",
+                        "tool_use_id": "tool_123",
+                        "content": tool_output,
+                    }
+                ],
+            },
+        ]
+
+        # Create pipeline with SmartCrusher (same as proxy does)
+        pipeline = TransformPipeline(
+            transforms=[
+                SmartCrusher(
+                    SmartCrusherConfig(
+                        enabled=True,
+                        min_tokens_to_crush=100,
+                        max_items_after_crush=15,
+                    ),
+                    ccr_config=CCRConfig(
+                        enabled=True,
+                        inject_retrieval_marker=True,
+                        min_items_to_cache=10,
+                    ),
+                ),
+            ],
+            provider=AnthropicProvider(),
+        )
+
+        # Apply pipeline (this is what the proxy does)
+        result = pipeline.apply(
+            messages=messages,
+            model="claude-sonnet-4-20250514",
+            model_limit=200000,
+        )
+
+        # Verify SmartCrusher was invoked (transform name starts with smart_crush)
+        smart_crush_applied = any(
+            t.startswith("smart_crush") or t.startswith("smart:")
+            for t in result.transforms_applied
+        )
+        assert smart_crush_applied, (
+            f"SmartCrusher should be in transforms: {result.transforms_applied}"
+        )
+
+        # Check if compression was actually performed (not skipped)
+        # Skip messages look like "smart:skip:reason(100->100)"
+        compression_was_skipped = any(
+            "skip" in t.lower() for t in result.transforms_applied if "smart:" in t.lower()
+        )
+
+        # If compression happened, verify TOIN and store
+        if not compression_was_skipped:
+            # Verify compression store has the entry
+            store = get_compression_store()
+            stats = store.get_stats()
+            assert stats["entry_count"] >= 1, "Should have cached entry"
+
+            # Verify TOIN recorded the compression
+            signature = ToolSignature.from_items(items)
+            pattern = fresh_toin._patterns.get(signature.structure_hash)
+            assert pattern is not None, (
+                "TOIN should have recorded compression event. "
+                "If this fails, SmartCrusher is not calling TOIN.record_compression."
+            )
+            assert pattern.total_compressions >= 1, "Should have at least 1 compression"
+        else:
+            # Compression was skipped - this is expected for some data patterns
+            # The important thing is that SmartCrusher was invoked and made a decision
+            # The other tests verify the full loop when compression does happen
+            pass
+
+    def test_retrieval_through_proxy_updates_toin_field_semantics(
+        self, fresh_toin, client_with_optimization
+    ):
+        """CRITICAL: Verify retrieval through proxy updates TOIN field semantics.
+
+        This tests the full feedback loop:
+        1. Store compressed content (simulating prior compression)
+        2. Retrieve through proxy endpoint
+        3. Verify TOIN learned field semantics from retrieved items
+        """
+        from headroom.telemetry import ToolSignature
+
+        # Create items with distinctive field types
+        items = [
+            {
+                "id": i,
+                "error_code": 500 if i % 10 == 0 else 200,
+                "timestamp": f"2024-01-{i:02d}T00:00:00Z",
+                "message": f"Log entry {i}",
+            }
+            for i in range(50)
+        ]
+        original_content = json.dumps(items)
+        compressed_content = json.dumps(items[:10])
+
+        # Get the signature hash
+        signature = ToolSignature.from_items(items)
+
+        # Store in compression store with correct metadata
+        store = get_compression_store()
+        hash_key = store.store(
+            original=original_content,
+            compressed=compressed_content,
+            original_item_count=50,
+            compressed_item_count=10,
+            tool_name="logs_api",
+            tool_signature_hash=signature.structure_hash,
+            compression_strategy="smart_sample",
+        )
+
+        # Pre-record some compressions in TOIN (needed for pattern to exist)
+        for _ in range(3):
+            fresh_toin.record_compression(
+                tool_signature=signature,
+                original_count=50,
+                compressed_count=10,
+                original_tokens=5000,
+                compressed_tokens=1000,
+                strategy="smart_sample",
+            )
+
+        # Retrieve through proxy endpoint
+        response = client_with_optimization.post("/v1/retrieve", json={"hash": hash_key})
+        assert response.status_code == 200
+
+        # Process pending feedback (this is what triggers TOIN learning)
+        # Note: get_compression_store is already imported at module level
+        store = get_compression_store()
+        store.process_pending_feedback()
+
+        # Verify TOIN learned field semantics
+        pattern = fresh_toin._patterns.get(signature.structure_hash)
+        assert pattern is not None, "Pattern should exist after compression and retrieval"
+
+        # CRITICAL ASSERTION: This catches the bug where compression_store
+        # wasn't passing retrieved_items to TOIN
+        assert len(pattern.field_semantics) > 0, (
+            "TOIN should have learned field semantics from retrieved items. "
+            "If this fails, the production code path "
+            "(CompressionStore.process_pending_feedback -> TOIN.record_retrieval) "
+            "is not passing retrieved_items."
+        )
+
+        # Verify specific field types were learned
+        field_names = list(pattern.field_semantics.keys())
+        assert len(field_names) > 0, "Should have learned at least one field"
+
+    def test_full_proxy_ccr_feedback_loop(self, fresh_toin, client_with_optimization):
+        """CRITICAL: Test the complete CCR feedback loop through proxy.
+
+        This is the most important integration test - it verifies:
+        1. Compression happens and TOIN records it
+        2. Retrieval happens and TOIN learns from it
+        3. Future recommendations reflect the learning
+        """
+        from headroom.telemetry import ToolSignature
+
+        # Create items for the full feedback loop test
+        items = [
+            {
+                "id": i,
+                "score": 1000 - i,
+                "category": f"cat_{i % 5}",
+                "status": "active" if i % 2 == 0 else "inactive",
+            }
+            for i in range(100)
+        ]
+        signature = ToolSignature.from_items(items)
+
+        # Store content directly (simulating what SmartCrusher does)
+        # This ensures we have entries regardless of whether compression was triggered
+        store = get_compression_store()
+        hash_key = store.store(
+            original=json.dumps(items),
+            compressed=json.dumps(items[:15]),
+            original_item_count=100,
+            compressed_item_count=15,
+            tool_name="search_api",
+            tool_signature_hash=signature.structure_hash,
+            compression_strategy="smart_sample",
+        )
+
+        # Record compressions in TOIN (simulating what SmartCrusher does)
+        for _ in range(3):
+            fresh_toin.record_compression(
+                tool_signature=signature,
+                original_count=100,
+                compressed_count=15,
+                original_tokens=5000,
+                compressed_tokens=1000,
+                strategy="smart_sample",
+            )
+
+        # Step 2: Retrieve through proxy endpoint
+        response = client_with_optimization.post(
+            "/v1/retrieve",
+            json={"hash": hash_key, "query": "category:cat_1"},
+        )
+        assert response.status_code == 200
+
+        # Process feedback (this triggers TOIN learning)
+        store.process_pending_feedback()
+
+        # Step 3: Verify TOIN learned
+        pattern = fresh_toin._patterns.get(signature.structure_hash)
+        assert pattern is not None, "Pattern should exist"
+        assert pattern.total_compressions >= 1, "Should have compression count"
+        assert pattern.total_retrievals >= 1, "Should have retrieval count"
+
+        # Step 4: Verify field semantics were learned
+        assert len(pattern.field_semantics) > 0, (
+            "TOIN should learn field semantics through the full proxy CCR loop. "
+            "This is the ultimate integration test - if this fails, "
+            "the production feedback loop is broken."
+        )
+
+        # Step 5: Get recommendation (verifies learning is usable)
+        recommendation = fresh_toin.get_recommendation(signature, "find category")
+        assert recommendation.confidence >= 0, "Recommendation should have confidence"

tests/test_toin_field_learning.py

@@ -0,0 +1,734 @@
+"""Tests for TOIN Field-Level Learning.
+
+These tests PROVE that field-level learning actually works:
+1. FieldSemantics correctly infers types from retrieval patterns
+2. TOIN populates field_semantics from retrievals
+3. SmartCrusher uses learned semantics to detect important items
+4. End-to-end: important items are preserved based on learned behavior
+
+No hardcoded patterns - all learning is behavior-based.
+"""
+
+import hashlib
+
+import pytest
+
+from headroom.telemetry import (
+    ToolIntelligenceNetwork,
+    ToolPattern,
+    ToolSignature,
+    reset_toin,
+)
+from headroom.telemetry.models import FieldSemantics
+
+
+@pytest.fixture(autouse=True)
+def reset_globals():
+    """Reset global state before each test."""
+    reset_toin()
+    yield
+    reset_toin()
+
+
+def _hash_value(value) -> str:
+    """Hash a value the same way TOIN does."""
+    value_str = str(value)
+    return hashlib.sha256(value_str.encode()).hexdigest()[:8]
+
+
+def _hash_field(field_name: str) -> str:
+    """Hash a field name the same way TOIN does."""
+    return hashlib.sha256(field_name.encode()).hexdigest()[:8]
+
+
+class TestFieldSemanticsLearning:
+    """Test that FieldSemantics correctly learns from retrieval patterns."""
+
+    def test_identifier_type_inference(self):
+        """PROVES: Field used with exact-match queries + high uniqueness = identifier."""
+        fs = FieldSemantics(field_hash="test123")
+
+        # Simulate: user retrieves items by unique IDs (exact match queries)
+        # Each ID is different - high uniqueness
+        for i in range(10):
+            fs.record_retrieval_value(_hash_value(f"id_{i}"), operator="=")
+
+        # Simulate compression stats: all values are unique
+        fs.record_compression_stats(
+            unique_values=100,
+            total_values=100,  # uniqueness ratio = 1.0
+            most_common_value_hash=_hash_value("id_0"),
+            most_common_frequency=0.01,  # No dominant value
+        )
+        fs.record_compression_stats(
+            unique_values=100,
+            total_values=100,
+            most_common_value_hash=_hash_value("id_1"),
+            most_common_frequency=0.01,
+        )
+
+        # Now infer the type
+        fs.infer_type()
+
+        # VERIFY: Should be classified as identifier
+        assert fs.inferred_type == "identifier", (
+            f"Expected 'identifier' but got '{fs.inferred_type}'. "
+            "High uniqueness + exact match queries should = identifier"
+        )
+        assert fs.confidence > 0.5, "Should have reasonable confidence"
+
+    def test_error_indicator_type_inference(self):
+        """PROVES: Field with dominant default + retrievals for non-default = error_indicator."""
+        fs = FieldSemantics(field_hash="status_field")
+
+        # Simulate: most items have status="success" (the default)
+        # But user only retrieves items with status="error" or "failed"
+        error_hash = _hash_value("error")
+        failed_hash = _hash_value("failed")
+        success_hash = _hash_value("success")
+
+        # User retrieves "error" and "failed" values (non-default)
+        for _ in range(5):
+            fs.record_retrieval_value(error_hash, operator="=")
+            fs.record_retrieval_value(failed_hash, operator="=")
+
+        # Compression stats: 90% have "success" (the default)
+        fs.record_compression_stats(
+            unique_values=3,  # "success", "error", "failed"
+            total_values=100,
+            most_common_value_hash=success_hash,
+            most_common_frequency=0.9,  # 90% are "success"
+        )
+        fs.record_compression_stats(
+            unique_values=3,
+            total_values=100,
+            most_common_value_hash=success_hash,
+            most_common_frequency=0.9,
+        )
+
+        fs.infer_type()
+
+        # VERIFY: Should be error_indicator
+        assert fs.inferred_type == "error_indicator", (
+            f"Expected 'error_indicator' but got '{fs.inferred_type}'. "
+            "Dominant default + retrieval of non-default = error indicator"
+        )
+        assert fs.default_value_hash == success_hash, "Default should be 'success'"
+        assert fs.confidence > 0.5, "Should have reasonable confidence"
+
+    def test_status_type_inference(self):
+        """PROVES: Low cardinality + specific values retrieved = status."""
+        fs = FieldSemantics(field_hash="state_field")
+
+        # Simulate: field has few unique values (low cardinality)
+        # User retrieves the same few values repeatedly
+        pending_hash = _hash_value("pending")
+        active_hash = _hash_value("active")
+
+        for _ in range(6):
+            fs.record_retrieval_value(pending_hash, operator="=")
+        for _ in range(4):
+            fs.record_retrieval_value(active_hash, operator="=")
+
+        # Compression stats: only 5 unique values across 100 items
+        fs.record_compression_stats(
+            unique_values=5,
+            total_values=100,  # uniqueness ratio = 0.05 (very low)
+            most_common_value_hash=None,
+            most_common_frequency=0.3,  # No overwhelming default
+        )
+        fs.record_compression_stats(
+            unique_values=5,
+            total_values=100,
+            most_common_value_hash=None,
+            most_common_frequency=0.3,
+        )
+
+        fs.infer_type()
+
+        # VERIFY: Should be status
+        assert fs.inferred_type == "status", (
+            f"Expected 'status' but got '{fs.inferred_type}'. "
+            "Low cardinality + specific values retrieved = status"
+        )
+
+    def test_score_type_inference(self):
+        """PROVES: Range queries = score type."""
+        fs = FieldSemantics(field_hash="relevance_field")
+
+        # Simulate: user queries with range operators (top-N behavior)
+        for _ in range(8):
+            fs.record_retrieval_value(_hash_value("0.95"), operator=">")
+        for _ in range(4):
+            fs.record_retrieval_value(_hash_value("0.90"), operator=">=")
+
+        # Compression stats
+        fs.record_compression_stats(
+            unique_values=50,
+            total_values=100,
+            most_common_value_hash=None,
+            most_common_frequency=0.1,
+        )
+        fs.record_compression_stats(
+            unique_values=50,
+            total_values=100,
+            most_common_value_hash=None,
+            most_common_frequency=0.1,
+        )
+
+        fs.infer_type()
+
+        # VERIFY: Should be score
+        assert fs.inferred_type == "score", (
+            f"Expected 'score' but got '{fs.inferred_type}'. "
+            "Range queries (>, >=) should = score type"
+        )
+
+    def test_content_type_inference(self):
+        """PROVES: Contains/text search queries = content type."""
+        fs = FieldSemantics(field_hash="description_field")
+
+        # Simulate: user does text search on this field
+        for i in range(10):
+            fs.record_retrieval_value(_hash_value(f"search_term_{i}"), operator="contains")
+
+        # Compression stats: high uniqueness (different descriptions)
+        fs.record_compression_stats(
+            unique_values=90,
+            total_values=100,
+            most_common_value_hash=None,
+            most_common_frequency=0.05,
+        )
+        fs.record_compression_stats(
+            unique_values=90,
+            total_values=100,
+            most_common_value_hash=None,
+            most_common_frequency=0.05,
+        )
+
+        fs.infer_type()
+
+        # VERIFY: Should be content
+        assert fs.inferred_type == "content", (
+            f"Expected 'content' but got '{fs.inferred_type}'. "
+            "Contains queries should = content type"
+        )
+
+    def test_is_value_important_for_error_indicator(self):
+        """PROVES: For error_indicator, non-default values are important."""
+        fs = FieldSemantics(field_hash="status")
+
+        error_hash = _hash_value("error")
+        success_hash = _hash_value("success")
+
+        # Set up as error_indicator
+        fs.inferred_type = "error_indicator"
+        fs.confidence = 0.8
+        fs.default_value_hash = success_hash
+        fs.important_value_hashes = [error_hash]
+
+        # VERIFY
+        assert fs.is_value_important(error_hash), "Error value should be important"
+        assert not fs.is_value_important(success_hash), "Default value should NOT be important"
+
+    def test_is_value_important_for_status(self):
+        """PROVES: For status fields, retrieved values are important."""
+        fs = FieldSemantics(field_hash="state")
+
+        pending_hash = _hash_value("pending")
+        unknown_hash = _hash_value("never_retrieved")
+
+        # Set up as status
+        fs.inferred_type = "status"
+        fs.confidence = 0.7
+        fs.value_retrieval_frequency = {pending_hash: 5}
+
+        # VERIFY
+        assert fs.is_value_important(pending_hash), "Retrieved value should be important"
+        assert not fs.is_value_important(unknown_hash), "Never-retrieved value should NOT be important"
+
+
+class TestTOINFieldLearningIntegration:
+    """Test that TOIN correctly integrates field-level learning."""
+
+    def test_record_retrieval_populates_field_semantics(self):
+        """PROVES: record_retrieval with items actually populates field_semantics."""
+        toin = ToolIntelligenceNetwork()
+
+        # Create a tool signature
+        items = [
+            {"id": "123", "status": "ok", "value": 100},
+            {"id": "456", "status": "error", "value": 200},
+        ]
+        sig = ToolSignature.from_items(items)
+
+        # Record retrieval with items - THIS IS WHERE LEARNING HAPPENS
+        toin.record_retrieval(
+            tool_signature_hash=sig.structure_hash,
+            retrieval_type="full",
+            query="status=error",
+            query_fields=["status"],
+            retrieved_items=items,
+        )
+
+        # VERIFY: pattern should have field_semantics populated
+        pattern = toin._patterns.get(sig.structure_hash)
+        assert pattern is not None, "Pattern should exist"
+        assert len(pattern.field_semantics) > 0, (
+            f"field_semantics should be populated after retrieval. "
+            f"Got: {pattern.field_semantics}"
+        )
+
+        # Check that field hashes match expected fields
+        expected_field_hashes = {_hash_field(f) for f in ["id", "status", "value"]}
+        actual_field_hashes = set(pattern.field_semantics.keys())
+        assert expected_field_hashes == actual_field_hashes, (
+            f"Expected field hashes {expected_field_hashes}, got {actual_field_hashes}"
+        )
+
+    def test_repeated_retrievals_trigger_type_inference(self):
+        """PROVES: After multiple retrievals, TOIN infers field types."""
+        toin = ToolIntelligenceNetwork()
+
+        # Simulate items with status field
+        items = [
+            {"status": "success"},
+            {"status": "success"},
+            {"status": "success"},
+            {"status": "error"},
+        ]
+        sig = ToolSignature.from_items(items)
+
+        # Simulate 6 retrievals (enough to trigger inference at retrieval 5)
+        for _ in range(6):
+            # User always retrieves items with error status
+            toin.record_retrieval(
+                tool_signature_hash=sig.structure_hash,
+                retrieval_type="full",
+                query="status=error",
+                query_fields=["status"],
+                retrieved_items=[{"status": "error"}],
+            )
+            # Also record compression to get enough data for inference
+            toin.record_compression(
+                tool_signature=sig,
+                original_count=100,
+                compressed_count=10,
+                original_tokens=1000,
+                compressed_tokens=100,
+                strategy="top_n",
+                items=items,
+            )
+
+        # VERIFY: After enough retrievals, type should be inferred
+        pattern = toin._patterns.get(sig.structure_hash)
+        status_hash = _hash_field("status")
+        assert status_hash in pattern.field_semantics, "Status field should be tracked"
+
+        status_sem = pattern.field_semantics[status_hash]
+        # The type should be inferred (not unknown) after enough data
+        assert status_sem.retrieval_count >= 6, f"Should have 6+ retrievals, got {status_sem.retrieval_count}"
+
+    def test_get_recommendation_includes_field_semantics(self):
+        """PROVES: get_recommendation returns learned field_semantics."""
+        toin = ToolIntelligenceNetwork()
+
+        # Set up pattern with learned field_semantics
+        items = [{"id": "123", "status": "ok"}]
+        sig = ToolSignature.from_items(items)
+
+        # Record enough data
+        for i in range(10):
+            toin.record_retrieval(
+                tool_signature_hash=sig.structure_hash,
+                retrieval_type="full",
+                query=f"id={i}",
+                query_fields=["id"],
+                retrieved_items=[{"id": str(i), "status": "ok"}],
+            )
+            toin.record_compression(
+                tool_signature=sig,
+                original_count=100,
+                compressed_count=10,
+                original_tokens=1000,
+                compressed_tokens=100,
+                strategy="top_n",
+                items=items,
+            )
+
+        # Get recommendation
+        hint = toin.get_recommendation(sig)
+
+        # VERIFY: hint should include field_semantics
+        assert hint is not None, "Should get a recommendation"
+        # field_semantics might be empty if confidence is too low,
+        # but the attribute should exist
+        assert hasattr(hint, "field_semantics"), "Hint should have field_semantics attribute"
+
+    def test_field_semantics_persisted_correctly(self):
+        """PROVES: field_semantics survives to_dict/from_dict round-trip."""
+        # ToolPattern is already imported at module level from headroom.telemetry
+
+        # Create pattern with field_semantics
+        pattern = ToolPattern(tool_signature_hash="test123")
+
+        # Add field semantics
+        fs = FieldSemantics(field_hash="field123")
+        fs.inferred_type = "error_indicator"
+        fs.confidence = 0.8
+        fs.important_value_hashes = ["value1", "value2"]
+        fs.default_value_hash = "default"
+        pattern.field_semantics["field123"] = fs
+
+        # Round-trip through dict
+        d = pattern.to_dict()
+        pattern2 = ToolPattern.from_dict(d)
+
+        # VERIFY: field_semantics preserved
+        assert "field123" in pattern2.field_semantics, "field_semantics should be preserved"
+        fs2 = pattern2.field_semantics["field123"]
+        assert fs2.inferred_type == "error_indicator"
+        assert fs2.confidence == 0.8
+        assert fs2.important_value_hashes == ["value1", "value2"]
+
+
+class TestSmartCrusherUsesLearnedSemantics:
+    """Test that SmartCrusher actually uses learned field semantics."""
+
+    def test_detect_items_by_learned_semantics_finds_important_items(self):
+        """PROVES: _detect_items_by_learned_semantics correctly identifies items."""
+        from headroom.transforms.smart_crusher import _detect_items_by_learned_semantics
+
+        # Create field semantics that knows "error" is important
+        status_hash = _hash_field("status")
+        error_hash = _hash_value("error")
+        success_hash = _hash_value("success")
+
+        fs = FieldSemantics(field_hash=status_hash)
+        fs.inferred_type = "error_indicator"
+        fs.confidence = 0.8
+        fs.default_value_hash = success_hash
+        fs.important_value_hashes = [error_hash]
+        fs.value_retrieval_frequency = {error_hash: 10}
+
+        field_semantics = {status_hash: fs}
+
+        # Test items - index 1 has error status
+        items = [
+            {"status": "success", "message": "all good"},
+            {"status": "error", "message": "something failed"},  # <-- This should be detected
+            {"status": "success", "message": "also good"},
+        ]
+
+        # VERIFY
+        important_indices = _detect_items_by_learned_semantics(items, field_semantics)
+        assert 1 in important_indices, (
+            f"Index 1 (error status) should be detected as important. "
+            f"Got indices: {important_indices}"
+        )
+        assert 0 not in important_indices, "Index 0 (success) should not be important"
+        assert 2 not in important_indices, "Index 2 (success) should not be important"
+
+    def test_detect_items_handles_empty_semantics(self):
+        """PROVES: Function handles edge cases gracefully."""
+        from headroom.transforms.smart_crusher import _detect_items_by_learned_semantics
+
+        items = [{"status": "ok"}]
+
+        # Empty semantics
+        assert _detect_items_by_learned_semantics(items, {}) == []
+        assert _detect_items_by_learned_semantics(items, None) == []
+        assert _detect_items_by_learned_semantics([], {"x": FieldSemantics(field_hash="x")}) == []
+
+    def test_detect_items_requires_confidence(self):
+        """PROVES: Low confidence semantics are ignored."""
+        from headroom.transforms.smart_crusher import _detect_items_by_learned_semantics
+
+        status_hash = _hash_field("status")
+        error_hash = _hash_value("error")
+
+        fs = FieldSemantics(field_hash=status_hash)
+        fs.inferred_type = "error_indicator"
+        fs.confidence = 0.1  # TOO LOW
+        fs.important_value_hashes = [error_hash]
+
+        items = [{"status": "error"}]
+
+        # VERIFY: Low confidence = ignored
+        result = _detect_items_by_learned_semantics(items, {status_hash: fs})
+        assert result == [], "Low confidence semantics should be ignored"
+
+
+class TestEndToEndFieldLearning:
+    """End-to-end tests proving the full learning pipeline works."""
+
+    def test_full_pipeline_learns_and_applies(self):
+        """PROVES: End-to-end learning from retrieval to compression."""
+        toin = ToolIntelligenceNetwork()
+
+        # PHASE 1: Learning - User retrieves items with error status
+        # Simulating: "show me all failed items"
+        items_with_errors = [
+            {"id": "1", "status": "success", "data": "..."},
+            {"id": "2", "status": "error", "data": "..."},  # Retrieved
+            {"id": "3", "status": "success", "data": "..."},
+            {"id": "4", "status": "failed", "data": "..."},  # Retrieved
+        ]
+        sig = ToolSignature.from_items(items_with_errors)
+
+        # User keeps retrieving error/failed items (learning behavior)
+        for _ in range(5):
+            toin.record_retrieval(
+                tool_signature_hash=sig.structure_hash,
+                retrieval_type="full",
+                query="status!=success",
+                query_fields=["status"],
+                retrieved_items=[
+                    {"id": "2", "status": "error", "data": "..."},
+                    {"id": "4", "status": "failed", "data": "..."},
+                ],
+            )
+            toin.record_compression(
+                tool_signature=sig,
+                original_count=100,
+                compressed_count=10,
+                original_tokens=1000,
+                compressed_tokens=100,
+                strategy="top_n",
+                items=[
+                    {"id": str(i), "status": "success" if i % 5 != 0 else "error", "data": "..."}
+                    for i in range(100)
+                ],
+            )
+
+        # PHASE 2: Verify learning occurred
+        pattern = toin._patterns.get(sig.structure_hash)
+        assert pattern is not None
+        assert len(pattern.field_semantics) > 0, "Should have learned field semantics"
+
+        # Check status field was learned
+        status_hash = _hash_field("status")
+        if status_hash in pattern.field_semantics:
+            status_sem = pattern.field_semantics[status_hash]
+            # Error value should be tracked
+            error_hash = _hash_value("error")
+            failed_hash = _hash_value("failed")
+            assert error_hash in status_sem.important_value_hashes or \
+                   failed_hash in status_sem.important_value_hashes, \
+                   "Error/failed values should be marked as important"
+
+    def test_recommendation_hint_includes_learned_semantics(self):
+        """PROVES: TOIN recommendation includes learned field semantics for SmartCrusher."""
+        toin = ToolIntelligenceNetwork()
+
+        # Set up sufficient learning
+        items = [{"status": "ok", "id": "123"}]
+        sig = ToolSignature.from_items(items)
+
+        # Create pattern with confident field semantics by directly manipulating internal state
+        # (This is a test - in real code, patterns are created via record_* methods)
+        pattern = ToolPattern(tool_signature_hash=sig.structure_hash)
+        toin._patterns[sig.structure_hash] = pattern
+
+        status_hash = _hash_field("status")
+        fs = FieldSemantics(field_hash=status_hash)
+        fs.inferred_type = "error_indicator"
+        fs.confidence = 0.8  # High confidence
+        fs.retrieval_count = 10
+        fs.important_value_hashes = [_hash_value("error")]
+        pattern.field_semantics[status_hash] = fs
+
+        # Ensure pattern has enough data for recommendation
+        pattern.total_compressions = 10
+        pattern.sample_size = 10  # Required by min_samples_for_recommendation
+        pattern.confidence = 0.5
+
+        # Get recommendation
+        hint = toin.get_recommendation(sig)
+
+        # VERIFY
+        assert hint is not None
+        assert len(hint.field_semantics) > 0, (
+            f"Recommendation should include field_semantics. "
+            f"Got: {hint.field_semantics}"
+        )
+        assert status_hash in hint.field_semantics
+
+
+class TestFieldSemanticsMemoryBounds:
+    """Test that memory bounds are enforced."""
+
+    def test_important_values_bounded(self):
+        """PROVES: important_value_hashes stays within bounds."""
+        fs = FieldSemantics(field_hash="test")
+
+        # Add more values than MAX_IMPORTANT_VALUES
+        for i in range(fs.MAX_IMPORTANT_VALUES + 20):
+            fs.record_retrieval_value(_hash_value(f"value_{i}"))
+
+        # VERIFY: bounded
+        assert len(fs.important_value_hashes) <= fs.MAX_IMPORTANT_VALUES
+
+    def test_value_frequency_bounded(self):
+        """PROVES: value_retrieval_frequency stays within bounds."""
+        fs = FieldSemantics(field_hash="test")
+
+        # Add more values than MAX_VALUE_FREQUENCY_ENTRIES
+        for i in range(fs.MAX_VALUE_FREQUENCY_ENTRIES + 20):
+            fs.record_retrieval_value(_hash_value(f"value_{i}"))
+
+        # VERIFY: bounded
+        assert len(fs.value_retrieval_frequency) <= fs.MAX_VALUE_FREQUENCY_ENTRIES
+
+
+class TestProductionCodePath:
+    """Integration tests for the ACTUAL production code path.
+
+    These tests verify that CompressionStore -> TOIN integration works,
+    not just TOIN in isolation. This is critical because the unit tests
+    can pass while the production integration is broken.
+    """
+
+    def test_compression_store_passes_items_to_toin(self):
+        """PROVES: CompressionStore.process_pending_feedback passes retrieved_items to TOIN.
+
+        This is the integration test that would have caught the original bug
+        where retrieved_items was never passed to TOIN in production.
+        """
+        import json
+
+        from headroom.cache.compression_store import CompressionStore
+        from headroom.telemetry.toin import get_toin, reset_toin
+
+        reset_toin()
+        toin = get_toin()
+
+        # Create a store with feedback enabled
+        store = CompressionStore(max_entries=100, default_ttl=300, enable_feedback=True)
+
+        # Store some compressed content with items that have distinct field values
+        items = [
+            {"id": "123", "status": "success", "value": 100},
+            {"id": "456", "status": "error", "value": 200},
+            {"id": "789", "status": "success", "value": 300},
+        ]
+        compressed_json = json.dumps(items)
+        original_json = json.dumps(items * 10)  # Original was bigger
+
+        # Create a tool signature hash for this structure
+        sig = ToolSignature.from_items(items)
+
+        hash_key = store.store(
+            original=original_json,
+            compressed=compressed_json,
+            original_tokens=1000,
+            compressed_tokens=100,
+            original_item_count=30,
+            compressed_item_count=3,
+            tool_name="test_api",
+            tool_call_id="call_123",
+            tool_signature_hash=sig.structure_hash,
+            compression_strategy="top_n",
+        )
+
+        # Simulate a retrieval (this triggers the feedback loop)
+        store.retrieve(hash_key, query="status=error")
+
+        # Process pending feedback - THIS IS WHERE THE BUG WAS
+        store.process_pending_feedback()
+
+        # VERIFY: TOIN should have received the items and learned from them
+        pattern = toin._patterns.get(sig.structure_hash)
+        assert pattern is not None, "TOIN should have a pattern for this tool"
+
+        # The key assertion: field_semantics should be populated
+        # This would have FAILED before the fix because retrieved_items wasn't passed
+        assert len(pattern.field_semantics) > 0, (
+            "TOIN should have learned field semantics from the retrieved items. "
+            "If this fails, CompressionStore is not passing retrieved_items to TOIN."
+        )
+
+        # Verify specific fields were learned
+        id_hash = _hash_field("id")
+        status_hash = _hash_field("status")
+        value_hash = _hash_field("value")
+
+        learned_fields = set(pattern.field_semantics.keys())
+        expected_fields = {id_hash, status_hash, value_hash}
+        assert expected_fields == learned_fields, (
+            f"Expected fields {expected_fields}, got {learned_fields}"
+        )
+
+    def test_compression_store_handles_wrapped_arrays(self):
+        """PROVES: CompressionStore correctly extracts items from wrapped arrays."""
+        import json
+
+        from headroom.cache.compression_store import CompressionStore
+        from headroom.telemetry.toin import get_toin, reset_toin
+
+        reset_toin()
+        toin = get_toin()
+
+        store = CompressionStore(max_entries=100, default_ttl=300, enable_feedback=True)
+
+        # Content wrapped in {"results": [...]} pattern
+        items = [{"name": "test", "score": 0.95}]
+        wrapped_content = json.dumps({"results": items, "total": 1})
+
+        sig = ToolSignature.from_items(items)
+
+        hash_key = store.store(
+            original=wrapped_content,
+            compressed=wrapped_content,
+            original_tokens=100,
+            compressed_tokens=100,
+            original_item_count=1,
+            compressed_item_count=1,
+            tool_name="search_api",
+            tool_call_id="call_456",
+            tool_signature_hash=sig.structure_hash,
+            compression_strategy="top_n",
+        )
+
+        store.retrieve(hash_key)
+        store.process_pending_feedback()
+
+        # VERIFY: Items were extracted from wrapped structure
+        pattern = toin._patterns.get(sig.structure_hash)
+        assert pattern is not None
+        assert len(pattern.field_semantics) > 0, (
+            "TOIN should extract items from wrapped arrays like {'results': [...]}"
+        )
+
+    def test_compression_store_handles_invalid_json(self):
+        """PROVES: CompressionStore gracefully handles invalid JSON."""
+        from headroom.cache.compression_store import CompressionStore
+        from headroom.telemetry.toin import get_toin, reset_toin
+
+        reset_toin()
+        get_toin()  # Initialize TOIN for feedback loop
+
+        store = CompressionStore(max_entries=100, default_ttl=300, enable_feedback=True)
+
+        # Store invalid JSON content
+        invalid_json = "not valid json {"
+
+        hash_key = store.store(
+            original=invalid_json,
+            compressed=invalid_json,
+            original_tokens=10,
+            compressed_tokens=10,
+            original_item_count=0,
+            compressed_item_count=0,
+            tool_name="broken_api",
+            tool_call_id="call_789",
+            tool_signature_hash="invalid123",
+            compression_strategy="none",
+        )
+
+        # This should not crash
+        store.retrieve(hash_key)
+        store.process_pending_feedback()
+
+        # VERIFY: No crash, pattern may or may not exist but no exception
+        # The main assertion is that we got here without exception

tests/test_toin_integration.py

@@ -170,6 +170,20 @@ class TestTOINIntegration:
             fresh_store.retrieve(entry.hash, query="find all items")
 
         # Step 4: Verify TOIN learned from retrievals
+        pattern = fresh_toin._patterns.get(signature.structure_hash)
+        assert pattern is not None, "Pattern should exist after retrievals"
+
+        # CRITICAL: Verify field-level learning actually happened
+        # This assertion would have caught the bug where compression_store
+        # wasn't passing retrieved_items to TOIN
+        assert len(pattern.field_semantics) > 0, (
+            "TOIN should learn field semantics from retrieved items. "
+            "If this fails, the production code path (CompressionStore -> TOIN) is broken."
+        )
+
+        # Verify retrieval stats were updated
+        assert pattern.total_retrievals >= 1, "Should have recorded retrievals"
+
         recommendation = fresh_toin.get_recommendation(signature, "find all items")
 
         # After many retrievals, TOIN should recommend more items

uv.lock

@@ -849,7 +849,7 @@ wheels = [
 
 [[package]]
 name = "headroom-ai"
-version = "0.2.9"
+version = "0.2.10"
 source = { editable = "." }
 dependencies = [
     { name = "litellm" },