"""Budget alerting for the ``apikey.budget_check`` cron.

Each :class:`ApiKeyAssignment` carries a ``monthly_budget_usd`` cap
declared on the original request (see ``ApiKeyAssignmentCreate`` —
``Field(..., ge=20)``, in USD). PR #251 (2026-05-12) replaced the
prior per-applicant ``cost_owner_person_id`` sum (which conflated
unrelated scope-owner-fallback attribution into a single key's bar)
with a per-AK dispatch via :func:`compute_assignment_spend` in
``apikey/spend.py`` — Foundry by RG, Bedrock by IAM-user principal-
tag, Vertex by GCP project scope, Bailian honest "data unavailable"
(Aliyun BSS limitation). The same helper backs the per-Key MTD
panel; both surfaces agree to the cent by construction.

Why a separate cron from ``cmdb.billing_sync``: that one alerts on
**cloud-account** budget breaches (a tenant-level finance signal).
This one alerts on **per-applicant API-key** breaches, which is a
chargeback signal — the applicant needs to know their key is about
to blow through its self-declared cap, and their manager needs to
know in case they're rubber-stamping budgets without follow-through.

Thresholds: 80% warning + 100% over-budget. Each is monotonic —
once it fires for a given (assignment, month), we land a dedup row
and stay quiet for the rest of the month. A 100% breach SUPPRESSES
the matching 80% (we land both rows so a refund-driven dip back
under 100% can't re-fire 80%). See :class:`ApiKeyBudgetAlertsDedup`
for the dedup contract.

Dispatch: in-app + Feishu card via :class:`NotificationService` to
**both** the applicant AND their direct manager (when the manager
exists and is active). The notification template is bilingual via
the ``apikey.budget_alert`` template registered in
``app/notifications/service.py``.

Audit: every actual send writes an ``apikey.budget_alert_sent`` row
on ``audit_logs`` with the assignment id, threshold, MTD spend, and
budget. Best-effort — an audit-write failure must not block the
alert path, mirroring cmdb billing_alerts.
"""
from __future__ import annotations

import logging
import uuid
from datetime import UTC, date, datetime
from decimal import Decimal
from typing import Any

from sqlalchemy import delete, select
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession

from app.audit.service import AuditService
from app.domains.apikey.models import ApiKeyAssignment, ApiKeyBudgetAlertsDedup
from app.identity.person_models import IdentityPerson
from app.notifications.service import NotificationEnvelope, NotificationService
from app.notifications.text_safety import safe_envelope_text

logger = logging.getLogger(__name__)


# Threshold ladder. Order matters: highest first so a 100% breach
# is reported as the over-budget signal even when the lower 80% row
# also lands. Storing both rows lets a future refund-driven dip
# back under 100% NOT re-fire 80% (the 80% row is already there).
_THRESHOLDS_PCT: tuple[int, ...] = (100, 80)


def _month_key(today: date) -> str:
    """YYYY-MM string used as the dedup key.

    Anchored to the cron-row's tz (Asia/Shanghai) externally; the
    handler passes ``today`` already converted, so this stays
    timezone-naive at the formatting layer.
    """
    return today.strftime("%Y-%m")


async def _compute_assignment_mtd_spend(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    month_start: date,
    month_end: date,
) -> tuple[Decimal, bool, str | None]:
    """Return ``(mtd_usd, data_available, data_caveat)`` for the
    assignment's product-attributable spend in the month window.

    Routes through :func:`compute_assignment_spend` so the cron's
    threshold compare uses the SAME per-AK number the panel
    (``/apikeys/assignments/{id}/spend``) shows — no drift between
    "alert fired at 100%" and "panel reads 92.3%". The legacy
    per-applicant ``cost_owner_person_id`` sum sat here pre-PR-251
    and produced the false-positive NexusCRM over-budget alerts.

    When the bucket cannot resolve a number (Bailian BSS limitation,
    Foundry without RG binding, Bedrock without IAM binding), the
    function returns ``Decimal('0'), False, <caveat>``. The cron is
    expected to interpret ``data_available=False`` as "skip threshold
    compare for this assignment this period" — see
    :func:`run_budget_check` for the dispatch.
    """
    from app.domains.apikey.spend import compute_assignment_spend

    spend = await compute_assignment_spend(
        session,
        assignment=assignment,
        month_start=month_start,
        month_end=month_end,
    )
    return spend["mtd_usd"], spend["data_available"], spend["data_caveat"]


