"""Scheduled jobs — CRUD + execution wrapper.

The service owns *persistence + telemetry*; the handler owns *work*.
``execute_job`` is the single place where a run starts, gets timed,
and the result lands in ``scheduled_job_runs`` + the owning job row.
Every caller (startup scheduler tick, manual "run now" button)
goes through it so the history is consistent.
"""
from __future__ import annotations

import asyncio
import json
import logging
from collections.abc import AsyncIterator
from contextlib import asynccontextmanager
from datetime import UTC, datetime
from typing import Any
from uuid import UUID

from sqlalchemy import delete as _sa_delete
from sqlalchemy import desc, select
from sqlalchemy.ext.asyncio import AsyncSession

from app.core.config import settings
from app.core.database import async_session_factory
from app.core.exceptions import (
    AppException,
    NotFoundError,
    format_exception_for_log,
    is_connection_lost_db_error,
    is_transaction_aborting_db_error,
)
from app.scheduled_jobs.chain import (
    CHAIN_ADVANCING_STATUSES,
    assert_no_cycle,
    load_parent_map,
    spawn_children,
)
from app.scheduled_jobs.hygiene import (
    integration_gate_state,
    should_record_gate_skip,
)
from app.scheduled_jobs.models import (
    RUN_STATUS_INTERRUPTED,
    TRIGGER_CHAINED,
    ScheduledJob,
    ScheduledJobRun,
)
from app.scheduled_jobs.registry import get_handler, list_handlers

logger = logging.getLogger(__name__)

MAX_RUN_HISTORY_PER_JOB = 100  # trim history on every successful write


# -- Shutdown-drain flag -------------------------------------------
# A cancelled run has two possible causes and they deserve different
# words on the run row: the worker's graceful drain (overwhelmingly the
# common one -- ~15 deploys a day) or a single-flight lock lost
# mid-hold while ``auto_renew`` was renewing it (rare: a Redis
# eviction, a failover, a manual DEL -- in which case
# ``distributed_lock``'s watchdog cancels the holder on purpose).
#
# The executor cannot tell them apart from the CancelledError alone,
# and "cancelled for one of two reasons" is a worse run row than
# either specific answer. The worker sets this flag immediately before
# it starts draining, which makes the distinction free.
_DRAINING = False


def mark_draining() -> None:
    """Record that this process has begun a graceful shutdown.

    Called by the worker before it drains in-flight jobs, so a run
    cancelled from here on can name the deploy rather than guess.
    """
    global _DRAINING
    _DRAINING = True


def is_draining() -> bool:
    return _DRAINING


# ── Queries ──────────────────────────────────────────────────────


# Category → handler-namespace prefixes. Exposed to the router as a
# Segmented filter so the admin page doesn't render a flat 20+ row
# table. Keep the top-level keys tight — fewer, wider buckets beat many
# narrow ones when the total number of jobs is under ~100.
JOB_CATEGORIES: dict[str, tuple[str, ...]] = {
    "hr": ("hris.", "hr."),
    "cmdb": ("cmdb.",),
    "notification": ("notification.",),
    "provisioning": ("provisioning.",),
}


def _category_prefixes(category: str) -> tuple[str, ...]:
    """Return the handler_key prefixes that belong to a category, or
    an empty tuple for unknown categories (treated as no-op filter)."""
    return JOB_CATEGORIES.get(category, ())


async def list_jobs(
    session: AsyncSession,
    *,
    page: int = 1,
    page_size: int = 20,
    keyword: str | None = None,
    enabled: bool | None = None,
    last_status: str | None = None,
    category: str | None = None,
    health: str | None = None,
) -> tuple[list[ScheduledJob], int]:
    """List scheduled jobs with optional filters + pagination.

    Return shape changed 2026-04-22: now ``(rows, total)`` instead of
    ``list[rows]`` so the router can emit the standard paginated
    envelope. Only one internal caller existed (the router) and it
    was updated in the same change.

    Filters:
      * ``keyword`` — substring match on name / display_name /
        handler_key. ``description`` is deliberately NOT searched:
        the canonical cmdb.grants_sync description references every
        provider (Azure/GCP/AWS/Aliyun) so a user search for "azure"
        would otherwise pull back AWS / GCP rows too. Searchable
        fields are the ones an admin can realistically type.
      * ``enabled`` — True|False to narrow to on/off jobs
      * ``last_status`` — 'ok'|'failed'|'running'|'skipped' — gated on
        ScheduledJob.last_status, the cached status from the most
        recent finished run.
      * ``category`` — one of ``hr``, ``cmdb``, ``notification``,
        ``provisioning``, or ``other``. ``other`` narrows to jobs whose
        handler_key doesn't match any known namespace prefix. Unknown
        values silently fall through (no filter applied).
      * ``health`` — one of the ``hygiene.JobHealth`` values. Health is
        a derived verdict (it depends on the handler registry, the
        integrations table, and a run aggregate), so it cannot be a SQL
        predicate. When this filter is set the query runs unpaginated,
        health is resolved for the matching rows, and the page is cut
        in Python. That is bounded by design: the ledger is ~350 rows
        and only ever grows by migration, so "load them all" costs one
        indexed scan and is honest, whereas a cached health column
        would have to be invalidated by three unrelated tables.
    """
    from sqlalchemy import and_, not_, or_
    from sqlalchemy import func as _func

    query = select(ScheduledJob).order_by(ScheduledJob.display_name)
    count_q = select(_func.count()).select_from(ScheduledJob)

    if keyword:
        pattern = f"%{keyword}%"
        cond = or_(
            ScheduledJob.name.ilike(pattern),
            ScheduledJob.display_name.ilike(pattern),
            ScheduledJob.handler_key.ilike(pattern),
        )
        query = query.where(cond)
        count_q = count_q.where(cond)
    if enabled is not None:
        query = query.where(ScheduledJob.enabled == enabled)
        count_q = count_q.where(ScheduledJob.enabled == enabled)
    if last_status:
        query = query.where(ScheduledJob.last_status == last_status)
        count_q = count_q.where(ScheduledJob.last_status == last_status)
    if category:
        prefixes = _category_prefixes(category)
        if prefixes:
            cat_cond = or_(
                *(ScheduledJob.handler_key.like(f"{p}%") for p in prefixes),
            )
            query = query.where(cat_cond)
            count_q = count_q.where(cat_cond)
        elif category == "other":
            # Everything that doesn't match any known prefix.
            known = [p for prefs in JOB_CATEGORIES.values() for p in prefs]
            other_cond = and_(
                *(not_(ScheduledJob.handler_key.like(f"{p}%")) for p in known),
            )
            query = query.where(other_cond)
            count_q = count_q.where(other_cond)

    if health:
        from app.scheduled_jobs.hygiene import annotate_health

        candidates = list((await session.execute(query)).scalars().all())
        resolved = await annotate_health(session, candidates)
        matched = [
            job for job in candidates if resolved.get(str(job.id)) == health
        ]
        start = (page - 1) * page_size
        return matched[start:start + page_size], len(matched)

    total = (await session.execute(count_q)).scalar() or 0
    rows = (
        await session.execute(
            query.offset((page - 1) * page_size).limit(page_size)
        )
    ).scalars().all()
    return list(rows), total


async def resolve_chain_parent_names(
    session: AsyncSession, jobs: list[ScheduledJob],
) -> dict[str, str]:
    """``{job_id: predecessor display_name}`` for a page of jobs.

    One query for the whole page. The list needs to render "接在 X 之后"
    where a cadence would otherwise go, and a per-row lookup would make
    the cheapest column on the page the most expensive.
    """
    wanted = {
        j.chain_after_job_id for j in jobs if j.chain_after_job_id is not None
    }
    if not wanted:
        return {}
    rows = (
        await session.execute(
            select(ScheduledJob.id, ScheduledJob.display_name).where(
                ScheduledJob.id.in_(wanted),
            ),
        )
    ).all()
    names = {str(row[0]): row[1] for row in rows}
    return {
        str(j.id): names[str(j.chain_after_job_id)]
        for j in jobs
        if j.chain_after_job_id is not None
        and str(j.chain_after_job_id) in names
    }


async def job_category_counts(session: AsyncSession) -> dict[str, int]:
    """Return ``{category: count}`` including ``total`` and ``other``.

    Works in a single trip — fetch every ``(handler_key, 1)`` tuple
    grouped client-side in Python. With ~20 handler_keys today this is
    trivial; if it ever grows we can switch to a CASE expression.
    """
    from sqlalchemy import func as _func

    rows = (
        await session.execute(
            select(ScheduledJob.handler_key, _func.count())
            .group_by(ScheduledJob.handler_key)
        )
    ).all()
    counts = {k: 0 for k in JOB_CATEGORIES}
    counts["other"] = 0
    total = 0
    for handler_key, n in rows:
        total += int(n)
        bucket = "other"
        for cat, prefixes in JOB_CATEGORIES.items():
            if any(handler_key.startswith(p) for p in prefixes):
                bucket = cat
                break
        counts[bucket] += int(n)
    counts["total"] = total
    return counts


async def get_job(session: AsyncSession, job_id: UUID) -> ScheduledJob:
    row = await session.get(ScheduledJob, job_id)
    if row is None:
        raise NotFoundError("ScheduledJob", str(job_id))
    return row


# ── Scheduler health (UI banner support) ──────────────────────────


async def scheduler_health_summary(
    session: AsyncSession,
) -> dict[str, Any]:
    """Surface scheduler-liveness signal to the admin UI.

    The same wedge condition that the worker watchdog
    (``app/worker.py``) catches and acts on (exit-1 → docker restart),
    this helper surfaces visually on ``/admin/scheduled-jobs`` so an
    admin scanning the page can spot the wedge BEFORE the watchdog's
    30-minute threshold fires — and so the wedge is loud even if the
    watchdog itself is silenced (e.g., manual override during a
    diagnostic window).

    Computed entirely from existing ``scheduled_jobs`` rows; no
    additional table. Three buckets:

      * ``ok``    — ``actual_gap_seconds`` ≤ 2 × ``interval_seconds``
      * ``late``  — between 2× and 5× interval (transient hiccup or
                    slow handler)
      * ``stale`` — > 5× interval (scheduler likely wedged)

    The 5× factor matches the watchdog's "fastest-job × ~60" rule of
    thumb at the 30-minute cap, while staying tolerant of one or two
    missed ticks for legitimate reasons (handler ran long, etc.).

    Returns:
        ``{
            "checked_at": iso8601,
            "fastest_interval_seconds": int | None,
            "fastest_last_run_at": iso8601 | None,
            "fastest_actual_gap_seconds": float | None,
            "stale_count": int,
            "late_count": int,
            "stale_jobs": [{"name", "interval_seconds", "actual_gap_seconds"}, …],
        }``

    Cron-only worker configs (no interval jobs) get all-None signals
    so the UI can hide the banner gracefully.
    """
    now = datetime.now(UTC)
    # Include never-run jobs (``last_run_at IS NULL``) — that's exactly
    # the wedge signal the banner exists to surface (a freshly added
    # interval job the scheduler never picked up). They're treated as
    # fully stale once their interval has elapsed since the job row was
    # created.
    rows = (
        await session.execute(
            select(ScheduledJob)
            .where(
                ScheduledJob.enabled.is_(True),
                ScheduledJob.interval_seconds.isnot(None),
            )
        )
    ).scalars().all()

    fastest_interval: int | None = None
    fastest_last: datetime | None = None
    fastest_gap: float | None = None
    stale_jobs: list[dict[str, Any]] = []
    late_count = 0
    stale_count = 0

    for row in rows:
        interval = int(row.interval_seconds or 0)
        if interval <= 0:
            continue
        # Reference point: prefer last_run_at; fall back to created_at
        # so a never-run job's age is measured against "how long it's
        # been waiting" rather than appearing infinitely stale (which
        # would force the banner to ALWAYS be lit on a fresh deploy).
        reference = row.last_run_at or row.created_at
        if reference is None:
            continue
        ref_cmp = reference.replace(tzinfo=None) if reference.tzinfo else reference
        now_cmp = now.replace(tzinfo=None)
        gap = (now_cmp - ref_cmp).total_seconds()
        if gap > 5 * interval:
            stale_count += 1
            stale_jobs.append({
                "name": row.name,
                "interval_seconds": interval,
                "actual_gap_seconds": int(gap),
                "never_ran": row.last_run_at is None,
            })
        elif gap > 2 * interval:
            late_count += 1
        if fastest_interval is None or interval < fastest_interval:
            fastest_interval = interval
            fastest_last = row.last_run_at
            fastest_gap = gap

    # Sort stale by worst staleness ratio so the UI banner can show
    # the most-suspicious entries first.
    stale_jobs.sort(
        key=lambda r: r["actual_gap_seconds"] / max(r["interval_seconds"], 1),
        reverse=True,
    )

    return {
        "checked_at": now.isoformat(),
        "fastest_interval_seconds": fastest_interval,
        "fastest_last_run_at": (
            fastest_last.isoformat() if fastest_last is not None else None
        ),
        "fastest_actual_gap_seconds": (
            int(fastest_gap) if fastest_gap is not None else None
        ),
        "stale_count": stale_count,
        "late_count": late_count,
        # Cap at top 10 to keep payloads small even on a wedge.
        "stale_jobs": stale_jobs[:10],
    }


