"""Helix worker entrypoint.

Runs one APScheduler side, driven entirely from DB rows:

**Generic scheduled jobs** — one APScheduler trigger per row in
``scheduled_jobs`` (HRIS incremental sync, overdue-leaver check,
domain-expiry alert, cloud-tenant sync, …). Handlers register via
``@register_scheduled_job`` in :mod:`app.scheduled_jobs.handlers`, and
the service's ``execute_job`` records every run in
``scheduled_job_runs`` so the admin console can show history +
last-run metrics.

Migration 045 retired the per-``cmdb_provider_accounts`` cron loop
that used to live here — cloud tenants are now ``integrations(kind=
'cloud')`` rows swept by the ``cmdb.cloud_sync`` scheduled job.

Reloads its config every hour so admin changes from the UI (pause a
job, change cron, tweak config JSON) take effect without a worker
restart.
"""

from __future__ import annotations

import asyncio
import datetime
import logging
import signal
from pathlib import Path
from typing import Any

from app.core.safe_logging import configure_external_http_log_safety

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s %(levelname)s [helix-worker] %(message)s",
)
configure_external_http_log_safety()
logger = logging.getLogger(__name__)

# Re-read all schedulers' config from DB every hour to pick up UI changes
SCHEDULER_RELOAD_INTERVAL = 3600

# Written from the worker event loop and consumed by Docker's healthcheck.
# A fresh container has its own /tmp, so existence proves startup passed the
# handler-registration boot guard; mtime proves the event loop is still
# advancing. Keep the health threshold in docker-compose.prod.yml comfortably
# above this cadence.
WORKER_HEARTBEAT_PATH = Path("/tmp/helix-worker-heartbeat")
WORKER_HEARTBEAT_INTERVAL_SECONDS = 30

# Graceful-shutdown drain window. On SIGTERM/SIGINT we pause the
# scheduler (stop dispatching NEW jobs) and give the jobs already
# in-flight this many seconds to reach a clean ``await`` boundary
# before we cancel the stragglers. Kept below docker's default 10 s
# SIGTERM→SIGKILL grace so the drain always completes within the
# container stop window.
#
# Why this exists: APScheduler's ``AsyncIOExecutor`` runs each cron
# job as a bare ``loop.create_task``. If we returned from ``_run``
# without stopping the scheduler, ``asyncio.run``'s teardown would
# (a) let ``_process_jobs`` dispatch a fresh job during the
# ``shutdown_default_executor`` await window and (b) tear down the
# loop's default ThreadPoolExecutor underneath any job still mid-flight.
# Either path lands a job inside ``asyncio.to_thread`` AFTER the default
# executor is gone, and Python 3.12's ``loop._check_default_executor``
# raises ``RuntimeError: Executor shutdown has been called`` — observed
# in prod as ``cmdb.resource_sync <id>/iam: Executor shutdown has been
# called`` on every worker restart that overlapped a running cloud sync.
SHUTDOWN_DRAIN_GRACE_SECONDS = 8

# Job tasks currently executing under :func:`_run_scheduled_job`.
# Drained on graceful shutdown so in-flight work finishes (or is
# cleanly cancelled) while the event loop + default executor are still
# alive — never raced against ``asyncio.run``'s executor teardown.
_inflight_jobs: set[asyncio.Task[Any]] = set()

# ``scheduled_jobs.id`` (as str) of every run currently executing in THIS
# process, maintained alongside ``_inflight_jobs``. Read by
# :func:`_register_scheduled_jobs` to tell "we missed a beat" apart from
# "the beat is still running" — see the missed-beat backfill there.
_inflight_job_ids: set[str] = set()

# ── Scheduler-liveness watchdog tuning ────────────────────────────
#
# Why it exists: docker-compose's ``restart: unless-stopped`` only
# recovers crashes that exit the worker process. The 2026-05 incident
# (worker scheduler silently wedged for 3+ days across 10+ prod deploys
# while the process kept running and ``docker compose up -d`` claimed
# "healthy") is the case this watchdog catches.
#
# Strategy: a foreground asyncio task (separate from APScheduler's
# event loop) queries the worst-stale enabled interval job's last
# DISPATCH every minute. If it's older than the threshold
# AND the worker has been up long enough that the job should have
# fired by now, log a fatal error and trigger graceful shutdown
# with exit code 1. Docker's ``restart: unless-stopped`` recreates
# the container.
#
# The check intentionally runs OUTSIDE APScheduler — using a
# scheduled-job-based watchdog would have the dead-scheduler
# blind spot (the very mechanism that's broken can't observe itself).
#
# Conservative thresholds chosen to keep restart loops loose: a
# worker that genuinely can't start its scheduler will restart at
# most every ~35 min (5 min grace + 30 min staleness), giving log
# inspection plenty of time and not melting the EC2 host with a
# tight crash loop.
SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS = 300  # 5 min
SCHEDULER_WATCHDOG_STALE_THRESHOLD_SECONDS = 1800  # 30 min
SCHEDULER_WATCHDOG_CHECK_INTERVAL_SECONDS = 60


# ══ Generic scheduled jobs ════════════════════════════════════════


async def _assert_all_enabled_handlers_registered() -> None:
    """Crash boot if any enabled ``scheduled_jobs`` row's handler is unknown.

    Why this exists: the worker registers handlers via decorator side-
    effects from a hand-maintained list of ``import`` statements in
    :func:`_run`. The API process imports everything transitively through
    its router tree, so an admin testing locally never sees the problem
    — the worker process's narrower import graph quietly drops any
    handler whose module isn't on the list.

    The pre-2026-05 behaviour was a silent ``logger.warning`` + skip
    inside :func:`_register_scheduled_jobs`. That hid
    ``identity_sync.dispatch`` for **9 days** in prod (only the
    "fastest enabled interval job" watchdog catches dead schedulers,
    and a faster healthy job — ``provisioning.dispatch`` at 30 s —
    masked the wedge). PR #316 added this guard so any future
    forgotten cross-domain import crashes startup loudly, docker
    restarts the worker, and the failure is visible within minutes
    instead of days.

    Surfaces ALL missing handler_keys at once (not just the first) so
    the operator gets the full picture in one log line rather than
    fix-redeploy-discover-next ping-pong.
    """
    from sqlalchemy import select

    from app.core.database import async_session_factory
    from app.scheduled_jobs.models import ScheduledJob
    from app.scheduled_jobs.registry import get_handler

    async with async_session_factory() as db:
        rows = list(
            (
                await db.execute(
                    select(ScheduledJob.name, ScheduledJob.handler_key)
                    .where(ScheduledJob.enabled.is_(True))
                )
            ).all()
        )

    missing: list[tuple[str, str]] = [
        (name, key) for (name, key) in rows if get_handler(key) is None
    ]
    if not missing:
        return
    pretty = ", ".join(f"{n}({k})" for n, k in missing)
    raise RuntimeError(
        "scheduled-jobs boot guard: the following enabled rows have no "
        f"registered handler in this worker process: {pretty}. This is "
        "almost always because the handler module isn't imported by "
        "app.worker._run() — cross-domain handlers under app.domains.* "
        "must be added there explicitly (the API process gets them "
        "transitively via app.main, the worker does not)."
    )