async def _try_claim_dedup_slot(
    session: AsyncSession,
    *,
    assignment_id: uuid.UUID,
    month_key: str,
    threshold_pct: int,
    mtd_usd: Decimal,
    budget_usd: Decimal,
) -> bool:
    """Atomically claim the (assignment, month, threshold) dedup slot.

    Returns ``True`` when the INSERT succeeded — this run owns the slot
    and should dispatch the notification.  Returns ``False`` when an
    ``IntegrityError`` means a previous run already fired OR a concurrent
    run beat us to the INSERT — dispatch must be skipped in both cases.

    Inverting the old SELECT-then-INSERT pattern eliminates the TOCTOU
    window where two concurrent budget-check runs could both pass the
    SELECT, both dispatch, and only then race for the INSERT.  With
    claim-first the check and the insert are a single atomic step.

    Uses a SAVEPOINT (``begin_nested``) so the IntegrityError rolls back
    only this slot's write and leaves the caller's outer transaction
    unaffected — the same session-safety rationale as the old
    ``_record_dedup_send``.
    """
    now = datetime.now(UTC)
    try:
        async with session.begin_nested():
            session.add(ApiKeyBudgetAlertsDedup(
                id=uuid.uuid4(),
                assignment_id=assignment_id,
                billing_month=month_key,
                threshold_pct=threshold_pct,
                last_alerted_at=now,
                last_alerted_amount_usd=mtd_usd,
                last_alerted_budget_usd=budget_usd,
            ))
    except IntegrityError:
        return False
    return True


async def _release_dedup_slot(
    session: AsyncSession,
    *,
    assignment_id: uuid.UUID,
    month_key: str,
    threshold_pct: int,
) -> None:
    """Release a previously-claimed dedup slot so the next cron tick
    can retry the alert.

    Called from the dispatch path when EVERY recipient failed (no
    outbox row was ever written), since the standard outbox retry
    can't recover what was never persisted. Pre-launch hardening
    2026-05-16 (LS3 P0): without this release, a recipient-resolver
    bug / DB flush failure on the first dispatch attempt locked the
    month-key dedup slot, silencing every subsequent retry until the
    next billing month — applicant + manager never got the alert.

    Wrapped in a SAVEPOINT so a DELETE failure (FK race) cannot
    corrupt the outer transaction.
    """
    try:
        async with session.begin_nested():
            await session.execute(
                delete(ApiKeyBudgetAlertsDedup).where(
                    ApiKeyBudgetAlertsDedup.assignment_id == assignment_id,
                    ApiKeyBudgetAlertsDedup.billing_month == month_key,
                    ApiKeyBudgetAlertsDedup.threshold_pct == threshold_pct,
                )
            )
    except Exception:  # noqa: BLE001
        logger.exception(
            "apikey.budget_check: failed to release dedup slot for "
            "assignment=%s month=%s threshold=%d — slot may stay locked",
            assignment_id, month_key, threshold_pct,
        )


async def _record_alert_audit(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    threshold_pct: int,
    month_key: str,
    mtd_usd: Decimal,
    budget_usd: Decimal,
) -> None:
    """Append an ``apikey.budget_alert_sent`` audit row.

    Best-effort — an audit-write failure must not block the alert path.
    Uses a savepoint so the failure cannot corrupt the caller's outer
    transaction or invalidate the already-written dedup row.
    """
    try:
        async with session.begin_nested():
            audit = AuditService(session)
            await audit.log(
                trace_id=(
                    f"apikey.budget_alert.{assignment.id}."
                    f"{month_key}.{threshold_pct}"
                ),
                actor_type="system",
                action="apikey.budget_alert_sent",
                status="success",
                resource_type="api_key_assignment",
                resource_id=str(assignment.id),
                detail={
                    "request_number": assignment.request_number,
                    "threshold_pct": threshold_pct,
                    "month": month_key,
                    "mtd_usd": str(mtd_usd),
                    "budget_usd": str(budget_usd),
                },
            )
    except Exception:  # noqa: BLE001
        logger.exception(
            "apikey.budget_alert_sent audit write failed (non-fatal)",
        )


