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

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

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

第二数据源(0.58.19):被丢弃的畸形 tool_call 参数(kind=malformed)——这类失败
整轮不入 messages(防投毒级联),loop 落 usage_events(kind=tool_malformed),
在此并入同一聚合口径(signature=归一化 JSON 报错,sample=损坏参数首尾片段)。

第三数据源(0.58.21):run 级终态错误(kind=run)—— LLM 请求层/构建期直接抛异常
(RateLimitError 余额不足、认证失败等),整轮无 tool 消息,_run_agent_bg 落
usage_events(kind=run_error),在此按归一化错误签名聚合(tool 名固定 "(run)")。
其中 provider 级致命错误(余额/认证)另走 `alert_provider_critical` 即时邮件,
不等日巡检 —— 这类错误会让该 provider 上所有用户的所有 run 全挂。

失败判定(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 math
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 错误消息,返回超阈值的聚集。

    阈值语义:同签名 >= min_count 次 且 跨 >= min_tasks 个 task —— 单 task 内
    模型试错几次就自愈的正常噪音不触发;跨 task 复现的才是平台性问题。

    时间分布:每个聚集带 `daily`(从 now 往回按 24h 分桶的次数,旧→新,
    非日历日)和 `count_24h`(= daily 尾桶)—— 修复部署后看尾桶是否归零,
    区分「还在发生」和「窗口内的存量记录」。排序:近 24h 活跃的在前
    (count_24h 降序),其后按 count 降序。
    同步阻塞(DB 查询),asyncio 调用方放 to_thread/executor。
    """
    now = datetime.now(timezone.utc)
    cutoff = now - timedelta(days=days)
    n_buckets = max(1, math.ceil(days))
    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()
        # 第二段:被丢弃的畸形 tool_call 参数(kind=tool_malformed)。这类失败整轮
        # 不入 messages(防投毒),loop._log_malformed_args 落在 usage_events,
        # 是它们进面板/巡检邮件的唯一路径。
        mrows = s.execute(
            text(
                "select task_id, user_id, created_at, "
                "       units->>'tool' as tool_name, "
                "       units->>'err' as err, "
                "       units->>'head' as head, "
                "       units->>'tail' as tail "
                "from usage_events "
                "where kind = 'tool_malformed' and created_at >= :cutoff"
            ),
            {"cutoff": cutoff},
        ).fetchall()
        # 第三段:run 级终态错误(kind=run)。LLM 层抛异常时整轮无 tool 消息,
        # tasks.run_error 只留最后一次,usage_events(kind=run_error)才是完整留痕。
        rrows = s.execute(
            text(
                "select task_id, user_id, created_at, "
                "       units->>'err' as err "
                "from usage_events "
                "where kind = 'run_error' and created_at >= :cutoff"
            ),
            {"cutoff": cutoff},
        ).fetchall()

    agg: Dict[Tuple[str, str], Dict[str, Any]] = {}

    def _add(
        tool_name: str, kind: str, sig_line: str, sample: str,
        task_id: Any, user_id: Any, created_at: datetime,
    ) -> None:
        # DB 列若是 naive timestamp(存 UTC),补 tzinfo 才能和 now 做减法
        ts = created_at if created_at.tzinfo else created_at.replace(tzinfo=timezone.utc)
        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(),
                "first_at": ts,
                "last_at": ts,
                "sample": sample[:300],
                "daily": [0] * n_buckets,
            }
        c["count"] += 1
        c["tasks"].add(task_id)
        c["users"].add(user_id)
        # 分桶:距 now 每满 24h 退一桶,尾桶 = 近 24h(时钟漂移/边界值 clamp 进首尾桶)
        age_days = int((now - ts).total_seconds() // 86400)
        c["daily"][n_buckets - 1 - min(n_buckets - 1, max(0, age_days))] += 1
        if ts < c["first_at"]:
            c["first_at"] = ts
        if ts > c["last_at"]:
            c["last_at"] = ts
            c["sample"] = sample[:300]

    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
        _add(tool_name, kind, sig_line, content, task_id, user_id, created_at)

    for task_id, user_id, created_at, tool_name, err, head, tail in mrows:
        _add(
            tool_name, "malformed", err or "?",
            f"{head or ''} … {tail or ''}",
            task_id, user_id, created_at,
        )

    for task_id, user_id, created_at, err in rrows:
        _add("(run)", "run", err or "?", err or "", task_id, user_id, created_at)

    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"],
            "count_24h": c["daily"][-1],
            "daily": c["daily"],
            "task_count": len(c["tasks"]),
            "user_count": len(c["users"]),
            "first_at": c["first_at"].isoformat(),
            "last_at": c["last_at"].isoformat(),
            "sample": c["sample"],
        })
    # 活跃的(近 24h 还在发生)排前面,已安静的沉底 —— 面板/邮件都先看还在烧的
    out.sort(key=lambda x: (x["count_24h"], x["count"]), reverse=True)
    return out


# ── provider 级致命错误即时告警 ──
# 命中判据:错误文案含余额/配额/认证类关键词 —— 这类错误不是单任务偶发,而是该
# provider 上所有 run 全挂(如 Zai 余额不足),等日巡检的 5 次/2 task 阈值太慢。
# 冷却:同归一化签名 6h 内只发一封(进程内存态,重启清零 —— 重启后再发一封可接受,
# 比引入持久化状态表划算)。
_CRITICAL_RE = re.compile(
    r"余额不足|无可用资源包|请充值|欠费"
    r"|insufficient[_ ](?:quota|balance|funds)"
    r"|exceeded your current quota"
    r"|AuthenticationError|invalid[_ ]api[_ ]?key|api key.{0,20}(?:invalid|expired)",
    re.IGNORECASE,
)
_ALERT_COOLDOWN_S = 6 * 3600
_alerted_at: Dict[str, datetime] = {}


def alert_provider_critical(err: str, *, task_id: Any = None, model_profile: str = "") -> bool:
    """run 错误若属 provider 级致命(余额/认证)→ 立即邮件开发者;返回是否已发。

    同步阻塞(SMTP),只应在 worker 线程调用(_run_agent_bg 的 except 路径本就在
    to_thread 里)。所有失败静默 —— 告警绝不能反过来在错误路径上再抛异常。
    """
    try:
        if not err or not _CRITICAL_RE.search(err):
            return False
        sig = _normalize(err)
        now = datetime.now(timezone.utc)
        last = _alerted_at.get(sig)
        if last and (now - last).total_seconds() < _ALERT_COOLDOWN_S:
            return False
        _alerted_at[sig] = now
        import os

        from tools.send_email import send_email_smtp, smtp_configured
        print(f"[runerror] provider-critical task={task_id} mp={model_profile} "
              f"err={err[:200]}", flush=True)
        dev_email = os.getenv("ZCBOT_DEVELOPER_EMAIL", "").strip()
        if not (dev_email and smtp_configured()):
            print("[runerror] ZCBOT_DEVELOPER_EMAIL/SMTP 未配,仅日志", flush=True)
            return False
        body = "\n".join([
            "检出 provider 级致命错误(余额/配额/认证),该 provider 上的 run 可能全部失败:",
            "",
            f"错误: {err[:500]}",
            f"模型档: {model_profile or '?'}",
            f"任务: {task_id or '?'}",
            "",
            f"同类签名 {_ALERT_COOLDOWN_S // 3600}h 内不再重复告警;"
            "完整聚合见 admin 工具失败面板(kind=run)。",
        ])
        send_email_smtp(dev_email, f"[zcbot] provider 级错误:{sig[:60]}", body)
        return True
    except Exception:
        return False


def format_alert(clusters: List[Dict[str, Any]], days: float) -> str:
    """聚集列表 → 告警邮件正文(纯文本)。"""
    lines = [f"近 {days:g} 天内检出 {len(clusters)} 类工具失败聚集(近 24h 活跃的在前):", ""]
    for c in clusters:
        n24 = c.get("count_24h", 0)
        lines.append(
            f"- [{c['tool']}/{c['kind']}] x{c['count']}"
            f"(近24h {n24} 次{',已安静' if not n24 else ''},"
            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)