async def _run_scheduled_job(job_id_str: str) -> None:
    """APScheduler callback — delegates to :func:`execute_job`.

    Kept at module scope (not a closure) so APScheduler's pickle-free
    scheduling works and error tracebacks mention a recognisable name.
    """
    import uuid

    from app.scheduled_jobs.service import execute_job

    # Register this task so graceful shutdown can drain it before the
    # event loop tears down the default executor (see
    # ``SHUTDOWN_DRAIN_GRACE_SECONDS``). ``current_task()`` is the task
    # APScheduler's AsyncIOExecutor created for this coroutine.
    task = asyncio.current_task()
    if task is not None:
        _inflight_jobs.add(task)
    _inflight_job_ids.add(job_id_str)
    try:
        await execute_job(uuid.UUID(job_id_str), triggered_by="scheduler")
    except asyncio.CancelledError:
        # Cancelled by graceful shutdown (or APScheduler stop). This is
        # an orderly stop, not a job failure — execute_job's own session
        # context rolls back any open transaction. Don't synthesise a
        # failed run row or fire the failure-notifier; the job re-runs on
        # the next tick (all handlers are idempotent). Re-raise so the
        # cancellation propagates as asyncio expects.
        raise
    except Exception as exc:  # noqa: BLE001
        logger.exception(
            "scheduled job %s raised OUTSIDE the wrapper (no run row written)",
            job_id_str,
        )
        # Best-effort: synthesise a failed-run row + fire the
        # failure-notifier so the watchdog isn't the only signal for
        # wrapper-bypass crashes. Without this, a session-factory
        # error (or any exception before ``execute_job``'s inner
        # try/except takes over) left no scheduled_job_runs entry and
        # the failure-notifier stayed silent for up to 60×interval.
        try:
            from datetime import UTC as _UTC
            from datetime import datetime as _datetime

            from app.core.database import async_session_factory
            from app.core.exceptions import format_exception_for_log
            from app.scheduled_jobs.failure_notifier import (
                maybe_notify_failure,
            )
            from app.scheduled_jobs.models import (
                ScheduledJob,
                ScheduledJobRun,
            )
            # Canonical render handles httpx transport errors / AppException
            # whose ``str(exc)`` is empty — without it the wrapper-bypass
            # row + failure-notifier card both surfaced just the bare
            # exception class with no body.
            rendered = format_exception_for_log(exc, max_len=500)
            async with async_session_factory() as _s:
                job = await _s.get(ScheduledJob, uuid.UUID(job_id_str))
                if job is not None:
                    now = _datetime.now(_UTC)
                    _s.add(
                        ScheduledJobRun(
                            job_id=job.id,
                            triggered_by="scheduler",
                            status="failed",
                            started_at=now,
                            finished_at=now,
                            error_message=(
                                f"execute_job raised outside wrapper: "
                                f"{rendered}"
                            ),
                        )
                    )
                    await _s.commit()
                    await maybe_notify_failure(
                        job_id=job.id,
                        handler_key=job.handler_key,
                        error_message=rendered,
                        triggered_by="scheduler",
                    )
        except Exception:  # noqa: BLE001
            logger.exception(
                "wrapper-bypass synthetic-failure row write also failed",
            )
    finally:
        if task is not None:
            _inflight_jobs.discard(task)
        _inflight_job_ids.discard(job_id_str)


