"""Compression Feedback Loop for learning optimal compression strategies. This module analyzes retrieval patterns from the CompressionStore to learn what kinds of compression work well and what doesn't. It provides hints to SmartCrusher to improve compression over time. Key insight from ACON research: Learn compression guidelines by analyzing failures. When compression causes the LLM to retrieve more data, that's a signal that we compressed too aggressively. Features: - Track retrieval rates per tool type - Learn common search queries for each tool - Adjust compression aggressiveness based on patterns - Provide hints: max_items, fields to preserve, etc. Usage: feedback = CompressionFeedback(compression_store) # Get hints before compressing hints = feedback.get_compression_hints("github_search_repos") # hints = {"max_items": 50, "preserve_fields": ["id", "name"], ...} # Apply hints in SmartCrusher config config = SmartCrusherConfig(max_items=hints.get("max_items", 15)) """ from __future__ import annotations import re import threading import time from dataclasses import dataclass, field from typing import TYPE_CHECKING, Any from .compression_strategy_outcomes import CompressionStrategyOutcomes if TYPE_CHECKING: from .compression_store import CompressionStore, RetrievalEvent @dataclass class LocalToolPattern: """Learned patterns for a specific tool type (local feedback). MEDIUM FIX #18: Renamed from ToolPattern to avoid confusion with headroom.telemetry.toin.ToolPattern which serves a different purpose: - LocalToolPattern: Local feedback patterns keyed by tool_name - toin.ToolPattern: Cross-user TOIN patterns keyed by tool_signature_hash """ tool_name: str # Retrieval statistics total_compressions: int = 0 total_retrievals: int = 0 full_retrievals: int = 0 # Retrieved entire original content search_retrievals: int = 0 # Used search within content # Query analysis common_queries: dict[str, int] = field(default_factory=dict) queried_fields: dict[str, int] = field(default_factory=dict) # Strategy analysis - track which strategies work for this tool strategy_compressions: dict[str, int] = field(default_factory=dict) strategy_retrievals: dict[str, int] = field(default_factory=dict) # Signature hash tracking - correlate with TOIN patterns signature_hashes: set[str] = field(default_factory=set) # Timing last_compression: float = 0.0 last_retrieval: float = 0.0 # Calculated metrics @property def retrieval_rate(self) -> float: """Fraction of compressions that resulted in retrieval.""" if self.total_compressions == 0: return 0.0 return self.total_retrievals / self.total_compressions @property def full_retrieval_rate(self) -> float: """Fraction of retrievals that were full (not search).""" if self.total_retrievals == 0: return 0.0 return self.full_retrievals / self.total_retrievals @property def search_rate(self) -> float: """Fraction of retrievals that used search.""" if self.total_retrievals == 0: return 0.0 return self.search_retrievals / self.total_retrievals def strategy_retrieval_rate(self, strategy: str) -> float: """Get retrieval rate for a specific compression strategy.""" return self.strategy_outcomes.retrieval_rate(strategy) def best_strategy(self) -> str | None: """Find the strategy with lowest retrieval rate (most successful).""" return self.strategy_outcomes.best_strategy() @property def strategy_outcomes(self) -> CompressionStrategyOutcomes: """Strategy outcome view backed by this pattern's public counters.""" return CompressionStrategyOutcomes( compressions=self.strategy_compressions, retrievals=self.strategy_retrievals, ) def record_strategy_compression(self, strategy: str) -> None: """Record strategy compression outcome.""" outcomes = self.strategy_outcomes outcomes.record_compression(strategy) self.strategy_compressions = outcomes.compressions self.strategy_retrievals = outcomes.retrievals def record_strategy_retrieval(self, strategy: str) -> None: """Record strategy retrieval outcome.""" outcomes = self.strategy_outcomes outcomes.record_retrieval(strategy) self.strategy_compressions = outcomes.compressions self.strategy_retrievals = outcomes.retrievals @dataclass class CompressionHints: """Hints for optimizing compression of a specific tool's output.""" # Item count hints max_items: int = 15 # Default from SmartCrusher min_items: int = 3 suggested_items: int | None = None # Calculated optimal # Field preservation preserve_fields: list[str] = field(default_factory=list) # Compression aggressiveness (0.0 = aggressive, 1.0 = conservative) aggressiveness: float = 0.7 # Reasoning reason: str = "" # Whether to skip compression entirely skip_compression: bool = False # Recommended compression strategy based on local learning recommended_strategy: str | None = None class CompressionFeedback: """Learn from retrieval patterns to improve compression. This class analyzes retrieval events from CompressionStore and builds tool-specific patterns. These patterns inform compression decisions. Design principles: - High retrieval rate (>50%) → compress less aggressively - Full retrieval dominates → data is unique, skip compression - Search retrieval dominates → keep compressed, add search capability - Frequent queries → preserve fields mentioned in queries """ # Thresholds for adjusting compression HIGH_RETRIEVAL_THRESHOLD = 0.5 # 50% retrieval = too aggressive MEDIUM_RETRIEVAL_THRESHOLD = 0.2 # 20% retrieval = acceptable MIN_SAMPLES_FOR_HINTS = 5 # Need at least 5 events to make recommendations def __init__( self, store: CompressionStore | None = None, enable_learning: bool = True, analysis_interval: float = 60.0, ): """Initialize feedback analyzer. Args: store: CompressionStore to analyze. If None, uses global store. enable_learning: Whether to update patterns from events. analysis_interval: Interval in seconds between re-analyzing store events. """ self._store = store self._enable_learning = enable_learning self._lock = threading.Lock() # Learned patterns per tool self._tool_patterns: dict[str, LocalToolPattern] = {} # Time-based tracking self._last_analysis: float = 0.0 self._analysis_interval: float = analysis_interval self._last_event_timestamp: float = ( 0.0 # Track last processed event to avoid double-counting ) # Global statistics self._total_compressions: int = 0 self._total_retrievals: int = 0 @property def store(self) -> CompressionStore: """Get the compression store (lazy load global if not set).""" if self._store is None: from .compression_store import get_compression_store self._store = get_compression_store() return self._store def record_compression( self, tool_name: str | None, original_count: int, compressed_count: int, strategy: str | None = None, tool_signature_hash: str | None = None, ) -> None: """Record that a compression occurred. Called by SmartCrusher after compressing to track compression events. Args: tool_name: Name of the tool whose output was compressed. original_count: Original item count. compressed_count: Compressed item count. strategy: Compression strategy used (e.g., "SMART_SAMPLE", "TOP_N"). tool_signature_hash: Hash from ToolSignature for correlation with TOIN. """ if not self._enable_learning or not tool_name: return with self._lock: self._total_compressions += 1 if tool_name not in self._tool_patterns: self._tool_patterns[tool_name] = LocalToolPattern(tool_name=tool_name) pattern = self._tool_patterns[tool_name] pattern.total_compressions += 1 pattern.last_compression = time.time() # Track strategy usage if strategy: pattern.record_strategy_compression(strategy) # Track signature hash for TOIN correlation if tool_signature_hash: pattern.signature_hashes.add(tool_signature_hash) # CRITICAL FIX: Use deterministic truncation for signature_hashes # Sort lexicographically to ensure consistent behavior across runs if len(pattern.signature_hashes) > 100: sorted_hashes = sorted(pattern.signature_hashes)[:100] pattern.signature_hashes = set(sorted_hashes) def record_retrieval( self, event: RetrievalEvent, strategy: str | None = None, ) -> None: """Record a retrieval event for pattern learning. Called by CompressionStore when content is retrieved. Args: event: The retrieval event to record. strategy: Compression strategy that was used (for tracking success rates). """ if not self._enable_learning: return tool_name = event.tool_name if not tool_name: return with self._lock: self._total_retrievals += 1 if tool_name not in self._tool_patterns: self._tool_patterns[tool_name] = LocalToolPattern(tool_name=tool_name) pattern = self._tool_patterns[tool_name] pattern.total_retrievals += 1 pattern.last_retrieval = time.time() if event.retrieval_type == "full": pattern.full_retrievals += 1 else: pattern.search_retrievals += 1 # Track strategy retrievals (for success rate calculation) if strategy: pattern.record_strategy_retrieval(strategy) # Track query patterns if event.query: query_lower = event.query.lower() pattern.common_queries[query_lower] = pattern.common_queries.get(query_lower, 0) + 1 # HIGH: Limit common_queries dict to prevent unbounded growth if len(pattern.common_queries) > 100: sorted_queries = sorted( pattern.common_queries.items(), key=lambda x: x[1], reverse=True, )[:100] pattern.common_queries = dict(sorted_queries) # Extract potential field names from query self._extract_field_hints(pattern, event.query) def _truncate_strategy_dicts(self, pattern: LocalToolPattern) -> None: """Truncate strategy counters using the shared strategy outcome domain.""" outcomes = pattern.strategy_outcomes outcomes.prune() pattern.strategy_compressions = outcomes.compressions pattern.strategy_retrievals = outcomes.retrievals def _extract_field_hints(self, pattern: LocalToolPattern, query: str) -> None: """Extract potential field names from search queries. Common patterns: - "field:value" or "field=value" - JSON field names like "status", "error", "id" """ # Look for field:value patterns field_patterns = re.findall(r"(\w+)[=:]", query) for field_name in field_patterns: pattern.queried_fields[field_name] = pattern.queried_fields.get(field_name, 0) + 1 # Look for common JSON field names common_fields = [ "id", "name", "status", "error", "message", "type", "code", "result", "value", "data", "items", "count", ] query_lower = query.lower() for common_field in common_fields: if common_field in query_lower: pattern.queried_fields[common_field] = ( pattern.queried_fields.get(common_field, 0) + 1 ) # HIGH: Limit queried_fields dict to prevent unbounded growth if len(pattern.queried_fields) > 50: sorted_fields = sorted( pattern.queried_fields.items(), key=lambda x: x[1], reverse=True, )[:50] pattern.queried_fields = dict(sorted_fields) def get_compression_hints( self, tool_name: str | None, ) -> CompressionHints: """Get compression hints for a specific tool based on learned patterns. Args: tool_name: Name of the tool to get hints for. Returns: CompressionHints with recommended settings. """ hints = CompressionHints() if not tool_name: hints.reason = "No tool name provided, using defaults" return hints with self._lock: pattern = self._tool_patterns.get(tool_name) if pattern is None: hints.reason = f"No pattern data for {tool_name}, using defaults" return hints # Need minimum samples for reliable hints if pattern.total_compressions < self.MIN_SAMPLES_FOR_HINTS: hints.reason = ( f"Insufficient data ({pattern.total_compressions} samples), " f"need {self.MIN_SAMPLES_FOR_HINTS}" ) return hints # Calculate hints based on retrieval rate retrieval_rate = pattern.retrieval_rate if retrieval_rate > self.HIGH_RETRIEVAL_THRESHOLD: # High retrieval = compress less aggressively if pattern.full_retrieval_rate > 0.8: # Almost all retrievals are full → skip compression hints.skip_compression = True hints.reason = ( f"Very high full retrieval rate ({pattern.full_retrieval_rate:.0%}), " f"recommending skip compression" ) else: # Mix of full and search → increase items hints.max_items = 50 hints.suggested_items = 40 hints.aggressiveness = 0.3 hints.reason = ( f"High retrieval rate ({retrieval_rate:.0%}), " f"recommending less aggressive compression" ) elif retrieval_rate > self.MEDIUM_RETRIEVAL_THRESHOLD: # Medium retrieval = slightly less aggressive hints.max_items = 30 hints.suggested_items = 25 hints.aggressiveness = 0.5 hints.reason = ( f"Medium retrieval rate ({retrieval_rate:.0%}), " f"recommending moderate compression" ) else: # Low retrieval = current compression is working hints.max_items = 15 hints.suggested_items = 10 hints.aggressiveness = 0.7 hints.reason = ( f"Low retrieval rate ({retrieval_rate:.0%}), current compression is effective" ) # Add field preservation hints based on common queries if pattern.queried_fields: # Get top 5 most queried fields sorted_fields = sorted( pattern.queried_fields.items(), key=lambda x: x[1], reverse=True, )[:5] hints.preserve_fields = [f for f, _ in sorted_fields] # Recommend the best strategy based on local retrieval patterns best = pattern.best_strategy() if best: hints.recommended_strategy = best return hints def get_all_patterns(self) -> dict[str, LocalToolPattern]: """Get all learned tool patterns. Returns: Dict mapping tool names to their patterns. HIGH FIX: Returns deep copies to prevent external mutation of internal state. """ import copy as copy_module with self._lock: # Deep copy to prevent external code from modifying internal state return copy_module.deepcopy(self._tool_patterns) def get_stats(self) -> dict[str, Any]: """Get feedback statistics for monitoring. Returns: Dict with feedback statistics. """ with self._lock: return { "total_compressions": self._total_compressions, "total_retrievals": self._total_retrievals, "global_retrieval_rate": ( self._total_retrievals / self._total_compressions if self._total_compressions > 0 else 0.0 ), "tools_tracked": len(self._tool_patterns), "tool_patterns": { name: { "compressions": p.total_compressions, "retrievals": p.total_retrievals, "retrieval_rate": p.retrieval_rate, "full_rate": p.full_retrieval_rate, "search_rate": p.search_rate, "common_queries": list(p.common_queries.keys())[:5], "queried_fields": list(p.queried_fields.keys())[:5], } for name, p in self._tool_patterns.items() }, } def analyze_from_store(self) -> None: """Analyze retrieval events from the store. This pulls recent events from CompressionStore and updates patterns. Useful for catching up after restart or periodic refresh. HIGH FIX: All timestamp reads/writes happen under lock to prevent race conditions where another thread could cause events to be missed or double-counted. """ if not self._enable_learning: return # Rate limit analysis - check under lock for thread safety now = time.time() with self._lock: if now - self._last_analysis < self._analysis_interval: return # Mark that we're starting analysis (prevents concurrent analysis) self._last_analysis = now last_ts = self._last_event_timestamp # Fetch events outside lock (store has its own lock) events = self.store.get_retrieval_events(limit=1000) # Filter events to only process new ones (avoid double-counting) new_events = [e for e in events if e.timestamp > last_ts] if new_events: # Find the maximum timestamp from new events max_timestamp = max(e.timestamp for e in new_events) for event in new_events: self.record_retrieval(event) # Update the timestamp AFTER processing - under lock for atomicity with self._lock: # Only update if our max_timestamp is greater than current # (another thread may have processed newer events) if max_timestamp > self._last_event_timestamp: self._last_event_timestamp = max_timestamp def clear(self) -> None: """Clear all learned patterns. Mainly for testing.""" with self._lock: self._tool_patterns.clear() self._total_compressions = 0 self._total_retrievals = 0 self._last_analysis = 0.0 self._last_event_timestamp = 0.0 # Global feedback instance (lazy initialization) _compression_feedback: CompressionFeedback | None = None _feedback_lock = threading.Lock() def get_compression_feedback() -> CompressionFeedback: """Get the global compression feedback instance. Returns: Global CompressionFeedback instance. """ global _compression_feedback if _compression_feedback is None: with _feedback_lock: if _compression_feedback is None: _compression_feedback = CompressionFeedback() return _compression_feedback def reset_compression_feedback() -> None: """Reset the global compression feedback. Mainly for testing.""" global _compression_feedback with _feedback_lock: if _compression_feedback is not None: _compression_feedback.clear() _compression_feedback = None