"""The two crons behind the exception ledger.

``automation_exception.collect`` walks the five collectors and folds
their answers in. It runs often — every 30 minutes — because the point
of the ledger is that a failure becomes visible and attributable
without anyone going looking, and because the ingest is cheap: five
indexed reads and a handful of upserts.

``automation_exception.reconcile`` is the daily self-check. Collectors
only touch clusters they reported on, so a cluster whose source stopped
producing rows entirely (a cron that was deleted, an integration that
was removed) would keep its last computed counters forever. The
reconciler recomputes every non-closed cluster from its occurrences,
which is also the repair path if a counter ever drifts.

Both are savepoint-isolated per source, for one specific reason: an
empty candidate list from a broken collector looks exactly like "all
clear", and all-clear triggers absence-based recovery. The collector
layer marks its own failures ``partial`` and the engine refuses to
close anything on a partial pass; the savepoint here is the second
line, keeping one source's exception from costing the other four their
writes or poisoning the session.
"""
from __future__ import annotations

import logging
from datetime import UTC, datetime
from typing import Any

from sqlalchemy import or_, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.txn import best_effort
from app.domains.automation_exception.collectors import COLLECTORS, run_collector
from app.domains.automation_exception.ingest import (
    CAUSE_SOURCE_MANUAL,
    CAUSE_SOURCE_NONE,
    ingest_source,
    recompute_exceptions,
)
from app.domains.automation_exception.models import AutomationException
from app.domains.automation_exception.taxonomy import CAUSE_UNKNOWN
from app.scheduled_jobs.registry import register_scheduled_job
from app.scheduled_jobs.slo import SLOTarget

logger = logging.getLogger(__name__)