async def _register_scheduled_jobs(scheduler) -> int:
    """Load every enabled ``scheduled_jobs`` row and attach a trigger.

    Safe to call on reload — removes all previously registered jobs
    whose id starts with ``sched_`` before re-attaching from DB, so a
    row disabled via the UI disappears from the scheduler within the
    next reload tick.
    """
    from apscheduler.triggers.cron import CronTrigger
    from apscheduler.triggers.interval import IntervalTrigger
    from sqlalchemy import select

    from app.core.database import async_session_factory
    from app.scheduled_jobs.models import ScheduledJob
    from app.scheduled_jobs.registry import get_handler

    # Drop any previously-attached generic-scheduler jobs
    for job in list(scheduler.get_jobs()):
        if job.id.startswith("sched_"):
            job.remove()

    attached = 0
    # Deliberately NOT swallowing DB-load failures: at startup the
    # caller (start_schedulers) does NOT wrap this call, so a re-raise
    # crashes the worker and lets the container orchestrator retry —
    # better than silently running with 0 jobs scheduled. The hourly
    # reload loop in start_schedulers DOES wrap the call in try/except
    # and continues with the previously-attached jobs on failure.
    async with async_session_factory() as db:
        rows = list(
            (
                await db.execute(
                    select(ScheduledJob).where(ScheduledJob.enabled.is_(True))
                )
            )
            .scalars()
            .all()
        )

    for row in rows:
        if get_handler(row.handler_key) is None:
            logger.warning(
                "scheduled-jobs: row %s handler_key=%r has no registered handler — skipping",
                row.name, row.handler_key,
            )
            continue

        trigger: Any
        if row.trigger_type == "cron" and row.cron_expr:
            try:
                trigger = CronTrigger.from_crontab(
                    row.cron_expr, timezone=row.timezone or "UTC",
                )
            except (ValueError, KeyError) as e:
                logger.warning(
                    "scheduled-jobs: invalid cron %r on %s: %s — skipping",
                    row.cron_expr, row.name, e,
                )
                continue
        elif row.trigger_type == "interval" and row.interval_seconds:
            trigger = IntervalTrigger(
                seconds=int(row.interval_seconds),
                timezone=row.timezone or "UTC",
            )
        elif row.trigger_type == "chained":
            # Intentional: a chained row has no clock. Its parent's
            # completion fires it (fast path) and
            # ``scheduler.chain_advance`` guarantees it (reconciler), so
            # there is nothing for APScheduler to attach. Skip QUIETLY —
            # a warning here would fire once per chained row on every
            # boot and every hourly reload, training operators to ignore
            # the one warning below that means something is broken.
            logger.debug(
                "scheduled-jobs: %s is chained after %s — no trigger attached",
                row.name, row.chain_after_job_id,
            )
            continue
        else:
            logger.warning(
                "scheduled-jobs: row %s has incomplete trigger config (%s) — skipping",
                row.name, row.trigger_type,
            )
            continue

        job_id = f"sched_{row.id}"
        add_kwargs: dict[str, Any] = {
            "trigger": trigger,
            "id": job_id,
            "replace_existing": True,
            "kwargs": {"job_id_str": str(row.id)},
            # Hard cap: only one instance of any given job at a time.
            # APScheduler's default is 1 too, but it silently drops the
            # next-tick run when the prior run is still going — and
            # silently drops it forever, with no row in
            # ``scheduled_job_runs`` to alarm on. ``coalesce=True``
            # collapses missed runs into one and ``misfire_grace_time``
            # bounds how late a coalesced run can fire (15 min — long
            # enough to absorb a lazy long-running job, short enough
            # that a 4-hour cron isn't perpetually shifted).
            "max_instances": 1,
            "coalesce": True,
            "misfire_grace_time": 900,
        }
        # For interval jobs, schedule the first run shortly after boot
        # (not ``now() + interval``) so a slow-cadence job doesn't
        # stay quiet for hours after a worker restart.
        #
        # For CRON jobs, fire shortly after boot ONLY if the next
        # scheduled fire after ``last_run_at`` is already in the past
        # (i.e. we genuinely missed a beat — likely because the worker
        # was being redeployed during the scheduled window).
        #
        # The previous heuristic (``last_run_at > 25h ago``) wrongly
        # treated long-cadence crons (monthly / weekly) as overdue on
        # every hourly reload: a monthly cron with ``cron_expr='5 3 1
        # * *'`` last ran 30 days ago, so the >25h check was True every
        # single hour for 29 days, re-firing the job daily. In
        # production this overwrote the AI-spending monthly summary
        # 7 times in 2026-05 and produced spurious ``period`` rollovers
        # (PR #626). Use the actual cron schedule via
        # ``CronTrigger.get_next_fire_time`` so monthly/weekly/quarterly
        # crons are treated correctly and daily ones still backfill.
        now_utc = datetime.datetime.now(datetime.UTC)
        if row.trigger_type == "interval":
            # #1774: only fire shortly after boot when the interval job is
            # genuinely overdue (never ran, or ``last_run_at + interval``
            # already passed). ``_register_scheduled_jobs`` runs on boot AND
            # on every hourly ``_reload_loop`` tick (replace_existing=True),
            # so an UNCONDITIONAL ``now + 60s`` reset every interval job's
            # next fire to ~60s after each reload — turning a configured
            # 4h/13h/24h cadence into an effective hourly run (vendor-sync
            # API churn, distorted run metrics, silently-ignored >1h
            # cadence). The cron branch already got this backfill check in
            # PR #626; the interval branch never did. Compute the honest
            # next fire from ``last_run_at + interval``.
            interval_s = int(row.interval_seconds or 0)
            last_iv = row.last_run_at
            if last_iv is not None and last_iv.tzinfo is None:
                last_iv = last_iv.replace(tzinfo=datetime.UTC)
            if last_iv is None or interval_s <= 0:
                # Never ran (or degenerate interval) → fire shortly after boot.
                add_kwargs["next_run_time"] = now_utc + datetime.timedelta(seconds=60)
            else:
                due_iv = last_iv + datetime.timedelta(seconds=interval_s)
                if due_iv <= now_utc:
                    # Missed a beat (likely a redeploy across the window) →
                    # backfill soon.
                    add_kwargs["next_run_time"] = now_utc + datetime.timedelta(seconds=60)
                else:
                    # Not due yet → honour the remaining cadence so a reload
                    # doesn't reset the clock and re-fire early.
                    add_kwargs["next_run_time"] = due_iv
        elif (
            row.trigger_type == "cron"
            and row.last_run_at
            # 2026-08-10: ``last_run_at`` is written by ``execute_job``'s
            # TERMINAL write-back, so while a run is in flight the column
            # still holds the PREVIOUS run's timestamp. A reload tick that
            # lands inside a long run therefore read "today's beat never
            # happened" and scheduled a backfill fire 90 s later —
            # ``max_instances=1`` suppressed it only if the run was still
            # going when it fired, so any run finishing inside that 90 s
            # window got re-executed immediately. Prod: 16 duplicate runs
            # in two months, gaps of 3–89 s after the previous run ended,
            # all on daily crons long enough to overlap a reload —
            # including the 2026-08-10 duplicate 24-minute AliCloud 海外
            # CloudMonitor pull that then failed on errno 2013. Skipping
            # the backfill for jobs running RIGHT NOW is the narrow fix:
            # the redeploy case this backfill exists for (worker killed
            # mid-window, nothing in flight after reboot) still fires.
            # Residual race: a run that ends between the row SELECT above
            # and this check is still re-fired — milliseconds wide, versus
            # the minutes-wide window it replaces.
            and str(row.id) not in _inflight_job_ids
        ):
            last = row.last_run_at
            if last.tzinfo is None:
                last = last.replace(tzinfo=datetime.UTC)
            # ``get_next_fire_time(previous, now)`` returns the next
            # fire strictly after ``now``. Bump ``last`` by 1s so we
            # ask for "next fire strictly AFTER the previous run"
            # rather than risk returning ``last`` itself.
            try:
                expected_next = trigger.get_next_fire_time(
                    None, last + datetime.timedelta(seconds=1),
                )
            except Exception:  # noqa: BLE001
                # If the schedule can't compute (corrupt expression),
                # fall back to "no backfill" — silently skipping a
                # missed run is better than re-firing forever.
                expected_next = None
            if expected_next is not None and expected_next < now_utc:
                add_kwargs["next_run_time"] = now_utc + datetime.timedelta(seconds=90)
        scheduler.add_job(_run_scheduled_job, **add_kwargs)
        logger.info(
            "scheduled-jobs: attached %s (%s) next_run=%s",
            row.name,
            f"interval {row.interval_seconds}s" if row.trigger_type == "interval"
            else f"cron {row.cron_expr}",
            scheduler.get_job(job_id).next_run_time,
        )
        attached += 1

    return attached


# ══ Boot + reload loop ════════════════════════════════════════════


def _build_scheduler():
    """Create and return a fresh AsyncIOScheduler (not yet started)."""
    from apscheduler.schedulers.asyncio import AsyncIOScheduler

    return AsyncIOScheduler(timezone="UTC")


