Add sre_agent/tools/rca_tools.py

sre_agent/tools/rca_tools.py

@@ -0,0 +1,537 @@
+"""
+Root Cause Analysis Tools for SRE Agent
+
+Implements multi-signal correlation and dependency-aware RCA.
+Based on:
+- AMER-RCL recursive reasoning pattern (arxiv:2601.02732)
+- RCACopilot statistical pre-filter + LLM interpretation (arxiv:2507.03224)
+- TrioXpert multi-dimensional system status representation (arxiv:2506.10043)
+"""
+
+import json
+from datetime import datetime, timedelta
+from smolagents import Tool
+
+
+class RootCauseCorrelatorTool(Tool):
+    """Multi-signal correlation engine for root cause analysis."""
+    name = "rca_correlator"
+    description = """Correlates anomalies across metrics, logs, traces, and alerts to determine root cause.
+    
+    This is the PRIMARY RCA tool — feed it evidence from other tools (anomaly detector, log parser, etc.)
+    and it will:
+    1. Temporally align anomalies across signals
+    2. Score candidate root causes by evidence strength
+    3. Identify the propagation path (which component failed first, what cascaded)
+    4. Provide a ranked list of likely root causes with confidence scores
+    
+    The evidence should be a JSON object with any combination of:
+    - "metrics": list of metric anomalies (from timeseries_anomaly_detector)
+    - "logs": list of log findings (from log_parser)
+    - "alerts": list of active alerts (from alert_summary)
+    - "changes": list of recent changes (from change_correlator)
+    - "topology": service dependency info (from service_dependency_analyzer)
+    
+    Returns a ranked list of root cause hypotheses with evidence and confidence.
+    """
+    inputs = {
+        "evidence_json": {
+            "type": "string",
+            "description": "JSON object containing evidence from various signals. Keys can include: metrics, logs, alerts, changes, topology. Or 'auto' for a simulated incident.",
+        },
+        "service_name": {
+            "type": "string",
+            "description": "Primary service under investigation.",
+            "nullable": True,
+        },
+        "incident_description": {
+            "type": "string",
+            "description": "Brief description of the incident symptoms, e.g. 'High latency on checkout endpoint for last 30 minutes'.",
+            "nullable": True,
+        },
+    }
+    output_type = "string"
+
+    def _generate_sample_evidence(self, service_name: str) -> dict:
+        """Generate realistic multi-signal evidence for a simulated incident."""
+        return {
+            "metrics": [
+                {"metric": "cpu_utilization", "service": service_name, "anomaly_type": "spike", "value": 94.5, "threshold": 80, "timestamp": "2024-01-15T10:32:00Z", "severity": "critical"},
+                {"metric": "p99_latency_ms", "service": service_name, "anomaly_type": "spike", "value": 2500, "threshold": 500, "timestamp": "2024-01-15T10:33:00Z", "severity": "critical"},
+                {"metric": "error_rate", "service": service_name, "anomaly_type": "spike", "value": 12.5, "threshold": 1.0, "timestamp": "2024-01-15T10:33:30Z", "severity": "critical"},
+                {"metric": "memory_usage_pct", "service": "database-primary", "anomaly_type": "gradual_increase", "value": 91, "threshold": 85, "timestamp": "2024-01-15T10:25:00Z", "severity": "warning"},
+                {"metric": "connection_pool_active", "service": "database-primary", "anomaly_type": "spike", "value": 495, "threshold": 200, "timestamp": "2024-01-15T10:30:00Z", "severity": "critical"},
+                {"metric": "gc_pause_ms", "service": service_name, "anomaly_type": "spike", "value": 450, "threshold": 50, "timestamp": "2024-01-15T10:31:00Z", "severity": "warning"},
+            ],
+            "logs": [
+                {"service": "database-primary", "level": "ERROR", "message": "Connection pool exhausted: 500/500 connections active", "timestamp": "2024-01-15T10:30:15Z", "count": 47},
+                {"service": service_name, "level": "ERROR", "message": "Connection timeout to database after 30000ms", "timestamp": "2024-01-15T10:31:00Z", "count": 230},
+                {"service": service_name, "level": "CRITICAL", "message": "Circuit breaker OPEN for database-primary", "timestamp": "2024-01-15T10:32:00Z", "count": 15},
+                {"service": "api-gateway", "level": "ERROR", "message": f"Upstream {service_name} returned 503", "timestamp": "2024-01-15T10:32:30Z", "count": 180},
+                {"service": service_name, "level": "WARN", "message": "GC pause exceeded 200ms threshold", "timestamp": "2024-01-15T10:31:30Z", "count": 8},
+            ],
+            "alerts": [
+                {"name": f"{service_name}-high-error-rate", "severity": "critical", "fired_at": "2024-01-15T10:33:00Z", "status": "firing"},
+                {"name": "database-primary-connection-saturation", "severity": "critical", "fired_at": "2024-01-15T10:30:00Z", "status": "firing"},
+                {"name": f"{service_name}-high-latency", "severity": "warning", "fired_at": "2024-01-15T10:33:00Z", "status": "firing"},
+            ],
+            "changes": [
+                {"type": "deployment", "service": service_name, "version": "v2.4.1 → v2.5.0", "timestamp": "2024-01-15T10:00:00Z", "author": "ci-pipeline"},
+                {"type": "config_change", "service": "database-primary", "description": "max_connections reduced from 1000 to 500", "timestamp": "2024-01-15T09:45:00Z", "author": "dba-team"},
+            ],
+            "topology": {
+                "service": service_name,
+                "dependencies": [
+                    {"name": "database-primary", "type": "database", "protocol": "tcp", "critical": True},
+                    {"name": "cache-redis", "type": "cache", "protocol": "tcp", "critical": False},
+                    {"name": "auth-service", "type": "service", "protocol": "grpc", "critical": True},
+                ],
+                "dependents": [
+                    {"name": "api-gateway", "type": "gateway", "protocol": "http"},
+                    {"name": "web-frontend", "type": "frontend", "protocol": "http"},
+                ],
+            },
+        }
+
+    def forward(
+        self,
+        evidence_json: str,
+        service_name: str = "unknown-service",
+        incident_description: str = "",
+    ) -> str:
+        if evidence_json.strip().lower() == "auto":
+            evidence = self._generate_sample_evidence(service_name)
+            print(f"[RCACorrelator] Using simulated incident evidence for '{service_name}'")
+        else:
+            evidence = json.loads(evidence_json)
+
+        print(f"[RCACorrelator] Analyzing evidence for service '{service_name}': {incident_description}")
+
+        # ── 1. Temporal alignment ──
+        all_events = []
+        for m in evidence.get("metrics", []):
+            all_events.append({"timestamp": m.get("timestamp", ""), "type": "metric", "source": m.get("service", ""), "detail": f"{m['metric']}={m['value']}", "severity": m.get("severity", "info")})
+        for l in evidence.get("logs", []):
+            all_events.append({"timestamp": l.get("timestamp", ""), "type": "log", "source": l.get("service", ""), "detail": l["message"], "severity": l.get("level", "INFO").lower()})
+        for a in evidence.get("alerts", []):
+            all_events.append({"timestamp": a.get("fired_at", ""), "type": "alert", "source": a.get("name", ""), "detail": a["name"], "severity": a.get("severity", "info")})
+        for c in evidence.get("changes", []):
+            all_events.append({"timestamp": c.get("timestamp", ""), "type": "change", "source": c.get("service", ""), "detail": c.get("description", c.get("version", "")), "severity": "info"})
+
+        all_events.sort(key=lambda x: x.get("timestamp", ""))
+
+        # ── 2. Root cause hypothesis generation ──
+        hypotheses = []
+
+        # Hypothesis: Recent deployment caused the issue
+        changes = evidence.get("changes", [])
+        deployments = [c for c in changes if c.get("type") == "deployment"]
+        config_changes = [c for c in changes if c.get("type") == "config_change"]
+        
+        for deploy in deployments:
+            hypotheses.append({
+                "hypothesis": f"Recent deployment of {deploy.get('service', 'unknown')} ({deploy.get('version', 'unknown')}) introduced a regression",
+                "category": "deployment",
+                "evidence_for": [
+                    f"Deployment at {deploy.get('timestamp', 'unknown')} preceded incident",
+                    f"Service {deploy.get('service', 'unknown')} showing anomalies",
+                ],
+                "evidence_against": [],
+                "confidence": 0.0,
+                "remediation": f"Rollback {deploy.get('service', 'unknown')} to previous version",
+            })
+
+        for change in config_changes:
+            hypotheses.append({
+                "hypothesis": f"Configuration change on {change.get('service', 'unknown')}: {change.get('description', '')}",
+                "category": "config_change",
+                "evidence_for": [
+                    f"Config changed at {change.get('timestamp', 'unknown')} preceded incident",
+                    f"{change.get('description', 'Unknown change')}",
+                ],
+                "evidence_against": [],
+                "confidence": 0.0,
+                "remediation": f"Revert configuration change on {change.get('service', 'unknown')}",
+            })
+
+        # Hypothesis: Resource exhaustion
+        metrics = evidence.get("metrics", [])
+        resource_metrics = [m for m in metrics if m.get("metric") in ("cpu_utilization", "memory_usage_pct", "connection_pool_active", "disk_io_mbps")]
+        if resource_metrics:
+            # Find earliest resource anomaly
+            resource_metrics.sort(key=lambda x: x.get("timestamp", ""))
+            first_resource = resource_metrics[0]
+            hypotheses.append({
+                "hypothesis": f"Resource exhaustion on {first_resource.get('service', 'unknown')}: {first_resource['metric']} at {first_resource['value']}",
+                "category": "resource_exhaustion",
+                "evidence_for": [
+                    f"{m['metric']} on {m.get('service', 'unknown')} at {m['value']} (threshold: {m.get('threshold', 'N/A')})"
+                    for m in resource_metrics
+                ],
+                "evidence_against": [],
+                "confidence": 0.0,
+                "remediation": f"Scale up {first_resource.get('service', 'unknown')} or optimize resource usage",
+            })
+
+        # Hypothesis: Dependency failure
+        topology = evidence.get("topology", {})
+        deps = topology.get("dependencies", [])
+        for dep in deps:
+            dep_name = dep["name"]
+            dep_logs = [l for l in evidence.get("logs", []) if dep_name in l.get("service", "") or dep_name in l.get("message", "")]
+            dep_metrics = [m for m in metrics if dep_name in m.get("service", "")]
+            if dep_logs or dep_metrics:
+                hypotheses.append({
+                    "hypothesis": f"Dependency failure: {dep_name} ({dep['type']}) is degraded, causing cascading failure",
+                    "category": "dependency_failure",
+                    "evidence_for": [
+                        *[f"Log: {l['message'][:100]}" for l in dep_logs[:3]],
+                        *[f"Metric: {m['metric']}={m['value']} on {dep_name}" for m in dep_metrics[:3]],
+                    ],
+                    "evidence_against": [],
+                    "confidence": 0.0,
+                    "remediation": f"Investigate {dep_name}, check circuit breakers, consider failover",
+                    "critical_dependency": dep.get("critical", False),
+                })
+
+        # ── 3. Score hypotheses ──
+        for h in hypotheses:
+            score = 0.0
+            n_evidence = len(h["evidence_for"])
+            
+            # Base score from evidence count
+            score += min(n_evidence * 0.15, 0.6)
+            
+            # Boost for temporal precedence (changes before anomalies)
+            if h["category"] in ("deployment", "config_change"):
+                score += 0.2
+            
+            # Boost for critical dependencies
+            if h.get("critical_dependency"):
+                score += 0.15
+            
+            # Boost for resource exhaustion with multiple signals
+            if h["category"] == "resource_exhaustion" and n_evidence >= 3:
+                score += 0.2
+            
+            h["confidence"] = round(min(score, 0.95), 2)
+
+        # Sort by confidence
+        hypotheses.sort(key=lambda x: x["confidence"], reverse=True)
+
+        # ── 4. Determine propagation path ──
+        propagation_path = []
+        for event in all_events:
+            if event["severity"] in ("critical", "error"):
+                propagation_path.append({
+                    "timestamp": event["timestamp"],
+                    "component": event["source"],
+                    "signal_type": event["type"],
+                    "detail": event["detail"][:100],
+                })
+
+        result = {
+            "incident": {
+                "service": service_name,
+                "description": incident_description,
+                "signals_analyzed": {
+                    "metrics": len(evidence.get("metrics", [])),
+                    "logs": len(evidence.get("logs", [])),
+                    "alerts": len(evidence.get("alerts", [])),
+                    "changes": len(evidence.get("changes", [])),
+                },
+            },
+            "root_cause_hypotheses": hypotheses,
+            "primary_hypothesis": hypotheses[0] if hypotheses else None,
+            "propagation_timeline": propagation_path[:20],
+            "blast_radius": {
+                "directly_affected": [service_name],
+                "indirectly_affected": [d["name"] for d in topology.get("dependents", [])],
+                "total_services_impacted": 1 + len(topology.get("dependents", [])),
+            },
+            "recommended_actions": [
+                {"priority": i + 1, "action": h["remediation"], "confidence": h["confidence"]}
+                for i, h in enumerate(hypotheses[:5])
+            ],
+        }
+
+        print(f"[RCACorrelator] Generated {len(hypotheses)} hypotheses. Primary: {hypotheses[0]['hypothesis'][:80]}..." if hypotheses else "[RCACorrelator] No hypotheses generated")
+        return json.dumps(result, indent=2)
+
+
+class ServiceDependencyAnalyzerTool(Tool):
+    """Analyze service dependency topology for impact assessment."""
+    name = "service_dependency_analyzer"
+    description = """Analyzes service dependency topology to understand blast radius and failure propagation.
+    
+    Given a service name, returns:
+    - Upstream and downstream dependencies
+    - Critical path analysis (which dependencies are on the critical path)
+    - Single points of failure
+    - Recommended investigation order for incidents
+    
+    Use this EARLY in incident investigation to understand which services to check.
+    """
+    inputs = {
+        "service_name": {
+            "type": "string",
+            "description": "Service to analyze, e.g. 'payment-service'.",
+        },
+        "topology_json": {
+            "type": "string",
+            "description": "Optional: JSON service topology. If 'auto', uses a simulated microservice topology.",
+            "nullable": True,
+        },
+    }
+    output_type = "string"
+
+    def forward(self, service_name: str, topology_json: str = "auto") -> str:
+        if topology_json.strip().lower() == "auto":
+            # Simulated microservice topology
+            topology = {
+                "api-gateway": {"deps": ["auth-service", "user-service", "order-service", "payment-service"], "type": "gateway"},
+                "web-frontend": {"deps": ["api-gateway"], "type": "frontend"},
+                "mobile-bff": {"deps": ["api-gateway"], "type": "frontend"},
+                "auth-service": {"deps": ["user-db", "cache-redis", "jwt-signer"], "type": "service"},
+                "user-service": {"deps": ["user-db", "cache-redis"], "type": "service"},
+                "order-service": {"deps": ["order-db", "payment-service", "inventory-service", "notification-service"], "type": "service"},
+                "payment-service": {"deps": ["payment-db", "payment-gateway-ext", "fraud-detection"], "type": "service"},
+                "inventory-service": {"deps": ["inventory-db", "cache-redis"], "type": "service"},
+                "notification-service": {"deps": ["email-provider-ext", "sms-provider-ext", "kafka"], "type": "service"},
+                "fraud-detection": {"deps": ["ml-model-service", "fraud-db"], "type": "service"},
+                "ml-model-service": {"deps": ["model-store-s3"], "type": "service"},
+                "user-db": {"deps": [], "type": "database"},
+                "order-db": {"deps": [], "type": "database"},
+                "payment-db": {"deps": [], "type": "database"},
+                "inventory-db": {"deps": [], "type": "database"},
+                "fraud-db": {"deps": [], "type": "database"},
+                "cache-redis": {"deps": [], "type": "cache"},
+                "kafka": {"deps": [], "type": "queue"},
+                "jwt-signer": {"deps": [], "type": "infrastructure"},
+                "payment-gateway-ext": {"deps": [], "type": "external"},
+                "email-provider-ext": {"deps": [], "type": "external"},
+                "sms-provider-ext": {"deps": [], "type": "external"},
+                "model-store-s3": {"deps": [], "type": "storage"},
+            }
+        else:
+            topology = json.loads(topology_json)
+
+        print(f"[DependencyAnalyzer] Analyzing topology for '{service_name}' ({len(topology)} services)")
+
+        # Find direct dependencies
+        service_info = topology.get(service_name, {"deps": [], "type": "unknown"})
+        direct_deps = service_info.get("deps", [])
+
+        # Find transitive dependencies (BFS)
+        transitive_deps = set()
+        queue = list(direct_deps)
+        visited = set()
+        while queue:
+            current = queue.pop(0)
+            if current in visited:
+                continue
+            visited.add(current)
+            transitive_deps.add(current)
+            if current in topology:
+                for dep in topology[current].get("deps", []):
+                    if dep not in visited:
+                        queue.append(dep)
+
+        # Find reverse dependencies (who depends on this service)
+        dependents = []
+        transitive_dependents = set()
+        for svc, info in topology.items():
+            if service_name in info.get("deps", []):
+                dependents.append(svc)
+
+        # Transitive dependents (BFS reverse)
+        queue = list(dependents)
+        visited = set()
+        while queue:
+            current = queue.pop(0)
+            if current in visited:
+                continue
+            visited.add(current)
+            transitive_dependents.add(current)
+            for svc, info in topology.items():
+                if current in info.get("deps", []) and svc not in visited:
+                    queue.append(svc)
+
+        # Single points of failure
+        spofs = []
+        for dep in direct_deps:
+            dep_info = topology.get(dep, {})
+            if dep_info.get("type") in ("database", "external"):
+                spofs.append({"service": dep, "type": dep_info.get("type"), "reason": f"Single {dep_info.get('type')} dependency with no failover"})
+
+        # Critical path (dependencies that are on the path of all dependents)
+        critical_deps = [dep for dep in direct_deps if topology.get(dep, {}).get("type") in ("database", "infrastructure")]
+
+        result = {
+            "service": service_name,
+            "service_type": service_info.get("type", "unknown"),
+            "direct_dependencies": [
+                {"name": d, "type": topology.get(d, {}).get("type", "unknown")}
+                for d in direct_deps
+            ],
+            "transitive_dependencies": {
+                "count": len(transitive_deps),
+                "services": sorted(transitive_deps),
+            },
+            "direct_dependents": [
+                {"name": d, "type": topology.get(d, {}).get("type", "unknown")}
+                for d in dependents
+            ],
+            "blast_radius": {
+                "direct_impact": len(dependents),
+                "transitive_impact": len(transitive_dependents),
+                "affected_services": sorted(transitive_dependents),
+            },
+            "single_points_of_failure": spofs,
+            "critical_dependencies": critical_deps,
+            "investigation_order": [
+                {"step": 1, "action": f"Check {service_name} health, metrics, and recent deployments"},
+                *[{"step": i + 2, "action": f"Check dependency: {dep} ({topology.get(dep, {}).get('type', 'unknown')})"} for i, dep in enumerate(direct_deps)],
+                {"step": len(direct_deps) + 2, "action": f"Check impact on dependents: {', '.join(dependents[:5])}"},
+            ],
+        }
+
+        print(f"[DependencyAnalyzer] {service_name} has {len(direct_deps)} deps, {len(dependents)} dependents, blast radius: {len(transitive_dependents)} services")
+        return json.dumps(result, indent=2)
+
+
+class ChangeCorrelationTool(Tool):
+    """Correlate recent changes with incident timing."""
+    name = "change_correlator"
+    description = """Correlates recent infrastructure/code changes with incident timing to identify change-induced failures.
+    
+    Checks:
+    - Recent deployments (code releases, container updates)
+    - Configuration changes (env vars, feature flags, resource limits)
+    - Infrastructure changes (scaling events, DNS changes, cert rotations)
+    - Dependency updates (library versions, API version changes)
+    
+    Ranks changes by temporal proximity to incident and likelihood of causing the observed symptoms.
+    """
+    inputs = {
+        "incident_time": {
+            "type": "string",
+            "description": "When the incident started (ISO8601), e.g. '2024-01-15T10:30:00Z'.",
+        },
+        "service_name": {
+            "type": "string",
+            "description": "Service experiencing the incident.",
+            "nullable": True,
+        },
+        "lookback_hours": {
+            "type": "integer",
+            "description": "How many hours before incident to check for changes. Default: 24.",
+            "nullable": True,
+        },
+        "changes_json": {
+            "type": "string",
+            "description": "Optional: JSON array of recent changes. Or 'auto' for simulated change log.",
+            "nullable": True,
+        },
+    }
+    output_type = "string"
+
+    def forward(
+        self,
+        incident_time: str,
+        service_name: str = "unknown-service",
+        lookback_hours: int = 24,
+        changes_json: str = "auto",
+    ) -> str:
+        if changes_json.strip().lower() == "auto":
+            changes = [
+                {"type": "deployment", "service": service_name, "version": "v2.4.1 → v2.5.0", "timestamp": "2024-01-15T10:00:00Z", "author": "ci-pipeline", "description": "Added new payment processing endpoint with connection pooling changes"},
+                {"type": "config_change", "service": "database-primary", "timestamp": "2024-01-15T09:45:00Z", "author": "dba-team", "description": "Reduced max_connections from 1000 to 500 for memory optimization"},
+                {"type": "scaling", "service": service_name, "timestamp": "2024-01-15T08:00:00Z", "author": "hpa", "description": "Auto-scaled from 3 to 5 replicas due to morning traffic ramp"},
+                {"type": "deployment", "service": "auth-service", "version": "v1.8.0 → v1.8.1", "timestamp": "2024-01-15T07:00:00Z", "author": "ci-pipeline", "description": "Security patch for JWT validation"},
+                {"type": "config_change", "service": "api-gateway", "timestamp": "2024-01-14T22:00:00Z", "author": "platform-team", "description": "Updated rate limit from 1000 to 2000 req/s"},
+                {"type": "cert_rotation", "service": "tls-ingress", "timestamp": "2024-01-14T20:00:00Z", "author": "cert-manager", "description": "Automated TLS certificate rotation"},
+                {"type": "dependency_update", "service": service_name, "timestamp": "2024-01-14T16:00:00Z", "author": "dependabot", "description": "Updated database driver from 4.2.1 to 5.0.0 (breaking API changes)"},
+            ]
+        else:
+            changes = json.loads(changes_json)
+
+        try:
+            incident_dt = datetime.fromisoformat(incident_time.replace("Z", "+00:00"))
+        except ValueError:
+            incident_dt = datetime.utcnow()
+
+        print(f"[ChangeCorrelator] Correlating changes within {lookback_hours}h before incident at {incident_time}")
+
+        # Score each change
+        scored_changes = []
+        for change in changes:
+            try:
+                change_dt = datetime.fromisoformat(change["timestamp"].replace("Z", "+00:00"))
+            except ValueError:
+                continue
+
+            time_diff = (incident_dt - change_dt).total_seconds() / 3600  # hours
+
+            if time_diff < 0 or time_diff > lookback_hours:
+                continue
+
+            # Scoring
+            score = 0.0
+
+            # Temporal proximity (closer = higher score)
+            if time_diff < 1:
+                score += 0.4
+            elif time_diff < 4:
+                score += 0.3
+            elif time_diff < 12:
+                score += 0.15
+            else:
+                score += 0.05
+
+            # Change type risk
+            type_risk = {
+                "deployment": 0.3,
+                "config_change": 0.25,
+                "dependency_update": 0.25,
+                "scaling": 0.1,
+                "cert_rotation": 0.05,
+                "feature_flag": 0.2,
+            }
+            score += type_risk.get(change.get("type", ""), 0.1)
+
+            # Same service bonus
+            if change.get("service", "") == service_name:
+                score += 0.2
+
+            # Breaking change keyword detection
+            desc = change.get("description", "").lower()
+            risk_keywords = ["breaking", "major", "migration", "reduced", "limit", "pooling", "connection", "auth", "security"]
+            keyword_hits = sum(1 for kw in risk_keywords if kw in desc)
+            score += min(keyword_hits * 0.05, 0.15)
+
+            scored_changes.append({
+                **change,
+                "hours_before_incident": round(time_diff, 2),
+                "risk_score": round(min(score, 0.95), 2),
+                "risk_level": "high" if score > 0.6 else "medium" if score > 0.3 else "low",
+            })
+
+        scored_changes.sort(key=lambda x: x["risk_score"], reverse=True)
+
+        result = {
+            "incident_time": incident_time,
+            "service": service_name,
+            "lookback_hours": lookback_hours,
+            "changes_found": len(scored_changes),
+            "changes_ranked_by_risk": scored_changes,
+            "highest_risk_change": scored_changes[0] if scored_changes else None,
+            "recommendation": (
+                f"INVESTIGATE: {scored_changes[0]['type']} on {scored_changes[0].get('service', 'unknown')} at {scored_changes[0]['timestamp']} (risk score: {scored_changes[0]['risk_score']})"
+                if scored_changes and scored_changes[0]["risk_score"] > 0.5
+                else "No high-risk changes found in the lookback window."
+            ),
+        }
+
+        print(f"[ChangeCorrelator] Found {len(scored_changes)} changes, highest risk: {scored_changes[0]['risk_score'] if scored_changes else 'N/A'}")
+        return json.dumps(result, indent=2)