awoooi/apps/api/src/jobs/baseline_snapshot.py

"""
AWOOOI AIOps Phase 0 — 基線快照 Job
=====================================
拍攝 AI 自主化飛輪「啟動前現況」，作為 Phase 0→1 進展衡量基準。

快照涵蓋 ADR-080 診斷表中的 6 大指標：
  1. MCP 呼叫次數/24h（目標：> 0；現況預估：0）
  2. Playbook trust/confidence 分佈（目標：動態；現況：全靜態）
  3. 學習閉環觸發率（目標：≥ 99%；現況：0%，fire-and-forget）
  4. 告警分類 general 比例（目標：< 10%；現況：~ 41%）
  5. 修復動作 RESTART 比例（目標：< 40%；現況：~ 68%）
  6. 自動執行成功次數/24h（目標：> 0；現況：0）

儲存策略：
  - Redis Key `aiops:baseline:{timestamp_iso}` — 最新快照（TTL 永不過期）
  - Redis Key `aiops:baseline:latest` — 指向最新快照的時間戳（方便 API 讀取）

使用方式：
  python -m src.jobs.baseline_snapshot          # 直接執行（一次性）
  await take_baseline_snapshot()                 # 從程式碼呼叫

ADR-080: AI 自主化飛輪總綱
MASTER: docs/superpowers/specs/2026-04-15-MASTER-ai-autonomous-flywheel-v2.md §5 Phase 0

2026-04-15 ogt + Claude Sonnet 4.6 (亞太): Phase 0 — 初始建立
"""

from __future__ import annotations

import asyncio
import json
from datetime import timedelta

import structlog
from sqlalchemy import func, select, text

from src.core.redis_client import get_redis
from src.db.base import get_db_context
from src.db.models import (
    AutoRepairExecution,
    IncidentRecord,
    KnowledgeEntryRecord,
)
from src.utils.timezone import now_taipei

logger = structlog.get_logger(__name__)

# Redis 鍵
BASELINE_KEY_PREFIX = "aiops:baseline:"
BASELINE_LATEST_KEY = "aiops:baseline:latest"

# Playbook Redis 前綴（同 playbook_repository.py）
PLAYBOOK_KEY_PREFIX = "playbook:"


async def take_baseline_snapshot() -> dict:
    """
    拍攝一次完整基線快照並寫入 Redis。

    Returns:
        dict: 快照內容（含 snapshot_at 時間戳）
    """
    now = now_taipei()
    since_24h = now - timedelta(hours=24)
    ts_iso = now.isoformat()

    logger.info("baseline_snapshot_start", snapshot_at=ts_iso)

    snapshot = {
        "snapshot_at": ts_iso,
        "phase": "P0",
        "description": "AI 自主化飛輪 Phase 0 啟動前基線",
        "metrics": {},
    }

    # ── 1. MCP 呼叫次數/24h ───────────────────────────────────────────────
    # Phase 0 時 MCP 尚未接入任何決策流程 → 預期為 0
    # Phase 1 完成後此數字應 > 0（PreDecisionInvestigator 開始呼叫）
    mcp_calls_24h = await _count_mcp_calls_24h(since_24h)
    snapshot["metrics"]["mcp_calls_24h"] = mcp_calls_24h

    # ── 2. Playbook confidence 分佈（Redis 掃描）──────────────────────────
    playbook_stats = await _playbook_confidence_stats()
    snapshot["metrics"]["playbook"] = playbook_stats

    # ── 3. 學習閉環觸發率 + 其他 DB 指標 ─────────────────────────────────
    db_metrics = await _db_metrics(since_24h)
    snapshot["metrics"].update(db_metrics)

    # ── 4. 計算衍生指標 ───────────────────────────────────────────────────
    snapshot["metrics"]["learning_loop_rate"] = _compute_learning_rate(
        db_metrics.get("auto_repair_24h", 0),
        db_metrics.get("learning_writes_24h", 0),
    )

    # ── 寫入 Redis ─────────────────────────────────────────────────────────
    await _persist_to_redis(ts_iso, snapshot)

    logger.info(
        "baseline_snapshot_done",
        snapshot_at=ts_iso,
        mcp_calls_24h=mcp_calls_24h,
        playbook_total=playbook_stats.get("total", 0),
        incidents_24h=db_metrics.get("incidents_24h", 0),
        auto_repair_success_24h=db_metrics.get("auto_repair_success_24h", 0),
    )

    return snapshot