async def _reconcile_cloud_scheduled_jobs() -> None:
    """Ensure every cloud integration has the full set of lifecycle
    ``scheduled_jobs`` rows its code expects. Runs once per boot so a
    freshly-added handler (e.g. PR #24's ``cmdb.grants_sync``) lands
    in the DB without anyone clicking save on each integration."""
    from app.core.database import async_session_factory
    from app.domains.cmdb.scheduled_jobs_provisioning import (
        reconcile_all_cloud_integrations,
    )

    try:
        async with async_session_factory() as db:
            summary = await reconcile_all_cloud_integrations(db)
        logger.info(
            "scheduled_jobs reconcile: scanned=%d added=%d removed=%d failed=%s",
            summary["integrations_scanned"],
            summary["rows_added"],
            summary["rows_removed"],
            summary["failed_integration_ids"] or "none",
        )
    except Exception:
        logger.exception("scheduled_jobs reconcile on boot failed")


# Strong refs to first-run-seed tasks so they're not GC'd before
# completion. ``asyncio.create_task`` only holds a weak reference.
_FIRST_RUN_SEED_TASKS_PENDING: set[Any] = set()


async def _seed_first_run_read_model_jobs() -> None:
    """Fire first-run read-model jobs once without waiting for their cron.

    Why: the auditor's controls all read from ``auditor_*`` tables
    populated by their respective scheduled-jobs handlers. Daily/hourly
    crons leave the dashboard empty for up to 24h after a fresh deploy;
    quarterly review (only fires Jan/Apr/Jul/Oct 1st) can leave it empty
    for months. Users see "0 / 0" everywhere on day one and reasonably
    conclude the feature is broken. The same applies to a newly introduced
    cloud-scope health baseline: an existing subscription must not wait for
    tomorrow's daily refresh before its AI tab can explain its real data gap.

    Seed semantics: only jobs with ``last_run_at IS NULL`` get fired.
    Jobs that have run before are left alone — this is genuinely "first
    run", not "force resync". Fire-and-forget via ``asyncio.create_task``
    so the scheduler boot path is not blocked.
    """
    import asyncio

    from sqlalchemy import select

    from app.core.database import async_session_factory
    from app.scheduled_jobs.models import ScheduledJob
    from app.scheduled_jobs.service import execute_job

    try:
        async with async_session_factory() as db:
            never_run_jobs = (await db.execute(
            select(ScheduledJob).where(
                (
                    ScheduledJob.handler_key.like("auditor.%")
                    | (
                        ScheduledJob.handler_key
                        == "intelligence.cloud_scope_baseline_bootstrap"
                    )
                ),
                ScheduledJob.enabled.is_(True),
                    ScheduledJob.last_run_at.is_(None),
                )
            )).scalars().all()
    except Exception:
        logger.exception("first-run-seed: scan failed")
        return

    if not never_run_jobs:
        logger.info("first-run-seed: all read-model jobs have run before")
        return

    logger.info(
        "first-run-seed: queuing %d never-run auditor jobs: %s",
        len(never_run_jobs),
        [j.handler_key for j in never_run_jobs],
    )
    for job in never_run_jobs:
        # Schedule each as its own task so a slow upstream doesn't
        # serialize the rest. ``execute_job`` records all run state in
        # ``scheduled_job_runs`` so the next cron tick treats it as
        # already-ran (last_run_at is NOT NULL) and skips.
        async def _run_one(job_id=job.id, key=job.handler_key) -> None:
            try:
                await execute_job(
                    job_id,
                    triggered_by="first_run_seed",
                    # The bootstrap only compiles existing local facts. Give
                    # it one bounded dedicated lane so initial subscription
                    # pages do not wait behind vendor syncs; no other seed
                    # may bypass the shared pool-safety gate.
                    bypass_concurrency_gate=(
                        key == "intelligence.cloud_scope_baseline_bootstrap"
                    ),
                )
                logger.info("first-run-seed: %s completed", key)
            except Exception:
                logger.exception(
                    "first-run-seed: %s failed", key,
                )

        task = asyncio.create_task(_run_one())
        # Prevent "Task was destroyed but it is pending" warnings if
        # the worker shuts down mid-seed: keep a strong ref.
        _FIRST_RUN_SEED_TASKS_PENDING.add(task)
        task.add_done_callback(_FIRST_RUN_SEED_TASKS_PENDING.discard)


async def start_schedulers() -> tuple[Any, asyncio.Task[None]]:
    """Start APScheduler with generic jobs and reload hourly.

    Returns ``(scheduler, reload_task)`` so the caller can stop both on
    graceful shutdown (see :func:`_stop_schedulers`) — leaving the
    scheduler running while ``asyncio.run`` tears down the loop races
    in-flight jobs against the default-executor shutdown.
    """
    scheduler = _build_scheduler()
    scheduler.start()
    logger.info("scheduler started")

    await _reconcile_cloud_scheduled_jobs()

    # Boot guard: surface unknown-handler rows as a startup crash rather
    # than silently dropping them (see ``_assert_all_enabled_handlers_registered``).
    # Runs AFTER ``_reconcile_cloud_scheduled_jobs`` so newly-seeded cloud
    # lifecycle rows are part of the check, and BEFORE the first
    # ``_register_scheduled_jobs`` call so we never even reach the
    # "warning + skip" branch in the happy path.
    await _assert_all_enabled_handlers_registered()

    sched_count = await _register_scheduled_jobs(scheduler)
    logger.info(
        "scheduler: %d scheduled_jobs attached on startup",
        sched_count,
    )

    # First-run seed for read-model jobs — without this, a freshly-deployed
    # Helix can show either an auditor dashboard of 0/0 or a cloud subscription
    # AI tab without its deterministic explanation until the next natural cron.
    # Idempotent: only ever fires jobs whose last_run_at is still NULL.
    await _seed_first_run_read_model_jobs()

    async def _reload_loop() -> None:
        while True:
            await asyncio.sleep(SCHEDULER_RELOAD_INTERVAL)
            logger.info("scheduler reload: re-reading DB config")
            try:
                s = await _register_scheduled_jobs(scheduler)
                logger.info(
                    "scheduler reload: %d scheduled_jobs active",
                    s,
                )
            except Exception:
                logger.exception("scheduler reload failed")

    _reload_task = asyncio.create_task(_reload_loop())
    _reload_task.add_done_callback(
        lambda t: logger.error(
            "scheduler reload loop died unexpectedly", exc_info=t.exception(),
        ) if not t.cancelled() and t.exception() else None
    )

    return scheduler, _reload_task


