"""Failure-notification dispatch for scheduled jobs.

Round-15 (2026-05-09): IT admin had no in-app signal when a scheduled
job (CMDB IAM sync, HRIS incremental sync, …) failed — the only
existing notification path was a Feishu webhook tied to ``cmdb.sync``
which (a) requires Feishu config, (b) is single-channel, (c) doesn't
cover non-CMDB jobs, and (d) re-fired every cron tick when the same
provider problem repeated.

This module is the single entry point the service-layer write-back
calls after a run lands ``status='failed'``. It:

  * builds a stable failure *signature* (handler_key + first line of
    error, with UUIDs / numbers / timestamps stripped) so the same
    underlying problem hashes to the same key across consecutive
    runs;
  * compares that signature against every unacknowledged failure of
    the same job inside a rolling look-back window (not just the
    immediately-preceding run — see ``_LOOKBACK_HOURS``);
  * holds a per-job quiet period between cards so a job whose error
    text flaps cannot page admins on every alternation;
  * drops the broadcast entirely for operator-triggered ("run now")
    runs — the trigger endpoint hands the outcome straight back to
    the admin who pressed the button;
  * dispatches an in-app notification to every active IT admin /
    global admin only on the FIRST occurrence of a new signature
    (or when the previous failure has been acknowledged — admin
    has signalled they're aware → next failure deserves a fresh
    push for visibility).

Channels: ``feishu`` + ``email`` + outbox (the standard
``NotificationService.dispatch`` path; the outbox row is what the
NotificationCenter bell reads). All-best-effort: if dispatch fails the
job's run row is unaffected — the handler outcome is the source of
truth, the notification is a courtesy.
"""
from __future__ import annotations

import asyncio
import hashlib
import logging
import re
from datetime import UTC, datetime, timedelta
from typing import Literal
from uuid import UUID

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

from app.auth.models import AuthRole, AuthUserRole
from app.core.config import settings
from app.core.rbac import GLOBAL_ADMIN_ROLE, MANAGEMENT_ROLE
from app.identity.person_models import IdentityPerson
from app.notifications.service import (
    LocalizedText,
    NotificationEnvelope,
    NotificationService,
)
from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun
from app.scheduled_jobs.registry import get_handler
from app.scheduled_jobs.slo import DEFAULT_SLO, SLOTier

# ``async_session_factory`` is imported lazily inside the call sites
# below — top-level binding would capture a stale reference at import
# time and the test fixture's monkeypatch on
# ``app.core.database.async_session_factory`` would have no effect for
# direct service-layer calls. Matches the late-import convention in
# ``app.worker``.

logger = logging.getLogger(__name__)

# Per-recipient dispatch ceiling. Bounds the time the post-handler
# write-back path can block the cron loop when Feishu is slow / down.
# 10s is well above any healthy round-trip (median ~150ms) so a healthy
# Feishu never trips the wait_for; a wedged Feishu trips it cleanly and
# the retry worker re-delivers the outbox row.
_DISPATCH_TIMEOUT_S = 10.0


# Strip volatile chunks of the error message before hashing so
# "9 rows vs 20 existing" and "11 rows vs 18 existing" collapse to
# the same signature. Order matters: UUIDs (which contain digits and
# hyphens) before the bare-number stripper.
_UUID_RE = re.compile(
    r"\b[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}\b",
    re.IGNORECASE,
)
_ISO_DATETIME_RE = re.compile(
    r"\b\d{4}-\d{2}-\d{2}[T ]\d{2}:\d{2}:\d{2}(?:\.\d+)?(?:[+-]\d{2}:?\d{2}|Z)?\b",
)
_HEX_BLOB_RE = re.compile(r"\b0x[0-9a-f]+\b", re.IGNORECASE)
_INT_RE = re.compile(r"\b\d+\b")


def compute_failure_signature(handler_key: str, error_message: str | None) -> str:
    """Return a stable 16-char hex signature for a failure.

    Pure function — exposed so tests can assert that two adjacent
    "X rows vs Y existing" failures collapse to the same key. Empty /
    None error messages still produce a deterministic value (keyed
    only off ``handler_key``) so the dedup still works for handlers
    that fail without text.
    """
    raw = (error_message or "").strip()
    # Keep just the first line — multi-line tracebacks are noise; the
    # top-level summary is what differs between "kind of failure".
    if "\n" in raw:
        raw = raw.split("\n", 1)[0]
    raw = _UUID_RE.sub("<uuid>", raw)
    raw = _ISO_DATETIME_RE.sub("<ts>", raw)
    raw = _HEX_BLOB_RE.sub("<hex>", raw)
    raw = _INT_RE.sub("<n>", raw)
    raw = raw.lower()
    digest = hashlib.sha256(f"{handler_key}|{raw}".encode()).hexdigest()
    return digest[:16]


