src/alerts.py

"""
P10 · Alert Manager
Anomaly detection on dashboard metrics.
Generates structured alerts — in production would push to Slack/PagerDuty.
"""

from dataclasses import dataclass
from typing import Any


@dataclass
class Alert:
    id: str
    severity: str        # critical | warning | info
    category: str        # slo | cost | latency | eval | prompt
    service: str
    message: str
    value: float
    threshold: float
    recommendation: str

    def to_dict(self) -> dict:
        return {
            "id": self.id,
            "severity": self.severity,
            "category": self.category,
            "service": self.service,
            "message": self.message,
            "value": self.value,
            "threshold": self.threshold,
            "recommendation": self.recommendation,
        }


# ── Anomaly detectors ─────────────────────────────────────────────────────────
def check_slo_alerts(slo_data: dict) -> list[Alert]:
    alerts = []
    for svc, metrics in slo_data.items():
        burn = metrics["burn_rate"]
        if burn > 14.4:
            alerts.append(Alert(
                id=f"slo-{svc}-critical",
                severity="critical",
                category="slo",
                service=svc,
                message=f"Burn rate {burn}x — exhausts budget in <2h",
                value=burn,
                threshold=14.4,
                recommendation=f"Page on-call immediately. Check recent deploys for {svc}.",
            ))
        elif burn > 6:
            alerts.append(Alert(
                id=f"slo-{svc}-warning",
                severity="warning",
                category="slo",
                service=svc,
                message=f"Burn rate {burn}x — exhausts budget in <5 days",
                value=burn,
                threshold=6.0,
                recommendation=f"Investigate {svc} error rate. Review slow query logs.",
            ))
    return alerts


def check_latency_alerts(latency_data: dict) -> list[Alert]:
    alerts = []
    for svc, metrics in latency_data.items():
        p99 = metrics["p99_ms"]
        slo = metrics["slo_ms"]
        ratio = p99 / slo
        if ratio > 0.9:
            severity = "critical" if ratio > 1.0 else "warning"
            alerts.append(Alert(
                id=f"latency-{svc}",
                severity=severity,
                category="latency",
                service=svc,
                message=f"p99 {p99}ms is {ratio*100:.0f}% of {slo}ms SLO",
                value=p99,
                threshold=slo,
                recommendation=f"Check {svc} for slow queries or resource saturation.",
            ))
    return alerts


def check_cost_alerts(cost_data: dict) -> list[Alert]:
    alerts = []
    avg = cost_data.get("avg_daily", 0)
    daily = cost_data.get("daily", [])
    if len(daily) >= 2:
        today = daily[-1]["total"]
        yesterday = daily[-2]["total"]
        if yesterday > 0 and today > yesterday * 1.5:
            alerts.append(Alert(
                id="cost-spike",
                severity="warning",
                category="cost",
                service="llm-costs",
                message=f"Daily cost ${today:.4f} is 50%+ above yesterday ${yesterday:.4f}",
                value=today,
                threshold=yesterday * 1.5,
                recommendation="Check for unusual traffic or prompt loops causing extra tokens.",
            ))
    return alerts


def check_eval_alerts(eval_history: list[dict]) -> list[Alert]:
    alerts = []
    if len(eval_history) >= 2:
        current = eval_history[-1]["avg_composite"]
        previous = eval_history[-2]["avg_composite"]
        if previous > 0:
            drop_pct = ((previous - current) / previous) * 100
            if drop_pct > 10:
                alerts.append(Alert(
                    id="eval-regression",
                    severity="critical",
                    category="eval",
                    service="llm-quality",
                    message=f"Eval score dropped {drop_pct:.1f}% ({previous:.2f} → {current:.2f})",
                    value=current,
                    threshold=previous * 0.9,
                    recommendation="Block deployment. Review recent prompt changes or model updates.",
                ))
            elif drop_pct > 5:
                alerts.append(Alert(
                    id="eval-warning",
                    severity="warning",
                    category="eval",
                    service="llm-quality",
                    message=f"Eval score dropped {drop_pct:.1f}% — approaching regression threshold",
                    value=current,
                    threshold=previous * 0.95,
                    recommendation="Investigate prompt changes. Run extended test suite.",
                ))
    return alerts


def detect_all_anomalies(dashboard_data: dict) -> list[Alert]:
    """Run all anomaly detectors and return combined alert list."""
    alerts = []
    alerts += check_slo_alerts(dashboard_data.get("slo", {}))
    alerts += check_latency_alerts(dashboard_data.get("latency", {}))
    alerts += check_cost_alerts(dashboard_data.get("cost_metrics", {}))
    alerts += check_eval_alerts(dashboard_data.get("eval_history", []))

    # Sort: critical first, then warning, then info
    priority = {"critical": 0, "warning": 1, "info": 2}
    return sorted(alerts, key=lambda a: priority.get(a.severity, 3))