# ─────────────────────────────────────────────────────────────────────────────
# Internal helpers
# ─────────────────────────────────────────────────────────────────────────────


async def _count_mcp_calls_24h(since_24h) -> int:
    """
    MCP 呼叫次數/24h。

    Phase 0：無 MCP Calls Table → 從 audit_logs 嘗試計數。
    Phase 1 建立 PreDecisionInvestigator 後，此處改為查 mcp_tool_calls 表。
    """
    try:
        async with get_db_context() as db:
            # audit_logs 中 action='mcp_call' — Phase 0 預期 0 筆
            result = await db.execute(
                text(
                    "SELECT COUNT(*) FROM audit_logs "
                    "WHERE action = 'mcp_call' AND created_at >= :since"
                ),
                {"since": since_24h},
            )
            return result.scalar_one_or_none() or 0
    except Exception:
        logger.exception("baseline_mcp_count_error")
        return 0


async def _playbook_confidence_stats() -> dict:
    """
    掃描 Redis 中全部 Playbook，統計 ai_confidence 分佈。

    指標診斷：
    - avg_confidence ≈ 0.3 → 佐證「全靜態」現況（Phase 0 基線）
    - Phase 3 EWMA 上線後，此值應動態分散（std_dev 升高、avg 可能提升）
    """
    stats = {
        "total": 0,
        "approved": 0,
        "avg_confidence": 0.0,
        "min_confidence": None,
        "max_confidence": None,
        "never_used": 0,       # success_count + failure_count == 0
        "action_type_dist": {},
    }

    try:
        redis = get_redis()
        confidences: list[float] = []
        action_counts: dict[str, int] = {}

        async for key in redis.scan_iter(match=f"{PLAYBOOK_KEY_PREFIX}PB-*", count=200):
            raw = await redis.get(key)
            if not raw:
                continue
            try:
                pb = json.loads(raw)
            except json.JSONDecodeError:
                continue

            stats["total"] += 1

            if pb.get("status") == "approved":
                stats["approved"] += 1

            conf = pb.get("ai_confidence", 0.0) or 0.0
            confidences.append(conf)

            used = (pb.get("success_count", 0) or 0) + (pb.get("failure_count", 0) or 0)
            if used == 0:
                stats["never_used"] += 1

            # 統計 repair_steps 中首個 action_type（代表主要修復動作）
            steps = pb.get("repair_steps", [])
            if steps:
                first_action = steps[0].get("action_type", "unknown")
                action_counts[first_action] = action_counts.get(first_action, 0) + 1

        if confidences:
            stats["avg_confidence"] = round(sum(confidences) / len(confidences), 4)
            stats["min_confidence"] = round(min(confidences), 4)
            stats["max_confidence"] = round(max(confidences), 4)

        # RESTART 比例：佐證 ADR-080 診斷（目標 < 40%）
        total_actions = sum(action_counts.values())
        restart_count = action_counts.get("restart_service", 0)
        stats["restart_ratio"] = round(restart_count / total_actions, 4) if total_actions else 0.0
        stats["action_type_dist"] = action_counts

    except Exception:
        logger.exception("baseline_playbook_stats_error")

    return stats