async def list_admin_recipients(session: AsyncSession) -> list[IdentityPerson]:
    """Public alias of ``_list_admin_recipients``. Use from other
    scheduled-job handlers (e.g. ``asset.held_by_leaver_reminder``) so
    we don't import a leading-underscore symbol across modules. The
    private name is kept as a back-compat shim.
    """
    return await _list_admin_recipients(session)


async def _list_admin_recipients(session: AsyncSession) -> list[IdentityPerson]:
    """Active persons holding ``it_admin`` / ``global_admin``.

    Same shape as ``OffboardingEscalationService._list_it_admins`` — the
    queries diverged in passing on a couple of join styles in the past;
    keep this one minimal and well-typed.
    """
    rows = (
        await session.execute(
            select(IdentityPerson)
            .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id)
            .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
            .where(
                and_(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "active",
                    # work_email is required for both Feishu (open_id
                    # resolution) and email channels — a recipient with
                    # NULL/blank work_email lands in a 5-retry outbox
                    # storm per failure (each retry: 5min→30min→2h→8h→24h
                    # all failing with "recipient has no work_email").
                    # Filter at the source so ghost IT-admin rows never
                    # spawn the storm.
                    IdentityPerson.work_email.isnot(None),
                    IdentityPerson.work_email != "",
                    or_(
                        AuthRole.name == "it_admin",
                        AuthRole.name == GLOBAL_ADMIN_ROLE,
                    ),
                )
            )
            .distinct()
        )
    ).scalars().all()
    return list(rows)


async def _list_management_recipients(session: AsyncSession) -> list[IdentityPerson]:
    """Active persons holding the ``management`` read-only role.

    Used for P0-tier failure cc-ing — management is excluded from the
    normal admin-broadcast fan-out by design (see ROADMAP "management
    role"), but P0 SLO breaches are operations-critical and *should*
    surface in their view of the platform.
    """
    rows = (
        await session.execute(
            select(IdentityPerson)
            .join(AuthUserRole, AuthUserRole.user_id == IdentityPerson.id)
            .join(AuthRole, AuthRole.id == AuthUserRole.role_id)
            .where(
                and_(
                    IdentityPerson.deleted_at.is_(None),
                    IdentityPerson.employment_status == "active",
                    IdentityPerson.work_email.isnot(None),
                    IdentityPerson.work_email != "",
                    AuthRole.name == MANAGEMENT_ROLE,
                )
            )
            .distinct()
        )
    ).scalars().all()
    return list(rows)


# Tier → card emoji + urgency phrase. Picked at dispatch time so a
# handler whose SLO is re-tiered via code change reflects on the next
# failure without a backfill.
_TIER_EMOJI: dict[SLOTier, str] = {"P0": "🚨", "P1": "⚠️", "P2": "ℹ️"}
_TIER_URGENCY_ZH: dict[SLOTier, str] = {
    "P0": "紧急 — 请立即处理",
    "P1": "需关注 — 请尽快确认",
    "P2": "供参考 — 可在工作时间内处理",
}
_TIER_URGENCY_EN: dict[SLOTier, str] = {
    "P0": "Urgent — act now",
    "P1": "Needs attention — confirm soon",
    "P2": "For information — handle during working hours",
}
_TIER_URGENCY_JA: dict[SLOTier, str] = {
    "P0": "緊急 — 直ちにご対応ください",
    "P1": "要対応 — お早めにご確認ください",
    "P2": "参考情報 — 業務時間内にご対応ください",
}


def _resolve_tier(handler_key: str) -> SLOTier:
    """Look up the handler's SLO tier, falling through to DEFAULT_SLO."""
    handler = get_handler(handler_key)
    if handler is None:
        return DEFAULT_SLO.tier
    return handler.slo.tier


# ── Failure-storm aggregation (2026-07-19 post-mortem) ────────────────
# One infrastructure event (DB pool exhaustion, Redis outage, TiDB
# brown-out) fails MANY unrelated jobs at once — and because the dedup
# gate above is PER-JOB, every victim's first failure fanned its own
# card to every admin: the 2026-07-19 incident produced 6 cards in one
# burst and a fresh batch each hour as new victims joined. When at
# least ``settings.SCHEDULED_JOB_FAILURE_STORM_THRESHOLD`` DISTINCT
# jobs have a failed run inside the rolling window, per-job cards are
# collapsed into ONE "systemic incident" card; a Redis SET-NX marker
# (TTL 30 min) elects the single announcer among racing failures and
# suppresses the rest for the storm's duration.
#
# P0-tier handlers BYPASS the storm gate — a paging-tier breach
# (leaver escalation, process advance) must reach admins individually
# even mid-storm. Redis being down fails OPEN to the per-job path:
# worst case is the old behaviour (N cards), never silence.
_STORM_WINDOW_MINUTES = 10
_STORM_MARKER_KEY = "helix:sched_job:failure_storm"
_STORM_MARKER_TTL_SECS = 1800