async def _stop_schedulers(
    scheduler: Any,
    reload_task: asyncio.Task[None],
) -> None:
    """Gracefully stop the scheduler and drain in-flight jobs.

    Ordering is the whole point — every step runs while the event loop
    and its default ThreadPoolExecutor are still alive:

    1. Cancel the hourly reload loop (it must not re-attach jobs after
       we pause).
    2. ``scheduler.pause()`` — stops ``_process_jobs`` from dispatching
       any NEW job. Jobs already running keep going.
    3. Drain: give in-flight jobs ``SHUTDOWN_DRAIN_GRACE_SECONDS`` to
       finish naturally, then cancel any stragglers and await their
       cancellation. Cancelling here yields a clean ``CancelledError``
       inside the job (handled in :func:`_run_scheduled_job`) instead of
       the ``RuntimeError: Executor shutdown has been called`` that
       ``asyncio.run``'s teardown would raise if the job were still
       calling ``asyncio.to_thread`` after the default executor is gone.
    4. ``scheduler.shutdown(wait=False)`` — release jobstores/executors.
    """
    # Tell the executor a graceful shutdown has begun BEFORE anything is
    # cancelled. A run cancelled from here on can then say "worker
    # shutdown drain" on its run row instead of guessing — the only other
    # way to get cancelled is a single-flight lock lost mid-hold, and the
    # two want different words in the ledger.
    from app.scheduled_jobs import service as sched_service

    sched_service.mark_draining()

    reload_task.cancel()
    # Await its unwind so it can't re-attach jobs after we pause and so
    # ``asyncio.run``'s teardown isn't left a straggler task.
    try:
        await reload_task
    except (asyncio.CancelledError, Exception):  # noqa: BLE001
        pass

    # ``pause()`` halts new dispatch immediately; ``shutdown`` later
    # tears the scheduler down. ``getattr`` guards keep this defensive
    # against a half-built scheduler if startup raced a fast SIGTERM.
    try:
        if getattr(scheduler, "running", False):
            scheduler.pause()
    except Exception:  # noqa: BLE001
        logger.exception("scheduler.pause() during shutdown failed")

    inflight = [t for t in _inflight_jobs if not t.done()]
    if inflight:
        logger.info(
            "worker shutdown: draining %d in-flight scheduled job(s) "
            "(grace=%ds)",
            len(inflight),
            SHUTDOWN_DRAIN_GRACE_SECONDS,
        )
        _, pending = await asyncio.wait(
            inflight, timeout=SHUTDOWN_DRAIN_GRACE_SECONDS,
        )
        if pending:
            logger.warning(
                "worker shutdown: %d job(s) did not finish in %ds — "
                "cancelling",
                len(pending),
                SHUTDOWN_DRAIN_GRACE_SECONDS,
            )
            for t in pending:
                t.cancel()
            # Await the cancellations so the to_thread futures unwind
            # while the default executor is still up.
            await asyncio.gather(*pending, return_exceptions=True)

    try:
        scheduler.shutdown(wait=False)
    except Exception:  # noqa: BLE001
        logger.exception("scheduler.shutdown() during shutdown failed")
    logger.info("scheduler stopped")


# ══ Scheduler liveness watchdog ═══════════════════════════════════


async def _measure_scheduler_staleness(
    uptime_seconds: float | None = None,
) -> tuple[float, int, str] | None:
    """Liveness signal for the **worst-stale-AND-actionable** interval job.

    Returns ``(stale_seconds, interval_seconds, job_name)`` for the
    enabled interval job with the highest ``stale_seconds /
    interval_seconds`` ratio AMONG jobs that have already exited their
    per-job startup grace window (``uptime ≥ max(static_grace,
    2 * interval)``). Returns ``None`` when:

      * no enabled interval rows exist (cron-only worker config — the
        watchdog must abstain), OR
      * every interval row is still inside its per-job grace window
        (a freshly-booted worker whose fastest job hasn't had time to
        tick twice yet — too early to make a wedge call on anything).

    Why worst-stale, not fastest: pre-PR #316 this returned only the
    fastest-cadence job's staleness, on the assumption that anything
    wedging the scheduler would wedge that job first. That assumption
    breaks the moment a single handler's import path goes wrong — the
    other jobs keep firing because they're independent APScheduler
    triggers; only the affected job goes silent. In the 2026-05
    ``identity_sync.dispatch`` incident, the 30s ``provisioning.dispatch``
    was healthy, and the watchdog read green for 9 days while
    ``identity_sync.dispatch`` (60s) never fired once. Worst-stale
    measurement catches the per-job blind spot.

    Why per-job grace, not single-shared-grace: if the worst-stale job
    happens to be a slow one (e.g. a 13-hour interval that genuinely
    hasn't had time to tick yet at 2-hour uptime), a single shared
    grace would either be too short (false-positive on the slow job)
    or so long it masks any fast-job wedge that happens during the
    window. Filtering inside the helper, per-job, means the watchdog
    sees the worst-stale job AMONG THOSE THAT SHOULD HAVE FIRED BY
    NOW — a healthy answer in every state of the boot.

    The ``uptime_seconds`` arg is the watchdog loop's own uptime; the
    helper computes per-job grace = ``max(static_grace, 2 * interval)``
    and filters rows whose grace hasn't elapsed yet. Default ``None``
    means "no grace filter" — useful for tests that want to inspect
    the raw ratio output without the boot-time gate, and for legacy
    callers that don't have a uptime handy.

    Never-ran semantics: staleness is measured from the last DISPATCH
    (``max(job.last_run_at, latest scheduled_job_runs.started_at)``),
    falling back to ``job.created_at`` for a job row that has never
    been dispatched at all. A never-*completed* job is therefore not
    stale as long as the scheduler keeps dispatching it — see the
    inline comment at the ``reference`` computation for the prod
    deadlock that measuring completions instead caused. Combined with
    per-job grace filtering above, an actionable never-ran result
    only surfaces AFTER ``uptime ≥ 2 * interval`` — which is the
    correct "this job should have fired at least once by now" line.

    Returns the job name too so the wedge log message can identify
    which job is responsible — without it, debugging a multi-job wedge
    requires a separate DB round-trip.
    """
    from sqlalchemy import func, select  # noqa: PLC0415

    from app.core.database import async_session_factory  # noqa: PLC0415
    from app.scheduled_jobs.models import ScheduledJob, ScheduledJobRun  # noqa: PLC0415

    async with async_session_factory() as db:
        rows = list(
            (await db.execute(
                select(
                    ScheduledJob.id,
                    ScheduledJob.name,
                    ScheduledJob.last_run_at,
                    ScheduledJob.created_at,
                    ScheduledJob.interval_seconds,
                )
                .where(
                    ScheduledJob.enabled.is_(True),
                    ScheduledJob.interval_seconds.isnot(None),
                )
            )).all()
        )
        if not rows:
            return None
        # Latest DISPATCH per job, not latest completion. See the
        # ``reference`` comment below for why the job row alone is not
        # a usable liveness signal. Scoped to the handful of enabled
        # interval jobs so this rides ``ix_scheduled_job_runs_job_started``
        # instead of grouping the whole (20k-row) history every minute.
        last_dispatch = {
            row.job_id: row.started_at
            for row in (await db.execute(
                select(
                    ScheduledJobRun.job_id,
                    func.max(ScheduledJobRun.started_at).label("started_at"),
                )
                .where(ScheduledJobRun.job_id.in_([row.id for row in rows]))
                .group_by(ScheduledJobRun.job_id)
            )).all()
        }

    now_aware = datetime.datetime.now(datetime.UTC)
    worst_ratio = -1.0
    worst: tuple[float, int, str] | None = None
    for row in rows:
        interval = int(row.interval_seconds or 0)
        if interval <= 0:
            continue
        if uptime_seconds is not None:
            per_job_grace = max(
                SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS,
                2 * interval,
            )
            if uptime_seconds < per_job_grace:
                # This job hasn't had time to fire even ONCE yet —
                # can't be considered stale. Skip without blocking
                # measurement of faster jobs that ARE past their grace.
                continue
        # Reference point = the most recent evidence that APScheduler
        # DISPATCHED this job, which is what this watchdog exists to
        # measure. ``last_run_at`` alone is not that evidence: it is
        # written only in the terminal write-back (``service.py``
        # ``_execute_job_locked``), so a job that is dispatched on time
        # but never reaches a terminal state leaves it NULL forever.
        # Reading NULL as "+inf stale" then kills a worker whose
        # scheduler is provably alive — and the restart guarantees the
        # job never completes, so the next boot reads NULL again. That
        # deadlock took prod down for ~16h on 2026-08-17: the 60s
        # ``intelligence.cloud_scope_baseline_bootstrap`` (271 cloud
        # scopes, longer than the 300s grace) crash-looped the worker
        # every ~5m20s while its own 874 ``skipped`` run rows proved
        # the scheduler was dispatching it every single minute.
        # ``scheduled_job_runs.started_at`` IS written at dispatch —
        # including for skipped and abandoned runs — so it stays green
        # exactly when the scheduler is alive, and goes stale exactly
        # when it wedges (the 2026-05 incident this watchdog was built
        # for). ``created_at`` is the fallback for a job row that has
        # never been dispatched at all, matching the never-ran
        # semantics ``service.py`` already uses for the SLO surface.
        reference = max(
            (ts for ts in (row.last_run_at, last_dispatch.get(row.id)) if ts is not None),
            default=row.created_at,
        )
        if reference is None:
            stale = float("inf")
        else:
            # ``last_run_at`` is stored as ``DateTime(timezone=True)`` but
            # MySQL/TiDB does not preserve tzinfo — the column comes back
            # naive in UTC. Strip ``now``'s tzinfo for the diff so this
            # works regardless of how the column round-trips.
            now = (
                now_aware.replace(tzinfo=None)
                if reference.tzinfo is None
                else now_aware
            )
            stale = (now - reference).total_seconds()
        ratio = stale / interval
        if ratio > worst_ratio:
            worst_ratio = ratio
            worst = (stale, interval, str(row.name))
    return worst