async def analytics_summary(
    session: AsyncSession,
    *,
    window_days: int = 7,
) -> dict[str, Any]:
    """Aggregate dashboard data for the 定时任务 → 分析 tab.

    Single round-trip per slice, every aggregate is index-eligible
    (``ix_scheduled_job_runs_job_started`` covers ``started_at`` range
    scans). Categories are derived in Python from ``handler_key`` —
    same logic as ``job_category_counts`` so the two views agree.

    The window applies to *runs*. Job-level rollups (total / enabled /
    failure_count) are lifetime-current because that's the surface
    operators monitor on the list view too.
    """

    from sqlalchemy import case
    from sqlalchemy import func as _func

    from app.core.timeseries import day_axis

    now = datetime.now(UTC)
    # Round-62 P1 fix (off-by-one analytics drops today): use the
    # shared day-axis helper so the SQL window and the bucket loop
    # describe the same N day-aligned interval ending on today. The
    # inline ``window_start = now - timedelta(days=window_days)`` +
    # ``range(window_days)`` bucket pattern produced [today-N … today-1]
    # buckets, silently dropping today from the chart while including
    # it in the headline KPI.
    axis = day_axis(window_days=window_days, now=now)
    window_start = axis.window_start

    # ── Job-level rollups ────────────────────────────────────────
    total_jobs = (
        await session.execute(
            select(_func.count()).select_from(ScheduledJob)
        )
    ).scalar() or 0
    enabled_jobs = (
        await session.execute(
            select(_func.count()).select_from(ScheduledJob)
            .where(ScheduledJob.enabled.is_(True))
        )
    ).scalar() or 0

    # by trigger type (interval vs cron) — design dimension worth
    # surfacing for ops planning.
    trigger_rows = (
        await session.execute(
            select(ScheduledJob.trigger_type, _func.count())
            .group_by(ScheduledJob.trigger_type)
        )
    ).all()
    by_trigger_type = {row[0] or "unknown": int(row[1]) for row in trigger_rows}

    # last_status breakdown across CURRENT jobs (most recent known
    # state per job) — gives "snapshot of fleet health" complementing
    # the runs-window aggregate below.
    last_status_rows = (
        await session.execute(
            select(ScheduledJob.last_status, _func.count())
            .group_by(ScheduledJob.last_status)
        )
    ).all()
    by_last_status = {
        (row[0] or "never_run"): int(row[1]) for row in last_status_rows
    }

    # ── Per-handler_key rollup (used to derive per-category aggregates)
    handler_rollup = (
        await session.execute(
            select(
                ScheduledJob.handler_key,
                _func.count(ScheduledJob.id),
                _func.coalesce(_func.sum(ScheduledJob.run_count), 0),
                _func.coalesce(_func.sum(ScheduledJob.failure_count), 0),
            )
            .group_by(ScheduledJob.handler_key)
        )
    ).all()

    by_category_jobs: dict[str, int] = {k: 0 for k in JOB_CATEGORIES}
    by_category_jobs["other"] = 0
    lifetime_runs_by_category: dict[str, int] = {k: 0 for k in by_category_jobs}
    lifetime_failures_by_category: dict[str, int] = {k: 0 for k in by_category_jobs}
    for handler_key, n_jobs, run_sum, fail_sum in handler_rollup:
        bucket = "other"
        for cat, prefixes in JOB_CATEGORIES.items():
            if any(handler_key.startswith(p) for p in prefixes):
                bucket = cat
                break
        by_category_jobs[bucket] += int(n_jobs)
        lifetime_runs_by_category[bucket] += int(run_sum or 0)
        lifetime_failures_by_category[bucket] += int(fail_sum or 0)

    # ── Run-level rollups inside the window ─────────────────────
    # Total + status breakdown.
    window_status_rows = (
        await session.execute(
            select(ScheduledJobRun.status, _func.count())
            .where(ScheduledJobRun.started_at >= window_start)
            .group_by(ScheduledJobRun.status)
        )
    ).all()
    runs_by_status_window = {
        row[0]: int(row[1]) for row in window_status_rows
    }
    ok_in_window = runs_by_status_window.get("ok", 0)
    failed_in_window = runs_by_status_window.get("failed", 0)
    # Round-62 follow-up: align success_rate denominator with
    # ``slo_evaluator.evaluate_handler_slo`` — TERMINAL statuses only
    # (ok + failed). ``skipped`` is a single-flight collision (worker
    # tick collided with a prior run still holding the lock) and not
    # a real attempt; ``running`` is in-flight. Including them
    # depressed the cron-panel success_rate while the SLO-panel
    # success_rate (which uses terminal-only) read clean, so the two
    # /admin/health panels disagreed about the same handler. Keep
    # ``runs_by_status_window`` populated separately for the status-
    # breakdown UI bar — only the success_rate denominator changes.
    #
    # ``None`` (not 100%) on a true-zero window so the KPI renders ``—``
    # rather than a misleading green "perfect health" tile while the
    # scheduler is actually wedged. Frontend handles the null contract.
    runs_in_window = sum(runs_by_status_window.values())
    terminal_runs_in_window = ok_in_window + failed_in_window
    success_rate: float | None = (
        round(ok_in_window * 1000 / terminal_runs_in_window) / 10
        if terminal_runs_in_window
        else None
    )

    # Daily timeseries (day × status) — driver for the trend chart.
    # DATE(started_at) works on both TiDB and MySQL.
    daily_rows = (
        await session.execute(
            select(
                _func.date(ScheduledJobRun.started_at).label("day"),
                ScheduledJobRun.status,
                _func.count(),
            )
            .where(ScheduledJobRun.started_at >= window_start)
            .group_by("day", ScheduledJobRun.status)
            .order_by("day")
        )
    ).all()
    daily_map: dict[str, dict[str, int]] = {}
    for day, status, n in daily_rows:
        key = day.isoformat() if hasattr(day, "isoformat") else str(day)
        # Every status the window can contain needs a slot here AND a
        # line on the chart. ``runs_in_window`` is the sum over all
        # statuses, so a status that exists in the data but not in this
        # dict makes the chart quietly disagree with the headline number
        # — the failure mode the ``running`` series was added to fix.
        bucket = daily_map.setdefault(
            key,
            {"ok": 0, "failed": 0, "skipped": 0, "running": 0, "interrupted": 0},
        )
        bucket[status] = int(n)
    # Fill every day in window so frontend can draw a continuous line.
    daily_series: list[dict[str, Any]] = []
    for d in axis.days:
        key = d.isoformat()
        slot = daily_map.get(key, {})
        daily_series.append({
            "date": key,
            "ok": slot.get("ok", 0),
            "failed": slot.get("failed", 0),
            "interrupted": slot.get("interrupted", 0),
            "skipped": slot.get("skipped", 0),
            "running": slot.get("running", 0),
        })

    # Hourly distribution over the window (24 bins). Useful for
    # spotting a cron's preferred minute / on-hour bias.
    hourly_rows = (
        await session.execute(
            select(
                _func.hour(ScheduledJobRun.started_at).label("h"),
                _func.count(),
            )
            .where(ScheduledJobRun.started_at >= window_start)
            .group_by("h")
            .order_by("h")
        )
    ).all()
    hourly_map = {int(h): int(n) for h, n in hourly_rows}
    hourly_distribution = [
        {"hour": h, "count": hourly_map.get(h, 0)} for h in range(24)
    ]

    # triggered_by source breakdown (scheduler / manual / startup).
    triggered_by_rows = (
        await session.execute(
            select(
                case(
                    (
                        ScheduledJobRun.triggered_by.like("manual:%"),
                        "manual",
                    ),
                    else_=ScheduledJobRun.triggered_by,
                ).label("src"),
                _func.count(),
            )
            .where(ScheduledJobRun.started_at >= window_start)
            .group_by("src")
        )
    ).all()
    by_triggered_by = {row[0]: int(row[1]) for row in triggered_by_rows}

    # Top 10 failing jobs (lifetime failure_count). Cheap join-back to
    # fetch display_name + handler_key for label rendering.
    top_failing = (
        await session.execute(
            select(
                ScheduledJob.id,
                ScheduledJob.name,
                ScheduledJob.display_name,
                ScheduledJob.handler_key,
                ScheduledJob.failure_count,
                ScheduledJob.run_count,
                ScheduledJob.last_status,
            )
            .where(ScheduledJob.failure_count > 0)
            .order_by(desc(ScheduledJob.failure_count))
            .limit(10)
        )
    ).all()
    top_failing_jobs = [
        {
            "id": str(row[0]),
            "name": row[1],
            "display_name": row[2],
            "handler_key": row[3],
            "failure_count": int(row[4]),
            "run_count": int(row[5]),
            "last_status": row[6],
        }
        for row in top_failing
    ]

    # Slowest 10 jobs by average duration in the window.
    slowest_rows = (
        await session.execute(
            select(
                ScheduledJob.id,
                ScheduledJob.name,
                ScheduledJob.display_name,
                ScheduledJob.handler_key,
                _func.avg(ScheduledJobRun.duration_ms).label("avg_ms"),
                _func.max(ScheduledJobRun.duration_ms).label("max_ms"),
                _func.count(ScheduledJobRun.id).label("n"),
            )
            .join(ScheduledJobRun, ScheduledJobRun.job_id == ScheduledJob.id)
            .where(
                ScheduledJobRun.started_at >= window_start,
                ScheduledJobRun.duration_ms.isnot(None),
            )
            .group_by(
                ScheduledJob.id,
                ScheduledJob.name,
                ScheduledJob.display_name,
                ScheduledJob.handler_key,
            )
            .order_by(desc("avg_ms"))
            .limit(10)
        )
    ).all()
    slowest_jobs = [
        {
            "id": str(row[0]),
            "name": row[1],
            "display_name": row[2],
            "handler_key": row[3],
            "avg_duration_ms": int(row[4] or 0),
            "max_duration_ms": int(row[5] or 0),
            "run_count": int(row[6]),
        }
        for row in slowest_rows
    ]

    # Unacked failure backlog (visible-red queue).
    unacked = (
        await session.execute(
            select(_func.count()).select_from(ScheduledJobRun)
            .where(
                ScheduledJobRun.status == "failed",
                ScheduledJobRun.acknowledged_at.is_(None),
            )
        )
    ).scalar() or 0

    return {
        "window_days": window_days,
        "checked_at": now.isoformat(),
        "totals": {
            "total_jobs": int(total_jobs),
            "enabled_jobs": int(enabled_jobs),
            "disabled_jobs": int(total_jobs) - int(enabled_jobs),
            "runs_in_window": int(runs_in_window),
            "ok_in_window": int(ok_in_window),
            "failed_in_window": int(failed_in_window),
            "success_rate": success_rate,
            "unacked_failures": int(unacked),
        },
        "by_trigger_type": by_trigger_type,
        "by_last_status": by_last_status,
        "by_category_jobs": by_category_jobs,
        "lifetime_runs_by_category": lifetime_runs_by_category,
        "lifetime_failures_by_category": lifetime_failures_by_category,
        "runs_by_status_window": runs_by_status_window,
        "daily_series": daily_series,
        "hourly_distribution": hourly_distribution,
        "by_triggered_by": by_triggered_by,
        "top_failing_jobs": top_failing_jobs,
        "slowest_jobs": slowest_jobs,
    }


