"""每周把助手的回归题库跑一遍,并和上一次比。

这道 cron 存在的理由是:助手的质量此前只有两个信号 —— 员工按的那个 👎,
和有人偶尔手打一句话试试。前者只覆盖"糟到有人愿意点一下"的那一小撮,
后者不可重复也没人记下来。**没有人知道昨天的一次改动让助手变好了还是变差了。**

跑法与手动跑完全同一条路径(`runner.execute_run`):模拟员工身份、走飞书
生产提示词与工具集、干跑派发器 —— 发不出卡片,建不了申请,写不了记录。

失败处理遵循本仓定时任务约定:**handler 必须 raise**,由框架的
`maybe_notify_failure` 去 page IT。这里有一个具体的判据要小心 —— "跑完了但
通过率很低"**不是** handler 失败,那是产品变差了,要走告警而不是走任务红灯;
真正的失败是"这一跑压根没跑起来"(模型没配、题库空了、执行中途崩了)。
把两者混为一谈,会让一次内容回归看起来像一次基础设施故障,找的人都不一样。
"""
from __future__ import annotations

import logging
from typing import Any

from sqlalchemy.ext.asyncio import AsyncSession

from app.assistant.evaluation.constants import (
    DEFAULT_CONCURRENCY,
    DEFAULT_REPEATS,
    MAX_CONCURRENCY,
    MAX_REPEATS,
    REGRESSION_ALERT_DROP_PP,
    RUN_FAILED,
    RUN_PARTIAL,
    TRIGGER_SCHEDULED,
)
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

#: A run left ``running`` for longer than this had its executor go away — a
#: deploy restarting the API process mid-run is the ordinary cause. Generous
#: relative to the run budget so a slow-but-live run is never reaped.
_STALE_RUN_SECONDS = 6 * 3600