async def _scheduler_watchdog_loop(stop_event: asyncio.Event) -> int:
    """Watch for APScheduler wedging; signal shutdown if detected.

    Returns 0 when the loop exits cleanly (``stop_event`` set by
    SIGINT/SIGTERM); returns 1 when the watchdog itself triggered the
    shutdown because the scheduler appears wedged.

    **Known gap (not fixed by this watchdog)**: a hung asyncio event
    loop (e.g., a sync sleep / blocking syscall in a handler) freezes
    THIS task too. The watchdog only catches "process alive but
    APScheduler stopped dispatching". A separate process-level
    healthcheck would be needed for the loop-hung case; that's
    deferred — the dominant 2026-05 incident was scheduler-wedge,
    not loop-hang.
    """
    startup_time = datetime.datetime.now(datetime.UTC)
    logger.info(
        "scheduler watchdog: armed (grace=%ds, stale_threshold=%ds, "
        "check_interval=%ds)",
        SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS,
        SCHEDULER_WATCHDOG_STALE_THRESHOLD_SECONDS,
        SCHEDULER_WATCHDOG_CHECK_INTERVAL_SECONDS,
    )
    while not stop_event.is_set():
        try:
            await asyncio.wait_for(
                stop_event.wait(),
                timeout=SCHEDULER_WATCHDOG_CHECK_INTERVAL_SECONDS,
            )
            return 0  # clean external shutdown
        except TimeoutError:
            pass

        uptime = (
            datetime.datetime.now(datetime.UTC) - startup_time
        ).total_seconds()
        if uptime < SCHEDULER_WATCHDOG_STARTUP_GRACE_SECONDS:
            continue

        try:
            measurement = await _measure_scheduler_staleness(
                uptime_seconds=uptime,
            )
        except Exception:
            # DB hiccup — log but do NOT exit. The watchdog itself
            # must never destabilise an otherwise-healthy worker.
            logger.exception("scheduler watchdog: staleness check failed")
            continue

        if measurement is None:
            # No interval-based jobs past their per-job grace window —
            # nothing actionable to measure yet. Note this also covers
            # the early-boot path where the static grace passed but
            # every individual job's ``2 * interval`` per-job grace
            # hasn't (e.g., a worker whose fastest interval is 5 min).
            continue
        stale_seconds, worst_interval, worst_name = measurement
        # Per-job stale threshold: max(static 30 min, 60 × interval).
        # The 60× factor means "this job has missed 60 consecutive ticks"
        # — way past any plausible "slow handler" / "transient API
        # hiccup" pattern. For a 30s job that's 30 min (matches the
        # legacy static threshold); for a 60s job (e.g. identity_sync.
        # dispatch) that's 60 min — tight enough to surface the 9-day
        # silent miss within the same hour; for a 1h cron-equivalent
        # interval that's 60h, still bounded so the wedge eventually
        # auto-recovers via docker restart. Per-job grace is already
        # applied inside ``_measure_scheduler_staleness``; the
        # ``measurement is None`` branch above is the boot-time gate.
        effective_threshold = max(
            SCHEDULER_WATCHDOG_STALE_THRESHOLD_SECONDS,
            60 * worst_interval,
        )
        if stale_seconds > effective_threshold:
            logger.error(
                "scheduler watchdog: enabled interval job %r "
                "(interval=%ds) has not run for %.0fs (threshold %ds, "
                "uptime %.0fs). APScheduler appears wedged. Exiting "
                "with code 1 so docker-compose's ``restart: "
                "unless-stopped`` recreates the worker container.",
                worst_name,
                worst_interval,
                stale_seconds,
                effective_threshold,
                uptime,
            )
            stop_event.set()
            return 1
    return 0