async def _storm_gate(
    session: AsyncSession,
) -> tuple[Literal["announce", "suppress"], list[str], int] | None:
    """Detect a cross-job failure storm and elect the announcer.

    Returns ``None`` when below threshold (normal per-job path),
    ``("announce", names, n)`` for the ONE caller that wins the marker
    (it must send the aggregated card), or ``("suppress", names, n)``
    for every other caller during the storm window.

    Any Redis error fails open (returns ``None``) — degraded noise
    beats silence.
    """
    threshold = int(settings.SCHEDULED_JOB_FAILURE_STORM_THRESHOLD)
    window_start = datetime.now(UTC) - timedelta(minutes=_STORM_WINDOW_MINUTES)
    distinct_job_ids = (
        await session.execute(
            select(ScheduledJobRun.job_id)
            .where(
                ScheduledJobRun.status == "failed",
                ScheduledJobRun.started_at >= window_start,
                # Operator "run now" re-runs are excluded: a person
                # re-triggering three jobs by hand while debugging is
                # not a systemic infrastructure event, and counting
                # them can push a healthy platform over the threshold
                # and silence the per-job cards that DO matter.
                ~ScheduledJobRun.triggered_by.like("manual:%"),
            )
            .distinct()
        )
    ).scalars().all()
    if len(distinct_job_ids) < threshold:
        return None
    names = (
        await session.execute(
            select(ScheduledJob.display_name, ScheduledJob.name)
            .where(ScheduledJob.id.in_(list(distinct_job_ids)))
        )
    ).all()
    display_names = sorted(dn or n for dn, n in names)
    try:
        from app.core.redis_client import get_redis  # noqa: PLC0415
        won = await get_redis().set(
            _STORM_MARKER_KEY,
            datetime.now(UTC).isoformat(),
            nx=True,
            ex=_STORM_MARKER_TTL_SECS,
        )
    except Exception:  # noqa: BLE001 — fail open to per-job cards
        logger.warning(
            "failure-storm marker unavailable (Redis error) — falling "
            "back to per-job notifications", exc_info=True,
        )
        return None
    verdict: Literal["announce", "suppress"] = "announce" if won else "suppress"
    return (verdict, display_names, len(distinct_job_ids))


async def _dispatch_storm_card(
    session: AsyncSession,
    *,
    affected_names: list[str],
    affected_count: int,
    trigger_handler_key: str,
    error_message: str | None,
) -> None:
    """Send the ONE aggregated systemic-incident card to all admins."""
    recipients = await _list_admin_recipients(session)
    if not recipients:
        return
    from app.integrations.service import _scrub_message  # noqa: PLC0415
    try:
        error_brief = _scrub_message(error_message or "").strip().split("\n", 1)[0][:240]
    except Exception:  # noqa: BLE001
        error_brief = (error_message or "").strip().split("\n", 1)[0][:240]
    shown = affected_names[:10]
    more = affected_count - len(shown)
    job_list = "\n".join(f"· {n}" for n in shown) + (
        f"\n· … 等共 {affected_count} 个任务" if more > 0 else ""
    )
    body_text = (
        f"{_STORM_WINDOW_MINUTES} 分钟内有 {affected_count} 个定时任务失败,"
        "已聚合为一条通知(单个任务的失败卡片在风暴期间静默,"
        "运行历史不受影响)。这通常指向一次系统性事件"
        "(数据库连接池 / Redis / 网络),而非多个独立故障。\n\n"
        f"受影响任务:\n{job_list}\n\n"
        f"样例错误({trigger_handler_key}):{error_brief or '(无错误信息)'}"
    )
    envelope_kwargs = {
        "title": f"🌊 [P1] 定时任务集中失败({affected_count} 个)— 疑似系统性事件",
        "body_text": body_text,
        "channels": ["feishu", "email"],
        "context": {
            "affected_count": affected_count,
            "window_minutes": _STORM_WINDOW_MINUTES,
            "trigger_handler_key": trigger_handler_key,
        },
        "url": "/admin/scheduled-jobs/runs?status=failed",
        "card_subtitle": LocalizedText(
            zh="[P1] 需关注 — 多个定时任务在同一窗口失败,请优先排查共同根因。",
            en=(
                "[P1] Needs attention — several scheduled jobs failed in the "
                "same window; look for a shared root cause first."
            ),
            ja=(
                "[P1] 要対応 — 複数のスケジュールジョブが同じ時間帯に失敗しました。"
                "まず共通の原因をご確認ください。"
            ),
        ),
        "related_resource_type": "scheduled_job",
    }
    trace_id = f"scheduled_job.failure_storm:{datetime.now(UTC).strftime('%Y%m%d%H%M')}"
    await asyncio.gather(
        *[
            _dispatch_one(
                recipient=r,
                envelope_kwargs=envelope_kwargs,
                trace_id=trace_id,
                handler_key="failure_storm",
            )
            for r in recipients
        ],
        return_exceptions=True,
    )
    logger.info(
        "failure-storm card dispatched to %d admin(s): %d job(s) failed "
        "within %d min (trigger=%s)",
        len(recipients), affected_count, _STORM_WINDOW_MINUTES,
        trigger_handler_key,
    )


