"""``auditor.self_attestation_advice`` — AI advice for the access self-check (PR5).

Generates one advice row per holder for the currently-open
self-attestation campaign, in batches. Batched because a campaign
snapshots every active employee (~500 people) and each holder is one
model call — doing that inside the request that opens the campaign would
time out, and doing it lazily on page load would put a 40-second wait in
front of the person we are asking for a favour.

The job is a no-op while
``access_review.self_attestation.ai_advice_enabled`` is off, which is
the default. Failures are per-holder and recorded on the row
(``failure_code``); ONE holder failing does not fail the run, because a
holder without a narrative still has a perfectly usable review page.

**But a run in which nobody got advice AND the model channel was the
reason does fail.** "Tolerate a partial failure" and "stay silent while
the feature is entirely down" are different claims, and the first was
quietly doing duty for the second: with the channel unreachable, every
holder trips ``no_client`` / ``call_failed``, the summary reads
``annotated=0, failed=500``, and the scheduled-jobs ledger records a
SUCCESS — so nothing pages anyone and the next campaign is asked to
trust advice that was never generated. ``FAIL_NO_ITEMS`` deliberately
does NOT count toward this: a holder with nothing to review is a fact
about that holder, not about the channel.
"""
from __future__ import annotations

import logging
from typing import Any

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

from app.core.txn import best_effort
from app.scheduled_jobs.registry import register_scheduled_job

logger = logging.getLogger(__name__)

SELF_ATTESTATION_ADVICE_HANDLER = "auditor.self_attestation_advice"

#: Holders per run. At ~500 employees a campaign takes a few nights to
#: fill in, which is fine: the campaign window is weeks, and the
#: deterministic priority ordering works from day one without any advice.
_DEFAULT_BATCH = 60

#: Failure codes that mean "the model channel is down", as opposed to
#: "this holder had nothing to say". Only these arm the total-failure
#: raise at the end of the run — ``FAIL_NO_ITEMS`` and ``FAIL_UNGROUNDED``
#: do not. ``ungrounded`` is excluded even at 100%: that is the grounding
#: gate DOING ITS JOB, and a gate that pages the on-call every time it
#: fires is a gate somebody eventually turns off.
_CHANNEL_FAILURE_CODES = frozenset({"no_client", "call_failed"})


@register_scheduled_job(
    key=SELF_ATTESTATION_ADVICE_HANDLER,
    display_name="访问自查 · AI 建议生成",
    description=(
        "每天 01:30 UTC。给当前开着的半年度访问自查 campaign 分批生成 AI 建议"
        "(每人一行):一段摘要 + 每个访问类别一个 keep/release 初步建议。事实来自 "
        "campaign 自己的条目 + PR4 的使用信号;模型不能引用事实里没有的数字、不能"
        "发明类别,也不能把「无法确认使用情况」说成「闲置」(代码级校验)。建议落在"
        "持有人决定**旁边**,永不落进决定里。开关 "
        "access_review.self_attestation.ai_advice_enabled 默认关,关着时本任务空转。"
    ),
)
async def run_self_attestation_advice(
    db: AsyncSession, config: dict[str, Any],
) -> dict[str, Any]:
    from app.domains.auditor import access_certification_service as acs
    from app.domains.auditor.models import (  # noqa: PLC0415
        AccessCertificationItem,
        AuditorSelfAttestationAdvice,
    )
    from app.domains.auditor.self_attestation_ai import (  # noqa: PLC0415
        advice_enabled,
        annotate_holder,
    )

    summary: dict[str, Any] = {
        "enabled": False, "campaign_id": None,
        "holders_annotated": 0, "holders_failed": 0, "remaining": 0,
        "channel_failures": 0,
    }
    if not await advice_enabled(db):
        return summary
    summary["enabled"] = True

    cert = await acs.latest_open_self_attestation(db)
    if cert is None:
        return summary
    summary["campaign_id"] = str(cert.id)

    try:
        batch = int(config.get("batch") or _DEFAULT_BATCH)
    except (TypeError, ValueError):
        batch = _DEFAULT_BATCH

    done = set((await db.execute(
        select(AuditorSelfAttestationAdvice.person_id).where(
            AuditorSelfAttestationAdvice.certification_id == cert.id,
            # Only rows that actually produced a narrative count as done.
            # A row that failed is retried on the next run — a transient
            # vendor error must not permanently deny one person their
            # advice while everyone else gets it.
            AuditorSelfAttestationAdvice.failure_code.is_(None),
        )
    )).scalars().all())

    rows = (await db.execute(
        select(AccessCertificationItem).where(
            AccessCertificationItem.certification_id == cert.id,
        )
    )).scalars().all()

    by_person: dict[Any, list[Any]] = {}
    display: dict[Any, str] = {}
    for row in rows:
        if row.person_id is None or row.person_id in done:
            continue
        by_person.setdefault(row.person_id, []).append(row)
        display.setdefault(
            row.person_id,
            row.holder_display_name or row.holder_email or str(row.person_id),
        )

    targets = sorted(by_person, key=lambda pid: -len(by_person[pid]))
    summary["remaining"] = max(len(targets) - batch, 0)

    for person_id in targets[:batch]:
        # SAVEPOINT per holder. The loop shares one session, so a bare
        # try/except around a body that FLUSHES would let one holder's
        # DB error poison the session and silently fail every holder
        # after it — the ``continue`` would keep running against a dead
        # transaction. ``best_effort`` isolates the failure and lets
        # transaction-aborting errors (1205/1213) propagate, which is
        # the only correct split.
        advice = None
        async with best_effort(
            db, label=f"self-attestation advice for {person_id}", log=logger,
        ) as step:
            advice = await annotate_holder(
                db,
                certification_id=cert.id,
                person_id=person_id,
                items=by_person[person_id],
                holder_display=display[person_id],
            )
            await db.flush()
        # ``step.ok`` only — reading ``advice`` after a rollback would
        # touch an expired instance and raise MissingGreenlet.
        if not step.ok:
            summary["holders_failed"] += 1
            continue
        if advice is not None and advice.failure_code:
            summary["holders_failed"] += 1
            if advice.failure_code in _CHANNEL_FAILURE_CODES:
                summary["channel_failures"] += 1
        else:
            summary["holders_annotated"] += 1

    await db.commit()

    # Commit FIRST, then decide whether to raise. Each per-holder row is
    # real evidence of what was attempted and carries its own
    # ``failure_code``; rolling them back to signal "the run failed"
    # would delete the only record of WHY, and the next run would find
    # the same holders unannotated with nothing to explain it.
    if summary["holders_annotated"] == 0 and summary["channel_failures"] > 0:
        raise RuntimeError(
            f"self-attestation advice produced nothing for campaign "
            f"{summary['campaign_id']}: {summary['channel_failures']} of "
            f"{summary['holders_failed']} holders failed on the model "
            f"channel (no_client / call_failed). The feature is down, not "
            f"degraded — advice rows exist with their failure_code set."
        )
    return summary