# ── Schedule forecast (timeline tab) ─────────────────────────────

# A job whose individual fires are too dense to plot as discrete dots
# is rendered as a continuous "band" on the timeline instead. Two
# triggers: (a) an interval shorter than this threshold, or (b) any
# trigger that would enumerate more than ``FORECAST_POINT_CAP`` fires
# inside the window. Both collapse to ``dense=True`` so the frontend
# draws one band rather than hundreds of overlapping markers.
DENSE_INTERVAL_SECONDS = 1800  # 30 min
FORECAST_POINT_CAP = 48
# Hard ceiling on the past-run detail rows pulled back for the sparse
# (non-dense) jobs — the timeline only plots dots for those, and the
# window is short, so this is generous headroom, not a real limit.
FORECAST_PAST_DETAIL_CAP = 2000


def _category_of(handler_key: str) -> str:
    """Bucket a handler_key into one of the UI categories (same rule
    as ``job_category_counts`` / ``analytics_summary``)."""
    for cat, prefixes in JOB_CATEGORIES.items():
        if any(handler_key.startswith(p) for p in prefixes):
            return cat
    return "other"


def _as_utc(dt: datetime | None) -> datetime | None:
    """Normalize a possibly-naive DB datetime to tz-aware UTC.

    MySQL/TiDB ``DATETIME`` round-trips strip tzinfo even though the
    column is declared ``timezone=True``; a freshly-loaded ``last_run_at``
    therefore arrives naive. Comparing a naive datetime against the
    tz-aware ``window_end`` raises ``TypeError`` — so coerce here.
    """
    if dt is None:
        return None
    return dt.replace(tzinfo=UTC) if dt.tzinfo is None else dt.astimezone(UTC)


