"""Data models for privacy-preserving telemetry. These models capture PATTERNS, not DATA. We never store actual values, user queries, or identifiable information. """ from __future__ import annotations import hashlib import json from dataclasses import dataclass, field from typing import Any, Literal # Type alias for field semantic types FieldSemanticType = Literal[ "unknown", "identifier", "error_indicator", "score", "status", "temporal", "content", ] @dataclass class FieldDistribution: """Statistics about a field's distribution (no actual values). This captures the SHAPE of the data, not the data itself. """ field_name_hash: str # SHA256[:8] of field name (anonymized) field_type: Literal["string", "numeric", "boolean", "array", "object", "null", "mixed"] # String field statistics avg_length: float | None = None unique_ratio: float | None = None # 0.0 = constant, 1.0 = all unique entropy: float | None = None # Shannon entropy normalized to [0, 1] looks_like_id: bool = False # High entropy + consistent format # Numeric field statistics has_variance: bool = False variance_bucket: Literal["zero", "low", "medium", "high"] | None = None has_negative: bool = False is_integer: bool = True has_outliers: bool = False # Values > 2σ from mean # Array field statistics avg_array_length: float | None = None # Derived insights is_likely_score: bool = False # Monotonic, bounded, high variance is_likely_timestamp: bool = False # Sequential, numeric, consistent intervals is_likely_status: bool = False # Low cardinality categorical def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "field_name_hash": self.field_name_hash, "field_type": self.field_type, "avg_length": self.avg_length, "unique_ratio": self.unique_ratio, "entropy": self.entropy, "looks_like_id": self.looks_like_id, "has_variance": self.has_variance, "variance_bucket": self.variance_bucket, "has_negative": self.has_negative, "is_integer": self.is_integer, "has_outliers": self.has_outliers, "avg_array_length": self.avg_array_length, "is_likely_score": self.is_likely_score, "is_likely_timestamp": self.is_likely_timestamp, "is_likely_status": self.is_likely_status, } @classmethod def from_dict(cls, data: dict[str, Any]) -> FieldDistribution: """Create from dictionary.""" return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) @dataclass class ToolSignature: """Anonymized signature of a tool's output structure. This identifies SIMILAR tools across users without revealing tool names. Two tools with the same field structure will have the same signature. """ # Structural hash (based on field types and names) # MEDIUM FIX #15: Uses SHA256[:24] (96 bits) for better collision resistance structure_hash: str # SHA256[:24] of sorted field names + types # Schema characteristics field_count: int has_nested_objects: bool has_arrays: bool max_depth: int # Field type distribution string_field_count: int = 0 numeric_field_count: int = 0 boolean_field_count: int = 0 array_field_count: int = 0 object_field_count: int = 0 # Pattern indicators (without revealing actual field names) has_id_like_field: bool = False has_score_like_field: bool = False has_timestamp_like_field: bool = False has_status_like_field: bool = False has_error_like_field: bool = False has_message_like_field: bool = False def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "structure_hash": self.structure_hash, "field_count": self.field_count, "has_nested_objects": self.has_nested_objects, "has_arrays": self.has_arrays, "max_depth": self.max_depth, "string_field_count": self.string_field_count, "numeric_field_count": self.numeric_field_count, "boolean_field_count": self.boolean_field_count, "array_field_count": self.array_field_count, "object_field_count": self.object_field_count, "has_id_like_field": self.has_id_like_field, "has_score_like_field": self.has_score_like_field, "has_timestamp_like_field": self.has_timestamp_like_field, "has_status_like_field": self.has_status_like_field, "has_error_like_field": self.has_error_like_field, "has_message_like_field": self.has_message_like_field, } @staticmethod def _calculate_depth(value: Any, current_depth: int = 1, max_depth_limit: int = 10) -> int: """Recursively calculate the depth of a nested structure. MEDIUM FIX #12: Actually calculate max_depth instead of hardcoding 1. """ if current_depth >= max_depth_limit: return current_depth # Prevent infinite recursion if isinstance(value, dict): if not value: return current_depth return max( ToolSignature._calculate_depth(v, current_depth + 1, max_depth_limit) for v in value.values() ) elif isinstance(value, list): if not value: return current_depth # Sample first few items in arrays to avoid O(n) traversal sample_items = value[:3] return max( ToolSignature._calculate_depth(item, current_depth + 1, max_depth_limit) for item in sample_items ) else: return current_depth @staticmethod def _matches_pattern( key_lower: str, patterns: list[str], original_key: str | None = None ) -> bool: """Check if key matches patterns using word boundary matching. MEDIUM FIX #14: Prevent false positives like "hidden" matching "id". Uses word boundary logic: pattern must be at start/end or surrounded by non-alphanumeric characters (underscore, hyphen, or boundary). Args: key_lower: The field name in lowercase patterns: List of patterns to match against original_key: The original field name (for camelCase detection) """ import re for pattern in patterns: # Exact match if key_lower == pattern: return True # Pattern at start with delimiter: "id_something" or "id-something" if key_lower.startswith(pattern + "_") or key_lower.startswith(pattern + "-"): return True # Pattern at end with delimiter: "user_id" or "user-id" if key_lower.endswith("_" + pattern) or key_lower.endswith("-" + pattern): return True # Pattern in middle with delimiters: "some_id_field" if f"_{pattern}_" in key_lower or f"-{pattern}-" in key_lower: return True if f"_{pattern}-" in key_lower or f"-{pattern}_" in key_lower: return True # camelCase detection: Look for capitalized pattern in original key # e.g., "userId" should match "id" (as "Id") if original_key: # Pattern capitalized (e.g., "Id" for "id") cap_pattern = pattern.capitalize() # Look for capital letter at start of pattern, preceded by lowercase camel_regex = rf"(?<=[a-z]){re.escape(cap_pattern)}(?=[A-Z]|$)" if re.search(camel_regex, original_key): return True return False @classmethod def from_items(cls, items: list[dict[str, Any]]) -> ToolSignature: """Create signature from sample items.""" if not items: # HIGH FIX: Generate unique hash for empty outputs to prevent # different tools' empty responses from colliding into one pattern. # Use a random component to ensure uniqueness across tool types. import uuid # MEDIUM FIX #15: Use 24 chars (96 bits) instead of 16 (64 bits) to reduce collision risk empty_hash = hashlib.sha256(f"empty:{uuid.uuid4()}".encode()).hexdigest()[:24] return cls( structure_hash=empty_hash, field_count=0, has_nested_objects=False, has_arrays=False, max_depth=0, ) # MEDIUM FIX #13: Analyze multiple items (up to 5) to get representative structure # This catches cases where items have varying schemas sample_items = items[:5] if len(items) >= 5 else items # Merge field info from all sampled items all_fields: dict[str, set[str]] = {} # field_name -> set of types seen for item in sample_items: if not isinstance(item, dict): continue for key, value in item.items(): if key not in all_fields: all_fields[key] = set() # Determine type if isinstance(value, str): all_fields[key].add("string") elif isinstance(value, bool): all_fields[key].add("boolean") elif isinstance(value, (int, float)): all_fields[key].add("numeric") elif isinstance(value, list): all_fields[key].add("array") elif isinstance(value, dict): all_fields[key].add("object") else: all_fields[key].add("null") # Build field_info with most common type per field field_info: list[tuple[str, str]] = [] string_count = 0 numeric_count = 0 boolean_count = 0 array_count = 0 object_count = 0 has_nested = False has_arrays = False # MEDIUM FIX #12: Calculate actual max_depth from sampled items max_depth = 1 for item in sample_items: if isinstance(item, dict): item_depth = cls._calculate_depth(item) max_depth = max(max_depth, item_depth) # Pattern detection (heuristic field name matching) has_id = False has_score = False has_timestamp = False has_status = False has_error = False has_message = False for key, types in all_fields.items(): key_lower = key.lower() # Use most specific type if multiple seen (prefer non-null) types_no_null = types - {"null"} if len(types_no_null) == 1: field_type = types_no_null.pop() elif len(types_no_null) > 1: # Multiple types seen - mark as mixed but pick one for counting # Priority: object > array > string > numeric > boolean for t in ["object", "array", "string", "numeric", "boolean"]: if t in types_no_null: field_type = t break else: field_type = "mixed" elif types: field_type = types.pop() # Only null seen else: field_type = "null" # Count field types if field_type == "string": string_count += 1 elif field_type == "boolean": boolean_count += 1 elif field_type == "numeric": numeric_count += 1 elif field_type == "array": array_count += 1 has_arrays = True elif field_type == "object": object_count += 1 has_nested = True field_info.append((key, field_type)) # MEDIUM FIX #14: Pattern detection with word boundary matching # Prevents false positives like "hidden" matching "id" # Pass original key for camelCase detection if cls._matches_pattern(key_lower, ["id", "uuid", "guid"], key) or key_lower.endswith( "key" ): has_id = True if cls._matches_pattern( key_lower, ["score", "rank", "rating", "relevance", "priority"], key ): has_score = True if ( cls._matches_pattern(key_lower, ["time", "date", "timestamp"], key) or key_lower.endswith("_at") or key_lower in ["created", "updated"] ): has_timestamp = True if cls._matches_pattern(key_lower, ["status", "state"], key) or key_lower in [ "level", "type", "kind", ]: has_status = True if cls._matches_pattern(key_lower, ["error", "exception", "fail", "warning"], key): has_error = True if cls._matches_pattern( key_lower, ["message", "msg", "text", "content", "body", "description"], key ): has_message = True # Create structure hash # MEDIUM FIX #15: Use 24 chars (96 bits) instead of 16 (64 bits) for collision resistance sorted_fields = sorted(field_info) hash_input = json.dumps(sorted_fields, sort_keys=True) structure_hash = hashlib.sha256(hash_input.encode()).hexdigest()[:24] return cls( structure_hash=structure_hash, field_count=len(field_info), has_nested_objects=has_nested, has_arrays=has_arrays, max_depth=max_depth, string_field_count=string_count, numeric_field_count=numeric_count, boolean_field_count=boolean_count, array_field_count=array_count, object_field_count=object_count, has_id_like_field=has_id, has_score_like_field=has_score, has_timestamp_like_field=has_timestamp, has_status_like_field=has_status, has_error_like_field=has_error, has_message_like_field=has_message, ) @classmethod def from_dict(cls, data: dict[str, Any]) -> ToolSignature: """Create from dictionary.""" return cls(**{k: v for k, v in data.items() if k in cls.__dataclass_fields__}) @dataclass class FieldSemantics: """Learned semantics for a field based on retrieval patterns. This is the evolution of TOIN - we learn WHAT fields mean from HOW users retrieve them. No hardcoded patterns, no assumptions. Learning process: 1. User retrieves items where field X has value Y 2. TOIN records: field_hash, value_hash, retrieval context 3. After N retrievals, TOIN infers: "This field behaves like an error indicator" 4. SmartCrusher uses this learned signal (O(1) lookup, zero latency) Privacy: All field names and values are hashed (SHA256[:8]). """ field_hash: str # SHA256[:8] of field name # Inferred semantic type (learned from retrieval patterns, NOT hardcoded) # These are behavioral categories, not syntactic patterns: # - "identifier": Users query by exact value (e.g., "show me item X") # - "error_indicator": Users retrieve when value != most common value # - "score": Users retrieve top-N by this field # - "status": Low cardinality, specific values trigger retrieval # - "temporal": Users query by time ranges # - "content": Users do text search on this field inferred_type: FieldSemanticType = "unknown" confidence: float = 0.0 # 0.0 = no data, 1.0 = high confidence # Value patterns (all hashed for privacy) # important_value_hashes: values that triggered retrieval # default_value_hash: most common value (probably NOT important) important_value_hashes: list[str] = field(default_factory=list) default_value_hash: str | None = None value_retrieval_frequency: dict[str, int] = field(default_factory=dict) # value_hash -> count # Value statistics (for inferring type) total_unique_values_seen: int = 0 total_values_seen: int = 0 most_common_value_frequency: float = 0.0 # Fraction of items with most common value # Query patterns (anonymized) # Tracks HOW users query this field (equals, not-equals, greater-than, etc.) query_operator_frequency: dict[str, int] = field(default_factory=dict) # operator -> count # Learning metadata retrieval_count: int = 0 compression_count: int = 0 # How many times we've seen this field in compression last_updated: float = 0.0 # Bounds for memory management MAX_IMPORTANT_VALUES: int = 50 MAX_VALUE_FREQUENCY_ENTRIES: int = 100 def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "field_hash": self.field_hash, "inferred_type": self.inferred_type, "confidence": self.confidence, "important_value_hashes": self.important_value_hashes[: self.MAX_IMPORTANT_VALUES], "default_value_hash": self.default_value_hash, "value_retrieval_frequency": dict( sorted( self.value_retrieval_frequency.items(), key=lambda x: x[1], reverse=True, )[: self.MAX_VALUE_FREQUENCY_ENTRIES] ), "total_unique_values_seen": self.total_unique_values_seen, "total_values_seen": self.total_values_seen, "most_common_value_frequency": self.most_common_value_frequency, "query_operator_frequency": self.query_operator_frequency, "retrieval_count": self.retrieval_count, "compression_count": self.compression_count, "last_updated": self.last_updated, } @classmethod def from_dict(cls, data: dict[str, Any]) -> FieldSemantics: """Create from dictionary.""" # Filter to valid fields only valid_fields = { "field_hash", "inferred_type", "confidence", "important_value_hashes", "default_value_hash", "value_retrieval_frequency", "total_unique_values_seen", "total_values_seen", "most_common_value_frequency", "query_operator_frequency", "retrieval_count", "compression_count", "last_updated", } filtered = {k: v for k, v in data.items() if k in valid_fields} return cls(**filtered) def record_retrieval_value(self, value_hash: str, operator: str = "=") -> None: """Record that a value was retrieved for this field. Args: value_hash: SHA256[:8] hash of the retrieved value. operator: Query operator used ("=", "!=", ">", "<", "contains", etc.) """ import time self.retrieval_count += 1 self.last_updated = time.time() # Track value frequency self.value_retrieval_frequency[value_hash] = ( self.value_retrieval_frequency.get(value_hash, 0) + 1 ) # Bound the frequency dict if len(self.value_retrieval_frequency) > self.MAX_VALUE_FREQUENCY_ENTRIES: sorted_items = sorted( self.value_retrieval_frequency.items(), key=lambda x: x[1], reverse=True, )[: self.MAX_VALUE_FREQUENCY_ENTRIES] self.value_retrieval_frequency = dict(sorted_items) # Track important values (values that get retrieved) if value_hash not in self.important_value_hashes: self.important_value_hashes.append(value_hash) if len(self.important_value_hashes) > self.MAX_IMPORTANT_VALUES: # Keep most frequently retrieved values self.important_value_hashes = sorted( self.important_value_hashes, key=lambda v: self.value_retrieval_frequency.get(v, 0), reverse=True, )[: self.MAX_IMPORTANT_VALUES] # Track query operators self.query_operator_frequency[operator] = self.query_operator_frequency.get(operator, 0) + 1 def record_compression_stats( self, unique_values: int, total_values: int, most_common_value_hash: str | None, most_common_frequency: float, ) -> None: """Record statistics from compression for type inference. Args: unique_values: Number of unique values seen for this field. total_values: Total number of items with this field. most_common_value_hash: Hash of the most common value. most_common_frequency: Fraction of items with the most common value. """ import time self.compression_count += 1 self.last_updated = time.time() # Update rolling statistics n = self.compression_count self.total_unique_values_seen = int( (self.total_unique_values_seen * (n - 1) + unique_values) / n ) self.total_values_seen = int((self.total_values_seen * (n - 1) + total_values) / n) self.most_common_value_frequency = ( self.most_common_value_frequency * (n - 1) + most_common_frequency ) / n # Track default value (most common) if most_common_value_hash and most_common_frequency > 0.5: self.default_value_hash = most_common_value_hash def infer_type(self) -> None: """Infer semantic type from accumulated statistics. This is the learning algorithm - purely data-driven, no hardcoded patterns. """ # Need minimum data to infer min_retrievals = 3 min_compressions = 2 if self.retrieval_count < min_retrievals or self.compression_count < min_compressions: self.inferred_type = "unknown" self.confidence = 0.0 return # Calculate metrics uniqueness_ratio = self.total_unique_values_seen / max(1, self.total_values_seen) has_dominant_default = self.most_common_value_frequency > 0.7 retrieval_diversity = len(self.value_retrieval_frequency) / max(1, self.retrieval_count) # Check query operator patterns total_ops = sum(self.query_operator_frequency.values()) equals_ratio = self.query_operator_frequency.get("=", 0) / max(1, total_ops) range_ratio = ( self.query_operator_frequency.get(">", 0) + self.query_operator_frequency.get("<", 0) + self.query_operator_frequency.get(">=", 0) + self.query_operator_frequency.get("<=", 0) ) / max(1, total_ops) contains_ratio = self.query_operator_frequency.get("contains", 0) / max(1, total_ops) # Inference logic (data-driven, no field name patterns) inferred: FieldSemanticType = "unknown" confidence = 0.0 # IDENTIFIER: High uniqueness + exact match queries if uniqueness_ratio > 0.8 and equals_ratio > 0.7: inferred = "identifier" confidence = min(0.9, uniqueness_ratio * equals_ratio) # ERROR_INDICATOR: Has dominant default + retrievals are for non-default values elif has_dominant_default and self.default_value_hash: # Check if retrieved values are different from default default_retrieval_count = self.value_retrieval_frequency.get(self.default_value_hash, 0) non_default_retrieval_ratio = 1 - ( default_retrieval_count / max(1, self.retrieval_count) ) if non_default_retrieval_ratio > 0.7: inferred = "error_indicator" confidence = min( 0.9, non_default_retrieval_ratio * self.most_common_value_frequency ) # STATUS: Low uniqueness + specific values retrieved elif uniqueness_ratio < 0.2 and retrieval_diversity < 0.5: inferred = "status" confidence = min(0.85, (1 - uniqueness_ratio) * (1 - retrieval_diversity)) # SCORE: Range queries or sorted access patterns elif range_ratio > 0.5: inferred = "score" confidence = min(0.85, range_ratio) # TEMPORAL: Range queries + high uniqueness (likely timestamps) elif range_ratio > 0.3 and uniqueness_ratio > 0.7: inferred = "temporal" confidence = min(0.8, range_ratio * uniqueness_ratio) # CONTENT: Contains/text search queries elif contains_ratio > 0.5: inferred = "content" confidence = min(0.85, contains_ratio) # Apply minimum confidence threshold if confidence < 0.3: inferred = "unknown" confidence = 0.0 self.inferred_type = inferred self.confidence = confidence def is_value_important(self, value_hash: str) -> bool: """Check if a specific value is considered important. A value is important if: 1. It's in the important_value_hashes list (has been retrieved) 2. It's NOT the default value (for error_indicator type) Args: value_hash: SHA256[:8] hash of the value to check. Returns: True if this value should be preserved during compression. """ # If we don't have enough data, be conservative if self.confidence < 0.3: return False # For error_indicator: non-default values are important if self.inferred_type == "error_indicator": if self.default_value_hash and value_hash != self.default_value_hash: return True # For any type: values that have been retrieved are important if value_hash in self.important_value_hashes: return True # For status: check if this value has been retrieved if self.inferred_type == "status": return value_hash in self.value_retrieval_frequency return False @dataclass class CompressionEvent: """Record of a single compression decision (anonymized). This captures WHAT happened, not WHAT the data was. """ # Tool identification (anonymized) tool_signature: ToolSignature # Compression metrics original_item_count: int compressed_item_count: int compression_ratio: float # compressed / original original_tokens: int compressed_tokens: int token_reduction_ratio: float # 1 - (compressed / original) # Strategy used strategy: str # "top_n", "time_series", "smart_sample", "skip", etc. strategy_reason: str | None = None # "high_variance", "has_score_field", etc. # Crushability analysis results crushability_score: float | None = None # 0.0 = don't crush, 1.0 = safe to crush crushability_reason: str | None = None # Field distributions (anonymized) field_distributions: list[FieldDistribution] = field(default_factory=list) # What was preserved kept_first_n: int = 0 kept_last_n: int = 0 kept_errors: int = 0 kept_anomalies: int = 0 kept_by_relevance: int = 0 kept_by_score: int = 0 # Timing timestamp: float = 0.0 processing_time_ms: float = 0.0 def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "tool_signature": self.tool_signature.to_dict(), "original_item_count": self.original_item_count, "compressed_item_count": self.compressed_item_count, "compression_ratio": self.compression_ratio, "original_tokens": self.original_tokens, "compressed_tokens": self.compressed_tokens, "token_reduction_ratio": self.token_reduction_ratio, "strategy": self.strategy, "strategy_reason": self.strategy_reason, "crushability_score": self.crushability_score, "crushability_reason": self.crushability_reason, "field_distributions": [f.to_dict() for f in self.field_distributions], "kept_first_n": self.kept_first_n, "kept_last_n": self.kept_last_n, "kept_errors": self.kept_errors, "kept_anomalies": self.kept_anomalies, "kept_by_relevance": self.kept_by_relevance, "kept_by_score": self.kept_by_score, "timestamp": self.timestamp, "processing_time_ms": self.processing_time_ms, } @dataclass class RetrievalStats: """Aggregated retrieval statistics for a tool signature. This tracks how often compression decisions needed correction. """ tool_signature_hash: str # Reference to ToolSignature.structure_hash # Retrieval counts total_compressions: int = 0 total_retrievals: int = 0 full_retrievals: int = 0 # Retrieved everything search_retrievals: int = 0 # Used search filter # Derived metrics @property def retrieval_rate(self) -> float: """Fraction of compressions that triggered 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 # Query pattern analysis (no actual queries, just patterns) query_field_frequency: dict[str, int] = field(default_factory=dict) # field_hash -> count def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "tool_signature_hash": self.tool_signature_hash, "total_compressions": self.total_compressions, "total_retrievals": self.total_retrievals, "full_retrievals": self.full_retrievals, "search_retrievals": self.search_retrievals, "retrieval_rate": self.retrieval_rate, "full_retrieval_rate": self.full_retrieval_rate, "query_field_frequency": self.query_field_frequency, } @dataclass class AnonymizedToolStats: """Complete anonymized statistics for a tool type. This is what gets aggregated across users to build the data flywheel. """ # Tool identification signature: ToolSignature # Compression statistics total_compressions: int = 0 total_items_seen: int = 0 total_items_kept: int = 0 avg_compression_ratio: float = 0.0 avg_token_reduction: float = 0.0 # Strategy distribution strategy_counts: dict[str, int] = field(default_factory=dict) # strategy -> count strategy_success_rate: dict[str, float] = field( default_factory=dict ) # strategy -> success rate # Retrieval statistics retrieval_stats: RetrievalStats | None = None # Learned optimal settings recommended_min_items: int | None = None recommended_preserve_fields: list[str] = field(default_factory=list) # field hashes skip_compression_recommended: bool = False # Confidence in recommendations sample_size: int = 0 confidence: float = 0.0 # 0.0 = no confidence, 1.0 = high confidence def to_dict(self) -> dict[str, Any]: """Convert to dictionary for serialization.""" return { "signature": self.signature.to_dict(), "total_compressions": self.total_compressions, "total_items_seen": self.total_items_seen, "total_items_kept": self.total_items_kept, "avg_compression_ratio": self.avg_compression_ratio, "avg_token_reduction": self.avg_token_reduction, "strategy_counts": self.strategy_counts, "strategy_success_rate": self.strategy_success_rate, "retrieval_stats": self.retrieval_stats.to_dict() if self.retrieval_stats else None, "recommended_min_items": self.recommended_min_items, "recommended_preserve_fields": self.recommended_preserve_fields, "skip_compression_recommended": self.skip_compression_recommended, "sample_size": self.sample_size, "confidence": self.confidence, } @classmethod def from_dict(cls, data: dict[str, Any]) -> AnonymizedToolStats: """Create from dictionary. Note: This method does not mutate the input dictionary. """ # Use .get() instead of .pop() to avoid mutating input signature_data = data.get("signature", {}) signature = ToolSignature.from_dict(signature_data) retrieval_data = data.get("retrieval_stats") retrieval_stats = None if retrieval_data: # Copy query_field_frequency to avoid mutation issues query_freq = retrieval_data.get("query_field_frequency", {}) retrieval_stats = RetrievalStats( tool_signature_hash=retrieval_data.get("tool_signature_hash", ""), total_compressions=retrieval_data.get("total_compressions", 0), total_retrievals=retrieval_data.get("total_retrievals", 0), full_retrievals=retrieval_data.get("full_retrievals", 0), search_retrievals=retrieval_data.get("search_retrievals", 0), query_field_frequency=dict(query_freq) if query_freq else {}, ) # Filter to only dataclass fields, excluding signature and retrieval_stats # which we've already handled excluded_keys = {"signature", "retrieval_stats"} filtered_data: dict[str, Any] = {} for k, v in data.items(): if k not in cls.__dataclass_fields__ or k in excluded_keys: continue # Deep copy mutable values to avoid corruption if caller modifies input if isinstance(v, dict): filtered_data[k] = dict(v) elif isinstance(v, list): filtered_data[k] = list(v) # type: ignore[assignment] else: filtered_data[k] = v return cls( signature=signature, retrieval_stats=retrieval_stats, **filtered_data, # type: ignore[arg-type] )