# ══ Main loop ═════════════════════════════════════════════════════


def _load_all_models() -> None:
    """Import every ORM model module up-front so SQLAlchemy's MetaData
    is complete before any query runs.

    Context: several cross-domain FKs are declared as string references
    (e.g. ``ForeignKey("identity_persons.id")`` in cmdb/email_group/
    provisioning/asset/...). SQLAlchemy resolves these lazily at
    ``configure_mappers()`` time. In the API process, every model is
    loaded transitively via the router tree (``auth_router`` → auth.models
    → identity.person_models), so resolution always succeeds. The worker
    has a much narrower import graph — only scheduled-job handlers —
    and some handlers touch ORM classes whose FK targets live in an
    unloaded module. Result: the first sync after a cold boot died with
    ``could not find table 'identity_persons'`` (prod 2026-04-24).
    Mirror ``tests/conftest.py``'s model-load pattern here so every
    worker process has the full MetaData regardless of handler order.
    """
    import app.approval.models  # noqa: F401
    import app.audit.models  # noqa: F401
    import app.auth.models  # noqa: F401
    import app.compliance.models  # noqa: F401
    import app.core.workflow_journal.models  # noqa: F401
    import app.domains.ai_labs.models  # noqa: F401
    import app.domains.ai_spending.models  # noqa: F401
    import app.domains.apikey.models  # noqa: F401
    import app.domains.asset.models  # noqa: F401
    import app.domains.bailian.models  # noqa: F401
    import app.domains.cmdb.models  # noqa: F401
    import app.domains.email_group.models  # noqa: F401
    import app.domains.hr.models  # noqa: F401
    import app.domains.identity_sync.models  # noqa: F401
    import app.domains.leaver_blocked.models  # noqa: F401
    import app.domains.offboarding.models  # noqa: F401
    import app.domains.permission.models  # noqa: F401
    import app.domains.provisioning.models  # noqa: F401
    import app.domains.relationships.models  # noqa: F401
    import app.identity.person_models  # noqa: F401
    import app.integrations.models  # noqa: F401
    import app.notifications.models  # noqa: F401
    import app.scheduled_jobs.models  # noqa: F401

    # Wire the Leaver Blocked SQLAlchemy event listener so background
    # jobs that mutate ``identity_persons.employment_status`` go through
    # the same hard-gate as the API path.
    from app.domains.leaver_blocked.listener import register_listener
    register_listener()
    # Wire the identity_persons audit listener (PR 2026-05-30) so every
    # background INSERT/UPDATE/DELETE on IdentityPerson — every HRIS
    # sync, every scheduled-job person mutation — lands an audit row.
    # See ``app/identity/audit_listener.py`` for the wangdi-case
    # post-mortem this closes.
    from app.identity.audit_listener import (
        register_listener as register_identity_audit_listener,
    )
    register_identity_audit_listener()
    # Wire the access-cycle ledger listener. Registering here is not
    # symmetry housekeeping — the provisioning worker performs the
    # canonical PENDING→ACTIVE and →REVOKED permission flips
    # (``_post_success_side_effects``), so a web-only registration would
    # leave every asynchronously-completed grant/revoke out of the
    # point-in-time holding ledger.
    from app.core.access_cycle_listener import (
        register_listener as register_access_cycle_listener,
    )
    register_access_cycle_listener()
    # Wire the asset-oversight notifier's AuditService.log post-log hook
    # so background asset mutations are subject to the same hook as the
    # API path. Background/cron actors are system (actor_user_id is
    # None) and get filtered out, but registering here keeps the two
    # processes symmetric and future-proofs a worker-run regional-admin
    # code path.
    from app.domains.asset.oversight_notifier import (
        register as register_asset_oversight_notifier,
    )
    register_asset_oversight_notifier()
    # Realized cloud-savings tracker: a lifecycle action (destroy/stop) can
    # be logged from either process — register here too so the outcome
    # ledger is written regardless of where the action ran.
    from app.domains.cmdb.savings_tracker import (
        register as register_cloud_savings_tracker,
    )
    register_cloud_savings_tracker()