def _forecast_upcoming(
    job: ScheduledJob,
    *,
    now: datetime,
    window_end: datetime,
    chain_offsets: dict[UUID, tuple[list[datetime], bool, int]] | None = None,
) -> tuple[list[datetime], bool]:
    """Return ``(upcoming_fires, dense)`` for one job inside
    ``(now, window_end]``.

    ``dense`` short-circuits enumeration: a sub-30-min interval is dense
    by definition, and any trigger that overflows ``FORECAST_POINT_CAP``
    inside the window flips dense mid-enumeration (the partial list is
    discarded — the frontend draws a band for dense jobs).

    A ``chained`` job has no clock of its own: its fires are its
    predecessor's fires shifted by however long that predecessor
    typically takes. ``chain_offsets`` carries the already-resolved
    ``(parent_fires, parent_dense, parent_duration_ms)`` per job id —
    the parent's *own* duration, because ``parent_fires`` is the
    parent's already-shifted list and so already carries everything
    above it. A seven-deep chain costs one pass rather than a recursive
    walk per member. The
    estimate is explicitly an estimate — the whole point of a chain is
    that the successor starts when the predecessor is *actually* done,
    which no forecast can know — but plotting nothing at all would make
    every chained job vanish from the timeline, and "invisible" is a
    worse answer than "approximately here".
    """
    fires: list[datetime] = []

    if job.trigger_type == TRIGGER_CHAINED:
        if not chain_offsets or job.id not in chain_offsets:
            return [], False
        parent_fires, parent_dense, offset_ms = chain_offsets[job.id]
        if parent_dense:
            return [], True
        from datetime import timedelta

        shift = timedelta(milliseconds=max(0, offset_ms))
        return [f + shift for f in parent_fires if f + shift <= window_end], False

    if job.trigger_type == "interval" and job.interval_seconds:
        interval = int(job.interval_seconds)
        if interval < DENSE_INTERVAL_SECONDS:
            return [], True
        from datetime import timedelta

        step = timedelta(seconds=interval)
        anchor = _as_utc(job.last_run_at) or now
        # Walk the anchor forward to the first fire strictly after ``now``
        # without looping second-by-second.
        if anchor <= now:
            gap = (now - anchor).total_seconds()
            steps = int(gap // interval) + 1
            t = anchor + step * steps
        else:
            t = anchor
        while t <= window_end:
            fires.append(t)
            if len(fires) > FORECAST_POINT_CAP:
                return [], True
            t = t + step
        return fires, False

    if job.trigger_type == "cron" and job.cron_expr:
        from apscheduler.triggers.cron import CronTrigger

        try:
            trig = CronTrigger.from_crontab(
                job.cron_expr, timezone=job.timezone or "UTC",
            )
        except (ValueError, KeyError):
            # A corrupt expression can't be plotted — surface nothing
            # rather than crash the whole forecast.
            return [], False
        # ``get_next_fire_time(previous, now)`` returns the next fire at
        # or after ``now`` and strictly after ``previous``. Seed with the
        # current instant, then feed each result back as ``previous`` to
        # step strictly forward.
        try:
            nxt = trig.get_next_fire_time(None, now)
            while nxt is not None and nxt <= window_end:
                fires.append(nxt.astimezone(UTC))
                if len(fires) > FORECAST_POINT_CAP:
                    return [], True
                nxt = trig.get_next_fire_time(nxt, nxt)
        except Exception:  # noqa: BLE001
            return fires, False
        return fires, False

    return [], False


def _next_fire_beyond_window(
    job: ScheduledJob, *, now: datetime, window_end: datetime,
) -> datetime | None:
    """First fire strictly after ``window_end``, or ``None``.

    Only called when a job has no fire inside the forecast window, to
    answer "nothing scheduled in the next 18h — so when *does* it run?".
    Same trigger maths as :func:`_forecast_upcoming`, just unbounded on
    the right, so the answer is the real next fire rather than a stored
    guess.

    Returns ``None`` for the two cases where "beyond the window" is not
    a truthful description:

      * **dense** jobs (sub-30-min interval) — their next fire is minutes
        away and always inside the window; ``upcoming`` is empty only
        because enumeration short-circuits into a band.
      * **chained** jobs — they have no clock of their own. A chained
        job whose predecessor does not fire in the whole window has no
        estimate worth showing; the forecast already labels chain times
        as estimates, and extrapolating past the window would stack a
        guess on a guess.
    """
    if job.trigger_type == TRIGGER_CHAINED:
        return None

    if job.trigger_type == "interval" and job.interval_seconds:
        interval = int(job.interval_seconds)
        if interval < DENSE_INTERVAL_SECONDS:
            return None
        from datetime import timedelta

        # Same anchor as ``_forecast_upcoming``: a never-run job is
        # anchored on ``now``, NOT on ``window_end``. Anchoring on the
        # window would push the answer a full window into the future —
        # a job that has never run and fires every 7d runs 7d from now,
        # not 7d after the end of an 18h chart.
        anchor = _as_utc(job.last_run_at) or now
        if anchor > window_end:
            return anchor
        gap = (window_end - anchor).total_seconds()
        steps = int(gap // interval) + 1
        return anchor + timedelta(seconds=interval) * steps

    if job.trigger_type == "cron" and job.cron_expr:
        from apscheduler.triggers.cron import CronTrigger

        try:
            trig = CronTrigger.from_crontab(
                job.cron_expr, timezone=job.timezone or "UTC",
            )
            # ``previous=window_end`` forces a fire strictly after the
            # window rather than one exactly on its edge (which
            # ``_forecast_upcoming`` would already have plotted).
            nxt = trig.get_next_fire_time(window_end, window_end)
        except Exception:  # noqa: BLE001
            # A corrupt expression can't be plotted — same posture as
            # ``_forecast_upcoming``: surface nothing, don't crash the
            # whole forecast.
            return None
        return nxt.astimezone(UTC) if nxt is not None else None

    return None


async def _typical_durations(
    session: AsyncSession, job_ids: list[UUID],
) -> dict[UUID, int]:
    """Average successful-run duration per job, in milliseconds.

    Feeds the chain forecast: a chained job's estimated start is its
    predecessor's fire plus how long that predecessor typically takes.
    ``ok`` and ``skipped`` only — a run killed at the timeout is real
    wall-clock but would push a chain estimate half an hour into the
    future on the strength of one bad night.
    """
    if not job_ids:
        return {}
    from sqlalchemy import func as _f

    rows = (
        await session.execute(
            select(ScheduledJobRun.job_id, _f.avg(ScheduledJobRun.duration_ms))
            .where(
                ScheduledJobRun.job_id.in_(job_ids),
                ScheduledJobRun.status.in_(("ok", "skipped")),
                ScheduledJobRun.duration_ms.isnot(None),
            )
            .group_by(ScheduledJobRun.job_id),
        )
    ).all()
    return {row[0]: int(row[1] or 0) for row in rows}


async def schedule_forecast(
    session: AsyncSession,
    *,
    past_hours: int = 6,
    future_hours: int = 18,
    category: str | None = None,
) -> dict[str, Any]:
    """Forward-looking + recent-past schedule for the timeline tab.

    Backs the 排程时间轴 tab on ``/admin/scheduled-jobs``. One round-trip
    returns, for every *enabled* job:

      * ``upcoming`` — future fire instants inside ``(now, window_end]``,
        computed from the live APScheduler trigger (cron) or interval
        arithmetic. Identical maths to the worker so the plotted times
        match what actually fires.
      * ``past_runs`` — recent executions inside ``[window_start, now]``
        (sparse jobs only; dense jobs get counts instead of dots).
      * ``dense`` — true for jobs too frequent to plot as discrete
        markers (rendered as a band by the frontend).
      * ``next_run_at`` — the job's next fire: ``upcoming[0]`` when one
        falls inside the window, otherwise the first fire *beyond*
        ``window_end`` so a quiet row still says when it next runs.
        ``None`` for dense and chained jobs (see
        :func:`_next_fire_beyond_window`). Computed from the live
        trigger — this is the authoritative next-run answer on the
        platform; ``scheduled_jobs`` has no stored next-run column.

    Plus a top-level ``load_buckets`` series: discrete upcoming-fire
    concurrency per hour bucket across the future window — the "are two
    heavy syncs about to collide" signal. ``dense_baseline`` reports how
    many always-on jobs run continuously underneath that curve.

    Window is ``[now - past_hours, now + future_hours]`` with
    ``window_start`` floored to the hour so the frontend's axis ticks
    land on clean boundaries. Only enabled jobs are included — the
    timeline answers "what is the platform going to run", and a disabled
    row runs nothing.
    """
    from datetime import timedelta

    from sqlalchemy import and_, case
    from sqlalchemy import func as _func

    past_hours = max(0, min(int(past_hours), 72))
    future_hours = max(1, min(int(future_hours), 168))

    now = datetime.now(UTC)
    window_start = (now - timedelta(hours=past_hours)).replace(
        minute=0, second=0, microsecond=0,
    )
    window_end = now + timedelta(hours=future_hours)

    # ── Enabled jobs (optionally category-filtered) ──────────────
    job_query = select(ScheduledJob).where(ScheduledJob.enabled.is_(True))
    if category:
        prefixes = _category_prefixes(category)
        if prefixes:
            from sqlalchemy import or_
            job_query = job_query.where(
                or_(*(ScheduledJob.handler_key.like(f"{p}%") for p in prefixes)),
            )
        elif category == "other":
            from sqlalchemy import not_
            known = [p for prefs in JOB_CATEGORIES.values() for p in prefs]
            job_query = job_query.where(
                and_(*(not_(ScheduledJob.handler_key.like(f"{p}%")) for p in known)),
            )
    jobs = (
        await session.execute(job_query.order_by(ScheduledJob.display_name))
    ).scalars().all()

    # ── Chain resolution ─────────────────────────────────────────
    # Resolve clock-triggered jobs first, then walk each chain outward
    # from its root so a member's estimate is its predecessor's fires
    # plus the predecessor's typical duration. Bounded by chain depth,
    # so a cyclic edge (only reachable by a direct DB write — the API
    # rejects it) terminates instead of spinning the forecast.
    from app.scheduled_jobs.chain import MAX_CHAIN_DEPTH

    typical_ms = await _typical_durations(session, [j.id for j in jobs])
    chain_offsets: dict[UUID, tuple[list[datetime], bool, int]] = {}
    resolved: dict[UUID, tuple[list[datetime], bool]] = {}
    for job in jobs:
        if job.trigger_type != TRIGGER_CHAINED:
            resolved[job.id] = _forecast_upcoming(
                job, now=now, window_end=window_end,
            )
    for _hop in range(MAX_CHAIN_DEPTH):
        progressed = False
        for job in jobs:
            if job.id in resolved or job.trigger_type != TRIGGER_CHAINED:
                continue
            parent_id = job.chain_after_job_id
            if parent_id is None or parent_id not in resolved:
                continue
            parent_fires, parent_dense = resolved[parent_id]
            # 🔴 The offset is the PARENT's own typical duration, not a
            # running total down the chain. ``resolved[parent_id]``
            # already carries every shift above it — a chained parent's
            # entry is its own predecessor's fires plus its own offset —
            # so adding the accumulated figure on top shifted every
            # grandchild by its grandparent's duration twice. Depth-1
            # chains were right, which is why it was invisible: the
            # three-deep cloud-IAM chains (#3467) were the first rows
            # this could be wrong for, and a forecast that is wrong only
            # past the second hop is a forecast nobody catches.
            chain_offsets[job.id] = (
                parent_fires,
                parent_dense,
                typical_ms.get(parent_id, 0),
            )
            resolved[job.id] = _forecast_upcoming(
                job, now=now, window_end=window_end,
                chain_offsets=chain_offsets,
            )
            progressed = True
        if not progressed:
            break

    # ── Per-job future fires + dense classification ──────────────
    per_job: dict[UUID, dict[str, Any]] = {}
    dense_ids: set[UUID] = set()
    for job in jobs:
        upcoming, dense = resolved.get(job.id) or ([], False)
        if dense:
            dense_ids.add(job.id)
        per_job[job.id] = {
            "job": job,
            "upcoming": upcoming,
            "dense": dense,
        }

    # ── Past-run aggregate (every selected job) ──────────────────
    selected_ids = list(per_job.keys())
    past_total: dict[UUID, int] = {}
    past_failed: dict[UUID, int] = {}
    if selected_ids:
        agg_rows = (
            await session.execute(
                select(
                    ScheduledJobRun.job_id,
                    _func.count(),
                    _func.coalesce(
                        _func.sum(
                            case(
                                (ScheduledJobRun.status == "failed", 1),
                                else_=0,
                            )
                        ),
                        0,
                    ),
                )
                .where(
                    ScheduledJobRun.job_id.in_(selected_ids),
                    ScheduledJobRun.started_at >= window_start,
                    ScheduledJobRun.started_at <= now,
                )
                .group_by(ScheduledJobRun.job_id)
            )
        ).all()
        for jid, total, failed in agg_rows:
            past_total[jid] = int(total)
            past_failed[jid] = int(failed or 0)

    # ── Past-run detail (sparse jobs only — these get plotted dots)
    past_detail: dict[UUID, list[dict[str, Any]]] = {}
    sparse_ids = [jid for jid in selected_ids if jid not in dense_ids]
    if sparse_ids:
        detail_rows = (
            await session.execute(
                select(
                    ScheduledJobRun.id,
                    ScheduledJobRun.job_id,
                    ScheduledJobRun.started_at,
                    ScheduledJobRun.status,
                    ScheduledJobRun.duration_ms,
                )
                .where(
                    ScheduledJobRun.job_id.in_(sparse_ids),
                    ScheduledJobRun.started_at >= window_start,
                    ScheduledJobRun.started_at <= now,
                )
                .order_by(ScheduledJobRun.started_at)
                .limit(FORECAST_PAST_DETAIL_CAP)
            )
        ).all()
        for run_id, jid, started_at, status, duration_ms in detail_rows:
            past_detail.setdefault(jid, []).append({
                "id": str(run_id),
                "started_at": _as_utc(started_at).isoformat(),
                "status": status,
                "duration_ms": int(duration_ms) if duration_ms is not None else None,
            })

    # ── Assemble per-job payload ─────────────────────────────────
    job_payloads: list[dict[str, Any]] = []
    from collections import Counter

    bucket_counter: Counter[str] = Counter()
    for jid, info in per_job.items():
        job: ScheduledJob = info["job"]
        upcoming: list[datetime] = info["upcoming"]
        dense: bool = info["dense"]
        next_run = (
            upcoming[0]
            if upcoming
            else _next_fire_beyond_window(job, now=now, window_end=window_end)
        )
        job_payloads.append({
            "id": str(job.id),
            "name": job.name,
            "display_name": job.display_name,
            "handler_key": job.handler_key,
            "category": _category_of(job.handler_key),
            "trigger_type": job.trigger_type,
            "interval_seconds": job.interval_seconds,
            "cron_expr": job.cron_expr,
            "timezone": job.timezone,
            "last_status": job.last_status,
            "chain_after_job_id": (
                str(job.chain_after_job_id) if job.chain_after_job_id else None
            ),
            # ``true`` when the plotted times are a chain estimate rather
            # than a clock — the frontend must not present a guess with
            # the same confidence as a cron fire.
            "chain_estimated": job.trigger_type == TRIGGER_CHAINED,
            "dense": dense,
            "upcoming": [t.isoformat() for t in upcoming],
            "past_runs": past_detail.get(jid, []),
            "past_total": past_total.get(jid, 0),
            "past_failed": past_failed.get(jid, 0),
            "next_run_at": next_run.isoformat() if next_run else None,
        })
        # Hourly concurrency over the FUTURE window — discrete fires only;
        # dense jobs are reported separately as a baseline so they don't
        # flatten the curve.
        for t in upcoming:
            bucket_key = t.replace(
                minute=0, second=0, microsecond=0,
            ).isoformat()
            bucket_counter[bucket_key] += 1

    # Continuous hourly axis for the future window so the frontend can
    # draw an unbroken bar series even across empty hours.
    load_buckets: list[dict[str, Any]] = []
    cursor = now.replace(minute=0, second=0, microsecond=0)
    while cursor <= window_end:
        key = cursor.isoformat()
        load_buckets.append({"start": key, "count": bucket_counter.get(key, 0)})
        cursor = cursor + timedelta(hours=1)

    return {
        "now": now.isoformat(),
        "window_start": window_start.isoformat(),
        "window_end": window_end.isoformat(),
        "past_hours": past_hours,
        "future_hours": future_hours,
        "dense_baseline": len(dense_ids),
        "jobs": job_payloads,
        "load_buckets": load_buckets,
    }


async def list_runs(
    session: AsyncSession, job_id: UUID, *, limit: int = 50,
) -> list[ScheduledJobRun]:
    rows = (
        await session.execute(
            select(ScheduledJobRun)
            .where(ScheduledJobRun.job_id == job_id)
            .order_by(desc(ScheduledJobRun.started_at))
            .limit(limit)
        )
    ).scalars().all()
    return list(rows)


async def list_all_runs(
    session: AsyncSession,
    *,
    page: int = 1,
    page_size: int = 50,
    status: str | None = None,
    category: str | None = None,
    keyword: str | None = None,
    triggered_by_prefix: str | None = None,
    since_hours: int | None = 24,
    include_acknowledged: bool = False,
) -> tuple[list[tuple[ScheduledJobRun, ScheduledJob]], int]:
    """Cross-job run listing for the "all recent executions" admin view.

    The admin console needs to see recent failures across every
    scheduled job on one screen — clicking into each row's drawer
    just to spot which category is broken is the friction the user
    complained about 2026-04-24. This function serves that view; it's
    read-only and admin-gated at the router layer.

    Filters:
      * ``status`` — exact match on ``run.status`` (``ok`` / ``failed``
        / ``running`` / ``skipped``).
      * ``category`` — same bucket semantics as ``list_jobs`` (joins
        on ``scheduled_jobs.handler_key`` prefixes).
      * ``keyword`` — substring match on the owning job's ``name`` /
        ``display_name`` / ``handler_key`` (same fields as
        ``list_jobs``, minus ``description`` for the same reason).
      * ``triggered_by_prefix`` — e.g. ``manual`` or ``scheduler``.
      * ``since_hours`` — only runs started within the last N hours.
        Default 24. Pass ``None`` to disable (then lean on pagination).
      * ``include_acknowledged`` — when ``False`` (default) and
        ``status='failed'``, hide rows whose ``acknowledged_at`` is
        not NULL. Other statuses are unaffected by this flag.
        Acknowledged rows still exist (we never rewrite history); they
        just don't pollute the default red queue.

    Returns ``(rows, total)`` where each row is the
    ``(ScheduledJobRun, ScheduledJob)`` pair — the router serializes
    both sides so the UI can show the job's display name without a
    second round-trip.
    """
    from datetime import datetime, timedelta

    from sqlalchemy import and_, not_, or_
    from sqlalchemy import func as _func

    base = (
        select(ScheduledJobRun, ScheduledJob)
        .join(ScheduledJob, ScheduledJob.id == ScheduledJobRun.job_id)
        .order_by(desc(ScheduledJobRun.started_at))
    )
    count_base = (
        select(_func.count())
        .select_from(ScheduledJobRun)
        .join(ScheduledJob, ScheduledJob.id == ScheduledJobRun.job_id)
    )

    conditions = []
    if status:
        conditions.append(ScheduledJobRun.status == status)
        # Hide acknowledged failed rows by default — the user-asked
        # behaviour is "an acknowledged failure should drop out of the
        # red queue without rewriting history". The flag only kicks in
        # for failed-status filter; non-failed statuses don't have ack
        # semantics.
        if status == "failed" and not include_acknowledged:
            conditions.append(ScheduledJobRun.acknowledged_at.is_(None))
    if since_hours is not None and since_hours > 0:
        cutoff = datetime.now(UTC) - timedelta(hours=since_hours)
        conditions.append(ScheduledJobRun.started_at >= cutoff)
    if triggered_by_prefix:
        conditions.append(
            ScheduledJobRun.triggered_by.like(f"{triggered_by_prefix}%"),
        )
    if keyword:
        pattern = f"%{keyword}%"
        conditions.append(or_(
            ScheduledJob.name.ilike(pattern),
            ScheduledJob.display_name.ilike(pattern),
            ScheduledJob.handler_key.ilike(pattern),
        ))
    if category:
        prefixes = _category_prefixes(category)
        if prefixes:
            conditions.append(or_(
                *(ScheduledJob.handler_key.like(f"{p}%") for p in prefixes),
            ))
        elif category == "other":
            known = [p for prefs in JOB_CATEGORIES.values() for p in prefs]
            conditions.append(and_(
                *(not_(ScheduledJob.handler_key.like(f"{p}%")) for p in known),
            ))

    if conditions:
        base = base.where(and_(*conditions))
        count_base = count_base.where(and_(*conditions))

    total = (await session.execute(count_base)).scalar() or 0
    rows = (
        await session.execute(
            base.offset((page - 1) * page_size).limit(page_size)
        )
    ).all()
    return [(r[0], r[1]) for r in rows], int(total)


async def get_run(session: AsyncSession, run_id: UUID) -> ScheduledJobRun:
    row = await session.get(ScheduledJobRun, run_id)
    if row is None:
        raise NotFoundError("ScheduledJobRun", str(run_id))
    return row


async def acknowledge_run(
    session: AsyncSession,
    run_id: UUID,
    *,
    actor_person_id: UUID,
    reason: str | None = None,
) -> ScheduledJobRun:
    """Mark a failed run as acknowledged by ``actor_person_id``.

    Refuses to ack non-failed runs (``ok`` / ``running`` / ``skipped``)
    — the operation only makes sense for the red queue. Idempotent:
    re-ack of an already-ack'd row is a no-op (returns the existing
    row unchanged); we don't reset ``acknowledged_at`` to "now"
    because the original timestamp is more useful for audit.

    The caller (router or MCP) is responsible for the audit log entry
    and ``await session.commit()``.
    """
    row = await get_run(session, run_id)
    if row.status != "failed":
        raise AppException(
            code="SCHEDULED_RUN_NOT_FAILED",
            message=(
                "Only failed runs can be acknowledged "
                f"(this run is status={row.status!r})."
            ),
            status_code=409,
        )
    if row.acknowledged_at is not None:
        return row
    row.acknowledged_at = datetime.now(UTC)
    row.acknowledged_by_id = str(actor_person_id)
    row.ack_reason = (reason or None)
    await session.flush()
    return row


# ── Admin update ─────────────────────────────────────────────────


async def update_job(
    session: AsyncSession,
    job_id: UUID,
    *,
    enabled: bool | None = None,
    interval_seconds: int | None = None,
    cron_expr: str | None = None,
    timezone: str | None = None,
    config: dict[str, Any] | None = None,
    display_name: str | None = None,
    description: str | None = None,
    chain_after_job_id: UUID | None = None,
    clear_chain: bool = False,
) -> ScheduledJob:
    row = await get_job(session, job_id)
    if enabled is not None:
        row.enabled = enabled
    if interval_seconds is not None:
        if interval_seconds < 30:
            raise AppException(
                code="INVALID_INTERVAL",
                message="interval_seconds 不能小于 30 秒,避免打爆下游。",
                status_code=400,
            )
        row.interval_seconds = interval_seconds
        row.trigger_type = "interval"
        row.cron_expr = None
    if cron_expr is not None:
        from apscheduler.triggers.cron import CronTrigger
        try:
            CronTrigger.from_crontab(cron_expr)
        except (ValueError, KeyError) as e:
            raise AppException(
                code="INVALID_CRON",
                message=f"无效 cron 表达式: {e}",
                status_code=400,
            ) from e
        row.cron_expr = cron_expr
        row.trigger_type = "cron"
        row.interval_seconds = None
    if timezone is not None:
        row.timezone = timezone
    if config is not None:
        # Defense-in-depth: refuse to repoint a cron row at a different
        # ``integration_id`` AND refuse to drop one that was
        # previously pinned. The cmdb.{billing,scopes,roles,grants}_sync
        # handler family keys off ``config['integration_id']``; an
        # admin-confused-deputy that flipped this could silently make
        # one cron iterate a different cloud account than its
        # ``display_name`` suggests, while audit attribution
        # (``last_result_summary`` rendered with the OLD integration's
        # name) still pointed at the original. Require deleting +
        # re-seeding the row through migration to change targets.
        #
        # NOTE: previously this guard fired only when the new config
        # SUPPLIED a different ``integration_id``. A PATCH that simply
        # OMITTED the field (e.g. ``{"timeout_secs": 60}``) wholesale-
        # overwrote ``row.config`` to the new dict — silently wiping
        # the pin and leaving the next handler invocation with no
        # target. Now we always preserve the previous pin: caller
        # may either match it explicitly or leave it out (we re-merge).
        prev_integration_id = (row.config or {}).get("integration_id")
        new_integration_id = config.get("integration_id")
        if prev_integration_id is not None:
            if (
                new_integration_id is not None
                and str(prev_integration_id) != str(new_integration_id)
            ):
                raise AppException(
                    code="SCHEDULED_JOB_INTEGRATION_PINNED",
                    message=(
                        f"Refusing to repoint scheduled job from integration "
                        f"{prev_integration_id} to {new_integration_id} via "
                        f"PATCH. Delete and re-seed the job via migration to "
                        f"change cloud-account targets."
                    ),
                    status_code=409,
                )
            # Re-merge the pin so a partial PATCH (no integration_id key)
            # doesn't silently wipe it.
            config = {**config, "integration_id": prev_integration_id}
        row.config = config
    if display_name is not None:
        row.display_name = display_name
    if description is not None:
        row.description = description

    # ── Chain edge ───────────────────────────────────────────────
    # ``clear_chain`` is a separate flag rather than "pass None",
    # because None is also what a PATCH that does not mention the
    # chain sends. Overloading absence with removal is how a partial
    # PATCH silently unhooks a job from its predecessor — the same
    # class of bug the integration-pin guard above exists to prevent.
    if clear_chain:
        row.chain_after_job_id = None
        if row.trigger_type == TRIGGER_CHAINED:
            # A chained row with no parent can never fire. Refuse to
            # leave it in that state silently: the caller has to give it
            # a clock back in the same request.
            raise AppException(
                code="SCHEDULED_JOB_CHAIN_ORPHANED",
                message=(
                    "解除依赖前请先给这个任务一个 cron 或 interval —— "
                    "chained 且没有上游的行永远不会触发。"
                ),
                status_code=400,
            )
    elif chain_after_job_id is not None:
        parent = await session.get(ScheduledJob, chain_after_job_id)
        if parent is None:
            raise NotFoundError("ScheduledJob", str(chain_after_job_id))
        parent_map = await load_parent_map(session)
        assert_no_cycle(
            job_id=job_id,
            new_parent_id=chain_after_job_id,
            parent_map=parent_map,
        )
        row.chain_after_job_id = chain_after_job_id
        # Chaining a job REPLACES its clock. Leaving the old cron in
        # place would give it two independent triggers, which is not a
        # richer schedule — it is a job that runs twice per cycle, once
        # in order and once not.
        row.trigger_type = TRIGGER_CHAINED
        row.cron_expr = None
        row.interval_seconds = None

    await session.flush()
    return row


# ── Execution wrapper — the single entry point for runs ──────────


# The single-flight lock must OUTLIVE the handler's own hard timeout.
# The lock is taken with ``auto_renew=True`` (2026-09-05) so a renewal
# watchdog keeps it alive for the whole critical section, but the TTL is
# still sized generously because renewal is best-effort: a handler that
# legitimately runs longer than the lock TTL while renewals are failing
# would release its lock mid-run and let a concurrent tick / second
# replica / manual "trigger now" enter the SAME handler — the exact double-run the single-flight lock exists to
# prevent (2× vendor API quota, 2× Feishu DMs for seat-sync). We therefore
# size the lock TTL to the job's own timeout plus a write-back buffer.
_LOCK_TTL_BUFFER_SECS = 300


# ── Process-global scheduled-job concurrency gate ─────────────────────
# APScheduler's AsyncIOExecutor launches every due cron as a bare
# ``loop.create_task`` — ``max_instances=1`` bounds a SINGLE job's
# self-overlap but nothing bounds how many DIFFERENT jobs run at once.
# Every running handler holds >=1 pooled DB connection for its whole run
# (``_execute_job_locked`` keeps ``work_session`` open across ``handler.run``,
# and handlers open further nested sessions), so an unbounded fan-out of
# clustered crons drains the 30-connection pool and every other DB consumer
# times out with "QueuePool limit ... reached" — the 2026-07-19 incident.
#
# This semaphore caps concurrent handler executions per process at
# ``settings.SCHEDULED_JOB_MAX_CONCURRENCY`` (default 8, always < the pool
# ceiling), making pool exhaustion by the scheduler structurally impossible:
# no matter how many crons come due in the same minute, at most N run at once
# and the rest await a slot (their APScheduler task is already alive, so they
# do not misfire). It gates EVERY entry into ``_execute_job_locked``
# regardless of trigger source (scheduler tick, boot-time overdue seed,
# admin "run now", or the cost_facts month-close drain), which is safe
# because NO handler awaits ``execute_job`` in-band — the only in-process
# fan-out dispatcher (``run_cost_facts_monthclose``) detaches its drain via
# ``background.spawn`` and returns immediately, so a gated execution never
# waits on another gated execution (no self-deadlock).
#
# Keyed by the running event loop so the singleton binds correctly under
# pytest's function-scoped loops (and stays a single entry in the one
# long-lived loop of the worker / API process).
_JOB_CONCURRENCY_GATES: dict[asyncio.AbstractEventLoop, asyncio.Semaphore] = {}

# Heavy sub-gate (2026-07-19 follow-up): ``weight="heavy"`` handlers
# (multi-minute cloud vendor sweeps) are additionally bounded by this
# smaller semaphore so a nightly cluster of cloud syncs can occupy at
# most ``SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY`` of the global slots —
# fast critical crons (provisioning dispatch every 30s, identity_sync
# dispatch every 60s, outbox retry) always have global slots they can
# never be starved out of. Acquisition order is STRICT: heavy gate
# first, THEN the global gate. A heavy job queueing on the heavy gate
# holds no global slot, so light jobs flow freely past it; and because
# every holder of the heavy gate only ever waits on the global gate
# (never the reverse), the two-level acquisition cannot deadlock.
_JOB_HEAVY_GATES: dict[asyncio.AbstractEventLoop, asyncio.Semaphore] = {}


def _job_concurrency_gate() -> asyncio.Semaphore:
    """The per-loop semaphore bounding concurrent handler executions."""
    loop = asyncio.get_running_loop()
    gate = _JOB_CONCURRENCY_GATES.get(loop)
    if gate is None:
        gate = asyncio.Semaphore(max(1, int(settings.SCHEDULED_JOB_MAX_CONCURRENCY)))
        _JOB_CONCURRENCY_GATES[loop] = gate
    return gate


def _job_heavy_gate() -> asyncio.Semaphore:
    """The per-loop sub-semaphore bounding concurrent HEAVY handlers.

    Sized ``min(HEAVY, MAX)`` so a misconfigured heavy limit can never
    exceed the global gate (which would make the sub-gate a no-op).
    """
    loop = asyncio.get_running_loop()
    gate = _JOB_HEAVY_GATES.get(loop)
    if gate is None:
        size = min(
            max(1, int(settings.SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY)),
            max(1, int(settings.SCHEDULED_JOB_MAX_CONCURRENCY)),
        )
        gate = asyncio.Semaphore(size)
        _JOB_HEAVY_GATES[loop] = gate
    return gate


@asynccontextmanager
async def _execution_slot(handler) -> AsyncIterator[int]:
    """Acquire the concurrency slot(s) for one handler execution.

    Yields the total milliseconds spent waiting for the slot(s) — 0 on
    the uncontended path — so the caller can surface queueing on the
    run row (``_gate_wait_ms``) and admins can see "the job was late
    because the gate was full" instead of guessing.
    """
    heavy = getattr(handler, "weight", "light") == "heavy"
    started = asyncio.get_running_loop().time()
    gate = _job_concurrency_gate()
    if heavy:
        async with _job_heavy_gate():
            async with gate:
                yield int((asyncio.get_running_loop().time() - started) * 1000)
    else:
        async with gate:
            yield int((asyncio.get_running_loop().time() - started) * 1000)


def _run_status_for_summary(result_summary: Any) -> str:
    """一次「没抛异常」的运行,该记 ``ok`` 还是 ``skipped``。

    Handler 返回 ``{"status": "skipped", ...}`` 意味着它**没有做事** ——
    撞上了别的锁、集成被停用、窗口内没有数据。把它记成 ``ok`` 会让执行历史
    说谎,而且这个谎是**承重的**:调用方(月结的 drain、运维、回填脚本)看见
    ``ok`` 就认为那段窗口已经同步过了。

    2026-08-16 实撞:一次 cost_facts 的补数连报两次 ``ok`` 却一行没写 ——
    内层 ``cloud_sync:cost_facts`` 锁还被一个被杀的运行占着。当时生产 30 天内
    **9,826 次「成功」里有 754 次(7.7%)是披着 ok 外衣的 skip**,其中 289 次
    是 ``cmdb.cost_facts_sync``。

    **只认顶层 status。** 那 754 次里有 435 次是*嵌套*的 skipped(例如可选的
    ``profile_sync`` 子步骤)挂在一次真正成功的运行上 —— 用子串匹配或深度搜索
    会把这 435 次正常成功一并降级,那比原来的问题更糟。

    ``failed`` / ``partial_failure`` 故意**不**在这里映射:本仓库的约定是
    handler 失败要 **raise**(这样 ``maybe_notify_failure`` 才会发告警);而且
    生产上顶层 failed 形状的返回是 **0 条**,凭空接上这条路只会新增一条没有
    实际产出者的告警通道。
    """
    if (
        isinstance(result_summary, dict)
        and result_summary.get("status") == "skipped"
    ):
        return "skipped"
    return "ok"


def _resolve_job_timeout_secs(job_config: dict[str, Any] | None) -> float:
    """Per-job hard timeout in seconds, clamped to ``[60, 7200]``.

    Shared by ``execute_job`` (to size the single-flight lock TTL) and
    ``_execute_job_locked`` (to arm ``asyncio.wait_for``) so the lock TTL
    and the handler timeout can never drift apart — the lock must always
    outlive the handler it guards.
    """
    raw = (job_config or {}).get("timeout_secs", 1800)
    try:
        secs = float(raw)
    except (TypeError, ValueError):
        secs = 1800.0
    return max(60.0, min(7200.0, secs))


# ── Transient-infrastructure retry (2026-07-19 follow-up) ─────────────
# One bounded in-slot retry for failures the NEXT attempt is genuinely
# likely to survive: DB pool checkout timeout (sqlalchemy.exc.TimeoutError
# — brief pool contention), lock-wait/deadlock (1205/1213/1305 — InnoDB
# says "restart transaction"), and connection drop (2006/2013 — TiDB node
# recycle). Every handler is required to be idempotent (the framework's
# single-flight lock + fresh-session lock-retry helpers already rely on
# that), so a clean re-run from current DB state is safe. Anything else
# (bad SQL, vendor 4xx, logic error) fails immediately — retry can't fix
# it and would just double vendor API quota burn.
#
# Budget guard: retry is offered ONLY when the first attempt died within
# ``_TRANSIENT_RETRY_MAX_ELAPSED_SECS`` — a sweep that ran 20 minutes
# before hitting a deadlock should surface as a failure (and re-run on
# its next tick) rather than doubling a near-timeout run. The worst-case
# extra wall time (MAX_ELAPSED + DELAY = 130s) stays well under the
# single-flight lock's ``_LOCK_TTL_BUFFER_SECS`` (300s) headroom, so a
# retried run can never outlive its lock.
_TRANSIENT_RETRY_MAX_ELAPSED_SECS = 120.0
_TRANSIENT_RETRY_DELAY_SECS = 10.0
assert (
    _TRANSIENT_RETRY_MAX_ELAPSED_SECS + _TRANSIENT_RETRY_DELAY_SECS
    < _LOCK_TTL_BUFFER_SECS
), "transient-retry budget must fit inside the lock-TTL buffer"


# ── Lock-acquire resilience (2026-08-13 incident) ─────────────────────
# Attempts at the single-flight lock ACQUIRE itself — distinct from the
# in-slot handler retry above, which only covers failures AFTER the lock
# is held. Four unrelated jobs (alicloud_intl.utilization,
# data_access.bigquery_apply_reconcile, provisioning.dispatch,
# onboarding.reconcile) failed inside ten minutes, every one of them on
# ``TimeoutError: Timeout connecting to server`` raised by the
# ``redis.set(nx=True)`` inside ``distributed_lock`` — one shared Redis
# reachability blip, not four bugs. ``core.redis_client`` now retries at
# the command level (reconnecting between attempts), which absorbs a
# reaped idle socket or a fast failover; this outer attempt covers a blip
# that outlasts that budget. Losing the acquire costs a whole tick and
# pages an admin, so one more try two seconds later is cheap. Strictly
# bounded at 2 — a genuine outage must still fail closed rather than run
# the handler without the single-flight guard.
_LOCK_ACQUIRE_ATTEMPTS = 2
_LOCK_ACQUIRE_RETRY_DELAY_SECS = 2.0


def single_flight_lock_key(job_id: UUID, lock_epoch: str | None) -> str:
    """The Redis key that serialises executions of one job row.

    Factored out of ``execute_job`` because the stale-run reaper needs
    to ask the *same* question from the other side — "is anybody
    actually inside this job's critical section right now?" — and a
    second hand-rolled copy of this key shape would be a silent
    divergence: the reaper would look up a key nobody holds and declare
    every live run orphaned.

    Keyed by job **id**, not handler key: nine ``billing_daily`` rows
    against nine different cloud accounts are independent, and a
    handler-keyed lock made eight of them skip on every boot backfill.
    A deliberate ``lock_epoch`` in the row's config appends a suffix so
    a deployment can abandon an overlong lease left by a killed process;
    the suffix is restricted to a short machine-safe token so an admin
    typo cannot mint unbounded Redis key shapes.
    """
    epoch = (lock_epoch or "").strip()
    if epoch and len(epoch) <= 64 and all(
        char.isalnum() or char in "-_" for char in epoch
    ):
        return f"sched_job:{job_id}:{epoch}"
    return f"sched_job:{job_id}"


def _is_transient_infra_error(exc: BaseException) -> bool:
    """True for infrastructure blips a fresh-session re-run survives."""
    from redis.exceptions import ConnectionError as RedisConnectionError
    from redis.exceptions import TimeoutError as RedisTimeoutError
    from sqlalchemy.exc import TimeoutError as SAPoolTimeoutError

    return (
        isinstance(exc, SAPoolTimeoutError)
        # Redis reachability blip — same class of event as a DB pool
        # checkout timeout: nothing about the handler is wrong, the next
        # attempt against a reconnected pool succeeds. Added 2026-08-13
        # after four unrelated jobs died on
        # ``TimeoutError: Timeout connecting to server``. The client-level
        # retry in ``core.redis_client`` absorbs most of these before they
        # ever reach here; this covers the residue (an outage that outlasts
        # the client's retry budget but is over by the time the in-slot
        # retry fires) and any handler that touches Redis directly.
        or isinstance(exc, RedisConnectionError | RedisTimeoutError)
        or is_transaction_aborting_db_error(exc)
        or is_connection_lost_db_error(exc)
    )


async def execute_job(
    job_id: UUID,
    *,
    triggered_by: str,
    config_override: dict[str, Any] | None = None,
    bypass_concurrency_gate: bool = False,
) -> dict[str, Any]:
    """Run a scheduled job by id, record the result, return a summary.

    Opens its own session — the caller (APScheduler or "run now"
    request) must NOT hold a transaction open while we do the work
    because it can be multi-minute.

    Single-flight: a Redis-backed advisory lock keyed by handler key
    serializes concurrent invocations. Without this guard, two parallel
    paths into ``execute_job`` (admin clicks "trigger now" twice,
    APScheduler tick fires while a manual run is mid-flight, two
    replicas dispatching the same cron job) would both enter the
    handler. For seat-sync handlers, that produces double-INSERT of
    active seat rows (the unique partial index added in migration 119
    catches THIS specific corruption, but vendor-side API quota is
    still burnt 2× and Feishu DM notifications go out 2×). For the
    provisioning dispatch worker it produces double driver calls. The
    lock is named ``helix:lock:sched_job:{handler_key}`` so concurrent
    runs of DIFFERENT jobs proceed in parallel — only same-handler
    overlap is blocked. ``LockBusy`` -> write a ``skipped`` run row
    and return.
    """
    from app.core.distributed_lock import LockBusy, distributed_lock

    # We need the handler_key + job-row state BEFORE the lock acquire
    # — for the "no such handler" / "disabled" early-returns to land
    # before we burn a lock attempt.
    async with async_session_factory() as session:
        job = await session.get(ScheduledJob, job_id)
        if job is None:
            raise NotFoundError("ScheduledJob", str(job_id))
        handler = get_handler(job.handler_key)
        if handler is None:
            error_message = (
                f"Handler {job.handler_key!r} 未注册 — "
                "可能是代码里的 register_scheduled_job 名字改了。"
            )
            logger.error(error_message)
            run_row = ScheduledJobRun(
                job_id=job.id,
                started_at=datetime.now(UTC),
                finished_at=datetime.now(UTC),
                duration_ms=0,
                status="failed",
                triggered_by=triggered_by,
                error_message=error_message,
            )
            session.add(run_row)
            job.last_status = "failed"
            job.last_error = error_message
            job.last_run_at = run_row.started_at
            # Round-62 follow-up: bump BOTH ``run_count`` and
            # ``failure_count`` to preserve the metrics invariant
            # ``run_count >= failure_count`` that
            # ``helix_scheduled_job_runs_total{status='ok'}`` =
            # ``run_count - failure_count`` depends on. The normal
            # terminal write-back below bumps both together; this
            # handler-vanished branch was the lone site that bumped
            # only failure_count, and under a rename-deploy the
            # ``ok`` counter would clamp to 0 (max(...,0) defence) —
            # a positive→0 transition Prometheus interprets as a
            # counter reset, NaN-ing ``rate()`` and firing every
            # downstream alert. Handler-vanished IS a real attempt
            # (consumes a tick + writes a failed run row), so the
            # accounting is also semantically correct.
            job.run_count = (job.run_count or 0) + 1
            job.failure_count = (job.failure_count or 0) + 1
            handler_key = job.handler_key
            await session.commit()
            # Notify on first sighting of "handler vanished" — same
            # dedup gate as a regular failure so a deploy with a
            # renamed handler doesn't spam admins on every cron tick.
            from app.scheduled_jobs.failure_notifier import maybe_notify_failure
            await maybe_notify_failure(
                job_id=job_id,
                handler_key=handler_key,
                error_message=error_message,
                triggered_by=triggered_by,
            )
            return {"status": "failed", "error": error_message}

        if not job.enabled:
            logger.info("scheduled job %s is disabled — skipping", job.name)
            return {"status": "skipped", "reason": "disabled"}

        # Integration pre-gate (2026-09-05). A row bound to a disabled or
        # deleted integration can do no work — every per-scope cloud
        # handler already discovers that and returns ``skipped``, but it
        # does so from INSIDE the handler, after the run has taken one of
        # the four ``heavy`` slots. Prod carries 35 such rows across 5
        # disabled integrations, all firing in the same pre-dawn hours
        # where live sweeps are already queueing behind the gate. Deciding
        # here costs one indexed read on a session we do not hold open,
        # and the handler-level checks stay in place as defence in depth
        # (this must remain an optimisation, never the only guard).
        gate_state = await integration_gate_state(session, job)
        if not gate_state.runnable:
            reason = gate_state.reason or "integration_unavailable"
            summary = {
                "status": "skipped",
                "reason": reason,
                "integration_id": gate_state.integration_id,
            }
            logger.info(
                "scheduled job %s skipped before the gate: %s (integration=%s)",
                job.name, reason, gate_state.integration_id,
            )
            # Record the TRANSITION only. A permanently-off integration
            # would otherwise write one identical row per tick forever,
            # and ``MAX_RUN_HISTORY_PER_JOB`` would evict the job's real
            # history within days — destroying the very evidence the
            # first row exists to preserve.
            #
            # ``last_run_at`` deliberately does NOT move on a suppressed
            # skip. It is tempting — it would make ``chain_advance``'s
            # overdue predicate self-clear for free — but the compliance
            # posture gate reads this exact column to decide whether a
            # control's signal is fresh
            # (``auditor/compliance_service.py::freshness``). A job that
            # advertised a fresh clock while doing no work would show a
            # control green on data nobody is collecting. That is the
            # failure #2217 spent real effort closing, and it is not
            # worth re-opening to save the reconciler a query: the
            # reconciler learned to exclude these rows instead (see
            # ``chain.plan_advances``).
            if should_record_gate_skip(job, reason):
                now_skip = datetime.now(UTC)
                session.add(
                    ScheduledJobRun(
                        job_id=job.id,
                        started_at=now_skip,
                        finished_at=now_skip,
                        duration_ms=0,
                        status="skipped",
                        triggered_by=triggered_by,
                        result_summary=_ensure_json_safe(summary),
                    ),
                )
                job.last_run_at = now_skip
                job.last_status = "skipped"
                job.last_duration_ms = 0
                job.last_error = None
                job.last_result_summary = _ensure_json_safe(summary)
                job.run_count = (job.run_count or 0) + 1
                await session.commit()
            return summary

        handler_key = job.handler_key
        # Size the single-flight lock to this job's hard timeout (+ buffer)
        # so a long-but-legitimate run can't outlive its lock. Resolved here
        # (while the job row is loaded) using the SAME helper the inner
        # executor uses to arm ``asyncio.wait_for``.
        lock_ttl = int(
            _resolve_job_timeout_secs(
                {**(job.config or {}), **(config_override or {})}
            )
        ) + _LOCK_TTL_BUFFER_SECS
        lock_epoch = str((job.config or {}).get("lock_epoch") or "").strip()

    # Acquire the single-flight lock keyed by JOB ID (not handler_key).
    # Using handler_key caused all 9 billing_daily integrations to
    # serialise on ONE lock: when the overdue-detect fires them all at
    # restart, 8 of the 9 skipped with LockBusy even though they are
    # completely independent (each has a different integration_id and
    # the inner per-integration lock already prevents double-runs for
    # the same scope). Per-job-id lock means concurrent runs of
    # cmdb.billing_daily.aws-sg and cmdb.billing_daily.aliyun-primary
    # proceed in parallel — same as they do in steady state, where each
    # fires at a different cron minute. TTL is generous — handler runs
    # can take several minutes — but bounded so a wedged process
    # auto-releases on Redis expiry. The lock spans the entire executor
    # including write-back; releasing only after _trim_runs commits
    # guarantees the next concurrent caller sees the new run row.
    # A deployment may deliberately advance a job's lock epoch after a
    # process was killed while holding an old, overlong lease. This is not a
    # general bypass: both the job ID and the configured epoch remain part of
    # the key, and normal jobs keep the stable key. Restrict the suffix to a
    # small machine-safe token so an admin config typo cannot create unbounded
    # Redis key shapes.
    lock_key = single_flight_lock_key(job_id, lock_epoch)
    for _lock_attempt in range(_LOCK_ACQUIRE_ATTEMPTS):
        # Sentinel so the outer except can distinguish "lock acquire
        # itself failed (Redis down)" from "lock acquired fine but the
        # inner _execute_job_locked raised" — the former needs a failed
        # run row written here; the latter has already written its own
        # success/failed row from inside _execute_job_locked. Round-4
        # verification P2 (2026-05-05). Reset on every attempt.
        lock_acquired_marker = False
        try:
            # Note: cannot use ``async with`` outer-scope across multiple
            # sessions because ``execute_job`` returns inside the body;
            # the asynccontextmanager handles the release on normal exit
            # and on raise. We DO want the early-return ("disabled") path
            # not to acquire the lock — that's why the lock-acquire is
            # placed after the disabled check above.
            async with distributed_lock(
                lock_key, ttl=lock_ttl, auto_renew=True,
            ):
                lock_acquired_marker = True
                # Bound concurrent connection-holding executions per process
                # so a cluster of due crons can't drain the DB pool
                # (2026-07-19 incident — see _job_concurrency_gate). Acquired
                # INSIDE the single-flight lock so LockBusy skips and the
                # "disabled" / "handler vanished" early-returns above never
                # consume a slot; only a real handler run does. Waiting here
                # does not eat the handler's own timeout budget — the
                # ``asyncio.wait_for`` clock inside _execute_job_locked
                # starts only after the slot is held. Heavy handlers
                # additionally pass the heavy sub-gate first — see
                # _job_heavy_gate for the starvation-protection rationale.
                # The one-shot cloud-baseline seed is the sole caller allowed
                # to bypass this queue. It has no vendor side effect, remains
                # protected by this job's single-flight lock and hard timeout,
                # and prevents a fresh subscription page being blank simply
                # because ordinary cloud sweeps occupy every shared slot.
                if bypass_concurrency_gate:
                    return await _execute_job_locked(
                        job_id=job_id,
                        handler=handler,
                        triggered_by=triggered_by,
                        config_override=config_override,
                        gate_wait_ms=0,
                    )
                async with _execution_slot(handler) as gate_wait_ms:
                    if gate_wait_ms > 5_000:
                        logger.info(
                            "scheduled job %s waited %dms for a concurrency "
                            "slot (gate=%d heavy=%d weight=%s)",
                            handler_key, gate_wait_ms,
                            settings.SCHEDULED_JOB_MAX_CONCURRENCY,
                            settings.SCHEDULED_JOB_HEAVY_MAX_CONCURRENCY,
                            getattr(handler, "weight", "light"),
                        )
                    return await _execute_job_locked(
                        job_id=job_id,
                        handler=handler,
                        triggered_by=triggered_by,
                        config_override=config_override,
                        gate_wait_ms=gate_wait_ms,
                    )
        except LockBusy:
            # Another worker is running the same handler. Write a skipped
            # row so admins can see the collision in /admin/scheduled-jobs
            # rather than a silent gap.
            async with async_session_factory() as session:
                now = datetime.now(UTC)
                run_row = ScheduledJobRun(
                    job_id=job_id,
                    started_at=now,
                    finished_at=now,
                    duration_ms=0,
                    status="skipped",
                    triggered_by=triggered_by,
                    error_message=(
                        f"Skipped: another execution of {handler_key!r} is "
                        "in progress (single-flight lock busy)."
                    ),
                )
                session.add(run_row)
                await session.commit()
            return {"status": "skipped", "reason": "single_flight_lock_busy"}
        except Exception as exc:  # noqa: BLE001
            if lock_acquired_marker:
                # Lock acquired fine; failure originated INSIDE the
                # inner executor (run-row insert, write-back, etc.).
                # _execute_job_locked already wrote a status='failed'
                # row in its own write-back path before raising — so
                # we must NOT write a second misleading "lock
                # infrastructure" row. Re-raise as-is.
                raise
            # The acquire itself raised. If it looks like a Redis
            # reachability blip and we still have an attempt left, sleep
            # briefly and try again BEFORE declaring the tick failed —
            # see _LOCK_ACQUIRE_ATTEMPTS. No run row is written for a
            # retried attempt: the tick has not failed yet, and writing
            # one would make a self-healing blip look like a failure in
            # /admin/scheduled-jobs.
            if (
                _lock_attempt + 1 < _LOCK_ACQUIRE_ATTEMPTS
                and _is_transient_infra_error(exc)
            ):
                logger.warning(
                    "execute_job %s: transient lock-acquire failure "
                    "(%s) — retrying in %.1fs",
                    handler_key,
                    format_exception_for_log(exc, max_len=200),
                    _LOCK_ACQUIRE_RETRY_DELAY_SECS,
                )
                await asyncio.sleep(_LOCK_ACQUIRE_RETRY_DELAY_SECS)
                continue
            # Redis / lock-infrastructure failure — the lock-acquire
            # itself raised (RedisConnectionError, network blip).
            # Without this branch the bare exception propagates to the
            # APScheduler caller and no run row is written, leaving a
            # silent gap in scheduled_job_runs history. Round-3 logic-
            # scan P2: write a failed run row before re-raising so ops
            # sees the infrastructure failure in /admin/scheduled-jobs.
            # We re-raise (not swallow) because the next tick should
            # retry — failing closed is correct here vs. running without
            # the single-flight guard.
            logger.exception(
                "execute_job %s: lock infrastructure error", handler_key,
            )
            infra_error = (
                f"Lock infrastructure error for {handler_key!r}: "
                f"{format_exception_for_log(exc, max_len=512)}"
            )
            async with async_session_factory() as session:
                now = datetime.now(UTC)
                run_row = ScheduledJobRun(
                    job_id=job_id,
                    started_at=now,
                    finished_at=now,
                    duration_ms=0,
                    status="failed",
                    triggered_by=triggered_by,
                    error_message=infra_error,
                )
                session.add(run_row)
                await session.commit()
            # Notify on lock-infra failures with the same dedup gate. The
            # signature for ``RedisConnectionError`` collapses across
            # consecutive ticks, so a sustained Redis outage produces ONE
            # admin alert rather than one per tick. Best-effort.
            from app.scheduled_jobs.failure_notifier import (
                maybe_notify_failure,
            )
            await maybe_notify_failure(
                job_id=job_id,
                handler_key=handler_key,
                error_message=infra_error,
                triggered_by=triggered_by,
            )
            raise
    # Unreachable: every path inside the loop either returns, re-raises or
    # ``continue``s, and the last attempt can never ``continue``. Present so
    # the function has no implicit ``None`` return if that ever changes.
    raise AssertionError("execute_job lock-acquire loop exited without result")


async def _execute_job_locked(
    *,
    job_id: UUID,
    handler,
    triggered_by: str,
    config_override: dict[str, Any] | None = None,
    gate_wait_ms: int = 0,
) -> dict[str, Any]:
    """Inner body of ``execute_job`` after the single-flight lock is
    acquired. Split out only so the lock contextmanager wraps the
    full lifecycle of the run, including write-back.

    ``gate_wait_ms`` — time the run queued on the concurrency gate
    before this call; surfaced on the run row's ``result_summary`` so
    prolonged queueing is visible in the admin run history.
    """
    async with async_session_factory() as session:
        job = await session.get(ScheduledJob, job_id)
        if job is None:
            raise NotFoundError("ScheduledJob", str(job_id))

        # Insert a "running" row first so a crashed worker leaves
        # evidence rather than a silent gap in history.
        run_row = ScheduledJobRun(
            job_id=job.id,
            started_at=datetime.now(UTC),
            status="running",
            triggered_by=triggered_by,
        )
        session.add(run_row)
        await session.flush()
        run_id = run_row.id
        run_started_at = run_row.started_at
        job_config = {**(job.config or {}), **(config_override or {})}
        await session.commit()

    # Execute outside the session that holds the ``running`` row —
    # the handler opens its own session(s) as needed. This avoids
    # long-held transactions and keeps the "running" marker visible
    # to other readers.
    started = datetime.now(UTC)
    error_message: str | None = None
    result_summary: dict[str, Any] | None = None
    # Pre-launch audit Round-29 fix: per-job hard timeout so a wedged
    # handler (DB deadlock, slow vendor API with no client timeout,
    # infinite paginator loop) can't pin a worker thread forever.
    # 30-minute cap is generous — the slowest handler in practice
    # (cmdb.billing_daily_sync on an 18-RD-member Aliyun tenant) takes
    # ~6 min; anything past 30 min is wedged. Configurable per-job via
    # ``config['timeout_secs']`` (clamped to [60, 7200]) so a future
    # long-running handler can extend without code change. APScheduler's
    # ``max_instances=1`` + ``misfire_grace_time=900`` prevent re-fire
    # but don't reclaim a hung handler; this is the missing reclaim path.
    timeout_secs = _resolve_job_timeout_secs(job_config)
    # Attempt loop: at most 2 attempts, and only when the first died of
    # a classified transient-infrastructure error within the elapsed
    # budget — see _is_transient_infra_error for the rationale. Each
    # attempt runs on a brand-new session (the failed one may hold a
    # rolled-back transaction or a dead connection).
    attempt = 0
    transient_retried = False
    while True:
        attempt += 1
        attempt_started = asyncio.get_running_loop().time()
        try:
            async with async_session_factory() as work_session:
                result_summary = await asyncio.wait_for(
                    handler.run(work_session, job_config),
                    timeout=timeout_secs,
                )
                await work_session.commit()
            status = _run_status_for_summary(result_summary)
        except asyncio.CancelledError:
            # The process is going away mid-run: the worker's graceful
            # drain cancels stragglers after its eight-second window so
            # they unwind while the event loop and default executor are
            # still alive. This is NOT a handler failure, and it is not
            # rare either — production deploys ~15 times a day, and every
            # heavy cloud sweep in flight at that moment gets cancelled.
            #
            # Before this branch the drain deliberately wrote nothing,
            # which left the row at ``running`` until the six-hour reaper
            # stamped it ``failed`` — 386 of 452 scheduler "failures" in
            # a 30-day production window, every one of them already
            # re-run successfully by the boot backfill minutes later.
            # Record what actually happened, then re-raise so the
            # cancellation propagates exactly as asyncio expects.
            await _mark_run_interrupted(
                run_id,
                started_at=started,
                reason=(
                    "interrupted: worker shutdown drain cancelled the run"
                    if is_draining()
                    else "interrupted: the run was cancelled mid-flight - "
                    "the single-flight lock was lost while it was held"
                ),
            )
            raise
        except TimeoutError:
            # Builtin TimeoutError from asyncio.wait_for — the per-job
            # hard timeout. NOT retryable: the handler consumed its whole
            # budget once already. (The DB pool's checkout timeout raises
            # sqlalchemy.exc.TimeoutError, which is NOT a builtin
            # TimeoutError subclass and lands in the branch below.)
            logger.error(
                "scheduled job %s timed out after %.0fs",
                handler.key, timeout_secs,
            )
            error_message = (
                f"Handler {handler.key!r} timed out after {timeout_secs:.0f}s "
                f"(configurable via config['timeout_secs'], clamped to "
                f"[60, 7200])."
            )
            status = "failed"
        except Exception as exc:  # noqa: BLE001
            elapsed = asyncio.get_running_loop().time() - attempt_started
            if (
                attempt == 1
                and _is_transient_infra_error(exc)
                and elapsed <= _TRANSIENT_RETRY_MAX_ELAPSED_SECS
            ):
                logger.warning(
                    "scheduled job %s hit transient infra error after "
                    "%.1fs (%s) — retrying once on a fresh session in %.0fs",
                    handler.key, elapsed, type(exc).__name__,
                    _TRANSIENT_RETRY_DELAY_SECS,
                )
                transient_retried = True
                await asyncio.sleep(_TRANSIENT_RETRY_DELAY_SECS)
                continue
            logger.exception("scheduled job %s failed", handler.key)
            # Pre-launch P1 fix (Round 45 follow-up, CR-211 + Round 46
            # merge): use canonical ``format_exception_for_log`` so
            # AppException / httpx transport raises with empty ``str()``
            # still get class + repr fallback, AND scrub credential
            # patterns before persisting (cloud SDK exceptions can echo
            # access keys / session tokens / signed URLs). Truncate to
            # 4096 AFTER scrub so redaction markers don't push the tail
            # off the column.
            from app.integrations.service import _scrub_message  # noqa: PLC0415
            error_message = _scrub_message(
                format_exception_for_log(exc, max_len=4096)
            )[:4096]
            status = "failed"
        break

    # Surface queueing + retry forensics on the run row without touching
    # any handler's own summary keys (underscore-prefixed = framework).
    if isinstance(result_summary, dict):
        if gate_wait_ms > 0:
            result_summary = {**result_summary, "_gate_wait_ms": gate_wait_ms}
        if transient_retried:
            result_summary = {**result_summary, "_transient_retry": 1}

    finished = datetime.now(UTC)
    duration_ms = int((finished - started).total_seconds() * 1000)

    # Write-back in a fresh session so transactional state from the
    # handler doesn't leak.
    async with async_session_factory() as session:
        run_row = await session.get(ScheduledJobRun, run_id)
        if run_row is not None:
            run_row.finished_at = finished
            run_row.duration_ms = duration_ms
            run_row.status = status
            run_row.error_message = error_message
            run_row.result_summary = _ensure_json_safe(result_summary)
        job = await session.get(ScheduledJob, job_id)
        if job is not None:
            job.last_run_at = run_started_at
            job.last_status = status
            job.last_duration_ms = duration_ms
            job.last_error = error_message
            job.last_result_summary = _ensure_json_safe(result_summary)
            job.run_count = (job.run_count or 0) + 1
            if status == "failed":
                job.failure_count = (job.failure_count or 0) + 1
        await session.commit()

        # Trim history — keep only the most recent MAX rows per job.
        await _trim_runs(session, job_id)
        await session.commit()

    # Failure notification — dedup-aware, best-effort, runs in its own
    # session. We deliberately do NOT raise on notification failures;
    # the run row is the source of truth and a notification glitch
    # must not be allowed to corrupt the executor's contract with
    # APScheduler / "trigger now" callers.
    if status == "failed":
        from app.scheduled_jobs.failure_notifier import maybe_notify_failure
        await maybe_notify_failure(
            job_id=job_id,
            handler_key=handler.key,
            error_message=error_message,
            triggered_by=triggered_by,
        )

    # Release any chained successors. This is the FAST path only — the
    # ``scheduler.chain_advance`` reconciler is what guarantees the chain
    # moves at all, so a failure here costs latency, never correctness,
    # and must not be allowed to turn a finished run into a failed one.
    chained: list[str] = []
    if status in CHAIN_ADVANCING_STATUSES:
        try:
            async with async_session_factory() as chain_session:
                chained = await spawn_children(
                    chain_session,
                    parent_id=job_id,
                    parent_status=status,
                    parent_run_id=run_id,
                )
        except Exception:  # noqa: BLE001 - reconciler is the guarantee
            logger.warning(
                "could not spawn chained successors of %s; the "
                "chain_advance reconciler will pick them up",
                handler.key, exc_info=True,
            )
    if chained:
        logger.info(
            "job %s released %d chained successor(s): %s",
            handler.key, len(chained), ", ".join(chained),
        )

    return {
        "status": status,
        "duration_ms": duration_ms,
        "result_summary": result_summary,
        "error": error_message,
    }


async def _mark_run_interrupted(
    run_id: UUID, *, started_at: datetime, reason: str,
) -> None:
    """Close an in-flight run row as ``interrupted``, best-effort.

    Runs on a **fresh, shielded, time-boxed** session. Every one of those
    three matters, and all three are about the same fact: this executes
    while the task is already cancelled and the process is on its way
    out.

    * *Fresh* — the handler's own session may be mid-transaction or
      poisoned by the cancellation.
    * *Time-boxed* — shutdown must never hang on this. Three seconds is
      far more than a single indexed UPDATE needs.
    * *Best-effort* — anything at all going wrong here, including a
      second cancellation, is swallowed. It is safe to give up because
      losing the write is not losing the fact: the row stays ``running``
      and the lock-liveness reaper closes it on its next hourly tick.
      That is why this is deliberately **not** ``asyncio.shield``-ed —
      a shielded write that outlives ``wait_for`` becomes a detached
      task the loop teardown reports as "Task was destroyed but it is
      pending", which is a scary log line bought for a case the reaper
      already covers.

    Deliberately does **not** touch ``run_count`` / ``failure_count`` /
    ``last_status`` on the job row — see ``RUN_STATUS_INTERRUPTED`` in
    ``models.py`` for why those columns must keep describing completed
    attempts only.
    """

    async def _write() -> None:
        finished = datetime.now(UTC)
        async with async_session_factory() as session:
            run_row = await session.get(ScheduledJobRun, run_id)
            if run_row is None or run_row.status != "running":
                return
            run_row.finished_at = finished
            run_row.duration_ms = int(
                (finished - started_at).total_seconds() * 1000,
            )
            run_row.status = RUN_STATUS_INTERRUPTED
            run_row.error_message = reason
            await session.commit()

    try:
        await asyncio.wait_for(_write(), timeout=3.0)
    except (Exception, asyncio.CancelledError):  # noqa: BLE001
        logger.warning(
            "could not mark run %s as interrupted; the stale-run reaper "
            "will pick it up", run_id, exc_info=True,
        )


async def _trim_runs(session: AsyncSession, job_id: UUID) -> None:
    """Delete runs beyond the retention limit (oldest first).

    Uses a single bulk DELETE instead of per-row ORM deletes: faster
    and atomic — a mid-loop FK exception can't leave a partial trim.
    """
    rows = (
        await session.execute(
            select(ScheduledJobRun.id)
            .where(ScheduledJobRun.job_id == job_id)
            .order_by(desc(ScheduledJobRun.started_at))
            .offset(MAX_RUN_HISTORY_PER_JOB)
        )
    ).scalars().all()
    if not rows:
        return
    await session.execute(
        _sa_delete(ScheduledJobRun).where(ScheduledJobRun.id.in_(rows))
    )


def _ensure_json_safe(value: Any) -> Any:
    """Coerce a result summary into JSON-serializable primitives.

    Handlers sometimes tuck non-str dict keys (UUIDs) or datetimes in
    their return value; json() at response time then fails. Round-
    tripping through json.dumps(default=str) keeps the shape while
    stringifying anything exotic.
    """
    if value is None:
        return None
    try:
        return json.loads(json.dumps(value, default=str))
    except (TypeError, ValueError):
        return {"__unserializable__": str(value)[:1000]}


# ── Handler discovery (for UI "available handlers" list) ─────────


def available_handlers() -> list[dict[str, str]]:
    return [
        {"key": h.key, "display_name": h.display_name, "description": h.description}
        for h in list_handlers()
    ]


__all__ = [
    "acknowledge_run",
    "available_handlers",
    "execute_job",
    "get_job",
    "get_run",
    "list_all_runs",
    "list_jobs",
    "list_runs",
    "schedule_forecast",
    "update_job",
]