async def _db_metrics(since_24h) -> dict:
    """
    從 PostgreSQL 取得核心計數指標。
    """
    metrics: dict = {
        "incidents_24h": 0,
        "incidents_total": 0,
        "general_alert_ratio": 0.0,
        "auto_repair_24h": 0,
        "auto_repair_success_24h": 0,
        "km_total": 0,
        "km_vectorized": 0,
        "learning_writes_24h": 0,
        "audit_logs_24h": 0,
    }

    try:
        async with get_db_context() as db:
            # Incident 數量（24h + 總計）
            r = await db.execute(
                select(func.count(IncidentRecord.incident_id)).where(
                    IncidentRecord.created_at >= since_24h
                )
            )
            metrics["incidents_24h"] = r.scalar_one_or_none() or 0

            r = await db.execute(select(func.count(IncidentRecord.incident_id)))
            metrics["incidents_total"] = r.scalar_one_or_none() or 0

            # general 告警比例（alert_category = 'general'）
            r = await db.execute(
                select(func.count()).where(
                    IncidentRecord.alert_category == "general"
                )
            )
            general_count = r.scalar_one_or_none() or 0
            total = metrics["incidents_total"]
            metrics["general_alert_ratio"] = round(general_count / total, 4) if total else 0.0

            # 自動修復執行（24h）
            r = await db.execute(
                select(func.count(AutoRepairExecution.id)).where(
                    AutoRepairExecution.created_at >= since_24h
                )
            )
            metrics["auto_repair_24h"] = r.scalar_one_or_none() or 0

            r = await db.execute(
                select(func.count(AutoRepairExecution.id)).where(
                    AutoRepairExecution.created_at >= since_24h,
                    AutoRepairExecution.success.is_(True),
                )
            )
            metrics["auto_repair_success_24h"] = r.scalar_one_or_none() or 0

            # KM 數量 + 向量化率
            r = await db.execute(select(func.count(KnowledgeEntryRecord.id)))
            metrics["km_total"] = r.scalar_one_or_none() or 0

            r = await db.execute(
                select(func.count()).where(
                    KnowledgeEntryRecord.embedding.is_not(None)
                )
            )
            metrics["km_vectorized"] = r.scalar_one_or_none() or 0

            # 學習寫入數（24h 內新增 KM）
            r = await db.execute(
                select(func.count()).where(
                    KnowledgeEntryRecord.created_at >= since_24h
                )
            )
            metrics["learning_writes_24h"] = r.scalar_one_or_none() or 0

            # audit_logs 24h 計數（Phase 0 預期 = 0）
            r = await db.execute(
                text(
                    "SELECT COUNT(*) FROM audit_logs WHERE created_at >= :since"
                ),
                {"since": since_24h},
            )
            metrics["audit_logs_24h"] = r.scalar_one_or_none() or 0

    except Exception:
        logger.exception("baseline_db_metrics_error")

    return metrics


def _compute_learning_rate(auto_repair_24h: int, learning_writes_24h: int) -> float:
    """
    學習閉環觸發率 = learning_writes_24h / auto_repair_24h。

    Phase 0 診斷：fire-and-forget → 比率為 0%（即使 auto_repair > 0，learning 也可能 = 0）
    Phase 3 修復後目標：≥ 99%
    """
    if auto_repair_24h == 0:
        return 0.0
    return round(min(learning_writes_24h / auto_repair_24h, 1.0), 4)


async def _persist_to_redis(ts_iso: str, snapshot: dict) -> None:
    """
    將快照寫入 Redis：
    - `aiops:baseline:{ts_iso}` — 歷史記錄（永不過期）
    - `aiops:baseline:latest` — 最新快照全量（永不過期）
    """
    try:
        redis = get_redis()
        payload = json.dumps(snapshot, ensure_ascii=False)

        # 歷史記錄（保留全部 snapshot）
        await redis.set(f"{BASELINE_KEY_PREFIX}{ts_iso}", payload)

        # 最新快照（供 API 快速讀取）
        await redis.set(BASELINE_LATEST_KEY, payload)

        logger.info("baseline_snapshot_persisted", key=BASELINE_LATEST_KEY)
    except Exception:
        logger.exception("baseline_persist_error")


# ─────────────────────────────────────────────────────────────────────────────
# Entry point（直接執行）
# ─────────────────────────────────────────────────────────────────────────────

async def _main() -> None:
    snapshot = await take_baseline_snapshot()
    print(json.dumps(snapshot, indent=2, ensure_ascii=False))


if __name__ == "__main__":
    asyncio.run(_main())