# Signature dedup look-back (2026-08-17 incident). Comparing only
# against the IMMEDIATELY preceding failure meant a job alternating
# between two error shapes — A, B, A, B — was "a different kind of
# failure" every single time and notified on every run. The AI daily
# brief did exactly that (INTELLIGENCE_OUTPUT_INVALID ⇄
# BAILIAN_LLM_BAD_JSON) and pushed 3 P1 cards in 36 minutes. The gate
# now asks "have admins already been told about THIS signature
# recently?" by scanning the job's recent unacked failures.
#
# 24h keeps a daily cron honest: yesterday's still-unacked failure
# silences today's repeat of the same problem, but a fresh problem
# (new signature) always speaks. The row cap bounds the scan.
_LOOKBACK_HOURS = 24
_LOOKBACK_LIMIT = 20


async def _should_notify(
    session: AsyncSession,
    *,
    job_id: UUID,
    new_signature: str,
    tier: SLOTier,
) -> bool:
    """Decide whether a notification should fire for the just-finished
    failure.

    Returns ``True`` if EITHER:

      * no prior failure of this job shares this signature inside the
        look-back window (new kind of failure → notify), OR
      * the most recent prior failure for this job was acknowledged
        (admin already saw a similar one and decided to silence it,
        but a fresh run failed afterwards → notify so they know
        the issue resurged).

    Returns ``False`` (skip notification) when EITHER:

      * an unacknowledged failure with this signature already exists
        in the window AND a card actually went out covering it — the
        "don't re-spam an unread alert" path, widened from "the
        previous run" to "any recent run" so an alternating pair of
        error shapes can't ping-pong past it; OR
      * this job pushed a card less than
        ``SCHEDULED_JOB_FAILURE_NOTIFY_COOLDOWN_MINUTES`` ago and the
        tier is not P0 — a per-job quiet period that bounds ANY burst,
        including one made of genuinely distinct signatures.

    The single-flight lock guarantees prior runs are fully written
    before we compare, so this lookup is race-free.
    """
    # We deliberately tolerate ``status='running'`` rows in between
    # (e.g. the same handler raced startup) by filtering on
    # ``status='failed'`` only. Manual "run now" rows are excluded:
    # they never notify (see ``maybe_notify_failure``), so treating one
    # as "admins already heard about this signature" would silence the
    # next scheduled failure that nobody has been told about.
    #
    # Tiebreaker on ``id`` so two rows with sub-second-equal
    # ``started_at`` (mostly synthetic in tests, but possible in
    # production if a manual trigger fires the same second as a cron
    # tick) get a deterministic order — without this MySQL/TiDB can
    # return them in arbitrary order and the dedup decision flips
    # depending on which row is picked as "prior".
    prev = (
        await session.execute(
            select(ScheduledJobRun)
            .where(ScheduledJobRun.job_id == job_id)
            .where(ScheduledJobRun.status == "failed")
            .where(~ScheduledJobRun.triggered_by.like("manual:%"))
            .where(
                ScheduledJobRun.started_at
                >= datetime.now(UTC) - timedelta(hours=_LOOKBACK_HOURS)
            )
            .order_by(
                desc(ScheduledJobRun.started_at),
                desc(ScheduledJobRun.id),
            )
            .limit(_LOOKBACK_LIMIT)
        )
    ).scalars().all()

    # ``prev[0]`` is the run we just wrote (status='failed'). If
    # there is no second row, this is the first failure in the
    # window → notify.
    if len(prev) < 2:
        return True
    prior_runs = list(prev[1:])

    # Ack is the operator's explicit "I have seen this" — it re-opens
    # the gate ahead of BOTH the cooldown and the signature scan, so a
    # freshly-acked problem that resurges is never silently swallowed.
    if prior_runs[0].acknowledged_at is not None:
        return True

    job = await session.get(ScheduledJob, job_id)
    handler_key = job.handler_key if job is not None else ""

    cooldown_minutes = int(settings.SCHEDULED_JOB_FAILURE_NOTIFY_COOLDOWN_MINUTES)
    if (
        tier != "P0"
        and cooldown_minutes > 0
        and job is not None
        and job.last_failure_notified_at is not None
    ):
        last_notified = job.last_failure_notified_at
        if last_notified.tzinfo is None:
            # MySQL/TiDB hands back naive datetimes on some drivers;
            # comparing naive against aware raises TypeError and the
            # whole notification path would be swallowed as "crashed".
            last_notified = last_notified.replace(tzinfo=UTC)
        if datetime.now(UTC) - last_notified < timedelta(minutes=cooldown_minutes):
            logger.info(
                "scheduled job %s: failure card suppressed by the %d-minute "
                "per-job quiet period (last card at %s)",
                handler_key, cooldown_minutes, last_notified.isoformat(),
            )
            return False

    # A prior failure may only silence this one if admins were actually
    # TOLD about it. A run the quiet period suppressed was never
    # announced, and letting it silence its own repeat would mean a
    # distinct kind of failure that happened to land inside a quiet
    # period is never surfaced at all — the cooldown must delay a card,
    # not delete it.
    #
    # There is no per-run "notified" column, but the two clocks already
    # on the rows answer it: a card is written after the run it reports,
    # so a failure that started AFTER the job's most recent card is one
    # no card has covered. (Imprecise in one direction: if a later run
    # did get a card, earlier suppressed runs read as announced too. The
    # admin has a card for that job in hand in that case, so the residue
    # is a masked second signature, never total silence.)
    announced_through = job.last_failure_notified_at if job is not None else None
    if announced_through is None:
        # Nothing has ever been pushed for this job, so nothing in the
        # window can be "already reported".
        return True
    if announced_through.tzinfo is None:
        announced_through = announced_through.replace(tzinfo=UTC)

    # ``run.error_message`` is the raw message; recompute its signature
    # here rather than persisting per-row so a bug in the signature
    # function can be fixed in code without a backfill. The signature
    # is keyed on the owning job's handler_key.
    for run in prior_runs:
        if run.acknowledged_at is not None:
            continue
        started = run.started_at
        if started is not None and started.tzinfo is None:
            started = started.replace(tzinfo=UTC)
        if started is not None and started > announced_through:
            continue  # suppressed, not announced — carries no silence
        if compute_failure_signature(handler_key, run.error_message) == new_signature:
            return False
    return True