def _build_envelope(
    *,
    recipient: IdentityPerson,
    applicant: IdentityPerson,
    assignment: ApiKeyAssignment,
    provider_name: str,
    threshold_pct: int,
    mtd_usd: Decimal,
    budget_usd: Decimal,
    pct_used: float,
) -> NotificationEnvelope:
    """Build the bilingual notification body for one recipient.

    The recipient may be the applicant themselves OR their manager —
    the body is the same shape; the title prefixes "[Helix] · API Key
    预算告警" and the threshold percentage. The Feishu card carries
    a button to /me/api-keys (applicant) or /admin/assets/api-keys
    (manager) so the recipient can act in one click.
    """
    is_self = recipient.id == applicant.id
    severity_tag = "🚨" if threshold_pct >= 100 else "⚠️"
    severity_label_zh = "已超出" if threshold_pct >= 100 else "即将超出"
    severity_label_en = "exceeded" if threshold_pct >= 100 else "approaching"

    # Sanitise EVERY user-controlled value before splicing — defends
    # against CR/LF injection (fake "— Helix IT 平台" signature
    # spoofing) and ``[r]…[/r]`` markup injection (attacker-styled
    # "system-emphasis" red+bold payloads). Same posture as the
    # ``_safe_text`` helper in azure_openai_autogrant.py — promoted
    # to the shared ``safe_envelope_text`` so every notification call
    # site applies identical sanitisation. Caught in security review
    # pass 2 of this PR.
    safe_display_name = safe_envelope_text(applicant.display_name)
    safe_work_email = safe_envelope_text(applicant.work_email, limit=120)
    applicant_label = (
        safe_display_name or safe_work_email or str(applicant.id)
    )
    safe_project_name = safe_envelope_text(assignment.project_name)
    # ``request_number`` is platform-controlled (server-minted via
    # ``AK-YYYYMMDD-XXXX`` regex) — not user-controlled — so it skips
    # sanitisation, but the conservative move is to apply it anyway
    # in case a future request-number formatter accepts user input.
    safe_request_number = safe_envelope_text(
        assignment.request_number, limit=64,
    )
    safe_provider_name = safe_envelope_text(provider_name, limit=128)

    title = (
        f"[Helix] {severity_tag} API Key 月度预算{severity_label_zh} "
        f"({threshold_pct}%) / API key budget {severity_label_en} ({threshold_pct}%)"
    )
    portal_url = (
        "/me/api-keys" if is_self else "/admin/assets/api-keys"
    )
    body_text = (
        f"{severity_tag} API Key 月度预算告警\n"
        f"申请人: {applicant_label}\n"
        f"申请号: {safe_request_number}\n"
        f"提供方: {safe_provider_name}\n"
        f"项目: {safe_project_name}\n"
        f"月度预算: ${budget_usd:,.2f} USD\n"
        f"本月已花费: ${mtd_usd:,.2f} USD ({pct_used:.1f}%)\n"
        f"阈值: {threshold_pct}%\n\n"
        f"请到 Helix {portal_url} 查看并决定是否调整预算或暂停使用。\n"
        f"如对帐单有任何疑问，请联系 IT 团队进行确认。\n\n"
        f"— Helix IT 平台\n"
        f"---\n"
        f"{severity_tag} API key monthly-budget alert\n"
        f"Applicant: {applicant_label}\n"
        f"Request: {safe_request_number}\n"
        f"Provider: {safe_provider_name}\n"
        f"Project: {safe_project_name}\n"
        f"Monthly budget: ${budget_usd:,.2f} USD\n"
        f"Spent so far this month: ${mtd_usd:,.2f} USD ({pct_used:.1f}%)\n"
        f"Threshold: {threshold_pct}%\n\n"
        f"Open Helix {portal_url} to review and decide whether to "
        f"adjust the budget or pause usage.\n"
        f"If you have any questions about the bill, please contact "
        f"the IT team for confirmation.\n\n"
        f"— Helix IT Platform"
    )
    from app.notifications.service import CardSection, CardStat, LocalizedText

    card_stats = [
        CardStat(
            label=LocalizedText(
                zh="本月已用",
                en="Spent this month",
                ja="今月の利用額",
            ),
            value=f"${mtd_usd:,.2f}",
            trend=f"{pct_used:.1f}%",
            trend_is_good=False,
            hint=LocalizedText(
                zh=f"阈值 {threshold_pct}% · 月度预算 ${budget_usd:,.2f}",
                en=(
                    f"Threshold {threshold_pct}% · "
                    f"monthly budget ${budget_usd:,.2f}"
                ),
                ja=(
                    f"しきい値 {threshold_pct}% · "
                    f"月次予算 ${budget_usd:,.2f}"
                ),
            ),
        ),
        CardStat(
            label=LocalizedText(
                zh="月度预算", en="Monthly budget", ja="月次予算",
            ),
            value=f"${budget_usd:,.2f}",
        ),
        CardStat(
            label=LocalizedText(
                zh="提供方", en="Provider", ja="プロバイダー",
            ),
            value=safe_provider_name or "—",
        ),
        CardStat(
            label=LocalizedText(zh="项目", en="Project", ja="プロジェクト"),
            value=safe_project_name or "—",
            hint=LocalizedText(
                zh=f"申请号 {safe_request_number}",
                en=f"Request no. {safe_request_number}",
                ja=f"申請番号 {safe_request_number}",
            ),
        ),
    ]
    over_budget = threshold_pct >= 100
    if is_self:
        section_body = LocalizedText(
            zh=(
                "请尽快确认是否调整预算 / 暂停使用，避免本月继续超支。\n"
                "如已和审批方对齐，请到管理后台调高预算后继续使用。\n\n"
                "如对帐单有任何疑问，请联系 IT 团队进行确认。"
            ),
            en=(
                "Please decide soon whether to raise the budget or pause "
                "usage, so this month does not overrun further.\n"
                "Once you have aligned with the approver, raise the budget "
                "in the admin console and carry on.\n\n"
                "If you have any questions about the bill, please contact "
                "the IT team for confirmation."
            ),
            ja=(
                "今月これ以上の超過を防ぐため、予算の引き上げまたは利用の"
                "一時停止をお早めにご判断ください。\n"
                "承認者との合意が取れている場合は、管理コンソールで予算を"
                "引き上げてからご利用を継続してください。\n\n"
                "請求内容についてご不明な点があれば、IT チームまでご確認"
                "ください。"
            ),
        )
    else:
        section_body = LocalizedText(
            zh=(
                f"申请人 [b]{applicant_label}[/b] 的 API Key 本月预算"
                f"{severity_label_zh}阈值。请确认是否同意继续使用。\n\n"
                f"如对帐单有任何疑问，请联系 IT 团队进行确认。"
            ),
            en=(
                f"The API key of applicant [b]{applicant_label}[/b] has "
                f"{'exceeded' if over_budget else 'nearly reached'} its "
                f"monthly budget threshold. Please confirm whether usage "
                f"may continue.\n\n"
                f"If you have any questions about the bill, please contact "
                f"the IT team for confirmation."
            ),
            ja=(
                f"申請者 [b]{applicant_label}[/b] の API キーが今月の予算"
                f"しきい値を"
                f"{'超過しました' if over_budget else '超過しそうです'}。"
                f"利用を継続してよいかご確認ください。\n\n"
                f"請求内容についてご不明な点があれば、IT チームまでご確認"
                f"ください。"
            ),
        )
    card_sections = [
        CardSection(
            title=(
                LocalizedText(
                    zh="已超出预算",
                    en="Budget exceeded",
                    ja="予算超過",
                )
                if over_budget
                else LocalizedText(
                    zh="即将超出预算",
                    en="Approaching budget limit",
                    ja="予算超過の見込み",
                )
            ),
            body=section_body,
            tone=("danger" if threshold_pct >= 100 else "warn"),
        ),
    ]

    return NotificationEnvelope(
        recipient=recipient,
        title=title,
        body_text=body_text,
        template_key="apikey.budget_alert",
        context={
            # Both the body-template ``str.format`` substitution AND
            # the subject substitution read these context keys; pass
            # only sanitised values so the template renderer cannot
            # reach an injected payload by either path.
            "applicant_name": applicant_label,
            "applicant_email": safe_work_email,
            "request_number": safe_request_number,
            "provider_name": safe_provider_name,
            "project_name": safe_project_name,
            "monthly_budget_usd": f"{budget_usd:,.2f}",
            "mtd_usd": f"{mtd_usd:,.2f}",
            "threshold_pct": str(threshold_pct),
            "pct_used": f"{pct_used:.1f}",
            "is_self": "true" if is_self else "false",
            "portal_url": portal_url,
        },
        url=portal_url,
        card_subtitle=LocalizedText(
            zh=f"{safe_provider_name} · 申请号 {safe_request_number}",
            en=f"{safe_provider_name} · Request {safe_request_number}",
            ja=f"{safe_provider_name} · 申請番号 {safe_request_number}",
        ),
        card_stats=card_stats,
        card_sections=card_sections,
        related_resource_type="api_key_assignment",
        related_resource_id=assignment.id,
    )