async def _run() -> int:
    """Run the worker until stopped. Returns exit code (0 normal, 1 watchdog)."""
    # Pre-load all model modules so FK string references resolve on the
    # first ORM query — see ``_load_all_models`` docstring for why.
    _load_all_models()

    # Make sure handlers register at import time so the generic
    # scheduler can resolve ``handler_key`` lookups against the registry.
    #
    # Cross-domain handlers (anything OUTSIDE ``app.scheduled_jobs.*``) MUST
    # be imported here too — the API process loads them transitively via
    # ``app.main``'s router/model imports, but the worker has a much
    # narrower import graph and won't see them otherwise. PR #316 caught
    # ``identity_sync.dispatch`` silently never firing for 9 days because
    # its handler lived under ``app.domains.identity_sync.scheduler_handler``
    # and was only imported by ``main.py``. The boot-time guard further
    # down (``_assert_all_enabled_handlers_registered``) is the second
    # layer that converts a future forgotten import into a hard crash
    # instead of a 9-day silent miss.
    #
    # ``app.integrations.drivers`` lives in this same list for the same
    # reason: the package's ``__init__.py`` imports every cloud/software
    # driver module so the ``@register_driver`` decorators fire. Before
    # PR #504 (2026-05-26) the worker reached this package by accident
    # via coordinator modules' top-level ``from app.integrations.drivers.X
    # import Y`` — fine on long-lived workers, broken on fresh boots
    # where the ``cmdb.resource_creator_sweep`` cron fired BEFORE any
    # coordinator-importing handler had run. Mirror of ``app.main:12``
    # makes the driver registry unconditional, not order-dependent.
    import app.domains.asset.scheduler_handlers  # noqa: F401
    import app.domains.attendance.scheduler_handlers  # noqa: F401
    import app.domains.automation_exception.scheduler_handlers  # noqa: F401
    import app.domains.bounty.scheduler_handlers  # noqa: F401
    import app.domains.identity_sync.scheduler_handler  # noqa: F401
    import app.domains.security_awareness.scheduler_handlers  # noqa: F401
    import app.domains.sso.scheduler_handlers  # noqa: F401
    import app.integrations.drivers  # noqa: F401
    import app.integrations.health_check  # noqa: F401
    import app.scheduled_jobs.access_drift_handler  # noqa: F401
    import app.scheduled_jobs.ai_labs_handlers  # noqa: F401
    import app.scheduled_jobs.ai_spending_handlers  # noqa: F401
    import app.scheduled_jobs.approval_registry_handlers  # noqa: F401
    import app.scheduled_jobs.asset_inventory_handlers  # noqa: F401
    import app.scheduled_jobs.assistant_eval_handlers  # noqa: F401
    import app.scheduled_jobs.assistant_gap_handlers  # noqa: F401
    import app.scheduled_jobs.assistant_room_handlers  # noqa: F401
    import app.scheduled_jobs.assistant_submission_handlers  # noqa: F401
    import app.scheduled_jobs.auditor_handlers  # noqa: F401
    import app.scheduled_jobs.bailian_handlers  # noqa: F401
    import app.scheduled_jobs.chain_handlers  # noqa: F401
    import app.scheduled_jobs.cloud_owner_handlers  # noqa: F401
    import app.scheduled_jobs.cmdb_dns_request_handlers  # noqa: F401
    import app.scheduled_jobs.cmdb_server_owner_handlers  # noqa: F401
    import app.scheduled_jobs.data_access_handlers  # noqa: F401
    import app.scheduled_jobs.data_access_lifecycle_handlers  # noqa: F401
    import app.scheduled_jobs.evidence_precompile_handler  # noqa: F401
    import app.scheduled_jobs.feed_celebrations_handlers  # noqa: F401
    import app.scheduled_jobs.feishu_handlers  # noqa: F401
    import app.scheduled_jobs.feishu_helpdesk_handlers  # noqa: F401
    import app.scheduled_jobs.finance_handlers  # noqa: F401
    import app.scheduled_jobs.gamification_handlers  # noqa: F401
    import app.scheduled_jobs.github_org_handlers  # noqa: F401
    import app.scheduled_jobs.google_handlers  # noqa: F401
    import app.scheduled_jobs.handlers  # noqa: F401
    import app.scheduled_jobs.hr_leave_handlers  # noqa: F401
    import app.scheduled_jobs.hr_leave_insights_handlers  # noqa: F401
    import app.scheduled_jobs.hygiene_handlers  # noqa: F401
    import app.scheduled_jobs.identity_handlers  # noqa: F401
    import app.scheduled_jobs.idle_reclaim_handlers  # noqa: F401
    import app.scheduled_jobs.insights_handlers  # noqa: F401
    import app.scheduled_jobs.intelligence_handlers  # noqa: F401
    import app.scheduled_jobs.jira_handlers  # noqa: F401
    import app.scheduled_jobs.knowledge_loop_handlers  # noqa: F401
    import app.scheduled_jobs.mdm_handlers  # noqa: F401
    import app.scheduled_jobs.nexuscrm_handlers  # noqa: F401
    import app.scheduled_jobs.offboarding_assurance_handlers  # noqa: F401
    import app.scheduled_jobs.onboarding_handlers  # noqa: F401
    import app.scheduled_jobs.partner_directory_handlers  # noqa: F401
    import app.scheduled_jobs.permission_lifecycle_handlers  # noqa: F401
    import app.scheduled_jobs.posture_summary_handler  # noqa: F401
    import app.scheduled_jobs.procurement_handlers  # noqa: F401
    import app.scheduled_jobs.role_change_handlers  # noqa: F401
    import app.scheduled_jobs.secquiz_handlers  # noqa: F401
    import app.scheduled_jobs.self_attestation_advice_handlers  # noqa: F401
    import app.scheduled_jobs.slack_handlers  # noqa: F401
    import app.scheduled_jobs.storage_handlers  # noqa: F401
    import app.scheduled_jobs.support_feishu_kb_handlers  # noqa: F401
    import app.scheduled_jobs.support_kb_gap_handlers  # noqa: F401
    import app.scheduled_jobs.support_ticket_handlers  # noqa: F401
    import app.scheduled_jobs.usage_signal_handlers  # noqa: F401
    import app.scheduled_jobs.wrapped_handlers  # noqa: F401
    import app.scheduled_jobs.zoom_handlers  # noqa: F401

    stop_event = asyncio.Event()

    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig, stop_event.set)

    logger.info("worker started")
    scheduler, reload_task = await start_schedulers()

    # Foreground scheduler-liveness watchdog. Lives in the same event
    # loop as the main heartbeat so a hung loop also takes the watchdog
    # down (which means stop_event never gets set and Docker's healthcheck
    # would need to catch that — out of scope for this loop, see
    # docker-compose.prod.yml). The watchdog catches the case where the
    # loop is alive but APScheduler has stopped dispatching, which is
    # the case the 2026-05 incident exposed.
    watchdog_task = asyncio.create_task(
        _scheduler_watchdog_loop(stop_event),
    )

    while not stop_event.is_set():
        # Reaching this loop proves ``start_schedulers`` (including the
        # enabled-handler boot guard) completed. Updating the file from this
        # event loop also lets Docker distinguish a live worker from a process
        # whose loop is wedged while PID 1 remains present.
        WORKER_HEARTBEAT_PATH.touch()
        logger.info("worker heartbeat")
        try:
            await asyncio.wait_for(
                stop_event.wait(),
                timeout=WORKER_HEARTBEAT_INTERVAL_SECONDS,
            )
        except TimeoutError:
            continue

    logger.info("worker stopping")

    # Stop the scheduler and drain in-flight jobs BEFORE returning — i.e.
    # before ``asyncio.run`` tears down the loop's default executor. This
    # is what prevents the ``Executor shutdown has been called`` race on
    # every restart that overlaps a running sync (see
    # ``SHUTDOWN_DRAIN_GRACE_SECONDS``).
    await _stop_schedulers(scheduler, reload_task)

    # Await the watchdog so we can propagate its exit signal. ``stop_event``
    # was either set by SIGINT/SIGTERM (watchdog returns 0) or by the
    # watchdog itself (returns 1 — propagate so docker sees non-zero exit
    # and the ``restart: unless-stopped`` policy recreates the container).
    try:
        watchdog_exit = await asyncio.wait_for(watchdog_task, timeout=5)
    except (TimeoutError, asyncio.CancelledError):
        watchdog_task.cancel()
        watchdog_exit = 0
    return watchdog_exit


def main() -> None:
    exit_code = asyncio.run(_run())
    if exit_code:
        raise SystemExit(exit_code)


if __name__ == "__main__":
    main()