async def _dispatch_one(
    *,
    recipient,
    envelope_kwargs: dict,
    trace_id: str,
    handler_key: str,
) -> None:
    """Per-recipient dispatch in its OWN session.

    Round-62 P0 fix: the previous shared-session design meant a
    ``wait_for`` timeout mid-``_deliver`` injected a CancelledError into
    the shared session, leaving it in ``PendingRollbackError`` state.
    The outer ``session.commit()`` then raised and was swallowed by the
    outer try — net effect: ZERO outbox rows persisted for the entire
    batch (already-delivered recipients lost their inbox card,
    ``last_failure_signature`` never updated, dedup state stuck).

    Each recipient now owns a session so a cancellation can poison
    only its own work. Durability is two-layered:

      * Common case — the timeout fires DURING the Feishu / email
        HTTP call after the outbox INSERT has flushed. The follow-up
        ``r_session.commit()`` lands cleanly, the row is persisted
        with ``status='pending'``, and the retry worker re-delivers.
      * Edge case — the cancel propagates into an in-flight SQLAlchemy
        statement (mid-INSERT, mid-flush). ``r_session`` enters
        ``PendingRollbackError`` and the follow-up commit raises,
        which we swallow on purpose so the cron loop is not blocked.
        The outbox row is LOST for this attempt AND the dedup-signature
        gate above will mark ``last_failure_signature`` after the
        gather() returns, so an identical-signature next tick will
        be deduped. The notification is permanently lost for that
        specific failure -- recovery requires the admin to ack the
        prior failure (which re-opens the dedup gate) or for a
        different-signature failure to occur. The transition out of
        identical-signature happens in practice within a deploy
        window (the underlying cause is fixed and the next failure
        carries a new traceback shape).
    """
    from app.core.database import async_session_factory
    try:
        async with async_session_factory() as r_session:
            r_svc = NotificationService(r_session)
            # savepoint-exempt: dedicated per-recipient session (line above)
            try:
                await asyncio.wait_for(
                    r_svc.dispatch(
                        NotificationEnvelope(recipient=recipient, **envelope_kwargs),
                        trace_id=trace_id,
                        actor_user_id=None,
                    ),
                    timeout=_DISPATCH_TIMEOUT_S,
                )
                await r_session.commit()
            except TimeoutError:
                logger.warning(
                    "scheduled job %s failure notification to %s "
                    "timed out after %.1fs — outbox row persists; "
                    "retry worker will re-deliver",
                    handler_key, recipient.work_email, _DISPATCH_TIMEOUT_S,
                )
                # savepoint-exempt: dedicated per-recipient session; a
                # SAVEPOINT adds a round-trip on a session whose dispatch was
                # just cancelled, and `best_effort` could re-raise where the
                # pre-image never did. Treated the same as its structural twin
                # `identity_sync/alert_notifier._dispatch_one`.
                # savepoint-exempt: dedicated own session; savepoint in finally would mask
                try:
                    await r_session.commit()
                except Exception:  # noqa: BLE001 — defence
                    pass
            except Exception as exc:  # noqa: BLE001 — best-effort
                logger.warning(
                    "scheduled job %s failure notification to %s failed: %s",
                    handler_key, recipient.work_email, exc,
                )
                # savepoint-exempt: dedicated per-recipient session; a
                # SAVEPOINT adds a round-trip on a session whose dispatch was
                # just cancelled, and `best_effort` could re-raise where the
                # pre-image never did. Treated the same as its structural twin
                # `identity_sync/alert_notifier._dispatch_one`.
                # savepoint-exempt: dedicated own session; savepoint in finally would mask
                try:
                    await r_session.commit()
                except Exception:  # noqa: BLE001 — defence
                    pass
    except Exception:  # noqa: BLE001 — defence: per-recipient must never raise
        logger.exception(
            "scheduled job %s per-recipient session for %s crashed",
            handler_key, getattr(recipient, "work_email", "?"),
        )


