Add multi-agent architecture implementation with detailed documentation

- Alert Ingestion Agent: Normalize and deduplicate alerts
- Correlation Agent: Group related alerts into incidents
- Analysis Agent: AI-powered root cause analysis with Ollama
- Response Agent: Generate recommendations and integrations

Includes comprehensive agent-to-agent communication patterns:
- Synchronous request-response for critical paths
- Asynchronous event publishing for integrations
- Shared data store (PostgreSQL + Redis)
- Support for Slack, PagerDuty, OpsGenie integrations

agents/__init__.py

@@ -0,0 +1,24 @@
+"""
+Multi-Agent System for Incident Management
+
+This module implements a distributed agent architecture where specialized agents
+collaborate to detect, correlate, analyze, and respond to infrastructure incidents.
+
+Agent Types:
+    1. AlertIngestionAgent - Normalizes and dedups raw alerts
+    2. CorrelationAgent - Groups related alerts into incidents
+    3. AnalysisAgent - Performs AI-powered root cause analysis
+    4. ResponseAgent - Generates actionable recommendations
+"""
+
+from .alert_agent import AlertIngestionAgent
+from .correlation_agent import CorrelationAgent
+from .analysis_agent import AnalysisAgent
+from .response_agent import ResponseAgent
+
+__all__ = [
+    "AlertIngestionAgent",
+    "CorrelationAgent",
+    "AnalysisAgent",
+    "ResponseAgent"
+]

agents/alert_agent.py

