fix(api): explain auto execute slo degradation

apps/api/src/services/ai_slo_calculator.py

@@ -23,7 +23,9 @@ from __future__ import annotations
 
 import json
 from dataclasses import dataclass, field
-from datetime import timedelta
+from datetime import datetime, timedelta
+from math import ceil
+from typing import Any
 
 import structlog
 from sqlalchemy import func, select, text
@@ -81,6 +83,7 @@ class SloReport:
     any_violated: bool = False
     calculated_at: str = field(default_factory=lambda: now_taipei().isoformat())
     window_days: int = SLO_WINDOW_DAYS
+    diagnostics: dict[str, Any] = field(default_factory=dict)
 
     def to_dict(self) -> dict:
         return {
@@ -99,6 +102,7 @@ class SloReport:
                 }
                 for m in self.metrics
             ],
+            "diagnostics": self.diagnostics,
         }
 
 
@@ -131,6 +135,11 @@ class AiSloCalculator:
                 slo1 = await self._calc_auto_success_rate(session, since)
                 slo2 = await self._calc_human_override_rate(session, since)
                 slo3 = await self._calc_false_neg_rate(session, since)
+                diagnostics = {}
+                if slo1.violated:
+                    diagnostics["auto_execute_success_rate"] = (
+                        await self._build_auto_success_diagnostics(session, since)
+                    )
 
             metrics = [slo1, slo2, slo3]
             any_violated = any(m.violated for m in metrics)
@@ -138,6 +147,7 @@ class AiSloCalculator:
             report = SloReport(
                 metrics=metrics,
                 any_violated=any_violated,
+                diagnostics=diagnostics,
             )
 
             logger.info(
@@ -189,6 +199,7 @@ class AiSloCalculator:
                     any_violated=data.get("any_violated", False),
                     calculated_at=data.get("calculated_at", ""),
                     window_days=data.get("window_days", SLO_WINDOW_DAYS),
+                    diagnostics=data.get("diagnostics", {}),
                 )
         except Exception as e:
             logger.warning("slo_cache_read_error", error=str(e))
@@ -403,6 +414,264 @@ class AiSloCalculator:
                 direction="below", sample_count=0, violated=False,
             )
 
