"""Smart statistical tool output compression for Headroom SDK. This module provides intelligent JSON compression based on statistical analysis rather than fixed rules. It analyzes data patterns and applies optimal compression strategies to maximize token reduction while preserving important information. SCOPE: SmartCrusher handles JSON arrays only. Non-JSON content (plain text, search results, logs, code, diffs) passes through UNCHANGED. TEXT COMPRESSION IS OPT-IN: For text-based content, Headroom provides standalone utilities that applications can use explicitly: - SearchCompressor: For grep/ripgrep output (file:line:content format) - LogCompressor: For build/test logs (pytest, npm, cargo output) - TextCompressor: For generic plain text with anchor preservation Applications should decide when and how to use text compression based on their specific needs. This design prevents lossy text compression from being applied automatically, which could lose important context in coding tasks. SCHEMA-PRESERVING: Output contains only items from the original array. No wrappers, no generated text, no metadata keys. This ensures downstream tools and parsers work unchanged. Safe V1 Compression Recipe - Always keeps: - First K items (default 3) - Last K items (default 2) - Error items (containing 'error', 'exception', 'failed', 'critical') - Anomalous numeric items (> 2 std from mean) - Items around detected change points - Top-K by score if score field present - Items with high relevance score to user query (via RelevanceScorer) Key Features: - RelevanceScorer: ML-powered or BM25-based relevance matching (replaces regex) - Variance-based change point detection (preserve anomalies) - Error item detection (never lose error messages) - Pattern detection (time series, logs, search results) - Strategy selection based on data characteristics """ from __future__ import annotations import hashlib import json import logging import math import re import statistics import threading from collections import Counter from dataclasses import dataclass, field from enum import Enum from typing import Any from ..cache.compression_feedback import CompressionFeedback, get_compression_feedback 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 ( compute_short_hash, create_tool_digest_marker, deep_copy_messages, safe_json_dumps, safe_json_loads, ) from .base import Transform logger = logging.getLogger(__name__) # Legacy patterns for backwards compatibility (extract_query_anchors) _UUID_PATTERN = re.compile( r"\b[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}\b" ) _NUMERIC_ID_PATTERN = re.compile(r"\b\d{4,}\b") # 4+ digit numbers (likely IDs) _HOSTNAME_PATTERN = re.compile( r"\b[a-zA-Z0-9][-a-zA-Z0-9]*\.[a-zA-Z0-9][-a-zA-Z0-9]*(?:\.[a-zA-Z]{2,})?\b" ) _QUOTED_STRING_PATTERN = re.compile(r"['\"]([^'\"]{1,50})['\"]") # Short quoted strings _EMAIL_PATTERN = re.compile(r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b") def extract_query_anchors(text: str) -> set[str]: """Extract query anchors from user text (legacy regex-based method). DEPRECATED: Use RelevanceScorer.score_batch() for better semantic matching. Query anchors are identifiers or values that the user is likely searching for. When crushing tool outputs, items matching these anchors should be preserved. Extracts: - UUIDs (e.g., "550e8400-e29b-41d4-a716-446655440000") - Numeric IDs (4+ digits, e.g., "12345", "1001234") - Hostnames (e.g., "api.example.com", "server-01.prod") - Quoted strings (e.g., 'Alice', "error_code") - Email addresses (e.g., "user@example.com") Args: text: User message text to extract anchors from. Returns: Set of anchor strings (lowercased for case-insensitive matching). """ anchors: set[str] = set() if not text: return anchors # UUIDs for match in _UUID_PATTERN.findall(text): anchors.add(match.lower()) # Numeric IDs for match in _NUMERIC_ID_PATTERN.findall(text): anchors.add(match) # Hostnames for match in _HOSTNAME_PATTERN.findall(text): # Filter out common false positives if match.lower() not in ("e.g", "i.e", "etc."): anchors.add(match.lower()) # Quoted strings for match in _QUOTED_STRING_PATTERN.findall(text): if len(match.strip()) >= 2: # Skip very short matches anchors.add(match.lower()) # Email addresses for match in _EMAIL_PATTERN.findall(text): anchors.add(match.lower()) return anchors def item_matches_anchors(item: dict, anchors: set[str]) -> bool: """Check if an item matches any query anchors (legacy method). DEPRECATED: Use RelevanceScorer for better matching. Args: item: Dictionary item from tool output. anchors: Set of anchor strings to match. Returns: True if any anchor is found in the item's string representation. """ if not anchors: return False item_str = str(item).lower() return any(anchor in item_str for anchor in anchors) def _hash_field_name(field_name: str) -> str: """Hash a field name to match TOIN's anonymized preserve_fields. TOIN stores field names as SHA256[:8] hashes for privacy. This function produces the same hash format. """ return hashlib.sha256(field_name.encode()).hexdigest()[:8] def _get_preserve_field_values( item: dict, preserve_field_hashes: list[str], ) -> list[tuple[str, Any]]: """Get values from item fields that match TOIN's preserve_field hashes. TOIN stores preserve_fields as hashed field names (SHA256[:8]). This function iterates over item fields, hashes each, and returns matching field names and values. Args: item: Dictionary item from tool output. preserve_field_hashes: List of SHA256[:8] hashed field names from TOIN. Returns: List of (field_name, value) tuples for fields that match. """ if not preserve_field_hashes or not item: return [] # Convert preserve_fields to set for O(1) lookup hash_set = set(preserve_field_hashes) matches = [] for field_name, value in item.items(): field_hash = _hash_field_name(field_name) if field_hash in hash_set: matches.append((field_name, value)) return matches def _item_has_preserve_field_match( item: dict, preserve_field_hashes: list[str], query_context: str, ) -> bool: """Check if item has a preserve_field value that matches query context. Args: item: Dictionary item from tool output. preserve_field_hashes: List of SHA256[:8] hashed field names from TOIN. query_context: User's query to match against field values. Returns: True if any preserve_field value matches the query context. """ if not query_context: return False query_lower = query_context.lower() for _field_name, value in _get_preserve_field_values(item, preserve_field_hashes): if value is not None: value_str = str(value).lower() if value_str in query_lower or query_lower in value_str: return True return False class CompressionStrategy(Enum): """Compression strategies based on data patterns.""" NONE = "none" # No compression needed SKIP = "skip" # Explicitly skip - not safe to crush TIME_SERIES = "time_series" # Keep change points, summarize stable CLUSTER_SAMPLE = "cluster" # Dedupe similar items TOP_N = "top_n" # Keep highest scored items SMART_SAMPLE = "smart_sample" # Statistical sampling with constants # ===================================================================== # STATISTICAL FIELD DETECTION (replaces hardcoded string patterns) # ===================================================================== # Instead of matching field names like "id", "score", "error", we use # statistical and structural properties of the data to detect field types. def _is_uuid_format(value: str) -> bool: """Check if a string looks like a UUID (structural pattern).""" if not isinstance(value, str) or len(value) != 36: return False # UUID format: 8-4-4-4-12 hex chars parts = value.split("-") if len(parts) != 5: return False expected_lens = [8, 4, 4, 4, 12] for part, expected_len in zip(parts, expected_lens): if len(part) != expected_len: return False if not all(c in "0123456789abcdefABCDEF" for c in part): return False return True def _calculate_string_entropy(s: str) -> float: """Calculate Shannon entropy of a string, normalized to [0, 1]. High entropy (>0.7) suggests random/ID-like content. Low entropy (<0.3) suggests repetitive/predictable content. """ if not s or len(s) < 2: return 0.0 # Count character frequencies freq: dict[str, int] = {} for c in s: freq[c] = freq.get(c, 0) + 1 # Calculate entropy import math entropy = 0.0 length = len(s) for count in freq.values(): p = count / length if p > 0: entropy -= p * math.log2(p) # Normalize by max possible entropy for this length max_entropy = math.log2(min(len(freq), length)) if max_entropy > 0: return entropy / max_entropy return 0.0 def _detect_sequential_pattern(values: list[Any], check_order: bool = True) -> bool: """Detect if numeric values form a sequential pattern (like IDs: 1,2,3,...). Returns True if values appear to be auto-incrementing or sequential. Args: values: List of values to check. check_order: If True, also check if values are in ascending order in the array. Score fields are often sorted descending, while IDs are ascending. """ if len(values) < 5: return False # Get numeric values nums = [] for v in values: if isinstance(v, (int, float)) and not isinstance(v, bool): nums.append(v) elif isinstance(v, str): try: nums.append(int(v)) except ValueError: pass if len(nums) < 5: return False # Check if sorted values form a near-sequence sorted_nums = sorted(nums) diffs = [sorted_nums[i + 1] - sorted_nums[i] for i in range(len(sorted_nums) - 1)] if not diffs: return False # If most differences are 1 (or small constant), it's sequential avg_diff = sum(diffs) / len(diffs) if 0.5 <= avg_diff <= 2.0: # Check consistency - sequential IDs have consistent spacing consistent_count = sum(1 for d in diffs if 0.5 <= d <= 2.0) is_sequential = consistent_count / len(diffs) > 0.8 # Additional check: IDs are typically in ASCENDING order in the array # Scores sorted by relevance are typically in DESCENDING order if check_order and is_sequential: # Check if original order is ascending (like IDs) ascending_count = sum(1 for i in range(len(nums) - 1) if nums[i] <= nums[i + 1]) is_ascending = ascending_count / (len(nums) - 1) > 0.7 return is_ascending # Only flag as sequential if ascending (ID-like) return is_sequential return False def _detect_id_field_statistically(stats: FieldStats, values: list[Any]) -> tuple[bool, float]: """Detect if a field is an ID field using statistical properties. Returns (is_id_field, confidence). ID fields have: - Very high uniqueness (>0.95) - Sequential numeric pattern OR UUID format OR high entropy strings """ # Must have high uniqueness if stats.unique_ratio < 0.9: return False, 0.0 confidence = 0.0 # Check for UUID format (structural detection) if stats.field_type == "string": sample_values = [v for v in values[:20] if isinstance(v, str)] uuid_count = sum(1 for v in sample_values if _is_uuid_format(v)) if sample_values and uuid_count / len(sample_values) > 0.8: return True, 0.95 # Check for high entropy (random string IDs) if sample_values: avg_entropy = sum(_calculate_string_entropy(v) for v in sample_values) / len( sample_values ) if avg_entropy > 0.7 and stats.unique_ratio > 0.95: confidence = 0.8 return True, confidence # Check for sequential numeric pattern if stats.field_type == "numeric": if _detect_sequential_pattern(values) and stats.unique_ratio > 0.95: return True, 0.9 # High uniqueness numeric with high range suggests ID if stats.min_val is not None and stats.max_val is not None: value_range = stats.max_val - stats.min_val if value_range > 0 and stats.unique_ratio > 0.95: return True, 0.85 # Very high uniqueness alone is a signal (even without other patterns) if stats.unique_ratio > 0.98: return True, 0.7 return False, 0.0 def _detect_score_field_statistically(stats: FieldStats, items: list[dict]) -> tuple[bool, float]: """Detect if a field is a score/ranking field using statistical properties. Returns (is_score_field, confidence). Score fields have: - Numeric type - Bounded range (0-1, 0-10, 0-100, or similar) - NOT sequential (unlike IDs) - Often the data appears sorted by this field (descending) """ if stats.field_type != "numeric": return False, 0.0 if stats.min_val is None or stats.max_val is None: return False, 0.0 confidence = 0.0 # Check for bounded range typical of scores stats.max_val - stats.min_val min_val, max_val = stats.min_val, stats.max_val # Common score ranges: [0,1], [0,10], [0,100], [-1,1], [0,5] is_bounded = False if 0 <= min_val <= 1 and 0 <= max_val <= 1: # [0,1] range is_bounded = True confidence += 0.4 elif 0 <= min_val <= 10 and 0 <= max_val <= 10: # [0,10] range is_bounded = True confidence += 0.3 elif 0 <= min_val <= 100 and 0 <= max_val <= 100: # [0,100] range is_bounded = True confidence += 0.25 elif -1 <= min_val and max_val <= 1: # [-1,1] range is_bounded = True confidence += 0.35 if not is_bounded: return False, 0.0 # Should NOT be sequential (IDs are sequential, scores are not) sample_values = [item.get(stats.name) for item in items[:50] if stats.name in item] if _detect_sequential_pattern(sample_values): return False, 0.0 # Check if data appears sorted by this field (descending = relevance sorted) # Filter out NaN/Inf which break comparisons values_in_order: list[float] = [] for item in items: if stats.name in item: val = item.get(stats.name) if isinstance(val, (int, float)) and math.isfinite(val): values_in_order.append(float(val)) if len(values_in_order) >= 5: # Check for descending sort descending_count = sum( 1 for i in range(len(values_in_order) - 1) if values_in_order[i] >= values_in_order[i + 1] ) if descending_count / (len(values_in_order) - 1) > 0.7: confidence += 0.3 # Score fields often have floating point values # Filter out NaN/Inf which can't be converted to int float_count = sum( 1 for v in values_in_order[:20] if isinstance(v, float) and math.isfinite(v) and v != int(v) ) if float_count > len(values_in_order[:20]) * 0.3: confidence += 0.1 return confidence >= 0.4, min(confidence, 0.95) def _detect_structural_outliers(items: list[dict]) -> list[int]: """Detect items that are structural outliers (error-like items). Instead of looking for "error" keywords, we detect: 1. Items with extra fields that others don't have 2. Items with rare status/state values 3. Items with significantly different structure Returns indices of outlier items. """ if len(items) < 5: return [] outlier_indices: list[int] = [] # 1. Detect items with extra fields # Find the "common" field set (fields present in >80% of items) field_counts: dict[str, int] = {} for item in items: if isinstance(item, dict): for key in item.keys(): field_counts[key] = field_counts.get(key, 0) + 1 n = len(items) common_fields = {k for k, v in field_counts.items() if v >= n * 0.8} rare_fields = {k for k, v in field_counts.items() if v < n * 0.2} for i, item in enumerate(items): if not isinstance(item, dict): continue item_fields = set(item.keys()) # Has rare fields that most items don't have has_rare = bool(item_fields & rare_fields) if has_rare: outlier_indices.append(i) continue # 2. Detect rare status/state values # Find fields that look like status fields (low cardinality, categorical) status_outliers = _detect_rare_status_values(items, common_fields) outlier_indices.extend(status_outliers) return list(set(outlier_indices)) def _detect_rare_status_values(items: list[dict], common_fields: set[str]) -> list[int]: """Detect items with rare values in status-like fields. A status field has low cardinality (few distinct values). If 95%+ have the same value, items with different values are interesting. """ outlier_indices: list[int] = [] # Find potential status fields (low cardinality) for field_name in common_fields: values = [ item.get(field_name) for item in items if isinstance(item, dict) and field_name in item ] # Skip if too few values or non-hashable try: unique_values = {str(v) for v in values if v is not None} except Exception: continue # Status field = low cardinality (2-10 distinct values) if not (2 <= len(unique_values) <= 10): continue # Count value frequencies value_counts: dict[str, int] = {} for v in values: key = str(v) if v is not None else "__none__" value_counts[key] = value_counts.get(key, 0) + 1 # Find the dominant value if not value_counts: continue max_count = max(value_counts.values()) total = len(values) # If one value dominates (>90%), others are interesting if max_count >= total * 0.9: dominant_value = max(value_counts.keys(), key=lambda k: value_counts[k]) for i, item in enumerate(items): if not isinstance(item, dict) or field_name not in item: continue item_value = str(item[field_name]) if item[field_name] is not None else "__none__" if item_value != dominant_value: outlier_indices.append(i) return outlier_indices # Error keywords for PRESERVATION guarantee (not crushability detection) # This is for the quality guarantee: "ALL error items are ALWAYS preserved" # regardless of how common they are. Used in _prioritize_indices(). _ERROR_KEYWORDS_FOR_PRESERVATION = frozenset( { "error", "exception", "failed", "failure", "critical", "fatal", "crash", "panic", "abort", "timeout", "denied", "rejected", } ) def _detect_error_items_for_preservation(items: list[dict]) -> list[int]: """Detect items containing error keywords for PRESERVATION guarantee. This is NOT for crushability analysis - it's for ensuring ALL error items are retained during compression. The quality guarantee is that error items are NEVER dropped, even if errors are common in the dataset. Uses keywords because error semantics are well-defined across domains. """ error_indices: list[int] = [] for i, item in enumerate(items): if not isinstance(item, dict): continue # Serialize item to check all content try: item_str = json.dumps(item).lower() except Exception: continue # Check if any error keyword is present for keyword in _ERROR_KEYWORDS_FOR_PRESERVATION: if keyword in item_str: error_indices.append(i) break 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. The key insight: if we don't have a reliable SIGNAL to determine which items are important, we should NOT crush at all. Signals include: - Score/rank fields (search results) - Error keywords (logs) - Numeric anomalies (metrics) - Low uniqueness (repetitive data where sampling is representative) High variability + No signal = DON'T CRUSH """ crushable: bool confidence: float # 0.0 to 1.0 reason: str signals_present: list[str] = field(default_factory=list) signals_absent: list[str] = field(default_factory=list) # Detailed metrics has_id_field: bool = False id_uniqueness: float = 0.0 avg_string_uniqueness: float = 0.0 has_score_field: bool = False error_item_count: int = 0 anomaly_count: int = 0 @dataclass class FieldStats: """Statistics for a single field across array items.""" name: str field_type: str # "numeric", "string", "boolean", "object", "array", "null" count: int unique_count: int unique_ratio: float is_constant: bool constant_value: Any = None # Numeric-specific stats min_val: float | None = None max_val: float | None = None mean_val: float | None = None variance: float | None = None change_points: list[int] = field(default_factory=list) # String-specific stats avg_length: float | None = None top_values: list[tuple[str, int]] = field(default_factory=list) @dataclass class ArrayAnalysis: """Complete analysis of an array.""" item_count: int field_stats: dict[str, FieldStats] detected_pattern: str # "time_series", "logs", "search_results", "generic" recommended_strategy: CompressionStrategy constant_fields: dict[str, Any] estimated_reduction: float crushability: CrushabilityAnalysis | None = None # Whether it's safe to crush @dataclass class CompressionPlan: """Plan for how to compress an array.""" strategy: CompressionStrategy keep_indices: list[int] = field(default_factory=list) constant_fields: dict[str, Any] = field(default_factory=dict) summary_ranges: list[tuple[int, int, dict]] = field(default_factory=list) cluster_field: str | None = None sort_field: str | None = None keep_count: int = 10 @dataclass class CrushResult: """Result from SmartCrusher.crush() method. Used by ContentRouter when routing JSON arrays to SmartCrusher. """ compressed: str original: str was_modified: bool strategy: str = "passthrough" @dataclass class SmartCrusherConfig: """Configuration for smart crusher. SCHEMA-PRESERVING: Output contains only items from the original array. No wrappers, no generated text, no metadata keys. """ enabled: bool = True min_items_to_analyze: int = 5 # Don't analyze tiny arrays min_tokens_to_crush: int = 200 # Only crush if > N tokens variance_threshold: float = 2.0 # Std devs for change point detection uniqueness_threshold: float = 0.1 # Below this = nearly constant similarity_threshold: float = 0.8 # For clustering similar strings max_items_after_crush: int = 15 # Target max items in output preserve_change_points: bool = True factor_out_constants: bool = False # Disabled - preserves original schema include_summaries: bool = False # Disabled - no generated text # Feedback loop integration use_feedback_hints: bool = True # Use learned patterns to adjust compression # LOW FIX #21: Make TOIN confidence threshold configurable # Minimum confidence required to apply TOIN recommendations toin_confidence_threshold: float = 0.5 class SmartAnalyzer: """Analyzes JSON arrays to determine optimal compression strategy.""" def __init__(self, config: SmartCrusherConfig | None = None): self.config = config or SmartCrusherConfig() def analyze_array(self, items: list[dict]) -> ArrayAnalysis: """Perform complete statistical analysis of an array.""" if not items or not isinstance(items[0], dict): return ArrayAnalysis( item_count=len(items) if items else 0, field_stats={}, detected_pattern="generic", recommended_strategy=CompressionStrategy.NONE, constant_fields={}, estimated_reduction=0.0, ) # Analyze each field field_stats = {} all_keys: set[str] = set() for item in items: if isinstance(item, dict): all_keys.update(item.keys()) for key in all_keys: field_stats[key] = self._analyze_field(key, items) # Detect pattern pattern = self._detect_pattern(field_stats, items) # Extract constants constant_fields = {k: v.constant_value for k, v in field_stats.items() if v.is_constant} # CRITICAL: Analyze crushability BEFORE selecting strategy crushability = self.analyze_crushability(items, field_stats) # Select strategy (respects crushability) strategy = self._select_strategy(field_stats, pattern, len(items), crushability) # Estimate reduction (0 if not crushable) if strategy == CompressionStrategy.SKIP: reduction = 0.0 else: reduction = self._estimate_reduction(field_stats, strategy, len(items)) return ArrayAnalysis( item_count=len(items), field_stats=field_stats, detected_pattern=pattern, recommended_strategy=strategy, constant_fields=constant_fields, estimated_reduction=reduction, crushability=crushability, ) def _analyze_field(self, key: str, items: list[dict]) -> FieldStats: """Analyze a single field across all items.""" values = [item.get(key) for item in items if isinstance(item, dict)] non_null_values = [v for v in values if v is not None] if not non_null_values: return FieldStats( name=key, field_type="null", count=len(values), unique_count=0, unique_ratio=0.0, is_constant=True, constant_value=None, ) # Determine type from first non-null value first_val = non_null_values[0] if isinstance(first_val, bool): field_type = "boolean" elif isinstance(first_val, (int, float)): field_type = "numeric" elif isinstance(first_val, str): field_type = "string" elif isinstance(first_val, dict): field_type = "object" elif isinstance(first_val, list): field_type = "array" else: field_type = "unknown" # Compute uniqueness str_values = [str(v) for v in values] unique_values = set(str_values) unique_count = len(unique_values) unique_ratio = unique_count / len(values) if values else 0 # Check if constant is_constant = unique_count == 1 constant_value = non_null_values[0] if is_constant else None stats = FieldStats( name=key, field_type=field_type, count=len(values), unique_count=unique_count, unique_ratio=unique_ratio, is_constant=is_constant, constant_value=constant_value, ) # Numeric-specific analysis if field_type == "numeric": # Filter out NaN and Infinity which break statistics functions nums = [v for v in non_null_values if isinstance(v, (int, float)) and math.isfinite(v)] if nums: try: stats.min_val = min(nums) stats.max_val = max(nums) stats.mean_val = statistics.mean(nums) stats.variance = statistics.variance(nums) if len(nums) > 1 else 0 stats.change_points = self._detect_change_points(nums) except (OverflowError, ValueError): # Extreme values that overflow - skip detailed statistics stats.min_val = None stats.max_val = None stats.mean_val = None stats.variance = 0 stats.change_points = [] # String-specific analysis elif field_type == "string": strs = [v for v in non_null_values if isinstance(v, str)] if strs: stats.avg_length = statistics.mean(len(s) for s in strs) stats.top_values = Counter(strs).most_common(5) return stats def _detect_change_points(self, values: list[float], window: int = 5) -> list[int]: """Detect indices where values change significantly.""" if len(values) < window * 2: return [] change_points = [] # Calculate overall statistics overall_std = statistics.stdev(values) if len(values) > 1 else 0 if overall_std == 0: return [] threshold = self.config.variance_threshold * overall_std # Sliding window comparison for i in range(window, len(values) - window): before_mean = statistics.mean(values[i - window : i]) after_mean = statistics.mean(values[i : i + window]) if abs(after_mean - before_mean) > threshold: change_points.append(i) # Deduplicate nearby change points if change_points: deduped = [change_points[0]] for cp in change_points[1:]: if cp - deduped[-1] > window: deduped.append(cp) return deduped return [] def _detect_pattern(self, field_stats: dict[str, FieldStats], items: list[dict]) -> str: """Detect the data pattern using STATISTICAL analysis (no hardcoded field names). Pattern detection: - TIME_SERIES: Has a temporal field (detected by value format) + numeric variance - LOGS: Has a high-cardinality string field + low-cardinality categorical field - SEARCH_RESULTS: Has a score-like field (bounded numeric, possibly sorted) - GENERIC: Default """ # Check for time series pattern using STRUCTURAL detection has_timestamp = self._detect_temporal_field(field_stats, items) numeric_fields = [k for k, v in field_stats.items() if v.field_type == "numeric"] has_numeric_with_variance = any( (field_stats[k].variance is not None and (field_stats[k].variance or 0) > 0) for k in numeric_fields ) if has_timestamp and has_numeric_with_variance: return "time_series" # Check for logs pattern using STATISTICAL detection # Logs have: high-cardinality string (message) + low-cardinality categorical (level) has_message_like = False has_level_like = False for _name, stats in field_stats.items(): if stats.field_type == "string": # High-cardinality string = likely message field if stats.unique_ratio > 0.5 and stats.avg_length and stats.avg_length > 20: has_message_like = True # Low-cardinality string = likely level/status field elif stats.unique_ratio < 0.1 and 2 <= stats.unique_count <= 10: has_level_like = True if has_message_like and has_level_like: return "logs" # Check for search results pattern using STATISTICAL score detection for _name, stats in field_stats.items(): is_score, confidence = _detect_score_field_statistically(stats, items) if is_score and confidence >= 0.5: return "search_results" return "generic" def _detect_temporal_field(self, field_stats: dict[str, FieldStats], items: list[dict]) -> bool: """Detect if any field contains temporal values (dates/timestamps). Uses STRUCTURAL detection based on value format, not field names. """ # Check string fields for ISO 8601 patterns iso_datetime_pattern = re.compile(r"^\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}") iso_date_pattern = re.compile(r"^\d{4}-\d{2}-\d{2}$") for name, stats in field_stats.items(): if stats.field_type == "string": # Sample some values sample_values = [ item.get(name) for item in items[:10] if isinstance(item.get(name), str) ] if sample_values: # Check if values look like dates/datetimes iso_count = sum( 1 for v in sample_values if v is not None and (iso_datetime_pattern.match(v) or iso_date_pattern.match(v)) ) if iso_count / len(sample_values) > 0.5: return True # Check numeric fields for Unix timestamp range elif stats.field_type == "numeric": if stats.min_val and stats.max_val: # Unix timestamps (seconds): 1000000000 to 2000000000 (roughly 2001-2033) # Unix timestamps (milliseconds): 1000000000000 to 2000000000000 is_unix_seconds = 1000000000 <= stats.min_val <= 2000000000 is_unix_millis = 1000000000000 <= stats.min_val <= 2000000000000 if is_unix_seconds or is_unix_millis: return True return False def analyze_crushability( self, items: list[dict], field_stats: dict[str, FieldStats], ) -> CrushabilityAnalysis: """Analyze whether it's SAFE to crush this array. The key insight: High variability + No importance signal = DON'T CRUSH. We use STATISTICAL detection (no hardcoded field names): 1. ID fields detected by uniqueness + sequential/UUID/entropy patterns 2. Score fields detected by bounded range + sorted order 3. Error items detected by structural outliers (rare fields, rare status values) 4. Numeric anomalies (importance signal) 5. Low uniqueness (safe to sample) Returns: CrushabilityAnalysis with decision and reasoning. """ signals_present: list[str] = [] signals_absent: list[str] = [] # 1. Detect ID field STATISTICALLY (no hardcoded field names) id_field_name = None id_uniqueness = 0.0 id_confidence = 0.0 for name, stats in field_stats.items(): values = [item.get(name) for item in items if isinstance(item, dict)] is_id, confidence = _detect_id_field_statistically(stats, values) if is_id and confidence > id_confidence: id_field_name = name id_uniqueness = stats.unique_ratio id_confidence = confidence has_id_field = id_field_name is not None and id_confidence >= 0.7 # 2. Detect score/rank field STATISTICALLY (no hardcoded field names) has_score_field = False for name, stats in field_stats.items(): is_score, confidence = _detect_score_field_statistically(stats, items) if is_score: has_score_field = True signals_present.append(f"score_field:{name}(conf={confidence:.2f})") break if not has_score_field: signals_absent.append("score_field") # 3. Detect error items via STRUCTURAL OUTLIERS (no hardcoded keywords) outlier_indices = _detect_structural_outliers(items) structural_outlier_count = len(outlier_indices) if structural_outlier_count > 0: signals_present.append(f"structural_outliers:{structural_outlier_count}") else: signals_absent.append("structural_outliers") # 3b. Also detect errors via keywords in content (for log/message-style data) # This catches errors that are in the content but not structural outliers # (e.g., Slack messages where error is in the text field) error_keyword_indices = _detect_error_items_for_preservation(items) keyword_error_count = len(error_keyword_indices) if keyword_error_count > 0 and structural_outlier_count == 0: signals_present.append(f"error_keywords:{keyword_error_count}") # Combined error count for crushability analysis error_count = max(structural_outlier_count, keyword_error_count) # 4. Count numeric anomalies (importance signal) anomaly_count = 0 anomaly_indices: set[int] = set() for stats in field_stats.values(): if stats.field_type == "numeric" and stats.mean_val is not None and stats.variance: std = stats.variance**0.5 if std > 0: threshold = self.config.variance_threshold * std for i, item in enumerate(items): val = item.get(stats.name) if isinstance(val, (int, float)): if abs(val - stats.mean_val) > threshold: anomaly_indices.add(i) anomaly_count = len(anomaly_indices) if anomaly_count > 0: signals_present.append(f"anomalies:{anomaly_count}") else: signals_absent.append("anomalies") # 5. Compute average string uniqueness (EXCLUDING statistically-detected ID fields) string_stats = [ s for s in field_stats.values() if s.field_type == "string" and s.name != id_field_name ] avg_string_uniqueness = ( statistics.mean(s.unique_ratio for s in string_stats) if string_stats else 0.0 ) # Compute uniqueness of non-ID numeric fields non_id_numeric_stats = [ s for s in field_stats.values() if s.field_type == "numeric" and s.name != id_field_name ] avg_non_id_numeric_uniqueness = ( statistics.mean(s.unique_ratio for s in non_id_numeric_stats) if non_id_numeric_stats else 0.0 ) # Combined uniqueness metric (including ID fields) max_uniqueness = max(avg_string_uniqueness, id_uniqueness, 0.0) # Non-ID content uniqueness (for detecting repetitive content with unique IDs) non_id_content_uniqueness = max(avg_string_uniqueness, avg_non_id_numeric_uniqueness) # 6. Check for change points (importance signal for time series) has_change_points = any( stats.change_points for stats in field_stats.values() if stats.field_type == "numeric" ) if has_change_points: signals_present.append("change_points") # DECISION LOGIC has_any_signal = len(signals_present) > 0 # Case 0: Repetitive content with unique IDs # If all non-ID fields are nearly constant, data is safe to sample # even if there's a unique ID field (e.g., status="success" for all items) if non_id_content_uniqueness < 0.1 and has_id_field: signals_present.append("repetitive_content") return CrushabilityAnalysis( crushable=True, confidence=0.85, reason="repetitive_content_with_ids", signals_present=signals_present, signals_absent=signals_absent, has_id_field=has_id_field, id_uniqueness=id_uniqueness, avg_string_uniqueness=avg_string_uniqueness, has_score_field=has_score_field, error_item_count=error_count, anomaly_count=anomaly_count, ) # Case 1: Low uniqueness - safe to sample (data is repetitive) if max_uniqueness < 0.3: return CrushabilityAnalysis( crushable=True, confidence=0.9, reason="low_uniqueness_safe_to_sample", signals_present=signals_present, signals_absent=signals_absent, has_id_field=has_id_field, id_uniqueness=id_uniqueness, avg_string_uniqueness=avg_string_uniqueness, has_score_field=has_score_field, error_item_count=error_count, anomaly_count=anomaly_count, ) # Case 2: High uniqueness + ID field + NO signal = DON'T CRUSH # This is the critical case: DB results, file listings, user lists if has_id_field and max_uniqueness > 0.8 and not has_any_signal: return CrushabilityAnalysis( crushable=False, confidence=0.85, reason="unique_entities_no_signal", signals_present=signals_present, signals_absent=signals_absent, has_id_field=has_id_field, id_uniqueness=id_uniqueness, avg_string_uniqueness=avg_string_uniqueness, has_score_field=has_score_field, error_item_count=error_count, anomaly_count=anomaly_count, ) # Case 3: High uniqueness + has signal = CRUSH using signal if max_uniqueness > 0.8 and has_any_signal: return CrushabilityAnalysis( crushable=True, confidence=0.7, reason="unique_entities_with_signal", signals_present=signals_present, signals_absent=signals_absent, has_id_field=has_id_field, id_uniqueness=id_uniqueness, avg_string_uniqueness=avg_string_uniqueness, has_score_field=has_score_field, error_item_count=error_count, anomaly_count=anomaly_count, ) # Case 4: Medium uniqueness + no signal = be cautious, don't crush if not has_any_signal: return CrushabilityAnalysis( crushable=False, confidence=0.6, reason="medium_uniqueness_no_signal", signals_present=signals_present, signals_absent=signals_absent, has_id_field=has_id_field, id_uniqueness=id_uniqueness, avg_string_uniqueness=avg_string_uniqueness, has_score_field=has_score_field, error_item_count=error_count, anomaly_count=anomaly_count, ) # Case 5: Medium uniqueness + has signal = crush with caution return CrushabilityAnalysis( crushable=True, confidence=0.5, reason="medium_uniqueness_with_signal", signals_present=signals_present, signals_absent=signals_absent, has_id_field=has_id_field, id_uniqueness=id_uniqueness, avg_string_uniqueness=avg_string_uniqueness, has_score_field=has_score_field, error_item_count=error_count, anomaly_count=anomaly_count, ) def _select_strategy( self, field_stats: dict[str, FieldStats], pattern: str, item_count: int, crushability: CrushabilityAnalysis | None = None, ) -> CompressionStrategy: """Select optimal compression strategy based on analysis.""" if item_count < self.config.min_items_to_analyze: return CompressionStrategy.NONE # CRITICAL: Check crushability first if crushability is not None and not crushability.crushable: return CompressionStrategy.SKIP if pattern == "time_series": # Check if there are change points worth preserving numeric_fields = [v for v in field_stats.values() if v.field_type == "numeric"] has_change_points = any(f.change_points for f in numeric_fields) if has_change_points: return CompressionStrategy.TIME_SERIES if pattern == "logs": # Check if messages are clusterable (low-medium uniqueness) message_field = next( (v for k, v in field_stats.items() if "message" in k.lower()), None ) if message_field and message_field.unique_ratio < 0.5: return CompressionStrategy.CLUSTER_SAMPLE if pattern == "search_results": return CompressionStrategy.TOP_N # Default: smart sampling return CompressionStrategy.SMART_SAMPLE def _estimate_reduction( self, field_stats: dict[str, FieldStats], strategy: CompressionStrategy, item_count: int ) -> float: """Estimate token reduction ratio.""" if strategy == CompressionStrategy.NONE: return 0.0 # Count constant fields (will be factored out) constant_ratio = sum(1 for v in field_stats.values() if v.is_constant) / len(field_stats) # Estimate based on strategy base_reduction = { CompressionStrategy.TIME_SERIES: 0.7, CompressionStrategy.CLUSTER_SAMPLE: 0.8, CompressionStrategy.TOP_N: 0.6, CompressionStrategy.SMART_SAMPLE: 0.5, }.get(strategy, 0.3) # Adjust for constants reduction = base_reduction + (constant_ratio * 0.2) return min(reduction, 0.95) class SmartCrusher(Transform): """ Intelligent tool output compression using statistical analysis. Unlike fixed-rule crushing, SmartCrusher: 1. Analyzes JSON structure and computes field statistics 2. Detects data patterns (time series, logs, search results) 3. Identifies constant fields to factor out 4. Finds change points in numeric data to preserve 5. Applies optimal compression strategy per data type 6. Uses RelevanceScorer for semantic matching of user queries This results in higher compression with lower information loss. """ name = "smart_crusher" def __init__( self, config: SmartCrusherConfig | None = None, relevance_config: RelevanceScorerConfig | None = None, scorer: RelevanceScorer | None = None, ccr_config: CCRConfig | None = None, ): self.config = config or SmartCrusherConfig() self.analyzer = SmartAnalyzer(self.config) # CCR (Compress-Cache-Retrieve) configuration # When no ccr_config provided, default to caching enabled but markers disabled # This maintains backward compatibility - callers must opt-in to markers if ccr_config is None: self._ccr_config = CCRConfig( enabled=True, # Still cache for potential retrieval inject_retrieval_marker=False, # Don't break JSON parsing by default ) else: self._ccr_config = ccr_config self._compression_store: CompressionStore | None = None # Feedback loop for learning compression patterns self._feedback: CompressionFeedback | None = None # CRITICAL FIX: Lock for thread-safe lazy initialization # Without this, multiple threads could call _get_* methods simultaneously # and potentially create redundant initialization calls. self._lazy_init_lock = threading.Lock() # Initialize relevance scorer if scorer is not None: self._scorer = scorer else: rel_config = relevance_config or RelevanceScorerConfig() # Build kwargs based on tier - BM25 params only apply to bm25 tier scorer_kwargs = {} if rel_config.tier == "bm25": scorer_kwargs = {"k1": rel_config.bm25_k1, "b": rel_config.bm25_b} elif rel_config.tier == "hybrid": scorer_kwargs = { "alpha": rel_config.hybrid_alpha, "adaptive": rel_config.adaptive_alpha, } self._scorer = create_scorer(tier=rel_config.tier, **scorer_kwargs) # Use threshold from config, or default from RelevanceScorerConfig rel_cfg = relevance_config or RelevanceScorerConfig() self._relevance_threshold = rel_cfg.relevance_threshold # NOTE: Error detection now uses structural outlier detection (_detect_structural_outliers) # instead of hardcoded keywords. This scales to any data domain. def crush(self, content: str, query: str = "") -> CrushResult: """Crush content string directly (for use by ContentRouter). This is a simplified interface for compressing a single content string, used by ContentRouter when routing JSON arrays to SmartCrusher. Args: content: JSON string content to compress. query: Query context for relevance-based compression. Returns: CrushResult with compressed content and metadata. """ compressed, was_modified, analysis_info = self._smart_crush_content( content, query_context=query ) return CrushResult( compressed=compressed, original=content, was_modified=was_modified, strategy=analysis_info or "passthrough", ) def _get_compression_store(self) -> CompressionStore: """Get the compression store for CCR (lazy initialization). CRITICAL FIX: Thread-safe double-checked locking pattern. """ if self._compression_store is None: with self._lazy_init_lock: # Double-check after acquiring lock if self._compression_store is None: self._compression_store = get_compression_store( max_entries=self._ccr_config.store_max_entries, default_ttl=self._ccr_config.store_ttl_seconds, ) return self._compression_store def _get_feedback(self) -> CompressionFeedback: """Get the feedback analyzer (lazy initialization). CRITICAL FIX: Thread-safe double-checked locking pattern. """ if self._feedback is None: with self._lazy_init_lock: if self._feedback is None: self._feedback = get_compression_feedback() return self._feedback def _get_telemetry(self) -> TelemetryCollector: """Get the telemetry collector (lazy initialization). CRITICAL FIX: Thread-safe double-checked locking pattern. """ # Use getattr to avoid hasattr race condition if getattr(self, "_telemetry", None) is None: with self._lazy_init_lock: if getattr(self, "_telemetry", None) is None: self._telemetry = get_telemetry_collector() return self._telemetry def _get_toin(self) -> ToolIntelligenceNetwork: """Get the TOIN instance (lazy initialization). CRITICAL FIX: Thread-safe double-checked locking pattern. """ # Use getattr to avoid hasattr race condition if getattr(self, "_toin", None) is None: with self._lazy_init_lock: if getattr(self, "_toin", None) is None: self._toin = get_toin() return self._toin def _record_telemetry( self, items: list[dict], result: list, analysis: ArrayAnalysis, plan: CompressionPlan, tool_name: str | None = None, ) -> None: """Record compression telemetry for the data flywheel. This collects anonymized statistics about compression patterns to enable cross-user learning and improve compression over time. Privacy guarantees: - No actual data values are stored - Tool names can be hashed - Only structural patterns are captured """ try: telemetry = self._get_telemetry() # Calculate what was kept kept_first_n = sum(1 for i in plan.keep_indices if i < 3) kept_last_n = sum(1 for i in plan.keep_indices if i >= len(items) - 2) # Count error items in result error_indices = set(_detect_error_items_for_preservation(items)) kept_errors = sum(1 for i in plan.keep_indices if i in error_indices) # Count anomalies (approximate from change points) anomaly_count = 0 for stats in analysis.field_stats.values(): if stats.change_points: anomaly_count += len(stats.change_points) kept_anomalies = min(anomaly_count, len(plan.keep_indices)) # Crushability info crushability_score = None crushability_reason = None if analysis.crushability: crushability_score = analysis.crushability.confidence crushability_reason = analysis.crushability.reason # Record the event telemetry.record_compression( items=items[:100], # Sample for structure analysis original_count=len(items), compressed_count=len(result), original_tokens=0, # Not available here compressed_tokens=0, # Not available here strategy=analysis.recommended_strategy.value, tool_name=tool_name, strategy_reason=analysis.detected_pattern, crushability_score=crushability_score, crushability_reason=crushability_reason, kept_first_n=kept_first_n, kept_last_n=kept_last_n, kept_errors=kept_errors, kept_anomalies=kept_anomalies, kept_by_relevance=0, # Would need to track separately kept_by_score=0, # Would need to track separately ) except Exception: # Telemetry should never break compression pass def _prioritize_indices( self, keep_indices: set[int], items: list[dict], 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. Priority order: 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. 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, 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 intentional to preserve the quality guarantee. A warning is logged when this happens to help diagnose cases where compression is less effective than expected. Args: keep_indices: Initial set of indices to keep. items: The items being compressed. 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). """ # Use provided max_items or fall back to config effective_max = max_items if max_items is not None else self.config.max_items_after_crush 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)) # Identify structural outlier indices using STATISTICAL detection # (items with rare fields or rare status values) outlier_indices = set(_detect_structural_outliers(items)) # Identify numeric anomalies (MUST keep ALL of them) anomaly_indices = set() if analysis and analysis.field_stats: for field_name, stats in analysis.field_stats.items(): if stats.field_type == "numeric" and stats.mean_val is not None and stats.variance: std = stats.variance**0.5 if std > 0: threshold = self.config.variance_threshold * std for i, item in enumerate(items): val = item.get(field_name) if isinstance(val, (int, float)): 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 | 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 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 remaining_slots = effective_max - len(prioritized) if remaining_slots > 0: # First 3 items for i in range(min(3, n)): if i not in prioritized and remaining_slots > 0: prioritized.add(i) remaining_slots -= 1 # Last 2 items for i in range(max(0, n - 2), n): if i not in prioritized and remaining_slots > 0: prioritized.add(i) remaining_slots -= 1 # Fill remaining slots with other important indices (by index order) if remaining_slots > 0: other_indices = sorted(keep_indices - prioritized) for i in other_indices: if remaining_slots <= 0: break prioritized.add(i) remaining_slots -= 1 return prioritized def should_apply( self, messages: list[dict[str, Any]], tokenizer: Tokenizer, **kwargs: Any, ) -> bool: """Check if any tool messages would benefit from smart crushing.""" if not self.config.enabled: return False for msg in messages: # OpenAI style: role="tool" if msg.get("role") == "tool": content = msg.get("content", "") if isinstance(content, str): tokens = tokenizer.count_text(content) if tokens > self.config.min_tokens_to_crush: # Check if it's JSON with arrays parsed, success = safe_json_loads(content) if success and self._has_crushable_arrays(parsed): return True # Anthropic style: role="user" with tool_result content blocks content = msg.get("content") if isinstance(content, list): for block in content: if isinstance(block, dict) and block.get("type") == "tool_result": tool_content = block.get("content", "") if isinstance(tool_content, str): tokens = tokenizer.count_text(tool_content) if tokens > self.config.min_tokens_to_crush: parsed, success = safe_json_loads(tool_content) if success and self._has_crushable_arrays(parsed): return True return False def _has_crushable_arrays(self, data: Any, depth: int = 0) -> bool: """Check if data contains arrays large enough to crush.""" if depth > 5: return False if isinstance(data, list): if len(data) >= self.config.min_items_to_analyze: if data and isinstance(data[0], dict): return True for item in data[:10]: # Check first few items if self._has_crushable_arrays(item, depth + 1): return True elif isinstance(data, dict): for value in data.values(): if self._has_crushable_arrays(value, depth + 1): return True return False def apply( self, messages: list[dict[str, Any]], tokenizer: Tokenizer, **kwargs: Any, ) -> TransformResult: """Apply smart crushing to messages.""" tokens_before = tokenizer.count_messages(messages) result_messages = deep_copy_messages(messages) transforms_applied: list[str] = [] markers_inserted: list[str] = [] warnings: list[str] = [] # Extract query context from recent user messages for relevance scoring query_context = self._extract_context_from_messages(result_messages) crushed_count = 0 for msg in result_messages: # OpenAI style if msg.get("role") == "tool": content = msg.get("content", "") if not isinstance(content, str): continue tokens = tokenizer.count_text(content) if tokens <= self.config.min_tokens_to_crush: continue crushed, was_modified, analysis_info = self._smart_crush_content( content, query_context ) if was_modified: original_hash = compute_short_hash(content) marker = create_tool_digest_marker(original_hash) msg["content"] = crushed + "\n" + marker crushed_count += 1 markers_inserted.append(marker) if analysis_info: transforms_applied.append(f"smart:{analysis_info}") # Anthropic style content = msg.get("content") if isinstance(content, list): for i, block in enumerate(content): if not isinstance(block, dict): continue if block.get("type") != "tool_result": continue tool_content = block.get("content", "") if not isinstance(tool_content, str): continue tokens = tokenizer.count_text(tool_content) if tokens <= self.config.min_tokens_to_crush: continue crushed, was_modified, analysis_info = self._smart_crush_content( tool_content, query_context ) if was_modified: original_hash = compute_short_hash(tool_content) marker = create_tool_digest_marker(original_hash) content[i]["content"] = crushed + "\n" + marker crushed_count += 1 markers_inserted.append(marker) if analysis_info: transforms_applied.append(f"smart:{analysis_info}") if crushed_count > 0: transforms_applied.insert(0, f"smart_crush:{crushed_count}") tokens_after = tokenizer.count_messages(result_messages) return TransformResult( messages=result_messages, tokens_before=tokens_before, tokens_after=tokens_after, transforms_applied=transforms_applied, markers_inserted=markers_inserted, warnings=warnings, ) def _extract_context_from_messages(self, messages: list[dict[str, Any]]) -> str: """Extract query context from recent messages for relevance scoring. Builds a context string from: - Recent user messages (what the user is asking about) - Recent tool call arguments (what data was requested) This context is used by RelevanceScorer to determine which items to preserve during crushing. Args: messages: Full message list. Returns: Context string for relevance scoring. """ context_parts: list[str] = [] # Look at last 5 user messages (most relevant to recent tool calls) user_message_count = 0 for msg in reversed(messages): if msg.get("role") == "user": content = msg.get("content") if isinstance(content, str): context_parts.append(content) elif isinstance(content, list): # Anthropic style - extract from text blocks for block in content: if isinstance(block, dict) and block.get("type") == "text": text = block.get("text", "") if text: context_parts.append(text) user_message_count += 1 if user_message_count >= 5: break # Also check assistant tool_calls for function arguments if msg.get("role") == "assistant" and msg.get("tool_calls"): for tc in msg.get("tool_calls", []): if isinstance(tc, dict): func = tc.get("function", {}) args = func.get("arguments", "") if isinstance(args, str) and args: context_parts.append(args) return " ".join(context_parts) def _smart_crush_content( self, content: str, query_context: str = "", tool_name: str | None = None ) -> tuple[str, bool, str]: """ Apply smart crushing to content. Handles both JSON (existing SmartCrusher logic) and plain text content (search results, logs, generic text) using specialized compressors. Args: content: Content to crush (JSON or plain text). query_context: Context string from user messages for relevance scoring. tool_name: Name of the tool that produced this output. Returns: Tuple of (crushed_content, was_modified, analysis_info). """ parsed, success = safe_json_loads(content) if not success: # Not JSON - pass through unchanged # Text compression utilities (SearchCompressor, LogCompressor, TextCompressor) # are available as standalone tools for applications to use explicitly return content, False, "" # Recursively process and crush arrays crushed, info, ccr_markers = self._process_value( parsed, query_context=query_context, tool_name=tool_name ) result = safe_json_dumps(crushed, indent=None) was_modified = result != content.strip() # CCR: Inject retrieval markers if compression happened and CCR is enabled if was_modified and ccr_markers and self._ccr_config.inject_retrieval_marker: for ccr_hash, original_count, compressed_count in ccr_markers: marker = self._ccr_config.marker_template.format( original_count=original_count, compressed_count=compressed_count, hash=ccr_hash, ) result += marker return result, was_modified, info def _process_value( self, value: Any, depth: int = 0, query_context: str = "", tool_name: str | None = None ) -> tuple[Any, str, list[tuple[str, int, int]]]: """Recursively process a value, crushing arrays where appropriate. Returns: Tuple of (processed_value, info_string, ccr_markers). ccr_markers is a list of (hash, original_count, compressed_count) tuples. """ info_parts = [] ccr_markers: list[tuple[str, int, int]] = [] if isinstance(value, list): # Check if this array should be crushed # Must have enough items AND all items must be dicts (not mixed types) all_dicts = value and all(isinstance(item, dict) for item in value) if len(value) >= self.config.min_items_to_analyze and all_dicts: crushed, strategy, ccr_hash = self._crush_array(value, query_context, tool_name) info_parts.append(f"{strategy}({len(value)}->{len(crushed)})") # Track CCR marker for later injection if ccr_hash: ccr_markers.append((ccr_hash, len(value), len(crushed))) return crushed, ",".join(info_parts), ccr_markers else: # Process items recursively processed = [] for item in value: p_item, p_info, p_markers = self._process_value( item, depth + 1, query_context, tool_name ) processed.append(p_item) if p_info: info_parts.append(p_info) ccr_markers.extend(p_markers) return processed, ",".join(info_parts), ccr_markers elif isinstance(value, dict): # Process values recursively processed_dict: dict[str, Any] = {} for k, v in value.items(): p_val, p_info, p_markers = self._process_value( v, depth + 1, query_context, tool_name ) processed_dict[k] = p_val if p_info: info_parts.append(p_info) ccr_markers.extend(p_markers) return processed_dict, ",".join(info_parts), ccr_markers else: return value, "", [] def _crush_array( self, items: list[dict], query_context: str = "", tool_name: str | None = None ) -> tuple[list, str, str | None]: """Crush an array using statistical analysis and relevance scoring. IMPORTANT: If crushability analysis determines it's not safe to crush (high variability + no importance signal), returns original array unchanged. TOIN-aware: Consults the Tool Output Intelligence Network for cross-user learned patterns. High retrieval rate across all users → compress less. Feedback-aware: Uses learned patterns to adjust compression aggressiveness. High retrieval rate for a tool → compress less aggressively. Returns: Tuple of (crushed_items, strategy_info, ccr_hash). ccr_hash is the hash for retrieval if CCR is enabled, None otherwise. """ # BOUNDARY CHECK: If already at or below max_items, no compression needed if len(items) <= self.config.max_items_after_crush: return items, "none:at_limit", None # Get feedback hints if enabled # THREAD-SAFETY: Use a local effective_max_items instead of mutating shared config effective_max_items = self.config.max_items_after_crush hints_applied = False toin_hint_applied = False # Create ToolSignature for TOIN lookup tool_signature = ToolSignature.from_items(items) # TOIN: Get cross-user learned recommendations toin = self._get_toin() toin_hint = toin.get_recommendation(tool_signature, query_context) if toin_hint.skip_compression: return items, f"skip:toin({toin_hint.reason})", None # Apply TOIN recommendations if from network or local learning toin_preserve_fields: list[str] = [] toin_recommended_strategy: str | None = None toin_compression_level: str | None = None # LOW FIX #21: Use configurable threshold instead of hardcoded 0.5 if ( toin_hint.source in ("network", "local") and toin_hint.confidence >= self.config.toin_confidence_threshold ): # TOIN recommendations take precedence over local feedback effective_max_items = toin_hint.max_items toin_preserve_fields = toin_hint.preserve_fields # Fields to never remove toin_hint_applied = True # Store strategy and compression level for later use if toin_hint.recommended_strategy != "default": toin_recommended_strategy = toin_hint.recommended_strategy 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() hints = feedback.get_compression_hints(tool_name) # Check if hints recommend skipping compression if hints.skip_compression: return items, f"skip:feedback({hints.reason})", None # Adjust max_items based on feedback if hints.suggested_items is not None: effective_max_items = hints.suggested_items hints_applied = True # Use preserve_fields from local feedback (hash them for TOIN compatibility) # Note: CompressionFeedback stores actual field names, but _plan methods # expect SHA256[:8] hashes for privacy-preserving comparison if hints.preserve_fields: toin_preserve_fields = [_hash_field_name(field) for field in hints.preserve_fields] # Use recommended_strategy from local feedback if not already set by TOIN if hints.recommended_strategy and not toin_recommended_strategy: toin_recommended_strategy = hints.recommended_strategy try: # Analyze the array (includes crushability check) analysis = self.analyzer.analyze_array(items) # CRITICAL: If not crushable, return original array unchanged if analysis.recommended_strategy == CompressionStrategy.SKIP: reason = "" if analysis.crushability: reason = f"skip:{analysis.crushability.reason}" return items, reason, None # Apply TOIN strategy recommendation if available # TOIN learns which strategies work best from cross-user patterns if toin_recommended_strategy: try: toin_strategy = CompressionStrategy(toin_recommended_strategy) # Only override if TOIN suggests a valid non-SKIP strategy if toin_strategy != CompressionStrategy.SKIP: analysis.recommended_strategy = toin_strategy except ValueError: pass # Invalid strategy name, keep analyzer's choice # Apply TOIN compression level to adjust effective_max_items if toin_compression_level: if toin_compression_level == "none": # Don't compress - return original return items, "skip:toin_level_none", None elif toin_compression_level == "conservative": # Be conservative - keep more items effective_max_items = max(effective_max_items, min(50, len(items) // 2)) elif toin_compression_level == "aggressive": # Be aggressive - keep fewer items effective_max_items = min(effective_max_items, 15) # Create compression plan with relevance scoring # Pass TOIN preserve_fields so items with those fields get priority # Pass effective_max_items for thread-safe compression plan = self._create_plan( analysis, items, query_context, preserve_fields=toin_preserve_fields or None, effective_max_items=effective_max_items, ) # Execute compression result = self._execute_plan(plan, items, analysis) # CCR: Store original content for retrieval if enabled ccr_hash = None if ( self._ccr_config.enabled and len(items) >= self._ccr_config.min_items_to_cache and len(result) < len(items) # Only cache if compression actually happened ): store = self._get_compression_store() original_json = json.dumps(items, default=str) compressed_json = json.dumps(result, default=str) ccr_hash = store.store( original=original_json, compressed=compressed_json, original_item_count=len(items), compressed_item_count=len(result), tool_name=tool_name, query_context=query_context, # CRITICAL: Pass the tool_signature_hash so retrieval events # can be correlated with compression events in TOIN tool_signature_hash=tool_signature.structure_hash, compression_strategy=analysis.recommended_strategy.value, ) # Record compression event for feedback loop if self.config.use_feedback_hints and tool_name: feedback = self._get_feedback() feedback.record_compression( tool_name=tool_name, original_count=len(items), compressed_count=len(result), strategy=analysis.recommended_strategy.value, tool_signature_hash=tool_signature.structure_hash, ) # Record telemetry for data flywheel self._record_telemetry( items=items, result=result, analysis=analysis, plan=plan, tool_name=tool_name, ) # TOIN: Record compression event for cross-user learning try: # Calculate token counts (approximate) original_tokens = len(json.dumps(items, default=str)) // 4 compressed_tokens = len(json.dumps(result, default=str)) // 4 toin.record_compression( tool_signature=tool_signature, original_count=len(items), compressed_count=len(result), original_tokens=original_tokens, 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 pass strategy_info = analysis.recommended_strategy.value if toin_hint_applied: toin_parts = [f"items={toin_hint.max_items}", f"conf={toin_hint.confidence:.2f}"] if toin_recommended_strategy: toin_parts.append(f"strategy={toin_recommended_strategy}") if toin_compression_level and toin_compression_level != "moderate": toin_parts.append(f"level={toin_compression_level}") strategy_info += f"(toin:{','.join(toin_parts)})" 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 def _create_plan( self, analysis: ArrayAnalysis, items: list[dict], query_context: str = "", preserve_fields: list[str] | None = None, effective_max_items: int | None = None, ) -> CompressionPlan: """Create a detailed compression plan using relevance scoring. Args: analysis: The array analysis results. items: The items to compress. query_context: Context string from user messages for relevance scoring. preserve_fields: TOIN-learned fields that users commonly retrieve. Items with values in these fields get higher priority. effective_max_items: Thread-safe max items limit (defaults to config value). """ # Use provided effective_max_items or fall back to config max_items = ( effective_max_items if effective_max_items is not None else self.config.max_items_after_crush ) plan = CompressionPlan( strategy=analysis.recommended_strategy, constant_fields=analysis.constant_fields if self.config.factor_out_constants else {}, ) # Handle SKIP - keep all items (shouldn't normally reach here) if analysis.recommended_strategy == CompressionStrategy.SKIP: plan.keep_indices = list(range(len(items))) return plan if analysis.recommended_strategy == CompressionStrategy.TIME_SERIES: plan = self._plan_time_series( analysis, items, plan, query_context, preserve_fields, max_items ) elif analysis.recommended_strategy == CompressionStrategy.CLUSTER_SAMPLE: plan = self._plan_cluster_sample( analysis, items, plan, query_context, preserve_fields, max_items ) elif analysis.recommended_strategy == CompressionStrategy.TOP_N: plan = self._plan_top_n( analysis, items, plan, query_context, preserve_fields, max_items ) else: # SMART_SAMPLE or NONE plan = self._plan_smart_sample( analysis, items, plan, query_context, preserve_fields, max_items ) return plan def _plan_time_series( self, analysis: ArrayAnalysis, items: list[dict], plan: CompressionPlan, query_context: str = "", preserve_fields: list[str] | None = None, max_items: int | None = None, ) -> CompressionPlan: """Plan compression for time series data. Keeps items around change points (anomalies) plus first/last items. Uses STATISTICAL outlier detection for important items. Uses RelevanceScorer for semantic matching of user queries. Args: preserve_fields: TOIN-learned fields that users commonly retrieve. Items where query_context matches these field values get priority. max_items: Thread-safe max items limit (defaults to config value). """ # Use provided max_items or fall back to config effective_max = max_items if max_items is not None else self.config.max_items_after_crush n = len(items) keep_indices = set() # 1. First 3 items for i in range(min(3, n)): keep_indices.add(i) # 2. Last 2 items for i in range(max(0, n - 2), n): keep_indices.add(i) # 3. Items around change points from numeric fields for stats in analysis.field_stats.values(): if stats.change_points: for cp in stats.change_points: # Keep a window around each change point for offset in range(-2, 3): idx = cp + offset if 0 <= idx < n: keep_indices.add(idx) # 4. Structural outlier items (STATISTICAL detection - no hardcoded keywords) outlier_indices = _detect_structural_outliers(items) keep_indices.update(outlier_indices) # 4b. Error items via KEYWORD detection (PRESERVATION GUARANTEE) # This is critical - errors must ALWAYS be preserved regardless of structure error_indices = _detect_error_items_for_preservation(items) keep_indices.update(error_indices) # 5. Items matching query anchors (DETERMINISTIC exact match) # Anchors provide reliable preservation for specific entity lookups (UUIDs, IDs, names) if query_context: anchors = extract_query_anchors(query_context) for i, item in enumerate(items): if item_matches_anchors(item, anchors): keep_indices.add(i) # 6. Items with high relevance to query context (PROBABILISTIC semantic match) if query_context: item_strs = [json.dumps(item, default=str) for item in items] scores = self._scorer.score_batch(item_strs, query_context) for i, score in enumerate(scores): if score.score >= self._relevance_threshold: keep_indices.add(i) # 6b. TOIN preserve_fields: boost items where query matches these fields # Note: preserve_fields are SHA256[:8] hashes, use helper to match if preserve_fields and query_context: for i, item in enumerate(items): if _item_has_preserve_field_match(item, preserve_fields, query_context): keep_indices.add(i) # Limit to effective_max while ALWAYS preserving outliers and anomalies keep_indices = self._prioritize_indices(keep_indices, items, n, analysis, effective_max) plan.keep_indices = sorted(keep_indices) return plan def _plan_cluster_sample( self, analysis: ArrayAnalysis, items: list[dict], plan: CompressionPlan, query_context: str = "", preserve_fields: list[str] | None = None, max_items: int | None = None, ) -> CompressionPlan: """Plan compression for clusterable data (like logs). Uses clustering plus STATISTICAL outlier detection. Uses RelevanceScorer for semantic matching of user queries. Args: preserve_fields: TOIN-learned fields that users commonly retrieve. Items where query_context matches these field values get priority. max_items: Thread-safe max items limit (defaults to config value). """ # Use provided max_items or fall back to config effective_max = max_items if max_items is not None else self.config.max_items_after_crush n = len(items) keep_indices = set() # 1. First 3 items for i in range(min(3, n)): keep_indices.add(i) # 2. Last 2 items for i in range(max(0, n - 2), n): keep_indices.add(i) # 3. Structural outlier items (STATISTICAL detection - no hardcoded keywords) outlier_indices = _detect_structural_outliers(items) keep_indices.update(outlier_indices) # 3b. Error items via KEYWORD detection (PRESERVATION GUARANTEE) # This is critical - errors must ALWAYS be preserved regardless of structure error_indices = _detect_error_items_for_preservation(items) keep_indices.update(error_indices) # 4. Cluster by message-like field and keep representatives # Find a high-cardinality string field (likely message field) message_field = None max_uniqueness = 0.0 for name, stats in analysis.field_stats.items(): if stats.field_type == "string" and stats.unique_ratio > max_uniqueness: # Prefer fields with moderate to high uniqueness (message-like) if stats.unique_ratio > 0.3: message_field = name max_uniqueness = stats.unique_ratio if message_field: plan.cluster_field = message_field # Simple clustering: group by first 50 chars of message clusters: dict[str, list[int]] = {} for i, item in enumerate(items): msg = str(item.get(message_field, ""))[:50] msg_hash = hashlib.md5(msg.encode()).hexdigest()[:8] if msg_hash not in clusters: clusters[msg_hash] = [] clusters[msg_hash].append(i) # Keep 1-2 representatives from each cluster for indices in clusters.values(): for idx in indices[:2]: keep_indices.add(idx) # 5. Items matching query anchors (DETERMINISTIC exact match) # Anchors provide reliable preservation for specific entity lookups (UUIDs, IDs, names) if query_context: anchors = extract_query_anchors(query_context) for i, item in enumerate(items): if item_matches_anchors(item, anchors): keep_indices.add(i) # 6. Items with high relevance to query context (PROBABILISTIC semantic match) if query_context: item_strs = [json.dumps(item, default=str) for item in items] scores = self._scorer.score_batch(item_strs, query_context) for i, score in enumerate(scores): if score.score >= self._relevance_threshold: keep_indices.add(i) # 6b. TOIN preserve_fields: boost items where query matches these fields # Note: preserve_fields are SHA256[:8] hashes, use helper to match if preserve_fields and query_context: for i, item in enumerate(items): if _item_has_preserve_field_match(item, preserve_fields, query_context): keep_indices.add(i) # Limit total while ALWAYS preserving outliers and anomalies keep_indices = self._prioritize_indices(keep_indices, items, n, analysis, effective_max) plan.keep_indices = sorted(keep_indices) return plan def _plan_top_n( self, analysis: ArrayAnalysis, items: list[dict], plan: CompressionPlan, query_context: str = "", preserve_fields: list[str] | None = None, max_items: int | None = None, ) -> CompressionPlan: """Plan compression for scored/ranked data. For data with a score/relevance field, that field IS the primary relevance signal. Our internal relevance scoring is SECONDARY - it's used to find potential "needle" items that the original scoring might have missed. Strategy: 1. Keep top N by score (the original system's relevance ranking) 2. Add structural outliers (errors, anomalies) 3. Add high-confidence relevance matches (needles the user is looking for) Args: preserve_fields: TOIN-learned fields that users commonly retrieve. Items where query_context matches these field values get priority. max_items: Thread-safe max items limit (defaults to config value). """ # Use provided max_items or fall back to config effective_max = max_items if max_items is not None else self.config.max_items_after_crush # Find score field using STATISTICAL detection (no hardcoded field names) score_field = None max_confidence = 0.0 for name, stats in analysis.field_stats.items(): is_score, confidence = _detect_score_field_statistically(stats, items) if is_score and confidence > max_confidence: score_field = name max_confidence = confidence if not score_field: return self._plan_smart_sample( analysis, items, plan, query_context, preserve_fields, effective_max ) plan.sort_field = score_field keep_indices = set() # 1. TOP N by score FIRST (the primary relevance signal) # The original system's score field is the authoritative ranking scored_items = [(i, item.get(score_field, 0)) for i, item in enumerate(items)] scored_items.sort(key=lambda x: x[1], reverse=True) # Reserve slots for outliers top_count = max(0, effective_max - 3) for idx, _ in scored_items[:top_count]: keep_indices.add(idx) # 2. Structural outlier items (STATISTICAL detection - no hardcoded keywords) outlier_indices = _detect_structural_outliers(items) keep_indices.update(outlier_indices) # 2b. Error items via KEYWORD detection (PRESERVATION GUARANTEE) # This is critical - errors must ALWAYS be preserved regardless of structure error_indices = _detect_error_items_for_preservation(items) keep_indices.update(error_indices) # 3. Items matching query anchors (DETERMINISTIC exact match) - ADDITIVE # Anchors provide reliable preservation for specific entity lookups (UUIDs, IDs, names) # These are ALWAYS preserved since they represent explicit user intent if query_context: anchors = extract_query_anchors(query_context) for i, item in enumerate(items): if i not in keep_indices and item_matches_anchors(item, anchors): keep_indices.add(i) # 4. HIGH-CONFIDENCE relevance matches (potential needles) - ADDITIVE only # Only add items that are NOT already in top N but match the query strongly # Use a higher threshold (0.5) since the score field already captures relevance if query_context: item_strs = [json.dumps(item, default=str) for item in items] scores = self._scorer.score_batch(item_strs, query_context) # Higher threshold and limit count to avoid adding everything high_threshold = max(0.5, self._relevance_threshold * 2) added_count = 0 max_relevance_adds = 3 # Limit additional relevance matches for i, score in enumerate(scores): if i not in keep_indices and score.score >= high_threshold: keep_indices.add(i) added_count += 1 if added_count >= max_relevance_adds: break # 4b. TOIN preserve_fields: boost items where query matches these fields # Note: preserve_fields are SHA256[:8] hashes, use helper to match if preserve_fields and query_context: for i, item in enumerate(items): if i not in keep_indices: # Only add if not already kept if _item_has_preserve_field_match(item, preserve_fields, query_context): keep_indices.add(i) plan.keep_count = len(keep_indices) plan.keep_indices = sorted(keep_indices) return plan def _plan_smart_sample( self, analysis: ArrayAnalysis, items: list[dict], plan: CompressionPlan, query_context: str = "", preserve_fields: list[str] | None = None, max_items: int | None = None, ) -> CompressionPlan: """Plan smart statistical sampling using STATISTICAL detection. Always keeps: - First K items (default 3) - Last K items (default 2) - Structural outliers (items with rare fields or rare status values) - Anomalous numeric items (> 2 std from mean) - Items around change points - Items with high relevance to query context (via RelevanceScorer) Uses STATISTICAL detection instead of hardcoded keywords. Args: preserve_fields: TOIN-learned fields that users commonly retrieve. Items where query_context matches these field values get priority. max_items: Thread-safe max items limit (defaults to config value). """ # Use provided max_items or fall back to config effective_max = max_items if max_items is not None else self.config.max_items_after_crush n = len(items) keep_indices = set() # 1. First K items (default 3) for i in range(min(3, n)): keep_indices.add(i) # 2. Last K items (default 2) for i in range(max(0, n - 2), n): keep_indices.add(i) # 3. Structural outlier items (STATISTICAL detection - no hardcoded keywords) outlier_indices = _detect_structural_outliers(items) keep_indices.update(outlier_indices) # 3b. Error items via KEYWORD detection (PRESERVATION GUARANTEE) # This is critical - errors must ALWAYS be preserved regardless of structure error_indices = _detect_error_items_for_preservation(items) keep_indices.update(error_indices) # 4. Anomalous numeric items (> 2 std from mean) for name, stats in analysis.field_stats.items(): if stats.field_type == "numeric" and stats.mean_val is not None and stats.variance: std = stats.variance**0.5 if std > 0: threshold = self.config.variance_threshold * std for i, item in enumerate(items): val = item.get(name) if isinstance(val, (int, float)): if abs(val - stats.mean_val) > threshold: keep_indices.add(i) # 5. Items around change points (if detected) if self.config.preserve_change_points: for stats in analysis.field_stats.values(): if stats.change_points: for cp in stats.change_points: # Keep items around change point for offset in range(-1, 2): idx = cp + offset if 0 <= idx < n: keep_indices.add(idx) # 6. Items matching query anchors (DETERMINISTIC exact match) # Anchors provide reliable preservation for specific entity lookups (UUIDs, IDs, names) if query_context: anchors = extract_query_anchors(query_context) for i, item in enumerate(items): if item_matches_anchors(item, anchors): keep_indices.add(i) # 7. Items with high relevance to query context (PROBABILISTIC semantic match) if query_context: item_strs = [json.dumps(item, default=str) for item in items] scores = self._scorer.score_batch(item_strs, query_context) for i, score in enumerate(scores): if score.score >= self._relevance_threshold: keep_indices.add(i) # 7b. TOIN preserve_fields: boost items where query matches these fields # Note: preserve_fields are SHA256[:8] hashes, use helper to match if preserve_fields and query_context: for i, item in enumerate(items): if _item_has_preserve_field_match(item, preserve_fields, query_context): keep_indices.add(i) # Limit to effective_max while ALWAYS preserving outliers and anomalies keep_indices = self._prioritize_indices(keep_indices, items, n, analysis, effective_max) plan.keep_indices = sorted(keep_indices) return plan def _execute_plan( self, plan: CompressionPlan, items: list[dict], analysis: ArrayAnalysis ) -> list: """Execute a compression plan and return crushed array. SCHEMA-PRESERVING: Returns only items from the original array. No wrappers, no generated text, no metadata keys. """ result = [] # Return only the kept items, preserving original schema for idx in sorted(plan.keep_indices): if 0 <= idx < len(items): # Copy item unchanged - no modifications to schema result.append(items[idx].copy()) return result def smart_crush_tool_output( content: str, config: SmartCrusherConfig | None = None, ccr_config: CCRConfig | None = None, ) -> tuple[str, bool, str]: """ Convenience function to smart-crush a single tool output. NOTE: CCR markers are DISABLED by default in this convenience function to maintain backward compatibility (output remains valid JSON). To enable CCR markers, pass a CCRConfig with inject_retrieval_marker=True. Args: content: The tool output content (JSON string). config: Optional SmartCrusher configuration. ccr_config: Optional CCR (Compress-Cache-Retrieve) configuration. By default, CCR is enabled (caching) but markers are disabled. Returns: Tuple of (crushed_content, was_modified, analysis_info). """ cfg = config or SmartCrusherConfig() # Default: CCR enabled for caching, but markers disabled for clean JSON output if ccr_config is None: ccr_cfg = CCRConfig( enabled=True, # Still cache for retrieval inject_retrieval_marker=False, # Don't break JSON output ) else: ccr_cfg = ccr_config crusher = SmartCrusher(cfg, ccr_config=ccr_cfg) return crusher._smart_crush_content(content)