async def _commit_signature(
    *,
    job_id: UUID,
    signature: str,
    mark_notified_at: bool,
    handler_key: str,
) -> None:
    """Persist ``last_failure_signature`` in a fresh session.

    Isolated from the dispatch path so a poisoned dispatch session
    cannot block the dedup-state write — without isolation, repeated
    identical failures would re-fan-out on every cron tick instead of
    once. ``mark_notified_at`` controls whether ``last_failure_notified_at``
    is also stamped (true on real send, false on dedup-suppressed).
    """
    from app.core.database import async_session_factory
    try:
        async with async_session_factory() as sig_session:
            job = await sig_session.get(ScheduledJob, job_id)
            if job is None:
                return
            job.last_failure_signature = signature
            if mark_notified_at:
                job.last_failure_notified_at = datetime.now(UTC)
            await sig_session.commit()
    except Exception:  # noqa: BLE001 — best-effort
        logger.exception(
            "scheduled job %s: failed to persist last_failure_signature",
            handler_key,
        )


async def maybe_notify_failure(
    *,
    job_id: UUID,
    handler_key: str,
    error_message: str | None,
    triggered_by: str | None = None,
) -> None:
    """Top-level entry point invoked by the service write-back.

    Best-effort. Opens its own session, computes the signature, checks
    the dedup gate, fans out via ``NotificationService.dispatch`` to
    every IT admin if needed (each in its own session, concurrently),
    and records ``last_failure_signature`` / ``last_failure_notified_at``
    on the owning job row in a fresh isolated session.

    ``triggered_by`` is the run row's provenance (``scheduler`` /
    ``startup`` / ``manual:<user_id>``). Omitting it is treated as
    "scheduled" — the safe default, because the cost of a missing
    argument must be a card too many, never a real failure gone quiet.

    Never raises — the only callers are inside the post-handler write-
    back path and we don't want a notification glitch to obscure the
    real handler outcome.
    """
    try:
        signature = compute_failure_signature(handler_key, error_message)
        # Operator-triggered runs never broadcast. ``POST
        # /scheduled-jobs/{id}/trigger`` executes synchronously and
        # returns the failed outcome to the admin who pressed the
        # button; fanning a P1 card to every admin on top of that turns
        # one person verifying a fix into an org-wide page — which is
        # exactly what four "run now" re-runs of the AI daily brief did
        # on 2026-08-17. The signature is still recorded so the job
        # dashboard shows what it is currently failing with.
        if triggered_by is not None and triggered_by.startswith("manual:"):
            logger.info(
                "scheduled job %s: manual run failed (%s) — outcome is "
                "returned to the triggering admin, skipping broadcast",
                handler_key, triggered_by,
            )
            await _commit_signature(
                job_id=job_id, signature=signature,
                mark_notified_at=False, handler_key=handler_key,
            )
            return
        from app.core.database import async_session_factory
        async with async_session_factory() as session:
            # ── Cross-job storm gate (see _storm_gate) ───────────────
            # Checked BEFORE the per-job dedup so even a first-occurrence
            # per-job signature is absorbed into the aggregated card
            # during a systemic event. P0-tier handlers bypass: a
            # paging-tier breach must reach admins individually.
            tier = _resolve_tier(handler_key)
            if tier != "P0":
                storm = await _storm_gate(session)
                if storm is not None:
                    verdict, affected_names, affected_count = storm
                    if verdict == "announce":
                        await _dispatch_storm_card(
                            session,
                            affected_names=affected_names,
                            affected_count=affected_count,
                            trigger_handler_key=handler_key,
                            error_message=error_message,
                        )
                    else:
                        logger.info(
                            "scheduled job %s: failure absorbed by "
                            "active failure-storm window (%d jobs) — "
                            "per-job notification suppressed",
                            handler_key, affected_count,
                        )
                    # Either way the per-job card is replaced by the
                    # storm handling; keep the dashboard signature
                    # truthful. mark_notified_at only for the announcer
                    # (its failure IS surfaced via the storm card).
                    await _commit_signature(
                        job_id=job_id, signature=signature,
                        mark_notified_at=(verdict == "announce"),
                        handler_key=handler_key,
                    )
                    return

            should = await _should_notify(
                session, job_id=job_id, new_signature=signature, tier=tier,
            )
            if not should:
                # Update last_failure_signature anyway so the dashboards
                # can still show "currently failing with signature X"
                # even when we suppressed the notification. Use a fresh
                # session via the helper so it stays isolated.
                logger.info(
                    "scheduled job %s: failure dedup hit "
                    "(signature=%s) — skipping IT admin notification",
                    handler_key, signature,
                )
                await _commit_signature(
                    job_id=job_id, signature=signature,
                    mark_notified_at=False, handler_key=handler_key,
                )
                return

            recipients = await _list_admin_recipients(session)
            if not recipients:
                logger.info(
                    "scheduled job %s failure: no eligible it_admin "
                    "recipients — skipping notification",
                    handler_key,
                )
                # Still mark the signature so future runs dedup.
                await _commit_signature(
                    job_id=job_id, signature=signature,
                    mark_notified_at=True, handler_key=handler_key,
                )
                return

            job = await session.get(ScheduledJob, job_id)
            if job is None:
                logger.warning(
                    "scheduled job %s vanished before notification dispatch",
                    handler_key,
                )
                return

            display = job.display_name or job.name or handler_key
            tier_emoji = _TIER_EMOJI.get(tier, "🚨")
            title_zh = f"{tier_emoji} [{tier}] 定时任务失败 — {display}"
            # Body intentionally compact — the bell card surface is
            # narrow. Detail (full traceback, prior runs) lives one
            # click away on the runs page that ``url`` deep-links to.
            # Scrub credential-shaped substrings before they land in
            # the Feishu card body — driver SDKs occasionally leak
            # AccessKeyId / STS token fragments / signing URLs in
            # error.message, and the failure-notifier card is the most
            # widely-distributed surface in the admin org.
            from app.integrations.service import _scrub_message  # noqa: PLC0415
            try:
                scrubbed_msg = _scrub_message(error_message or "")
            except Exception:  # noqa: BLE001 — defence: never fail the card
                scrubbed_msg = error_message or ""
            error_brief = scrubbed_msg.strip().split("\n", 1)[0][:240]
            # Round-62 P2 fix: store the template sentinel instead of
            # raw [r]...[/r] inline-markup tags. ``render_inbox_body``
            # short-circuits on a non-sentinel body_text and renders it
            # verbatim, leaking the markup into the admin bell card
            # body. The sentinel forces a template render so the card
            # body comes from ``scheduled_job.failure`` (which has the
            # ``{error_brief}`` placeholder + locale variants).
            from app.notifications.service import (  # noqa: PLC0415
                _SEE_TEMPLATE_SENTINEL,
                CardSection,
                CardStat,
                LocalizedText,
            )
            body_text = _SEE_TEMPLATE_SENTINEL
            # ``handler_key`` is code-controlled (registered via
            # ``@register_scheduled_job(key=...)``) so it can't carry
            # adversarial input, but URL-encode anyway to keep the
            # query-string parsing-safe — namespaced keys contain dots
            # that browsers handle fine but explicit encoding documents
            # the boundary.
            from urllib.parse import quote
            url_path = (
                f"/admin/scheduled-jobs/runs?keyword="
                f"{quote(handler_key, safe='')}&status=failed"
            )

            # NotificationService.dispatch is invoked PER-RECIPIENT in
            # its own session (see ``_dispatch_one``) to bound the
            # blast radius of a ``wait_for`` timeout — the shared-
            # session version would poison the whole batch's commit.
            # Rich-card stats: at-a-glance health snapshot for IT admins.
            # All numbers come from the ``scheduled_jobs`` row, no
            # additional queries — keeps the notifier loop cheap.

            run_count = int(job.run_count or 0)
            failure_count = int(job.failure_count or 0)
            success_rate = (
                f"{(1 - failure_count / run_count) * 100:.1f}%"
                if run_count > 0 else "—"
            )
            last_run_label = (
                job.last_run_at.strftime("%Y-%m-%d %H:%M")
                if job.last_run_at else "—"
            )
            card_stats = [
                CardStat(
                    label=LocalizedText(zh="任务", en="Job", ja="ジョブ"),
                    value=display,
                    hint=f"`{handler_key}`",
                ),
                CardStat(
                    label=LocalizedText(
                        zh="总成功率",
                        en="Overall success rate",
                        ja="全体の成功率",
                    ),
                    value=success_rate,
                    hint=LocalizedText(
                        zh=f"{run_count} 次运行 / {failure_count} 次失败",
                        en=f"{run_count} runs / {failure_count} failures",
                        ja=f"実行 {run_count} 回 / 失敗 {failure_count} 回",
                    ),
                ),
                CardStat(
                    label=LocalizedText(
                        zh="上次成功",
                        en="Last success",
                        ja="前回の成功",
                    ),
                    value=last_run_label,
                ),
                CardStat(
                    label=LocalizedText(
                        zh="触发方式", en="Trigger", ja="トリガー",
                    ),
                    value=(job.cron_expr or "interval"),
                ),
            ]
            card_sections = [
                CardSection(
                    title=LocalizedText(
                        zh="错误摘要",
                        en="Error summary",
                        ja="エラー概要",
                    ),
                    # ``error_brief`` is the provider's / handler's own
                    # message — a single language by nature, so it stays a
                    # bare str. Only the empty-case placeholder is localised.
                    body=(error_brief or LocalizedText(
                        zh="(无错误信息)",
                        en="(no error message)",
                        ja="（エラーメッセージなし）",
                    )),
                    tone="danger",
                ),
            ]

            # P0 tier escalates to ``management`` in addition to the
            # standard IT-admin fan-out. Dedup by id so an operator
            # holding both global_admin and management roles still
            # only gets one card.
            if tier == "P0":
                mgmt_recipients = await _list_management_recipients(session)
                existing_ids = {r.id for r in recipients}
                for r in mgmt_recipients:
                    if r.id not in existing_ids:
                        recipients.append(r)
                        existing_ids.add(r.id)

            urgency_zh = _TIER_URGENCY_ZH.get(tier, "")
            urgency_en = _TIER_URGENCY_EN.get(tier, "")
            urgency_ja = _TIER_URGENCY_JA.get(tier, "")
            _cta_zh = "请点击下方按钮跳转到运行历史确认 / 重跑。"
            _cta_en = "Open the run history below to check or re-run."
            _cta_ja = "下のボタンから実行履歴を開いて確認・再実行してください。"
            card_subtitle = LocalizedText(
                zh=f"[{tier}] {urgency_zh}\n{_cta_zh}" if urgency_zh else _cta_zh,
                en=f"[{tier}] {urgency_en}\n{_cta_en}" if urgency_en else _cta_en,
                ja=f"[{tier}] {urgency_ja}\n{_cta_ja}" if urgency_ja else _cta_ja,
            )

            # Freeze the recipient list (detach SA-bound state we need)
            # before closing the read session. ``_dispatch_one`` opens
            # its own session per recipient, so the only thing we carry
            # across the close boundary is the recipient object — which
            # we use only for ``recipient.id`` / ``recipient.work_email``
            # / ``recipient.preferred_language`` (read in dispatch
            # before any flush). SA detaches transparently here.
            envelope_kwargs = {
                "title": title_zh,
                "body_text": body_text,
                "channels": ["feishu", "email"],
                "template_key": "scheduled_job.failure",
                "context": {
                    "job_display_name": display,
                    "handler_key": handler_key,
                    "error_brief": error_brief,
                    "signature": signature,
                    "tier": tier,
                },
                "url": url_path,
                "card_subtitle": card_subtitle,
                "card_stats": card_stats,
                "card_sections": card_sections,
                "related_resource_type": "scheduled_job",
                "related_resource_id": UUID(str(job.id)),
            }
            # Carry the failure signature in the trace_id so the outbox
            # row + audit row ties back to the specific failing run a
            # future ``GET /api/v1/audit?trace_id=…`` lookup can resolve.
            # Pre-fix (Round-20 audit, finding N1): trace_id="" left the
            # notification orphaned in audit — operators could not drill
            # from "admin got pinged" back to "which run failed."
            trace_id = f"scheduled_job.failure:{handler_key}:{signature}"
            frozen_recipients = list(recipients)

        # ── Fan out concurrently, each recipient in its OWN session.
        # Round-62 P0 fix: shared-session loop let one cancelled
        # ``wait_for`` poison the entire batch's commit, losing all
        # already-delivered outbox rows + the dedup-state write.
        # Round-62 P1 fix: serial loop turned a Feishu outage × N
        # recipients into N × timeout wall-time on the cron loop;
        # ``gather`` collapses that back to ~1 × timeout total.
        await asyncio.gather(
            *[
                _dispatch_one(
                    recipient=recipient,
                    envelope_kwargs=envelope_kwargs,
                    trace_id=trace_id,
                    handler_key=handler_key,
                )
                for recipient in frozen_recipients
            ],
            return_exceptions=True,
        )

        # ── Persist dedup state in a fresh, isolated session — cannot
        # be poisoned by any of the dispatch sessions above.
        await _commit_signature(
            job_id=job_id, signature=signature,
            mark_notified_at=True, handler_key=handler_key,
        )
        logger.info(
            "scheduled job %s failure notification dispatched to "
            "%d admin(s) (signature=%s)",
            handler_key, len(frozen_recipients), signature,
        )
    except Exception:  # noqa: BLE001
        logger.exception(
            "scheduled job %s failure notification path itself crashed "
            "— suppressing so the run outcome is unaffected",
            handler_key,
        )


__all__ = ["compute_failure_signature", "maybe_notify_failure"]