async def evaluate_assignment(
    session: AsyncSession,
    *,
    assignment: ApiKeyAssignment,
    today: date,
) -> dict[str, Any]:
    """Check + fire (per threshold) for one assignment.

    Returns a per-assignment summary dict:
    ``{assignment_id, request_number, mtd_usd, budget_usd, pct_used,
    fired: [{threshold_pct, recipients_count}], skipped: [...]}``.
    Caller (the cron handler) aggregates these into the run-level
    summary persisted on ``scheduled_jobs.last_result_summary``.

    Skips silently when:
    - ``monthly_budget_usd <= 0`` (budget disabled).
    - ``assigned_to`` person row missing or marked left.
    - Threshold already fired this month (dedup hit).
    - MTD spend below the threshold.
    """
    summary: dict[str, Any] = {
        "assignment_id": str(assignment.id),
        "request_number": assignment.request_number,
        "fired": [],
        "skipped": [],
    }
    budget = Decimal(str(assignment.monthly_budget_usd or 0))
    if budget <= 0:
        summary["skipped"].append({"reason": "no-budget"})
        summary["mtd_usd"] = "0"
        summary["budget_usd"] = "0"
        summary["pct_used"] = 0.0
        return summary

    applicant = await session.get(IdentityPerson, assignment.assigned_to)
    if applicant is None:
        summary["skipped"].append({"reason": "applicant-not-found"})
        return summary
    if applicant.employment_status in ("left", "leaver_pending"):
        # A departing employee's API key budget alerts would be noise:
        # the offboarding sweep will revoke the IAM/RAM principal before
        # month-end anyway. Skip both "left" (already offboarded) and
        # "leaver_pending" (in-flight offboarding) to avoid spurious
        # alerts during the T+0 → T+left window.
        summary["skipped"].append({"reason": f"applicant-{applicant.employment_status}"})
        return summary

    month_start = today.replace(day=1)
    mtd, data_available, data_caveat = await _compute_assignment_mtd_spend(
        session,
        assignment=assignment,
        month_start=month_start,
        month_end=today,
    )
    summary["mtd_usd"] = str(mtd)
    summary["budget_usd"] = str(budget)
    # When the bucket cannot resolve a per-AK number (Bailian BSS
    # account-level only; Foundry without RG binding; Bedrock without
    # IAM binding), threshold compare is meaningless — there is no
    # honest signal we can ladder against. Skip the dispatch and
    # record the reason so compliance can audit "why didn't budget
    # fire on AK X this month".
    if not data_available:
        import logging as _logging
        _log = _logging.getLogger(__name__)
        # Per-AK spend is unavailable for Bailian (account-level BSS only),
        # Foundry without RG binding, and Bedrock without IAM user.  We
        # cannot compare against the declared budget, so the threshold check
        # is skipped.  Emit a WARNING so the ops team can see these keys are
        # un-monitored and act (e.g. set an integration-level cny_per_usd
        # override or migrate to an AK type that supports per-AK billing).
        if budget > 0:
            _log.warning(
                "budget_alerts: AK assignment %s has monthly_budget_usd=%.2f "
                "but per-AK spend data is unavailable (caveat: %s). "
                "Budget enforcement is skipped for this key. "
                "Provider: %s, assigned_to: %s",
                assignment.id,
                float(budget),
                data_caveat,
                assignment.provider_id,
                assignment.assigned_to,
            )
        summary["skipped"].append({
            "reason": "bucket-data-not-available",
            "caveat": data_caveat,
        })
        summary["pct_used"] = 0.0
        return summary
    # Decimal pct for the threshold comparison (no float boundary
    # drift: a $80.00 / $100.00 budget could otherwise yield
    # ``79.99999999999999`` and silently skip the 80% alert). The
    # rounded float is kept for the summary display + UI bar where
    # 2-decimal precision is what users see — the bar and the cron
    # MUST NOT disagree on which side of the threshold the spend
    # landed.
    pct_decimal = (mtd / budget * Decimal(100)) if budget > 0 else Decimal(0)
    pct_used = float(pct_decimal)
    summary["pct_used"] = round(pct_used, 2)

    month_key = _month_key(today)

    # Resolve provider + manager once per assignment (saves 2 DB
    # round-trips per threshold).
    from app.domains.apikey.models import ApiKeyProvider
    provider_row = await session.get(ApiKeyProvider, assignment.provider_id)
    provider_name = (
        provider_row.name if provider_row is not None else "API Key"
    )
    manager: IdentityPerson | None = None
    if applicant.manager_person_id is not None:
        manager_row = await session.get(
            IdentityPerson, applicant.manager_person_id,
        )
        if (
            manager_row is not None
            and manager_row.employment_status != "left"
        ):
            manager = manager_row

    # Walk thresholds high-to-low. Each independent — landing a 100%
    # row does NOT remove the 80% row's monotonic guarantee for THIS
    # month (we still want both rows persisted so post-month dips
    # don't re-spam). However, we only DISPATCH the highest crossed
    # threshold per run — finance doesn't need two notifications in
    # the same minute saying "you're at 110%" AND "you're at 88%".
    # The lower threshold's dedup row is still INSERTed so it doesn't
    # fire the next day either.
    notify_dispatched = False
    for threshold_pct in _THRESHOLDS_PCT:
        # Decimal compare — see ``pct_decimal`` definition above. Mixed
        # Decimal/int compares are exact in Python's decimal module.
        crossed = pct_decimal >= Decimal(threshold_pct)
        if not crossed:
            summary["skipped"].append({
                "threshold_pct": threshold_pct,
                "reason": "below-threshold",
            })
            continue
        # Claim the dedup slot atomically BEFORE resolving recipients or
        # dispatching.  The old SELECT-then-INSERT had a TOCTOU gap:
        # two concurrent budget-check runs could both pass the SELECT,
        # both dispatch, and then race for the INSERT.  Claim-first
        # makes the check and insert a single atomic step — only the
        # run that wins the INSERT dispatches.
        claimed = await _try_claim_dedup_slot(
            session,
            assignment_id=assignment.id,
            month_key=month_key,
            threshold_pct=threshold_pct,
            mtd_usd=mtd,
            budget_usd=budget,
        )
        if not claimed:
            summary["skipped"].append({
                "threshold_pct": threshold_pct,
                "reason": "already-fired-this-month",
            })
            continue

        # Recipient resolution — tri-state on ``notify_recipients``:
        #   * NULL → default (applicant + manager). Vast majority of
        #     rows take this path; preserves PR #240 baseline.
        #   * [] → admin-set explicit "suppress alerts". Dedup row +
        #     audit still record the threshold crossing (so the next
        #     tick doesn't re-evaluate), but no notification dispatch.
        #   * [<uuid>, ...] → exactly these persons. Applicant is NOT
        #     auto-included; admin explicitly chose every recipient.
        # Persons whose ``employment_status='left'`` are filtered out
        # AFTER override resolution.
        #
        # Round-31 audit B6 fix (2026-05-18): when an explicit override
        # resolves to ZERO recipients after the leaver filter (admin
        # listed N persons, all of whom have since offboarded), fall
        # back to the default applicant+manager path AND audit the
        # event. The prior behaviour silently dropped the alert
        # forever — the override list was sticky in the DB, every
        # leaver dropout made it more invalid, but the cron just
        # logged ``recipients=[]`` and moved on. Operators had no
        # signal until finance noticed the missing budget alerts.
        recipients: list[IdentityPerson] = []
        override = assignment.notify_recipients
        override_collapsed_to_leavers = False
        if override is None:
            recipients.append(applicant)
            if manager is not None:
                recipients.append(manager)
        elif len(override) == 0:
            # Explicit suppress — leave recipients empty. The dispatch
            # loop below becomes a no-op; dedup row still written.
            pass
        else:
            override_ids = [uuid.UUID(str(x)) for x in override]
            override_rows = await session.execute(
                select(IdentityPerson).where(
                    IdentityPerson.id.in_(override_ids),
                    IdentityPerson.employment_status != "left",
                )
            )
            recipients = list(override_rows.scalars().all())
            if not recipients:
                # Every override entry resolved to a leaver — fall back
                # to the applicant + manager default so SOMEONE gets the
                # alert. Mark the path so the audit row captures "we
                # used the fallback for AK X" — admin can then re-curate
                # the override list.
                override_collapsed_to_leavers = True
                recipients.append(applicant)
                if manager is not None:
                    recipients.append(manager)
                logger.warning(
                    "apikey.budget_check: notify_recipients override "
                    "for assignment %s resolved to 0 live persons "
                    "(every entry has employment_status='left'); "
                    "falling back to applicant+manager default. "
                    "Operator action: re-curate the override list.",
                    assignment.request_number,
                )

        # Only the FIRST crossed-and-not-yet-fired threshold (highest,
        # i.e. 100% before 80%) actually dispatches a notification.
        # Lower-threshold rows still INSERT so the next cron tick
        # treats them as "already fired" and skips.
        #
        # Pre-launch hardening 2026-05-16 (LS3 P0): per-recipient try
        # so one bad open_id / DB flush doesn't take down the whole
        # batch. If EVERY dispatch failed and no outbox row was
        # written, release the dedup slot so the next cron retries
        # — otherwise the month goes silent.
        #
        # ``this_threshold_notified`` tracks dispatch outcome for the
        # CURRENT iteration ONLY — the cross-iteration ``notify_dispatched``
        # below stays for the "we already shouted at 100%, skip 80%"
        # decision, but the per-threshold audit / summary fields must
        # reflect this iteration's actual outcome. Without this split
        # the 80% row records ``notified=True`` whenever a prior 100%
        # iteration dispatched, even though 80% itself was intentionally
        # suppressed — a forensic lie.
        this_threshold_notified = False
        if not notify_dispatched and recipients:
            notify_svc = NotificationService(session)
            dispatched_any = False
            for recipient in recipients:
                envelope = _build_envelope(
                    recipient=recipient,
                    applicant=applicant,
                    assignment=assignment,
                    provider_name=provider_name,
                    threshold_pct=threshold_pct,
                    mtd_usd=mtd,
                    budget_usd=budget,
                    pct_used=pct_used,
                )
                try:
                    await notify_svc.dispatch(envelope)
                    dispatched_any = True
                except Exception:  # noqa: BLE001
                    logger.exception(
                        "apikey.budget_check: dispatch failed for "
                        "assignment %s recipient %s (threshold %d%%) — "
                        "continuing with remaining recipients",
                        assignment.request_number, recipient.id,
                        threshold_pct,
                    )
            if dispatched_any:
                notify_dispatched = True
                this_threshold_notified = True
            else:
                # Every recipient errored before outbox write — release
                # the dedup slot so the cron can re-attempt; otherwise
                # the month-key block keeps the alert silent.
                logger.error(
                    "apikey.budget_check: ALL %d recipients failed for "
                    "assignment %s threshold %d%% — releasing dedup slot",
                    len(recipients), assignment.request_number,
                    threshold_pct,
                )
                await _release_dedup_slot(
                    session,
                    assignment_id=assignment.id,
                    month_key=month_key,
                    threshold_pct=threshold_pct,
                )
                # Round-3 review follow-up (LS3 P2-A): write a durable
                # audit row BEFORE break so the silent-fail trail
                # survives ``scheduled_job_runs`` row trimming. Without
                # this, the only forensic trace of "we tried to alert
                # and every recipient was unreachable" lives in cron
                # summary which ages out, and operators never know
                # the month went silent.
                try:
                    await _record_alert_audit(
                        session,
                        assignment=assignment,
                        threshold_pct=threshold_pct,
                        month_key=month_key,
                        mtd_usd=mtd,
                        budget_usd=budget,
                    )
                except Exception:  # noqa: BLE001
                    logger.exception(
                        "apikey.budget_check: audit-write failure on "
                        "all-recipients-failed path for assignment %s "
                        "threshold %d%%",
                        assignment.request_number, threshold_pct,
                    )
                summary["skipped"].append({
                    "threshold_pct": threshold_pct,
                    "reason": "all-recipients-failed-dedup-released",
                })
                # Round-1 review follow-up (Review-D P2-4): every
                # remaining threshold in this evaluation would hit the
                # exact same recipient set with the exact same failure
                # mode (open_id resolver down, channel outage). Short-
                # circuit the loop so we don't burn N more dispatch
                # attempts per threshold per assignment. The dedup
                # slot was already released, so next cron tick gets a
                # fresh chance — no permanent state lost.
                break

        # Dedup row was already written atomically by _try_claim_dedup_slot
        # above (before dispatch).  Record the audit row — including on the
        # SUPPRESSED lower-threshold pass where notify_dispatched is already
        # True from the 100% iteration. ``status="success"`` reflects "the
        # cron made the right decision"; the per-threshold ``notified`` flag
        # captures the actual dispatch outcome.
        await _record_alert_audit(
            session,
            assignment=assignment,
            threshold_pct=threshold_pct,
            month_key=month_key,
            mtd_usd=mtd,
            budget_usd=budget,
        )
        summary["fired"].append({
            "threshold_pct": threshold_pct,
            "recipients_count": len(recipients),
            # ``this_threshold_notified`` is per-iteration: True iff THIS
            # threshold's dispatch loop produced at least one delivered
            # outbox row. Lower thresholds that are intentionally
            # suppressed (because a higher threshold already fired this
            # tick) correctly report ``notified=False``. Suppressed-by-
            # admin assignments (recipients=[]) also report False. The
            # prior ``notify_dispatched`` use here was a forensic lie —
            # see BLOCKER B2 in the 2026-05-18 API Key audit.
            "notified": this_threshold_notified,
            # B6 marker: surfaces "we used the applicant+manager
            # fallback because every override entry is a leaver". Lets
            # operators grep ``summary["fired"]`` for stale override
            # lists across the fleet instead of hunting per-AK.
            "override_collapsed_to_leavers": override_collapsed_to_leavers,
        })

    return summary


__all__ = ["evaluate_assignment"]