Spaces:
Running
Running
| """ | |
| Data Consistency Insights for AI Debug System | |
| Analyzes data consistency across distributed components and detects: | |
| - Data flow across distributed nodes | |
| - State divergence (same data, different states) | |
| - Replication completion verification | |
| - Data synchronization issues | |
| Example insights: | |
| - "Data sent to 5 nodes, 4 confirmed, 1 pending" | |
| - "Data X saved on Node 1 but not Node 2" | |
| - "Replication lag detected across 3 nodes" | |
| """ | |
| import asyncio | |
| from collections import defaultdict | |
| from datetime import datetime, timedelta | |
| from typing import Any, Dict, List, Optional, Set, Tuple | |
| from sqlalchemy.orm import Session | |
| from sqlalchemy import and_, or_ | |
| from core.models import ( | |
| DebugEvent, | |
| DebugInsight, | |
| DebugStateSnapshot, | |
| DebugInsightType, | |
| DebugInsightSeverity, | |
| ) | |
| from core.structured_logger import StructuredLogger | |
| class ConsistencyInsightGenerator: | |
| """ | |
| Generates insights about data consistency across distributed components. | |
| Tracks data flow through the system and detects inconsistencies, | |
| replication delays, and synchronization issues. | |
| """ | |
| def __init__(self, db_session: Session): | |
| """ | |
| Initialize consistency insight generator. | |
| Args: | |
| db_session: SQLAlchemy database session | |
| """ | |
| self.logger = StructuredLogger(__name__) | |
| self.db = db_session | |
| async def analyze_data_flow( | |
| self, | |
| operation_id: str, | |
| component_ids: List[str], | |
| component_type: str = "agent", | |
| ) -> Optional[DebugInsight]: | |
| """ | |
| Analyze data flow across distributed components. | |
| Tracks how data propagates through the system and identifies | |
| any components that haven't received or acknowledged the data. | |
| Args: | |
| operation_id: Operation to analyze | |
| component_ids: Expected components in the flow | |
| component_type: Type of components | |
| Returns: | |
| Consistency insight or None | |
| """ | |
| try: | |
| # Find all state snapshots for this operation | |
| snapshots = ( | |
| self.db.query(DebugStateSnapshot) | |
| .filter( | |
| and_( | |
| DebugStateSnapshot.operation_id == operation_id, | |
| DebugStateSnapshot.component_type == component_type, | |
| ) | |
| ) | |
| .order_by(DebugStateSnapshot.captured_at) | |
| .all() | |
| ) | |
| if not snapshots: | |
| return None | |
| # Track which components have snapshots | |
| components_with_snapshots = set(s.component_id for s in snapshots) | |
| # Find missing components | |
| missing_components = set(component_ids) - components_with_snapshots | |
| if missing_components: | |
| # Data hasn't reached all components yet | |
| return DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.WARNING.value, | |
| title="Incomplete data propagation", | |
| description=f"Data sent to {len(component_ids)} nodes but only " | |
| f"{len(components_with_snapshots)} confirmed receipt", | |
| summary=f"Data sent to {len(component_ids)} nodes, {len(missing_components)} pending", | |
| evidence={ | |
| "operation_id": operation_id, | |
| "expected_components": component_ids, | |
| "confirmed_components": list(components_with_snapshots), | |
| "missing_components": list(missing_components), | |
| "propagation_rate": len(components_with_snapshots) / len(component_ids), | |
| }, | |
| confidence_score=0.95, | |
| suggestions=[ | |
| f"Check connectivity to missing nodes: {', '.join(missing_components)}", | |
| "Verify message queues are processing", | |
| "Check for network partitions", | |
| ], | |
| scope="distributed", | |
| affected_components=[ | |
| {"type": component_type, "id": comp_id} for comp_id in component_ids | |
| ], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| # All components have data - check timing consistency | |
| capture_times = [s.captured_at for s in snapshots if s.captured_at] | |
| if len(capture_times) > 1: | |
| min_time = min(capture_times) | |
| max_time = max(capture_times) | |
| replication_lag = (max_time - min_time).total_seconds() | |
| if replication_lag > 5.0: # 5 second threshold | |
| return DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.WARNING.value, | |
| title="Replication lag detected", | |
| description=f"Data replicated to all {len(component_ids)} nodes but with " | |
| f"{replication_lag:.1f}s delay between first and last", | |
| summary=f"Replication lag of {replication_lag:.1f}s across {len(component_ids)} nodes", | |
| evidence={ | |
| "operation_id": operation_id, | |
| "replication_lag_seconds": replication_lag, | |
| "first_confirmation": min_time.isoformat() if min_time else None, | |
| "last_confirmation": max_time.isoformat() if max_time else None, | |
| }, | |
| confidence_score=0.90, | |
| suggestions=[ | |
| "Investigate slow components", | |
| "Check network latency between nodes", | |
| "Review resource utilization", | |
| ], | |
| scope="distributed", | |
| affected_components=[ | |
| {"type": component_type, "id": comp_id} for comp_id in component_ids | |
| ], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| # All good - data consistent | |
| return DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.INFO.value, | |
| title="Data consistent across all nodes", | |
| description=f"Data successfully propagated to all {len(component_ids)} components", | |
| summary=f"Data sent to {len(component_ids)} nodes, all confirmed", | |
| evidence={ | |
| "operation_id": operation_id, | |
| "component_count": len(component_ids), | |
| "replication_complete": True, | |
| }, | |
| confidence_score=1.0, | |
| scope="distributed", | |
| affected_components=[ | |
| {"type": component_type, "id": comp_id} for comp_id in component_ids | |
| ], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| except Exception as e: | |
| self.logger.error( | |
| "Failed to analyze data flow", | |
| operation_id=operation_id, | |
| error=str(e), | |
| ) | |
| return None | |
| async def detect_state_divergence( | |
| self, | |
| operation_id: str, | |
| component_type: str = "agent", | |
| ) -> Optional[DebugInsight]: | |
| """ | |
| Detect state divergence across components. | |
| Identifies when the same data has different states on different nodes, | |
| indicating a synchronization issue. | |
| Args: | |
| operation_id: Operation to analyze | |
| component_type: Type of components | |
| Returns: | |
| Consistency insight or None | |
| """ | |
| try: | |
| # Get all state snapshots for this operation | |
| snapshots = ( | |
| self.db.query(DebugStateSnapshot) | |
| .filter( | |
| and_( | |
| DebugStateSnapshot.operation_id == operation_id, | |
| DebugStateSnapshot.component_type == component_type, | |
| ) | |
| ) | |
| .order_by(DebugStateSnapshot.captured_at.desc()) | |
| .all() | |
| ) | |
| if len(snapshots) < 2: | |
| return None # Need at least 2 components to compare | |
| # Get latest snapshot for each component | |
| latest_snapshots = {} | |
| for snapshot in snapshots: | |
| if snapshot.component_id not in latest_snapshots: | |
| latest_snapshots[snapshot.component_id] = snapshot | |
| # Compare states across components | |
| inconsistencies = self._compare_states(latest_snapshots) | |
| if inconsistencies: | |
| component_ids = list(latest_snapshots.keys()) | |
| return DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.CRITICAL.value, | |
| title="State divergence detected", | |
| description=f"Found {len(inconsistencies)} state inconsistencies across " | |
| f"{len(component_ids)} components", | |
| summary=f"Data differs across {len(component_ids)} components for {len(inconsistencies)} keys", | |
| evidence={ | |
| "operation_id": operation_id, | |
| "inconsistencies": inconsistencies, | |
| "affected_keys": list(inconsistencies.keys()), | |
| }, | |
| confidence_score=0.92, | |
| suggestions=[ | |
| "Review synchronization logic", | |
| "Check for concurrent modifications", | |
| "Verify conflict resolution mechanisms", | |
| "Investigate data partitioning issues", | |
| ], | |
| scope="distributed", | |
| affected_components=[ | |
| {"type": component_type, "id": comp_id} for comp_id in component_ids | |
| ], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| return None | |
| except Exception as e: | |
| self.logger.error( | |
| "Failed to detect state divergence", | |
| operation_id=operation_id, | |
| error=str(e), | |
| ) | |
| return None | |
| async def verify_replication_completion( | |
| self, | |
| operation_id: str, | |
| expected_replicas: int, | |
| component_type: str = "agent", | |
| ) -> Optional[DebugInsight]: | |
| """ | |
| Verify that data has been replicated to all expected nodes. | |
| Args: | |
| operation_id: Operation to verify | |
| expected_replicas: Number of expected replicas | |
| component_type: Type of components | |
| Returns: | |
| Consistency insight or None | |
| """ | |
| try: | |
| # Count unique components that have data | |
| component_count = ( | |
| self.db.query(DebugStateSnapshot) | |
| .filter( | |
| and_( | |
| DebugStateSnapshot.operation_id == operation_id, | |
| DebugStateSnapshot.component_type == component_type, | |
| ) | |
| ) | |
| .distinct(DebugStateSnapshot.component_id) | |
| .count() | |
| ) | |
| if component_count < expected_replicas: | |
| return DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.WARNING.value, | |
| title="Incomplete replication", | |
| description=f"Expected {expected_replicas} replicas but only {component_count} confirmed", | |
| summary=f"{expected_replicas - component_count} replicas missing", | |
| evidence={ | |
| "operation_id": operation_id, | |
| "expected_replicas": expected_replicas, | |
| "actual_replicas": component_count, | |
| "completion_rate": component_count / expected_replicas, | |
| }, | |
| confidence_score=0.98, | |
| suggestions=[ | |
| "Check replication logs", | |
| "Verify network connectivity", | |
| "Review replication configuration", | |
| ], | |
| scope="distributed", | |
| affected_components=[{"type": component_type}], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| # Replication complete | |
| return DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.INFO.value, | |
| title="Replication complete", | |
| description=f"Data successfully replicated to all {expected_replicas} nodes", | |
| summary=f"All {expected_replicas} replicas confirmed", | |
| evidence={ | |
| "operation_id": operation_id, | |
| "replica_count": component_count, | |
| }, | |
| confidence_score=1.0, | |
| scope="distributed", | |
| affected_components=[{"type": component_type}], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| except Exception as e: | |
| self.logger.error( | |
| "Failed to verify replication completion", | |
| operation_id=operation_id, | |
| error=str(e), | |
| ) | |
| return None | |
| async def analyze_sync_patterns( | |
| self, | |
| time_range: str = "last_1h", | |
| ) -> List[DebugInsight]: | |
| """ | |
| Analyze synchronization patterns across all operations. | |
| Identifies systemic issues with data synchronization. | |
| Args: | |
| time_range: Time range to analyze | |
| Returns: | |
| List of consistency insights | |
| """ | |
| try: | |
| insights = [] | |
| time_filter = self._parse_time_range(time_range) | |
| # Find operations with incomplete replication | |
| operations = ( | |
| self.db.query(DebugStateSnapshot.operation_id) | |
| .filter(DebugStateSnapshot.captured_at >= time_filter) | |
| .group_by(DebugStateSnapshot.operation_id) | |
| .all() | |
| ) | |
| for (operation_id,) in operations: | |
| # Count components per operation | |
| component_count = ( | |
| self.db.query(DebugStateSnapshot) | |
| .filter( | |
| and_( | |
| DebugStateSnapshot.operation_id == operation_id, | |
| DebugStateSnapshot.captured_at >= time_filter, | |
| ) | |
| ) | |
| .distinct(DebugStateSnapshot.component_id) | |
| .count() | |
| ) | |
| # If an operation has only 1 component, it might not be replicating | |
| if component_count == 1: | |
| insights.append( | |
| DebugInsight( | |
| insight_type=DebugInsightType.CONSISTENCY.value, | |
| severity=DebugInsightSeverity.INFO.value, | |
| title="Single-component operation detected", | |
| description=f"Operation {operation_id} has data on only 1 component", | |
| summary=f"Operation may not be configured for replication", | |
| evidence={"operation_id": operation_id}, | |
| confidence_score=0.70, | |
| suggestions=[ | |
| "Verify replication is configured", | |
| "Check if this is a local-only operation", | |
| ], | |
| scope="distributed", | |
| affected_components=[], | |
| generated_at=datetime.utcnow(), | |
| ) | |
| ) | |
| return insights | |
| except Exception as e: | |
| self.logger.error("Failed to analyze sync patterns", error=str(e)) | |
| return [] | |
| def _compare_states( | |
| self, | |
| snapshots: Dict[str, DebugStateSnapshot], | |
| ) -> Dict[str, Any]: | |
| """ | |
| Compare states across multiple components. | |
| Args: | |
| snapshots: Component ID -> Snapshot mapping | |
| Returns: | |
| Dictionary of inconsistent keys and their values | |
| """ | |
| inconsistencies = {} | |
| # Get all keys from the first snapshot | |
| first_snapshot = list(snapshots.values())[0] | |
| all_keys = set(first_snapshot.state_data.keys()) if first_snapshot.state_data else set() | |
| # For each key, compare values across all snapshots | |
| for key in all_keys: | |
| values = {} | |
| for comp_id, snapshot in snapshots.items(): | |
| if snapshot.state_data and key in snapshot.state_data: | |
| values[comp_id] = snapshot.state_data[key] | |
| # Check if all values are the same | |
| unique_values = set(str(v) for v in values.values()) | |
| if len(unique_values) > 1: | |
| inconsistencies[key] = { | |
| "values": values, | |
| "divergence_detected": True, | |
| } | |
| return inconsistencies | |
| def _parse_time_range(self, time_range: str) -> datetime: | |
| """Parse time range string to datetime.""" | |
| now = datetime.utcnow() | |
| if time_range == "last_1h": | |
| return now - timedelta(hours=1) | |
| elif time_range == "last_24h": | |
| return now - timedelta(hours=24) | |
| elif time_range == "last_7d": | |
| return now - timedelta(days=7) | |
| else: | |
| return now - timedelta(hours=1) | |