@@ -0,0 +1,235 @@
+"""
+Alert Ingestion Agent
+
+Receives raw alerts from monitoring systems, normalizes them, and deduplicates
+based on fingerprints. First agent in the processing pipeline.
+
+Communication: Synchronous → Correlation Agent
+Data Storage: PostgreSQL (alerts), Redis (cache)
+"""
+
+import hashlib
+import logging
+from typing import Dict, Any, Optional
+from datetime import datetime, timedelta
+import aioredis
+from sqlalchemy.orm import Session
+
+from src.models.database import Alert
+from src.schemas import AlertCreate
+
+logger = logging.getLogger(__name__)
+
+
+class AlertIngestionAgent:
+    """Ingests and normalizes raw alerts from monitoring systems"""
+    
+    def __init__(self, db: Session, redis_client: Optional[aioredis.Redis] = None):
+        self.db = db
+        self.redis = redis_client
+        self.cache_ttl = 300  # 5 minutes
+    
+    async def ingest_alert(self, raw_alert: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Main entry point for alert ingestion.
+        
+        Timeline:
+        T+0ms:   Receive raw alert from monitoring system
+        T+10ms:  Normalize to AIMS schema
+        T+20ms:  Calculate fingerprint
+        T+30ms:  Check Redis cache for duplicates
+        T+40ms:  Store in PostgreSQL if new
+        T+50ms:  Return normalized alert
+        
+        Args:
+            raw_alert: Alert from Prometheus, Grafana, CloudWatch, etc.
+            
+        Returns:
+            Normalized alert dict ready for correlation
+        """
+        logger.info(f"[ALERT_AGENT] Ingesting alert from {raw_alert.get('source', 'unknown')}")
+        
+        try:
+            # Step 1: Normalize alert (T+10ms)
+            normalized = self._normalize_alert(raw_alert)
+            logger.debug(f"[ALERT_AGENT] Normalized alert: {normalized.get('title')}")
+            
+            # Step 2: Calculate fingerprint (T+20ms)
+            fingerprint = self._generate_fingerprint(normalized)
+            logger.debug(f"[ALERT_AGENT] Fingerprint: {fingerprint}")
+            
+            # Step 3: Check for duplicates (T+30ms)
+            duplicate_count = await self._check_duplicate(fingerprint)
+            if duplicate_count:
+                logger.info(f"[ALERT_AGENT] Duplicate alert detected. Count: {duplicate_count}")
+                normalized['duplicate_count'] = duplicate_count
+                return {
+                    "status": "deduplicated",
+                    "alert": normalized,
+                    "action": "increment_counter"
+                }
+            
+            # Step 4: Store in database (T+40ms)
+            alert_id = await self._store_alert(normalized, fingerprint)
+            logger.info(f"[ALERT_AGENT] Alert stored with ID: {alert_id}")
+            
+            # Step 5: Cache fingerprint (T+45ms)
+            await self._cache_fingerprint(fingerprint)
+            
+            # Step 6: Return result (T+50ms)
+            return {
+                "status": "created",
+                "alert_id": alert_id,
+                "alert": normalized,
+                "fingerprint": fingerprint,
+                "processing_time_ms": 50
+            }
+            
+        except Exception as e:
+            logger.error(f"[ALERT_AGENT] Error ingesting alert: {e}", exc_info=True)
+            raise
+    
+    def _normalize_alert(self, raw_alert: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Convert alert from various formats to AIMS standard format.
+        
+        Supports:
+        - Prometheus AlertManager format
+        - Grafana alert format
+        - CloudWatch format
+        - Custom webhook format
+        """
+        source = raw_alert.get('source', 'unknown')
+        
+        if source == 'prometheus':
+            return self._normalize_prometheus(raw_alert)
+        elif source == 'grafana':
+            return self._normalize_grafana(raw_alert)
+        elif source == 'cloudwatch':
+            return self._normalize_cloudwatch(raw_alert)
+        else:
+            return self._normalize_generic(raw_alert)
+    
+    def _normalize_prometheus(self, alert: Dict[str, Any]) -> Dict[str, Any]:
+        """Normalize Prometheus format alert"""
+        return {
+            'source': 'prometheus',
+            'title': alert.get('alert', alert.get('name', 'Unknown')),
+            'description': alert.get('description', ''),
+            'severity': alert.get('severity', 'warning').lower(),
+            'service': alert.get('labels', {}).get('job', 'unknown'),
+            'category': alert.get('labels', {}).get('category', 'other'),
+            'metrics': alert.get('value'),
+            'timestamp': datetime.utcnow().isoformat(),
+            'labels': alert.get('labels', {}),
+            'status': 'new'
+        }
+    
+    def _normalize_grafana(self, alert: Dict[str, Any]) -> Dict[str, Any]:
+        """Normalize Grafana format alert"""
+        return {
+            'source': 'grafana',
+            'title': alert.get('title', 'Grafana Alert'),
+            'description': alert.get('message', ''),
+            'severity': alert.get('severity', 'warning').lower(),
+            'service': alert.get('service', 'unknown'),
+            'category': alert.get('category', 'other'),
+            'metrics': alert.get('data'),
+            'timestamp': datetime.utcnow().isoformat(),
+            'url': alert.get('ruleUrl'),
+            'status': 'new'
+        }
+    
+    def _normalize_cloudwatch(self, alert: Dict[str, Any]) -> Dict[str, Any]:
+        """Normalize CloudWatch format alert"""
+        return {
+            'source': 'cloudwatch',
+            'title': alert.get('AlarmName', 'CloudWatch Alarm'),
+            'description': alert.get('AlarmDescription', ''),
+            'severity': 'critical' if alert.get('StateValue') == 'ALARM' else 'warning',
+            'service': alert.get('Trigger', {}).get('Namespace', 'aws'),
+            'category': alert.get('Trigger', {}).get('MetricName', 'other'),
+            'metrics': alert.get('StateChangeTime'),
+            'timestamp': datetime.utcnow().isoformat(),
+            'status': 'new'
+        }
+    
+    def _normalize_generic(self, alert: Dict[str, Any]) -> Dict[str, Any]:
+        """Normalize generic/webhook format alert"""
+        return {
+            'source': alert.get('source', 'custom'),
+            'title': alert.get('title', alert.get('name', 'Alert')),
+            'description': alert.get('description', ''),
+            'severity': alert.get('severity', 'warning').lower(),
+            'service': alert.get('service', 'unknown'),
+            'category': alert.get('category', 'other'),
+            'metrics': alert.get('metrics', {}),
+            'timestamp': alert.get('timestamp', datetime.utcnow().isoformat()),
+            'status': 'new'
+        }
+    
+    def _generate_fingerprint(self, normalized_alert: Dict[str, Any]) -> str:
+        """
+        Generate unique fingerprint for deduplication.
+        Fingerprint = hash(service + alert_type + severity)
+        """
+        key_parts = [
+            normalized_alert.get('service', 'unknown'),
+            normalized_alert.get('title', 'unknown'),
+            normalized_alert.get('category', 'unknown')
+        ]
+        key_string = '|'.join(key_parts)
+        return hashlib.sha256(key_string.encode()).hexdigest()[:16]
+    
+    async def _check_duplicate(self, fingerprint: str) -> int:
+        """
+        Check if alert with same fingerprint already exists.
+        Returns counter of how many times this alert has fired.
+        """
+        if not self.redis:
+            return 0
+        
+        cache_key = f"alert:fingerprint:{fingerprint}"
+        try:
+            count = await self.redis.incr(cache_key)
+            await self.redis.expire(cache_key, self.cache_ttl)
+            return count
+        except Exception as e:
+            logger.warning(f"[ALERT_AGENT] Redis check failed: {e}")
+            return 0
+    
+    async def _store_alert(self, normalized_alert: Dict[str, Any], fingerprint: str) -> str:
+        """Store normalized alert in PostgreSQL"""
+        try:
+            alert = Alert(
+                source=normalized_alert['source'],
+                title=normalized_alert['title'],
+                description=normalized_alert['description'],
+                severity=normalized_alert['severity'],
+                service=normalized_alert['service'],
+                category=normalized_alert['category'],
+                status='new',
+                fingerprint=fingerprint,
+                metrics=normalized_alert.get('metrics', {}),
+                labels=normalized_alert.get('labels', {}),
+                created_at=datetime.utcnow()
+            )
+            self.db.add(alert)
+            self.db.commit()
+            self.db.refresh(alert)
+            return str(alert.id)
+        except Exception as e:
+            self.db.rollback()
+            logger.error(f"[ALERT_AGENT] Database error: {e}")
+            raise
+    
+    async def _cache_fingerprint(self, fingerprint: str) -> None:
+        """Cache fingerprint in Redis for fast duplicate detection"""
+        if not self.redis:
+            return
+        
+        try:
+            cache_key = f"alert:fingerprint:{fingerprint}"
+            await self.redis.expire(cache_key, self.cache_ttl)
+        except Exception as e:
+            logger.warning(f"[ALERT_AGENT] Cache update failed: {e}")

agents/analysis_agent.py

@@ -0,0 +1,285 @@
+"""
+Analysis Agent
+
+Performs AI-powered root cause analysis using Ollama LLM.
+Third agent in the processing pipeline.
+
+Communication: Receives from CorrelationAgent → Sends to ResponseAgent
+Data Storage: PostgreSQL (analysis results), Redis (LLM cache)
+"""
+
+import logging
+import json
+from typing import Dict, Any, Optional, List
+from datetime import datetime
+
+logger = logging.getLogger(__name__)
+
+
+class AnalysisAgent:
+    """Performs AI-powered analysis of incidents using Ollama"""
+    
+    def __init__(self, llm_client=None):
+        self.llm = llm_client
+        self.cache_enabled = True
+    
+    async def analyze_incident(self, incident: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Main entry point for analysis.
+        
+        Timeline:
+        T+400ms: Receive incident from CorrelationAgent
+        T+450ms: Prepare analysis context
+        T+500ms: Send to Ollama
+        T+500-1500ms: Ollama processes (1 second average)
+        T+1500ms: Parse LLM response
+        T+1550ms: Return analysis
+        
+        Args:
+            incident: Correlated incident with alerts
+            
+        Returns:
+            Analysis result with root cause and recommendations
+        """
+        logger.info(f"[ANALYSIS_AGENT] Analyzing incident: {incident.get('id')}")
+        
+        try:
+            # Step 1: Prepare context (T+450ms)
+            context = self._prepare_analysis_context(incident)
+            logger.debug(f"[ANALYSIS_AGENT] Context prepared: {len(json.dumps(context))} chars")
+            
+            # Step 2: Build prompt (T+480ms)
+            prompt = self._build_analysis_prompt(context)
+            logger.debug(f"[ANALYSIS_AGENT] Prompt built")
+            
+            # Step 3: Send to Ollama (T+500ms)
+            if not self.llm:
+                return self._fallback_analysis(incident)
+            
+            analysis_result = await self.llm.analyze_incident(
+                alerts=incident.get('alerts', []),
+                context=context,
+                prompt=prompt
+            )
+            logger.info(f"[ANALYSIS_AGENT] LLM analysis complete")
+            
+            # Step 4: Parse and structure response (T+1500ms)
+            structured_analysis = self._structure_analysis(analysis_result)
+            
+            # Step 5: Return result (T+1550ms)
+            return {
+                "status": "analyzed",
+                "incident_id": incident.get('id'),
+                "analysis": structured_analysis,
+                "processing_time_ms": 1150
+            }
+            
+        except Exception as e:
+            logger.error(f"[ANALYSIS_AGENT] Error analyzing incident: {e}", exc_info=True)
+            return self._fallback_analysis(incident)
+    
+    def _prepare_analysis_context(self, incident: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Prepare comprehensive context for LLM analysis.
+        
+        Includes:
+        - Alert timeline
+        - Metrics
+        - Service dependencies
+        - Historical patterns
+        """
+        alerts = incident.get('alerts', [])
+        
+        context = {
+            'incident_id': incident.get('id'),
+            'service': incident.get('service'),
+            'severity': incident.get('severity'),
+            'alert_count': len(alerts),
+            'time_range': {
+                'start': min(a.get('created_at', '') for a in alerts),
+                'end': max(a.get('created_at', '') for a in alerts)
+            },
+            'alerts_summary': [
+                {
+                    'title': a.get('title'),
+                    'category': a.get('category'),
+                    'severity': a.get('severity'),
+                    'timestamp': a.get('created_at')
+                }
+                for a in alerts
+            ],
+            'metrics': self._aggregate_metrics(alerts),
+            'patterns': self._detect_patterns(alerts)
+        }
+        
+        return context
+    
+    def _aggregate_metrics(self, alerts: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Aggregate metrics from all alerts in incident"""
+        metrics = {}
+        
+        for alert in alerts:
+            alert_metrics = alert.get('metrics', {})
+            if isinstance(alert_metrics, dict):
+                metrics.update(alert_metrics)
+        
+        return metrics
+    
+    def _detect_patterns(self, alerts: List[Dict[str, Any]]) -> List[str]:
+        """Detect common patterns in alerts"""
+        patterns = []
+        categories = [a.get('category') for a in alerts]
+        
+        # Pattern: Multiple resource alerts (CPU, Memory, Disk)
+        resource_alerts = [c for c in categories if c in ['cpu', 'memory', 'disk', 'io']]
+        if len(resource_alerts) >= 2:
+            patterns.append("resource_exhaustion")
+        
+        # Pattern: Connection/Network related
+        if any(c in ['connection', 'network', 'timeout'] for c in categories):
+            patterns.append("connectivity_issue")
+        
+        # Pattern: Application errors
+        if any(c in ['error', 'crash', 'exception'] for c in categories):
+            patterns.append("application_failure")
+        
+        return patterns
+    
+    def _build_analysis_prompt(self, context: Dict[str, Any]) -> str:
+        """Build prompt for Ollama LLM"""
+        return f"""
+You are an expert infrastructure analyst. Analyze this incident and provide:
+1. Root cause hypothesis
+2. Confidence level (0-100)
+3. Top 3 recommended actions
+4. Severity assessment
+
+Incident Context:
+- Service: {context.get('service')}
+- Severity: {context.get('severity')}
+- Alert Count: {context.get('alert_count')}
+- Patterns Detected: {', '.join(context.get('patterns', []))}
+
+Alert Timeline:
+{json.dumps(context.get('alerts_summary', []), indent=2)}
+
+Metrics:
+{json.dumps(context.get('metrics', {}), indent=2)}
+
+Provide analysis in JSON format with fields:
+- root_cause
+- confidence (0-100)
+- evidence (list of supporting facts)
+- actions (list of 3 recommended actions)
+- severity_assessment
+"""
+    
+    async def _structure_analysis(self, llm_response: str) -> Dict[str, Any]:
+        """Parse and structure LLM response"""
+        try:
+            # Extract JSON from LLM response
+            import re
+            json_match = re.search(r'\{.*\}', llm_response, re.DOTALL)
+            
+            if json_match:
+                analysis = json.loads(json_match.group())
+            else:
+                analysis = {
+                    'root_cause': llm_response,
+                    'confidence': 50,
+                    'evidence': [],
+                    'actions': []
+                }
+            
+            return {
+                'root_cause': analysis.get('root_cause', 'Unknown'),
+                'confidence': int(analysis.get('confidence', 50)),
+                'evidence': analysis.get('evidence', []),
+                'recommended_actions': analysis.get('actions', []),
+                'severity_assessment': analysis.get('severity_assessment', 'Medium'),
+                'timestamp': datetime.utcnow().isoformat()
+            }
+        except Exception as e:
+            logger.error(f"[ANALYSIS_AGENT] Error parsing LLM response: {e}")
+            return self._fallback_analysis_result()
+    
+    def _fallback_analysis(self, incident: Dict[str, Any]) -> Dict[str, Any]:
+        """Fallback analysis when LLM is unavailable"""
+        logger.warning(f"[ANALYSIS_AGENT] Using fallback analysis")
+        
+        return {
+            "status": "analyzed",
+            "incident_id": incident.get('id'),
+            "analysis": self._fallback_analysis_result(),
+            "mode": "fallback"
+        }
+    
+    def _fallback_analysis_result(self) -> Dict[str, Any]:
+        """Generate fallback analysis based on patterns"""
+        return {
+            'root_cause': 'Unable to determine - LLM unavailable',
+            'confidence': 0,
+            'evidence': [],
+            'recommended_actions': [
+                'Check application logs',
+                'Review service metrics',
+                'Check for recent deployments'
+            ],
+            'severity_assessment': 'Unknown',
+            'timestamp': datetime.utcnow().isoformat()
+        }
+    
+    async def analyze_alerts(self, alerts: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """
+        Analyze individual alerts for patterns and anomalies.
+        Used by AlertIngestionAgent for pre-processing.
+        """
+        logger.info(f"[ANALYSIS_AGENT] Analyzing {len(alerts)} alerts")
+        
+        try:
+            context = {
+                'alert_count': len(alerts),
+                'services': list(set(a.get('service') for a in alerts)),
+                'severities': list(set(a.get('severity') for a in alerts)),
+                'categories': list(set(a.get('category') for a in alerts))
+            }
+            
+            analysis = {
+                'context': context,
+                'patterns_detected': self._detect_patterns(alerts),
+                'risk_level': self._calculate_risk_level(alerts),
+                'recommended_priority': self._calculate_priority(alerts)
+            }
+            
+            return analysis
+        except Exception as e:
+            logger.error(f"[ANALYSIS_AGENT] Error in alert analysis: {e}")
+            return {'error': str(e)}
+    
+    def _calculate_risk_level(self, alerts: List[Dict[str, Any]]) -> str:
+        """Calculate overall risk level from alerts"""
+        if not alerts:
+            return 'low'
+        
+        critical_count = sum(1 for a in alerts if a.get('severity') == 'critical')
+        warning_count = sum(1 for a in alerts if a.get('severity') == 'warning')
+        
+        if critical_count >= 3:
+            return 'critical'
+        elif critical_count >= 1 or warning_count >= 5:
+            return 'high'
+        elif warning_count >= 2:
+            return 'medium'
+        else:
+            return 'low'
+    
+    def _calculate_priority(self, alerts: List[Dict[str, Any]]) -> int:
+        """Calculate priority (1-5) for handling"""
+        risk = self._calculate_risk_level(alerts)
+        priority_map = {
+            'critical': 1,
+            'high': 2,
+            'medium': 3,
+            'low': 5
+        }
+        return priority_map.get(risk, 4)

agents/correlation_agent.py

@@ -0,0 +1,238 @@
+"""
+Correlation Agent
+
+Groups related alerts into incidents by detecting patterns and similarities.
+Second agent in the processing pipeline.
+
+Communication: Receives from AlertIngestionAgent → Sends to AnalysisAgent
+Data Storage: PostgreSQL (incidents, correlations), Redis (pattern cache)
+"""
+
+import logging
+from typing import Dict, List, Any, Optional
+from datetime import datetime, timedelta
+import json
+
+logger = logging.getLogger(__name__)
+
+
+class CorrelationAgent:
+    """Correlates related alerts into incidents"""
+    
+    def __init__(self, db_session=None, redis_client=None):
+        self.db = db_session
+        self.redis = redis_client
+        self.correlation_window = 600  # 10 minutes
+        self.similarity_threshold = 0.7
+    
+    async def correlate_alerts(self, alert: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Main entry point for correlation.
+        
+        Timeline:
+        T+200ms: Receive alert from AlertIngestionAgent
+        T+250ms: Query for similar alerts in time window
+        T+300ms: Calculate similarity scores
+        T+350ms: Create new incident or update existing
+        T+400ms: Return correlation result
+        
+        Args:
+            alert: Normalized alert from AlertIngestionAgent
+            
+        Returns:
+            Correlation result with incident info
+        """
+        logger.info(f"[CORRELATION_AGENT] Processing alert: {alert.get('title')}")
+        
+        try:
+            # Step 1: Query similar alerts (T+250ms)
+            similar_alerts = await self._find_similar_alerts(alert)
+            logger.info(f"[CORRELATION_AGENT] Found {len(similar_alerts)} similar alerts")
+            
+            # Step 2: Calculate similarity (T+300ms)
+            correlation_score = self._calculate_correlation_score(alert, similar_alerts)
+            logger.debug(f"[CORRELATION_AGENT] Correlation score: {correlation_score}")
+            
+            # Step 3: Determine action (T+350ms)
+            if correlation_score >= self.similarity_threshold and similar_alerts:
+                # Update existing incident
+                incident_id = await self._update_incident(alert, similar_alerts)
+                action = "update"
+                logger.info(f"[CORRELATION_AGENT] Updated incident: {incident_id}")
+            else:
+                # Create new incident
+                incident_id = await self._create_incident(alert)
+                action = "create"
+                logger.info(f"[CORRELATION_AGENT] Created new incident: {incident_id}")
+            
+            # Step 4: Return result (T+400ms)
+            return {
+                "status": "correlated",
+                "incident_id": incident_id,
+                "action": action,
+                "correlation_score": correlation_score,
+                "similar_alerts_count": len(similar_alerts),
+                "processing_time_ms": 200
+            }
+            
+        except Exception as e:
+            logger.error(f"[CORRELATION_AGENT] Error correlating alert: {e}", exc_info=True)
+            raise
+    
+    async def _find_similar_alerts(self, alert: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Find alerts similar to current one within correlation window.
+        
+        Similarity criteria:
+        - Same service
+        - Same category (or related)
+        - Within last 10 minutes
+        - Same severity level or escalating
+        """
+        service = alert.get('service', 'unknown')
+        category = alert.get('category', 'unknown')
+        
+        # In real implementation, query from PostgreSQL:
+        # SELECT * FROM alerts WHERE
+        #   service = service AND
+        #   created_at > NOW() - INTERVAL '10 minutes' AND
+        #   (category = category OR category IN related_categories)
+        
+        similar = []
+        # Placeholder for database query
+        logger.debug(f"[CORRELATION_AGENT] Querying for alerts: service={service}, category={category}")
+        
+        return similar
+    
+    def _calculate_correlation_score(self, alert: Dict[str, Any], similar_alerts: List) -> float:
+        """
+        Calculate how much this alert correlates with existing ones.
+        Score 0.0-1.0 where 1.0 = definitely same incident
+        
+        Scoring factors:
+        - Service match (40%)
+        - Time proximity (30%)
+        - Metric similarity (20%)
+        - Severity escalation (10%)
+        """
+        if not similar_alerts:
+            return 0.0
+        
+        score = 0.0
+        alert_service = alert.get('service', 'unknown')
+        alert_category = alert.get('category', 'unknown')
+        alert_severity = alert.get('severity', 'warning')
+        
+        service_matches = sum(1 for a in similar_alerts if a.get('service') == alert_service)
+        category_matches = sum(1 for a in similar_alerts if a.get('category') == alert_category)
+        
+        # Service match (40%)
+        if similar_alerts:
+            score += (service_matches / len(similar_alerts)) * 0.4
+        
+        # Category match (30%)
+        if similar_alerts:
+            score += (category_matches / len(similar_alerts)) * 0.3
+        
+        # Time proximity (20%)
+        most_recent = max(similar_alerts, key=lambda x: x.get('created_at', ''))
+        time_diff = datetime.utcnow() - datetime.fromisoformat(most_recent.get('created_at', ''))
+        if time_diff.seconds < 60:  # Within 1 minute
+            score += 0.2
+        elif time_diff.seconds < 300:  # Within 5 minutes
+            score += 0.1
+        
+        # Severity escalation (10%)
+        severity_levels = {'info': 1, 'warning': 2, 'critical': 3}
+        current_level = severity_levels.get(alert_severity, 1)
+        avg_level = sum(severity_levels.get(a.get('severity', 'info'), 1) for a in similar_alerts) / len(similar_alerts)
+        if current_level >= avg_level:
+            score += 0.1
+        
+        return min(score, 1.0)
+    
+    async def _create_incident(self, alert: Dict[str, Any]) -> str:
+        """
+        Create new incident from alert.
+        
+        Incident structure:
+        - title: Generated from alerts
+        - service: From alert
+        - severity: From alert
+        - status: OPEN
+        - created_alerts: [alert]
+        """
+        logger.info(f"[CORRELATION_AGENT] Creating incident from alert")
+        
+        # In real implementation:
+        # INSERT INTO incidents (title, service, severity, status, created_at)
+        # VALUES (title, service, severity, 'OPEN', NOW())
+        
+        incident = {
+            'id': 'incident_placeholder',
+            'title': f"Incident: {alert.get('title')}",
+            'service': alert.get('service'),
+            'severity': alert.get('severity'),
+            'status': 'OPEN',
+            'alerts': [alert],
+            'created_at': datetime.utcnow().isoformat()
+        }
+        
+        logger.info(f"[CORRELATION_AGENT] Incident created: {incident['id']}")
+        return incident['id']
+    
+    async def _update_incident(self, alert: Dict[str, Any], similar_alerts: List) -> str:
+        """
+        Update existing incident with new alert.
+        
+        Updates:
+        - Add alert to incident.alerts list
+        - Update severity if escalated
+        - Update updated_at timestamp
+        - Mark incident as active
+        """
+        if not similar_alerts:
+            return await self._create_incident(alert)
+        
+        logger.info(f"[CORRELATION_AGENT] Updating incident with new alert")
+        
+        # Get incident ID from one of the similar alerts
+        incident_id = similar_alerts[0].get('incident_id', 'unknown')
+        
+        # In real implementation:
+        # UPDATE incidents SET
+        #   severity = MAX(current_severity, new_alert_severity),
+        #   updated_at = NOW(),
+        #   alert_count = alert_count + 1
+        # WHERE id = incident_id
+        
+        logger.info(f"[CORRELATION_AGENT] Incident updated: {incident_id}")
+        return incident_id
+    
+    async def detect_cascading_failure(self, incident: Dict[str, Any]) -> Optional[Dict[str, Any]]:
+        """
+        Detect if incident is part of cascading failure pattern.
+        
+        Pattern detection:
+        - Multiple services affected
+        - Temporal correlation (alerts within seconds)
+        - Dependency chain (e.g., DB → Cache → API)
+        """
+        logger.info(f"[CORRELATION_AGENT] Analyzing for cascading failure pattern")
+        
+        alerts = incident.get('alerts', [])
+        if len(alerts) < 3:
+            return None
+        
+        services = set(a.get('service') for a in alerts)
+        logger.info(f"[CORRELATION_AGENT] Services affected: {services}")
+        
+        if len(services) > 1:
+            return {
+                "pattern": "cascading_failure",
+                "confidence": 0.85,
+                "affected_services": list(services),
+                "recommendation": "Check dependency chain - start with database/cache layer"
+            }
+        
+        return None

agents/response_agent.py

@@ -0,0 +1,406 @@
+"""
+Response Agent
+
+Generates actionable recommendations and formats responses for dashboard display
+and external integrations (Slack, PagerDuty, etc.).
+Fourth and final agent in the processing pipeline.
+
+Communication: Receives from AnalysisAgent → Sends to Dashboard/Integrations
+Data Storage: PostgreSQL (responses), Redis (response cache)
+"""
+
+import logging
+import json
+from typing import Dict, Any, List, Optional
+from datetime import datetime
+from enum import Enum
+
+logger = logging.getLogger(__name__)
+
+
+class ActionType(Enum):
+    """Types of recommended actions"""
+    SCALE_UP = "scale-up"
+    SCALE_DOWN = "scale-down"
+    RESTART = "restart"
+    CHECK_LOGS = "check-logs"
+    INCREASE_RESOURCES = "increase-resources"
+    ENABLE_CIRCUIT_BREAKER = "enable-circuit-breaker"
+    RATE_LIMIT = "rate-limit"
+    ROLLBACK = "rollback"
+    DRAIN_CONNECTIONS = "drain-connections"
+    INVESTIGATE = "investigate"
+
+
+class ResponseAgent:
+    """Generates recommendations and formats responses"""
+    
+    def __init__(self):
+        self.action_configs = self._load_action_configs()
+    
+    async def generate_response(self, analysis: Dict[str, Any], incident: Dict[str, Any]) -> Dict[str, Any]:
+        """
+        Main entry point for response generation.
+        
+        Timeline:
+        T+1550ms: Receive analysis from AnalysisAgent
+        T+1580ms: Generate recommendations
+        T+1600ms: Format for dashboard
+        T+1620ms: Format for integrations
+        T+1650ms: Publish events
+        T+1700ms: Return response
+        
+        Args:
+            analysis: AI analysis from AnalysisAgent
+            incident: Incident data
+            
+        Returns:
+            Formatted response ready for display and integrations
+        """
+        logger.info(f"[RESPONSE_AGENT] Generating response for incident: {incident.get('id')}")
+        
+        try:
+            # Step 1: Generate recommendations (T+1580ms)
+            recommendations = self._generate_recommendations(analysis, incident)
+            logger.info(f"[RESPONSE_AGENT] Generated {len(recommendations)} recommendations")
+            
+            # Step 2: Format for dashboard (T+1600ms)
+            dashboard_response = self._format_dashboard_response(
+                incident, analysis, recommendations
+            )
+            logger.debug(f"[RESPONSE_AGENT] Dashboard response formatted")
+            
+            # Step 3: Format for integrations (T+1620ms)
+            integration_payloads = self._format_integration_payloads(
+                incident, analysis, recommendations
+            )
+            logger.debug(f"[RESPONSE_AGENT] Integration payloads generated")
+            
+            # Step 4: Publish events (T+1650ms)
+            await self._publish_events(incident, recommendations)
+            logger.info(f"[RESPONSE_AGENT] Events published")
+            
+            # Step 5: Return response (T+1700ms)
+            return {
+                "status": "response_generated",
+                "incident_id": incident.get('id'),
+                "dashboard": dashboard_response,
+                "integrations": integration_payloads,
+                "processing_time_ms": 150
+            }
+            
+        except Exception as e:
+            logger.error(f"[RESPONSE_AGENT] Error generating response: {e}", exc_info=True)
+            raise
+    
+    def _generate_recommendations(self, analysis: Dict[str, Any], incident: Dict[str, Any]) -> List[Dict[str, Any]]:
+        """
+        Generate specific, actionable recommendations based on analysis.
+        
+        Recommendation structure:
+        - Action: What to do
+        - Priority: How urgent
+        - Evidence: Why we recommend it
+        - Impact: What will change
+        """
+        recommendations = []
+        root_cause = analysis.get('root_cause', '').lower()
+        confidence = analysis.get('confidence', 0)
+        severity = incident.get('severity', 'warning')
+        
+        logger.debug(f"[RESPONSE_AGENT] Root cause: {root_cause}, Confidence: {confidence}%")
+        
+        # Rule 1: CPU exhaustion → Scale up
+        if 'cpu' in root_cause and confidence > 70:
+            recommendations.append({
+                'action': ActionType.SCALE_UP.value,
+                'service': incident.get('service'),
+                'priority': 'high' if severity == 'critical' else 'medium',
+                'replicas': self._calculate_replicas(incident),
+                'evidence': [
+                    'High CPU usage detected',
+                    'Multiple CPU alerts correlated',
+                    'LLM confidence: {}%'.format(confidence)
+                ],
+                'expected_impact': 'Improved performance and reduced latency',
+                'estimated_recovery_time': '2-5 minutes'
+            })
+        
+        # Rule 2: Memory issues → Increase resources or restart
+        if 'memory' in root_cause or 'oom' in root_cause:
+            if confidence > 80:
+                recommendations.append({
+                    'action': ActionType.RESTART.value,
+                    'service': incident.get('service'),
+                    'priority': 'high',
+                    'graceful': True,
+                    'evidence': [
+                        'Memory pressure detected',
+                        'Out of memory conditions',
+                        'Application performance degraded'
+                    ],
+                    'expected_impact': 'Process restart, temporary service interruption',
+                    'estimated_recovery_time': '30-60 seconds'
+                })
+            else:
+                recommendations.append({
+                    'action': ActionType.INCREASE_RESOURCES.value,
+                    'service': incident.get('service'),
+                    'priority': 'medium',
+                    'memory_increase_gb': 4,
+                    'evidence': ['Memory usage trending up'],
+                    'expected_impact': 'More available memory for application',
+                    'estimated_recovery_time': 'Requires redeployment (5-10 minutes)'
+                })
+        
+        # Rule 3: Connection pool exhaustion
+        if 'connection' in root_cause or 'pool' in root_cause:
+            recommendations.append({
+                'action': ActionType.DRAIN_CONNECTIONS.value,
+                'service': incident.get('service'),
+                'priority': 'critical' if severity == 'critical' else 'high',
+                'steps': [
+                    'Increase max_connections parameter',
+                    'Enable connection recycling',
+                    'Implement connection timeout'
+                ],
+                'evidence': [
+                    'Connection pool at capacity',
+                    'Connection timeout errors',
+                    'Database connectivity issues'
+                ],
+                'expected_impact': 'Restored database connectivity',
+                'estimated_recovery_time': '1-2 minutes'
+            })
+        
+        # Rule 4: Network/DNS issues
+        if 'network' in root_cause or 'dns' in root_cause:
+            recommendations.append({
+                'action': ActionType.INVESTIGATE.value,
+                'service': incident.get('service'),
+                'priority': 'high',
+                'investigation_steps': [
+                    'Check DNS resolution',
+                    'Verify network connectivity',
+                    'Review firewall rules',
+                    'Check service discovery'
+                ],
+                'evidence': ['Network connectivity issues detected'],
+                'expected_impact': 'Identified networking bottleneck',
+                'estimated_recovery_time': '5-15 minutes investigation'
+            })
+        
+        # Rule 5: Recent deployment → Rollback
+        if 'deployment' in root_cause or 'version' in root_cause:
+            if confidence > 75:
+                recommendations.append({
+                    'action': ActionType.ROLLBACK.value,
+                    'service': incident.get('service'),
+                    'priority': 'critical',
+                    'to_version': self._get_previous_version(incident),
+                    'evidence': [
+                        'Incident occurred after deployment',
+                        'Performance metrics degraded',
+                        'Error rate increased'
+                    ],
+                    'expected_impact': 'Return to stable version',
+                    'estimated_recovery_time': '2-5 minutes'
+                })
+        
+        # Default: Investigate
+        if not recommendations:
+            recommendations.append({
+                'action': ActionType.INVESTIGATE.value,
+                'service': incident.get('service'),
+                'priority': 'medium',
+                'next_steps': [
+                    'Review application logs',
+                    'Check resource metrics',
+                    'Review recent changes',
+                    'Consult runbooks for service'
+                ],
+                'evidence': ['Root cause unclear - further investigation needed'],
+                'expected_impact': 'Identify actual root cause',
+                'estimated_recovery_time': '10-30 minutes'
+            })
+        
+        logger.info(f"[RESPONSE_AGENT] Generated {len(recommendations)} recommendations")
+        return recommendations
+    
+    def _format_dashboard_response(self, incident: Dict[str, Any], 
+                                   analysis: Dict[str, Any], 
+                                   recommendations: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Format response for dashboard display (Material-UI friendly)"""
+        return {
+            'incident_id': incident.get('id'),
+            'title': incident.get('title', 'Incident'),
+            'service': incident.get('service'),
+            'severity': incident.get('severity'),
+            'status': 'ANALYZING' if len(recommendations) < 1 else 'ANALYZED',
+            'analysis': {
+                'root_cause': analysis.get('root_cause'),
+                'confidence': f"{analysis.get('confidence', 0)}%",
+                'evidence': analysis.get('evidence', []),
+                'severity_assessment': analysis.get('severity_assessment')
+            },
+            'recommendations': [
+                {
+                    'id': i + 1,
+                    'action': r.get('action'),
+                    'priority': r.get('priority'),
+                    'description': self._action_to_description(r),
+                    'estimated_time': r.get('estimated_recovery_time'),
+                    'impact': r.get('expected_impact')
+                }
+                for i, r in enumerate(recommendations)
+            ],
+            'alert_count': len(incident.get('alerts', [])),
+            'timestamp': datetime.utcnow().isoformat(),
+            'ready_for_action': True
+        }
+    
+    def _format_integration_payloads(self, incident: Dict[str, Any],
+                                     analysis: Dict[str, Any],
+                                     recommendations: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Format payloads for external integrations (Slack, PagerDuty, etc.)"""
+        return {
+            'slack': self._format_slack_message(incident, analysis, recommendations),
+            'pagerduty': self._format_pagerduty_event(incident, analysis),
+            'opsgenie': self._format_opsgenie_alert(incident, analysis)
+        }
+    
+    def _format_slack_message(self, incident: Dict[str, Any],
+                             analysis: Dict[str, Any],
+                             recommendations: List[Dict[str, Any]]) -> Dict[str, Any]:
+        """Format for Slack notification"""
+        severity_color = {
+            'critical': '#FF0000',
+            'warning': '#FFA500',
+            'info': '#0099FF'
+        }
+        
+        return {
+            'text': f"🚨 Incident: {incident.get('title')}",
+            'attachments': [
+                {
+                    'color': severity_color.get(incident.get('severity'), '#999999'),
+                    'title': incident.get('title'),
+                    'fields': [
+                        {'title': 'Service', 'value': incident.get('service'), 'short': True},
+                        {'title': 'Severity', 'value': incident.get('severity').upper(), 'short': True},
+                        {'title': 'Root Cause', 'value': analysis.get('root_cause'), 'short': False},
+                        {'title': 'Confidence', 'value': f"{analysis.get('confidence')}%", 'short': True},
+                        {
+                            'title': 'Top Recommendation',
+                            'value': recommendations[0]['action'] if recommendations else 'Investigate',
+                            'short': True
+                        }
+                    ],
+                    'footer': 'AIMS - Autonomous Incident Management System'
+                }
+            ]
+        }
+    
+    def _format_pagerduty_event(self, incident: Dict[str, Any],
+                               analysis: Dict[str, Any]) -> Dict[str, Any]:
+        """Format for PagerDuty alert"""
+        severity_map = {
+            'critical': 'critical',
+            'warning': 'warning',
+            'info': 'info'
+        }
+        
+        return {
+            'routing_key': 'YOUR_PAGERDUTY_KEY',
+            'event_action': 'trigger',
+            'payload': {
+                'summary': incident.get('title'),
+                'severity': severity_map.get(incident.get('severity'), 'warning'),
+                'source': 'AIMS',
+                'custom_details': {
+                    'root_cause': analysis.get('root_cause'),
+                    'confidence': analysis.get('confidence'),
+                    'service': incident.get('service')
+                }
+            }
+        }
+    
+    def _format_opsgenie_alert(self, incident: Dict[str, Any],
+                              analysis: Dict[str, Any]) -> Dict[str, Any]:
+        """Format for OpsGenie alert"""
+        return {
+            'message': incident.get('title'),
+            'description': f"Root Cause: {analysis.get('root_cause')}",
+            'priority': self._map_to_opsgenie_priority(incident.get('severity')),
+            'source': 'AIMS',
+            'tags': [
+                incident.get('service'),
+                incident.get('severity'),
+                'aims'
+            ]
+        }
+    
+    async def _publish_events(self, incident: Dict[str, Any], 
+                             recommendations: List[Dict[str, Any]]) -> None:
+        """Publish events to Redis for async subscribers"""
+        events = [
+            {
+                'type': 'incident.analyzed',
+                'incident_id': incident.get('id'),
+                'timestamp': datetime.utcnow().isoformat()
+            },
+            {
+                'type': 'recommendations.generated',
+                'incident_id': incident.get('id'),
+                'count': len(recommendations),
+                'timestamp': datetime.utcnow().isoformat()
+            }
+        ]
+        
+        logger.debug(f"[RESPONSE_AGENT] Publishing {len(events)} events")
+        # In real implementation, publish to Redis pub/sub:
+        # for event in events:
+        #     await redis_client.publish('incidents', json.dumps(event))
+    
+    def _action_to_description(self, recommendation: Dict[str, Any]) -> str:
+        """Convert action dict to human-readable description"""
+        action = recommendation.get('action')
+        
+        descriptions = {
+            'scale-up': f"Scale up {recommendation.get('service')} to {recommendation.get('replicas')} replicas",
+            'restart': f"Gracefully restart {recommendation.get('service')} service",
+            'check-logs': "Review application logs for errors",
+            'increase-resources': f"Increase memory by {recommendation.get('memory_increase_gb')}GB",
+            'drain-connections': "Drain and reset database connections",
+            'rollback': f"Rollback to version {recommendation.get('to_version')}",
+            'investigate': "Further investigation required"
+        }
+        
+        return descriptions.get(action, action)
+    
+    def _calculate_replicas(self, incident: Dict[str, Any]) -> int:
+        """Calculate recommended replica count"""
+        # In real implementation, could base this on current load, auto-scaling policies, etc.
+        return 3
+    
+    def _get_previous_version(self, incident: Dict[str, Any]) -> str:
+        """Get previous stable version for rollback"""
+        # In real implementation, would query deployment history
+        return "v1.2.3"
+    
+    def _map_to_opsgenie_priority(self, severity: str) -> str:
+        """Map severity to OpsGenie priority"""
+        priority_map = {
+            'critical': 'P1',
+            'warning': 'P2',
+            'info': 'P3'
+        }
+        return priority_map.get(severity, 'P3')
+    
+    def _load_action_configs(self) -> Dict[str, Any]:
+        """Load action configurations"""
+        return {
+            'scale-up': {'max_replicas': 10, 'timeout': 300},
+            'restart': {'graceful_period': 30},
+            'rollback': {'timeout': 600}
+        }