@register_scheduled_job(
    key="assistant.eval_regression",
    display_name="助手 · 每周回归评测",
    description=(
        "每周跑一遍回归题库并与上一次终态运行对比;通过率明显下滑时告警 IT,"
        "并点名是哪几条用例从通过变成了不通过。顺带收尾上一周里被发布打断的运行。"
    ),
    # 几十条用例 × 3 遍,每一次都是一整轮模型 + 工具循环 —— 分钟到小时级,
    # 而且占着和员工同一个模型后端。归重。
    weight="heavy",
)
async def run_assistant_eval_regression(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from sqlalchemy import select

    from app.assistant.evaluation import service
    from app.assistant.evaluation.models import (
        AssistantEvalRun,
        AssistantEvalSuite,
    )
    from app.assistant.evaluation.runner import execute_run, reap_stale_runs
    from app.domains.support.copilot import copilot_available

    cfg = config or {}
    slug = str(cfg.get("suite") or "regression")
    repeats = _clamp(cfg.get("repeats"), DEFAULT_REPEATS, 1, MAX_REPEATS)
    concurrency = _clamp(
        cfg.get("concurrency"), DEFAULT_CONCURRENCY, 1, MAX_CONCURRENCY
    )

    reaped = await reap_stale_runs(db, stale_after_seconds=_STALE_RUN_SECONDS)

    if not await copilot_available(db):
        # 没配模型不是这道任务的错,但也不能报成 ok —— "配置从来没生效过"
        # 正是这种静默绿灯能藏一整个季度的地方。
        return {
            "status": "skipped",
            "reason": "assistant/LLM backend not configured",
            "stale_runs_reaped": reaped,
        }

    # 题库可能还没播过(全新部署,或者有人删了)。播一次是幂等的。
    await service.seed_builtin_suites(db)
    await db.commit()

    suite = (
        await db.execute(
            select(AssistantEvalSuite).where(AssistantEvalSuite.slug == slug)
        )
    ).scalar_one_or_none()
    if suite is None:
        raise RuntimeError(
            f"assistant.eval_regression: 题库 {slug!r} 不存在,而且播种之后仍然不存在。"
        )

    try:
        run = await service.start_run(
            db,
            suite_id=suite.id,
            repeats=repeats,
            concurrency=concurrency,
            trigger=TRIGGER_SCHEDULED,
            trace_id="assistant.eval_regression",
        )
    except Exception as exc:  # noqa: BLE001 — 换成任务红灯之前先说清是什么
        raise RuntimeError(
            f"assistant.eval_regression: 无法开始评测:{exc}"
        ) from exc
    run_id = run.id
    await db.commit()

    summary = await execute_run(run_id)
    if summary.get("status") == RUN_FAILED:
        raise RuntimeError(
            "assistant.eval_regression: 这一跑没有跑起来 —— "
            + str(summary.get("notes") or "原因未记录")
        )

    # execute_run 用的是自己的 session,所以这里要重新读一遍才能拿到汇总列。
    db.expire_all()
    fresh = await db.get(AssistantEvalRun, run_id)
    alert = None
    if fresh is not None:
        alert = await _maybe_alert(db, run=fresh)

    return {
        "status": "ok",
        "suite": slug,
        "run_id": str(run_id),
        "run_status": summary.get("status"),
        "pass_rate": summary.get("pass_rate"),
        "attempts": summary.get("attempts_total"),
        "passed": summary.get("attempts_passed"),
        "failed": summary.get("attempts_failed"),
        "errored": summary.get("attempts_errored"),
        "flaky_cases": summary.get("cases_flaky"),
        "partial": summary.get("status") == RUN_PARTIAL,
        "stale_runs_reaped": reaped,
        "alert": alert,
    }


def _clamp(raw: Any, default: int, low: int, high: int) -> int:
    try:
        value = int(raw)
    except (TypeError, ValueError):
        return default
    return max(low, min(value, high))


async def _maybe_alert(db: AsyncSession, *, run: Any) -> dict[str, Any] | None:
    """通过率相对上一次终态运行明显下滑时,告警 IT 管理员。

    只在**可比**的时候告警。样本太少的一次运行"掉 20 个点"可以什么都没发生,
    而一条会误报的告警,几周之内就会变成没人看的告警。
    """
    from app.assistant.evaluation import service
    from app.notifications import fanout
    from app.notifications.service import NotificationEnvelope
    from app.scheduled_jobs.failure_notifier import list_admin_recipients

    baseline = await service.previous_terminal_run(db, run=run)
    if baseline is None:
        return {"sent": False, "reason": "no_baseline"}

    diff = await service.compare_runs(db, run=run, baseline=baseline)
    drop = diff.get("drop_pp")
    if not diff.get("comparable"):
        return {"sent": False, "reason": "not_comparable", "drop_pp": drop}
    if drop is None or drop < REGRESSION_ALERT_DROP_PP:
        return {"sent": False, "reason": "within_threshold", "drop_pp": drop}

    regressed = diff.get("regressed") or []
    lines = "\n".join(
        f"{i + 1}. {item['question'][:100]}"
        f"{'…' if len(item['question']) > 100 else ''}"
        f"({item['before']}% → {item['after']}%)"
        for i, item in enumerate(regressed[:10])
    ) or "(没有单条用例的明确回退 —— 多半是整体波动或错误率上升。)"

    admins = await list_admin_recipients(db)
    if not admins:
        return {"sent": False, "reason": "no_admins", "drop_pp": drop}

    model_note = (
        "本次运行的模型与上次不同,先确认是不是换模型造成的。"
        if diff.get("model_changed")
        else "本次运行的模型与上次相同。"
    )
    envelopes = [
        NotificationEnvelope(
            recipient=admin,
            title=f"助手评测回退 · 通过率下降 {drop} 个百分点",
            body_text="",
            channels=["feishu", "email"],
            template_key="assistant.eval_regression",
            context={
                # ⚠️ The SHARED-case rates, not the two headlines. ``drop_pp``
                # is computed over the cases both runs actually scored; pairing
                # it with each run's own headline printed three numbers that did
                # not add up, in the one artefact an admin reads at 09:00.
                "pass_rate": str(diff.get("shared_pass_rate")),
                "baseline_pass_rate": str(diff.get("shared_baseline_pass_rate")),
                "shared_cases": str(diff.get("shared_cases") or 0),
                "drop_pp": str(drop),
                "regressed_count": str(len(regressed)),
                "regressed_lines": lines,
                "model_note": model_note,
            },
            related_resource_type="assistant_eval_run",
            related_resource_id=run.id,
            # 落在能看到这次运行明细的那一页,而不是一个还要再点两下的总览。
            url=f"/admin/assistant-eval?run={run.id}",
        )
        for admin in admins
    ]
    await fanout.enqueue_then_drain(
        db, envelopes, trace_id="", label="assistant.eval_regression"
    )
    return {
        "sent": True,
        "drop_pp": drop,
        "regressed": len(regressed),
        "admins": len(admins),
    }