+    async def _build_auto_success_diagnostics(self, session, since) -> dict[str, Any]:
+        """建立 W-1 auto_execute_success_rate 的可解釋診斷資料。"""
+        try:
+            result = await session.execute(
+                text("""
+                SELECT
+                    are.incident_id,
+                    are.playbook_id,
+                    are.playbook_name,
+                    are.success,
+                    are.error_message,
+                    are.created_at,
+                    COALESCE(
+                        inc.signals->0->>'alertname',
+                        inc.signals->0->'labels'->>'alertname',
+                        inc.signals->0->>'alert_name',
+                        inc.affected_services->>0,
+                        'unknown'
+                    ) AS alertname
+                FROM auto_repair_executions are
+                LEFT JOIN incidents inc ON inc.incident_id = are.incident_id
+                WHERE are.created_at >= :since
+                ORDER BY are.created_at ASC
+                """),
+                {"since": since},
+            )
+            rows = [dict(row._mapping) for row in result]
+            return build_auto_execute_success_diagnostics(
+                rows=rows,
+                now=now_taipei(),
+                threshold=SLO_AUTO_SUCCESS_MIN,
+                window_days=SLO_WINDOW_DAYS,
+                min_samples=SLO_MIN_SAMPLES,
+            )
+        except Exception as e:
+            logger.warning("slo1_diagnostics_error", error=str(e))
+            return {
+                "schema_version": "ai_slo_auto_execute_diagnostics_v1",
+                "status": "diagnostics_unavailable",
+                "error": str(e)[:200],
+            }
+
+
+def build_auto_execute_success_diagnostics(
+    rows: list[dict[str, Any]],
+    now: datetime,
+    threshold: float = SLO_AUTO_SUCCESS_MIN,
+    window_days: int = SLO_WINDOW_DAYS,
+    min_samples: int = SLO_MIN_SAMPLES,
+) -> dict[str, Any]:
+    """
+    從 auto_repair_executions rows 建立前端/Telegram 可讀的 W-1 診斷。
+
+    此函式保持純邏輯，讓 watchdog 與 API 可以共用同一份語義，也方便
+    單元測試鎖住 rolling-window 回綠推估。
+    """
+    sorted_rows = sorted(rows, key=lambda r: r.get("created_at") or now)
+    total = len(sorted_rows)
+    success = sum(1 for row in sorted_rows if bool(row.get("success")))
+    failed = total - success
+    rate = (success / total) if total else None
+    failures = [row for row in sorted_rows if not bool(row.get("success"))]
+    failure_groups = _build_failure_groups(failures)
+    sealed_groups = [
+        group for group in failure_groups
+        if str(group.get("closure_status", "")).startswith("sealed_")
+    ]
+    open_groups = [
+        group for group in failure_groups
+        if not str(group.get("closure_status", "")).startswith("sealed_")
+    ]
+    projected_green_at, projection_reason = _project_auto_success_green_at(
+        rows=sorted_rows,
+        now=now,
+        threshold=threshold,
+        window_days=window_days,
+        min_samples=min_samples,
+    )
+
+    if failed == 0:
+        status = "green"
+    elif open_groups:
+        status = "needs_investigation"
+    elif sealed_groups:
+        status = "sealed_waiting_window"
+    else:
+        status = "insufficient_diagnostics"
+
+    return {
+        "schema_version": "ai_slo_auto_execute_diagnostics_v1",
+        "status": status,
+        "summary": {
+            "total": total,
+            "success": success,
+            "failed": failed,
+            "rate": rate,
+            "threshold": threshold,
+            "window_days": window_days,
+            "min_samples": min_samples,
+        },
+        "top_failure_groups": failure_groups[:5],
+        "sealed_failure_group_count": len(sealed_groups),
+        "open_failure_group_count": len(open_groups),
+        "immediate_successes_needed": _successes_needed_now(success, total, threshold),
+        "projected_green_at": projected_green_at.isoformat() if projected_green_at else None,
+        "projection_reason": projection_reason,
+        "next_action": _auto_execute_diagnostics_next_action(status),
+    }
+
+
+def _build_failure_groups(failures: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    groups: dict[tuple[str, str, str, str], dict[str, Any]] = {}
+    for row in failures:
+        alertname = str(row.get("alertname") or "unknown")
+        playbook_id = str(row.get("playbook_id") or "unknown")
+        playbook_name = str(row.get("playbook_name") or "unknown")
+        error_signature = _auto_repair_error_signature(row.get("error_message"))
+        key = (alertname, playbook_id, playbook_name, error_signature)
+        group = groups.setdefault(
+            key,
+            {
+                "alertname": alertname,
+                "playbook_id": playbook_id,
+                "playbook_name": playbook_name,
+                "error_signature": error_signature,
+                "count": 0,
+                "first_seen": None,
+                "last_seen": None,
+                "example_incident_id": row.get("incident_id"),
+            },
+        )
+        group["count"] += 1
+        created_at = row.get("created_at")
+        if isinstance(created_at, datetime):
+            if group["first_seen"] is None or created_at < group["first_seen"]:
+                group["first_seen"] = created_at
+            if group["last_seen"] is None or created_at > group["last_seen"]:
+                group["last_seen"] = created_at
+
+    enriched = []
+    for group in groups.values():
+        closure = _classify_auto_repair_failure_closure(group)
+        enriched.append({
+            **group,
+            "first_seen": group["first_seen"].isoformat() if group["first_seen"] else None,
+            "last_seen": group["last_seen"].isoformat() if group["last_seen"] else None,
+            **closure,
+        })
+
+    return sorted(enriched, key=lambda item: item["count"], reverse=True)
+
+
+def _auto_repair_error_signature(error_message: Any) -> str:
+    error = str(error_message or "").strip().lower()
+    if not error:
+        return "missing_error_message"
+    if "unsupported scheme" in error and "docker restart" in error:
+        return "legacy_ssh_docker_restart"
+    if "nodes" in error and "not found" in error:
+        return "k3s_node_target_not_found"
+    if "http error" in error:
+        return "http_error"
+    if "timeout" in error:
+        return "timeout"
+    compact = " ".join(error.split())
+    return compact[:120] or "unknown_error"
+
+
+def _classify_auto_repair_failure_closure(group: dict[str, Any]) -> dict[str, str]:
+    signature = str(group.get("error_signature") or "")
+    alertname = str(group.get("alertname") or "")
+    playbook_name = str(group.get("playbook_name") or "")
+    text = f"{alertname} {playbook_name}".lower()
+
+    if signature == "legacy_ssh_docker_restart":
+        return {
+            "closure_status": "sealed_by_mcp_grant",
+            "closure_label": "已封口：Docker restart 已改走 ssh_docker_restart/write MCP grant",
+            "recommended_action": "觀察後續 DockerContainerUnhealthy 執行，不回填舊歷史",
+        }
+
+    if signature == "k3s_node_target_not_found" and (
+        "stock" in text or "wooo.work" in text or "external" in text
+    ):
+        return {
+            "closure_status": "sealed_by_external_site_guard",
+            "closure_label": "已封口：外部站台告警已阻擋 K3s node PlayBook 誤配",
+            "recommended_action": "觀察 StockWoooWorkDown 是否改走 external_site_down / NO_ACTION",
+        }
+
+    return {
+        "closure_status": "open_failure_source",
+        "closure_label": "待調查：尚未匹配到已封口修復來源",
+        "recommended_action": "反查 incident truth-chain、PlayBook、MCP 執行紀錄",
+    }
+
+
+def _successes_needed_now(success: int, total: int, threshold: float) -> int:
+    if total <= 0 or threshold >= 1:
+        return 0
+    gap = (threshold * total) - success
+    if gap <= 0:
+        return 0
+    return max(0, ceil(gap / (1 - threshold)))
+
+
+def _project_auto_success_green_at(
+    rows: list[dict[str, Any]],
+    now: datetime,
+    threshold: float,
+    window_days: int,
+    min_samples: int,
+) -> tuple[datetime | None, str | None]:
+    window = timedelta(days=window_days)
+    current_rows = [
+        row for row in rows
+        if isinstance(row.get("created_at"), datetime)
+        and row["created_at"] >= now - window
+    ]
+    current_total = len(current_rows)
+    current_success = sum(1 for row in current_rows if bool(row.get("success")))
+
+    if current_total < min_samples:
+        return now, "sample_window_below_min"
+    if current_success / current_total >= threshold:
+        return now, "already_green"
+
+    candidates = sorted({
+        row["created_at"] + window + timedelta(seconds=1)
+        for row in current_rows
+        if row["created_at"] + window > now
+    })
+    for checkpoint in candidates:
+        active_rows = [
+            row for row in rows
+            if isinstance(row.get("created_at"), datetime)
+            and row["created_at"] >= checkpoint - window
+            and row["created_at"] <= checkpoint
+        ]
+        active_total = len(active_rows)
+        active_success = sum(1 for row in active_rows if bool(row.get("success")))
+        if active_total < min_samples:
+            return checkpoint, "sample_window_below_min"
+        if active_success / active_total >= threshold:
+            return checkpoint, "rolling_window_if_no_new_failures"
+
+    return None, "no_projection_available"
+
+
+def _auto_execute_diagnostics_next_action(status: str) -> str:
+    if status == "green":
+        return "keep_monitoring"
+    if status == "sealed_waiting_window":
+        return "observe_rolling_window_no_manual_restart"
+    if status == "needs_investigation":
+        return "investigate_open_failure_groups"
+    return "refresh_truth_chain_and_execution_logs"
+
 
 # ─────────────────────────────────────────────────────────────────────────────
 # Singleton