"""工具失败聚集巡检:同签名的工具报错反复出现 → 主动冒头。

背景(2026-07,task 9dcae061 终案的结构性教训):mermaid 渲染在生产挂了 90 天
0 成功(67 次超时 + 26 次 launch fail、烧掉数十万 token),没有任何机制发现,
靠人工扫 DB 才挖出来。本模块把「失败聚集」变成信号:扫 messages 里 role=tool
的错误结果,按 (工具名 + 归一化错误签名) 聚合,超阈值即算聚集。

纯只读查询、无新表无状态;告警通道由调用方决定(web/app.py 的巡检 loop 发
开发者邮箱,admin API 直接返给前端表格)。

失败判定(tool content 的三类标记,形态见 executor_docker/_host):
- `[Error` 开头       —— 执行器/工具层报错([Error]、[Error executing ...])
- `command timed out` —— shell/run_python 超时
- 尾部 `[exit N]` 且 N != 0 —— shell 非零退出
`[exit 0]` 但语义失败(如 "No mermaid charts found")不判 —— 无通用判据,不猜。
"""
from __future__ import annotations

import re
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional, Tuple

from sqlalchemy import text

from core.storage import session_scope

# 签名归一:同一类错误在不同 task/参数下的差异(路径/数字/uuid/十六进制)抹平,
# 让 "figures/a.png doesn't exist" 和 "figures/b.png doesn't exist" 聚成一条。
_RE_UUID = re.compile(r"[0-9a-fA-F]{8}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{4}-[0-9a-fA-F]{12}")
_RE_HEX = re.compile(r"0x[0-9a-fA-F]+")
_RE_PATH = re.compile(r"(?:[A-Za-z]:)?(?:[/\\][\w.\-一-鿿*]+){2,}")
_RE_NUM = re.compile(r"\d+")
_RE_WS = re.compile(r"\s+")

_EXIT_TAIL = re.compile(r"\[exit (\d+)\]\s*$")
_STREAM_MARKS = ("[stdout]", "[stderr]")


def _normalize(s: str) -> str:
    s = _RE_UUID.sub("<id>", s)
    s = _RE_HEX.sub("<hex>", s)
    s = _RE_PATH.sub("<path>", s)
    s = _RE_NUM.sub("N", s)
    s = _RE_WS.sub(" ", s).strip()
    return s[:120]


def _classify(content: str) -> Optional[Tuple[str, str]]:
    """返回 (kind, 原始签名行) 或 None(不算失败)。"""
    head = content.lstrip()
    # 超时判定在前:超时结果形如 "[Error] command timed out after 30s",
    # 让它归 timeout 而不是被 [Error 前缀截走(kind 对排查方向有指示意义)
    if "command timed out" in content:
        return "timeout", "command timed out"
    if head.startswith("[Error"):
        return "error", head.splitlines()[0]
    m = _EXIT_TAIL.search(content)
    if m and m.group(1) != "0":
        # 签名取 [exit N] 前最后一行有实际内容的输出(通常是真正的报错行)
        lines = [
            ln.strip() for ln in content.splitlines()[:-1]
            if ln.strip() and ln.strip() not in _STREAM_MARKS
        ]
        return "exit", (lines[-1] if lines else f"exit {m.group(1)}")
    return None


def scan_tool_failures(
    days: float = 7,
    min_count: int = 5,
    min_tasks: int = 2,
) -> List[Dict[str, Any]]:
    """扫近 `days` 天的 tool 错误消息,返回超阈值的聚集(count 降序)。

    阈值语义:同签名 >= min_count 次 且 跨 >= min_tasks 个 task —— 单 task 内
    模型试错几次就自愈的正常噪音不触发;跨 task 复现的才是平台性问题。
    同步阻塞(DB 查询),asyncio 调用方放 to_thread/executor。
    """
    cutoff = datetime.now(timezone.utc) - timedelta(days=days)
    rows = []
    with session_scope() as s:
        rows = s.execute(
            text(
                "select m.task_id, t.user_id, m.created_at, "
                "       m.payload->>'name' as tool_name, "
                "       m.payload->>'content' as content "
                "from messages m join tasks t on t.task_id = m.task_id "
                "where m.created_at >= :cutoff "
                "  and m.payload->>'role' = 'tool' "
                "  and (m.payload->>'content' like '[Error%' "
                "       or m.payload->>'content' like '%command timed out%' "
                "       or m.payload->>'content' like '%[exit %')"
            ),
            {"cutoff": cutoff},
        ).fetchall()

    agg: Dict[Tuple[str, str], Dict[str, Any]] = {}
    for task_id, user_id, created_at, tool_name, content in rows:
        if not content:
            continue
        hit = _classify(content)
        if hit is None:
            continue
        kind, sig_line = hit
        key = (tool_name or "?", _normalize(sig_line))
        c = agg.get(key)
        if c is None:
            c = agg[key] = {
                "tool": key[0],
                "signature": key[1],
                "kind": kind,
                "count": 0,
                "tasks": set(),
                "users": set(),
                "last_at": created_at,
                "sample": content[:300],
            }
        c["count"] += 1
        c["tasks"].add(task_id)
        c["users"].add(user_id)
        if created_at > c["last_at"]:
            c["last_at"] = created_at
            c["sample"] = content[:300]

    out = []
    for c in agg.values():
        if c["count"] < min_count or len(c["tasks"]) < min_tasks:
            continue
        out.append({
            "tool": c["tool"],
            "signature": c["signature"],
            "kind": c["kind"],
            "count": c["count"],
            "task_count": len(c["tasks"]),
            "user_count": len(c["users"]),
            "last_at": c["last_at"].isoformat(),
            "sample": c["sample"],
        })
    out.sort(key=lambda x: x["count"], reverse=True)
    return out


def format_alert(clusters: List[Dict[str, Any]], days: float) -> str:
    """聚集列表 → 告警邮件正文(纯文本)。"""
    lines = [f"近 {days:g} 天内检出 {len(clusters)} 类工具失败聚集:", ""]
    for c in clusters:
        lines.append(
            f"- [{c['tool']}/{c['kind']}] x{c['count']}"
            f"(task {c['task_count']} 个 / 用户 {c['user_count']} 人,"
            f"最近 {c['last_at']})"
        )
        lines.append(f"  签名: {c['signature']}")
        lines.append(f"  样例: {c['sample'][:200]}")
        lines.append("")
    lines.append("排查入口:RUN.md 故障兜底表;历史案例:mermaid loopback DROP(0.58.10)。")
    return "\n".join(lines)