@register_scheduled_job(
    key="automation_exception.collect",
    display_name="自动化异常 · 采集与归并",
    description=(
        "每 30 分钟走一遍五个来源采集器(集成连接体检 / 建号任务 / 身份同步条目 / "
        "通知出箱死信 / 定时任务失败),把「一次没走通的自动化」按 "
        "(来源, 作用域, 错误签名) 归并成异常对象,并把每一条失败源行登记为它的一次"
        "发生。**幂等**:发生记录以 (来源, 源行类型, 源行 id) 唯一,重复跑不会重复计数;"
        "父行的计数一律从子行**重算**而非累加。"
        "\n\n"
        "恢复由采集器各自负责表态:建号任务与集成体检可以用「这次没报」证明已恢复"
        "(它们扫的是完整的当前失败集);定时任务运行与身份同步条目是历史事件,"
        "「没报」什么都不能证明,只能由采集器显式给出「之后成功过」的证据。"
        "采集器自己失败时会把结果标为 partial,此时引擎**不做任何**基于缺席的关闭 —— "
        "否则一次读取异常就会把这个来源下所有未解决的异常静默判为已修复。"
        "\n\n"
        "本任务只登记与归并,不发通知、不改任何源域的数据。"
    ),
    slo=SLOTarget(tier="P1"),
)
async def run_automation_exception_collect(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    now = datetime.now(UTC)
    only = config.get("only_sources")
    wanted = {str(s) for s in only} if isinstance(only, list) and only else None

    per_source: list[dict[str, Any]] = []
    failed_sources: list[dict[str, str]] = []

    for source, _fn in COLLECTORS:
        if wanted is not None and source not in wanted:
            continue
        summary: dict[str, Any] | None = None
        async with best_effort(
            db, label=f"ingest automation exceptions from {source}", log=logger,
        ) as step:
            result = await run_collector(db, source=source, now=now)
            stats = await ingest_source(db, source=source, result=result, now=now)
            # Materialise the summary INSIDE the savepoint: on rollback
            # the ORM rows this touched are expired, and the dataclass is
            # the only thing safe to read afterwards.
            summary = stats.as_dict()
        if not step.ok:
            failed_sources.append(
                {"source": source, "error": step.reason() or "unknown"},
            )
            continue
        if summary is not None:
            per_source.append(summary)

    totals = {
        "candidates": sum(int(s["candidates"]) for s in per_source),
        "exceptions_created": sum(int(s["exceptions_created"]) for s in per_source),
        "exceptions_reopened": sum(int(s["exceptions_reopened"]) for s in per_source),
        "exceptions_resolved": sum(int(s["exceptions_resolved"]) for s in per_source),
        "occurrences_created": sum(int(s["occurrences_created"]) for s in per_source),
        "occurrences_recovered": sum(
            int(s["occurrences_recovered"]) for s in per_source
        ),
    }
    # A source whose collector could not read its own tables is the one
    # number that must not be buried: while it is non-zero the ledger is
    # not a complete picture of what is broken.
    if failed_sources:
        logger.warning(
            "automation_exception.collect: %d source(s) failed: %s",
            len(failed_sources),
            ", ".join(f["source"] for f in failed_sources),
        )
    return {
        **totals,
        "sources_ingested": len(per_source),
        "sources_failed": len(failed_sources),
        "source_failure_detail": failed_sources,
        "partial_sources": [
            s["source"] for s in per_source if s.get("partial")
        ],
        "by_source": per_source,
    }


@register_scheduled_job(
    key="automation_exception.reconcile",
    display_name="自动化异常 · 每日对账",
    description=(
        "每日 04:10 UTC 把**每一条**未关闭的异常从它的发生记录重算一遍:"
        "总次数 / 未恢复次数 / 受影响人数 / 严重度 / 状态。"
        "采集只会碰到本次报到的簇,所以一个源头彻底停止产出的簇"
        "(任务被删掉、集成被移除)会一直停留在最后一次算出的计数上;"
        "本任务是那种情况的收口,也是计数万一漂移时的修复路径。"
        "只读发生记录、只写父行的派生列,不触碰任何源域。"
    ),
    slo=SLOTarget(tier="P2"),
)
async def run_automation_exception_reconcile(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    now = datetime.now(UTC)
    resolved = await recompute_exceptions(db, exception_ids=None, now=now)
    return {"resolved_by_reconcile": resolved}


@register_scheduled_job(
    key="automation_exception.triage",
    display_name="自动化异常 · AI 分诊",
    description=(
        "每小时给「规则说不出根因」的异常补一个根因与一句解释。"
        "**按签名调用,不按发生次数** —— 一个出现 4000 次的签名是一次模型调用,"
        "答案缓存在异常行上,签名文字变了才会重问(签名变了本来就是另一个问题)。"
        "\n\n"
        "四种情况**不问模型**:人已经判过(人的判断说了算)、规则已经定了因"
        "(确定的胜过概率的,而且规则是照着真实厂商文本写的)、这个签名问过了"
        "(答案是缓存的)、行已经解决(没人在等解释)。所以规则表越长,这条任务"
        "花的钱越少 —— 方向是对的。"
        "\n\n"
        "模型只能从 `ROOT_CAUSES` **闭集**里选,解释里出现的每个数字都必须在事实包里,"
        "否则整条回复作废(一个靠编数字得出的根因不是「有点小错」,是不能信)。"
        "**分诊不关闭任何东西、不通知任何人、不覆盖人工判定**;"
        "每次调用都会记进 LLM 调用台账,可按 `automation_exception.triage` 归集。"
    ),
    slo=SLOTarget(tier="P2"),
)
async def run_automation_exception_triage(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.automation_exception.triage import (  # noqa: PLC0415
        needs_triage,
        triage_exception,
    )

    # Bounded per run, and the bound is the point: an unbounded batch on
    # the first tick after a bad deploy would fire one model call per
    # new signature with nothing to stop it. Untriaged rows are picked
    # up by the next hourly run; nothing is lost by going slowly.
    limit = config.get("max_per_run")
    batch = int(limit) if isinstance(limit, int | str) and str(limit).isdigit() else 25
    batch = max(1, min(batch, 200))

    # 🔴 The skip conditions live in the WHERE clause, not only in
    # ``needs_triage``. Filtering in Python while paginating in SQL
    # starves the queue: a row that ``needs_triage`` always rejects is
    # never stamped, so it stays in the result set forever — and
    # because the ordering is "worst first", a few dozen
    # already-classified high-volume rows would fill every page and the
    # rows that actually need a cause would never be reached. Triage
    # would run every hour and silently do nothing.
    #
    # ``needs_triage`` is still called below: it is the single
    # definition, and this WHERE is an optimisation that must agree
    # with it. A test asserts they agree.
    rows = (
        await db.execute(
            select(AutomationException)
            .where(
                AutomationException.resolved_at.is_(None),
                AutomationException.advice_generated_at.is_(None),
                AutomationException.cause_source != CAUSE_SOURCE_MANUAL,
                or_(
                    AutomationException.cause_source.in_(
                        [CAUSE_SOURCE_NONE, ""],
                    ),
                    AutomationException.root_cause == CAUSE_UNKNOWN,
                ),
            )
            # Never-attempted first, then worst first inside each group.
            #
            # 🔴 The second key alone starves the queue for the same
            # reason the WHERE clause above exists, just via a different
            # door: ``FAILURE_MODEL`` and ``FAILURE_UNGROUNDED`` record
            # the attempt on ``advice_model`` but deliberately do NOT
            # stamp ``advice_generated_at``, so a row the model keeps
            # answering ungroundedly stays in the result set forever.
            # Ordered by volume alone, a handful of such rows sit at the
            # top of every page and consume the whole hourly budget, and
            # the rows nobody has ever asked about are never reached.
            # ``advice_model IS NULL`` is exactly "we have not tried
            # this one yet", so this makes progress monotone while
            # keeping worst-first where it matters.
            .order_by(
                AutomationException.advice_model.is_(None).desc(),
                AutomationException.open_occurrence_count.desc(),
                AutomationException.last_seen_at.desc(),
            )
            .limit(batch * 2),
        )
    ).scalars().all()

    considered = 0
    triaged = 0
    failed = 0
    for row in rows:
        if triaged + failed >= batch:
            break
        if not needs_triage(row):
            continue
        considered += 1
        before = row.advice_generated_at
        await triage_exception(db, row)
        if row.advice_generated_at is not None and row.advice_generated_at != before:
            triaged += 1
        else:
            failed += 1
    await db.flush()
    return {
        "considered": considered,
        "triaged": triaged,
        "failed_or_skipped": failed,
    }


@register_scheduled_job(
    key="automation_exception.echo",
    display_name="自动化异常 · 回音与复发升级",
    description=(
        "两件事,方向相反。**回音**:一次没走通的自动化背后通常有个人在等 —— "
        "没拿到账号的新人、坐席申请死在厂商 500 里的员工。修好之后告诉他们一声,"
        "此前平台知道这件事却什么都不说,他们只能靠问 IT 才知道。"
        "**复发升级**:一个被解决之后又回来的问题,说明上次的处置没解决根因,"
        "所以告诉负责人,并把当前根因与建议的下一步一起带上。"
        "\n\n"
        "三条规矩:① 只在**真的修好**时发,且只发一次(`subject_notified_at` 让它是一次"
        "**状态转移**而不是一个状态,采集器再跑一遍也不会发第二条 —— 被告知两次的人"
        "会连第一条也不再相信);② **不新增打扰类目**,走既有的 `lifecycle` 类目,"
        "于是员工已有的静音与免打扰设置天然管住它(新增类目更省事,但等于悄悄给所有人"
        "塞一条他们从没同意过的推送);③ **不把他无法处置的失败告诉他** —— 给员工的那条"
        "只说「现在可以用了」,不说「它此前失败了 47 次」,后者是管理员的问题,"
        "讲给一个只是在等的人听,等于把一个已解决的问题变成一份新的担心。"
    ),
    slo=SLOTarget(tier="P2"),
)
async def run_automation_exception_echo(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.automation_exception.echo import (  # noqa: PLC0415
        escalate_recurrences,
        send_recovery_echoes,
    )

    echoes = await send_recovery_echoes(db, trace_id="")
    escalations = await escalate_recurrences(db, trace_id="")
    return {**echoes.as_dict(), "owners_escalated": escalations.owners_escalated}