fix(jobs): make last-run writes atomic and drain them properly on shutdown

Two review findings on 22bee28, both reproduced before fixing.

[1] Concurrent writes could lose or rewind a row. record_finish did
select-then-insert-or-update, and pipelines are separate scheduler jobs that can
overlap while sharing step ids -- data_collector belongs to all four. Reproduced
both halves: two sessions that SELECT before either INSERTs make the second
commit raise IntegrityError, which _persist_job_run swallows, so the run
silently vanishes; and a later write carrying an OLDER finished_at rewound the
row from 12:00 back to 09:00, dragging the status with it, so the panel would
report a stale outcome as the latest.

Now a single atomic INSERT ... ON CONFLICT (job_name) DO UPDATE, guarded by
WHERE job_run_state.finished_at < excluded.finished_at so an older completion
can never overwrite a newer one. Dialect-specific because prod is Postgres and
tests are SQLite; both support it (SQLite >= 3.24, ours is 3.45). updated_at is
set explicitly, as the model's onupdate hook does not fire for a core upsert,
and get_map now uses populate_existing since core writes leave any
previously-loaded ORM instance stale in the identity map.

[2] Shutdown could dispose the engine underneath a pending write.
scheduler.shutdown(wait=False) returns before APScheduler dispatches its
completion events, and those events are what create persist tasks -- so a single
snapshot of the task set missed writes still to be queued. flush_job_run_persists
now settles briefly for pending callbacks, then drains in a loop until the set
stays empty, with the deadline still bounding total shutdown time. Left
shutdown(wait=False) alone deliberately: waiting would block a deploy restart
behind a long-running scan.

Six regression tests: interleaved first writes, older-never-rewinds,
newer-still-wins, a task queued mid-drain, prompt return when idle, and giving
up rather than hanging shutdown.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>

alembic/versions/031_job_run_state.py

@@ -0,0 +1,51 @@
+"""Durable last-run state per scheduled job
+
+Revision ID: 031
+Revises: 030
+Create Date: 2026-08-08 00:00:00.000000
+
+Job run state lived only in an in-memory dict in ``app.scheduler``, so every
+process restart wiped it. Admin → Jobs could then only report "Active" with no
+indication of whether a job had ever run, or how it ended — which is exactly
+the information an operator opens that page for.
+
+One row per job, upserted on ``job_name``. Not history: ``system_events``
+already grows unbounded with no retention job, and a second append-only
+operational table would repeat that debt.
+
+The table starts empty; each job populates its row the next time it finishes.
+No backfill from ``system_events`` — that table only records warning/error
+outcomes and uses a different status vocabulary, so seeding from it would
+invent successful runs that never happened.
+"""
+from typing import Sequence, Union
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "031"
+down_revision: Union[str, None] = "030"
+branch_labels: Union[str, Sequence[str], None] = None
+depends_on: Union[str, Sequence[str], None] = None
+
+
+def upgrade() -> None:
+    op.create_table(
+        "job_run_state",
+        sa.Column("id", sa.Integer(), nullable=False),
+        sa.Column("job_name", sa.String(length=64), nullable=False),
+        sa.Column("status", sa.String(length=32), nullable=False),
+        sa.Column("started_at", sa.DateTime(timezone=True), nullable=True),
+        sa.Column("finished_at", sa.DateTime(timezone=True), nullable=False),
+        sa.Column("processed", sa.Integer(), nullable=True),
+        sa.Column("total", sa.Integer(), nullable=True),
+        sa.Column("message", sa.Text(), nullable=True),
+        sa.Column("updated_at", sa.DateTime(timezone=True), nullable=False),
+        sa.PrimaryKeyConstraint("id"),
+        sa.UniqueConstraint("job_name", name="uq_job_run_state_job_name"),
+    )
+
+
+def downgrade() -> None:
+    op.drop_table("job_run_state")

app/job_catalog.py

@@ -0,0 +1,207 @@
+"""Job topology: names, labels, pipeline membership, categories, ordering.
+
+The single source of truth for *what the jobs are*, as opposed to how they run.
+It deliberately imports nothing from ``app`` so both ``app.scheduler`` and
+``app.services.admin_service`` can import it at module level -- admin_service
+otherwise has to do ``from app.scheduler import ...`` inside functions to dodge a
+cycle.
+
+The pipeline step lists live here rather than in the scheduler because three
+separate things need them and used to keep private copies: the runner, the
+``PIPELINE_MEMBERS`` set the admin API reports, and the UI's grouping.  Steps are
+``(step_name, coroutine_name)``; ``_run_pipeline`` resolves the coroutine late
+out of the scheduler's own globals, so nothing here depends on those functions
+existing.
+"""
+
+from __future__ import annotations
+
+# ---------------------------------------------------------------------------
+# Pipelines
+# ---------------------------------------------------------------------------
+
+_DAILY_PIPELINE_STEPS = [
+    ("data_collector", "collect_ohlcv"),
+    ("benchmark_collector", "collect_benchmark"),
+    ("sentiment_collector", "collect_sentiment"),
+    ("market_regime", "compute_market_regime"),
+    # Observational only — display/alerts; not trade selection.
+    ("regime_monitor", "compute_regime_monitor"),
+    # Alerts after regime so quadrant changes reach Telegram in the morning.
+    # Dispatcher is change-driven; quiet days stay quiet. Setup alerts still
+    # fire on the near-close pipeline after the qualifying scan.
+    ("alerts", "dispatch_alerts_job"),
+]
+
+# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (incremental
+# ingestion overlaps the latest stored session), then the only daily
+# qualifying R:R scan, then Telegram immediately so manual fills can still hit
+# MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan
+# may see ~15:15 prices — immaterial for a 12-1 momentum signal.
+#
+# US early-close days (~3/year, 13:00 ET close): this job runs post-close and
+# entries behave like stale_close (still acceptable per execution-recovery matrix).
+# No exchange calendar dependency.
+_NEAR_CLOSE_PIPELINE_STEPS = [
+    # Must land today's in-progress bar (~20 min behind live), or the scan falls
+    # back to the previous close and execution degrades to the stale_close floor.
+    ("data_collector", "collect_ohlcv_for_scan"),
+    ("rr_scanner", "scan_rr"),
+    # Straight after the scan so shadow entries mark at the same near-close
+    # prices the discretionary book is looking at.
+    ("shadow_book", "run_shadow_book"),
+    ("alerts", "dispatch_alerts_job"),
+]
+
+# After close (~16:45 ET Mon–Fri): fresh OHLCV fetch so outcomes resolve on the
+# final bar, not the near-close partial bar, then outcome/paper close.
+_AFTER_CLOSE_PIPELINE_STEPS = [
+    ("data_collector", "collect_ohlcv_final"),
+    ("outcome_evaluator", "evaluate_outcomes"),
+]
+
+# Intraday (light): keep prices current and resolve outcomes through the day,
+# without the expensive scan/sentiment. The dashboard recomputes live R:R from
+# the latest price, so refreshing OHLCV is enough to stop prices lagging; the
+# outcome step also closes paper trades that hit their stop/target intraday.
+_INTRADAY_PIPELINE_STEPS = [
+    ("data_collector", "collect_ohlcv"),
+    ("outcome_evaluator", "evaluate_outcomes"),
+]
+
+# Ordered by trading day, not alphabetically: this is the sequence an operator
+# reads down the page, and it drives the UI's ordering too.
+PIPELINE_STEPS: dict[str, list[tuple[str, str]]] = {
+    "daily_pipeline": _DAILY_PIPELINE_STEPS,
+    "intraday_pipeline": _INTRADAY_PIPELINE_STEPS,
+    "near_close_pipeline": _NEAR_CLOSE_PIPELINE_STEPS,
+    "after_close_pipeline": _AFTER_CLOSE_PIPELINE_STEPS,
+}
+
+# Derived, never hand-maintained: this used to be a literal set in admin_service
+# duplicating the four lists above from another module, with nothing asserting
+# the two agreed.
+PIPELINE_MEMBERS: frozenset[str] = frozenset(
+    step for steps in PIPELINE_STEPS.values() for step, _ in steps
+)
+
+
+def _pipelines_by_member() -> dict[str, tuple[str, ...]]:
+    """Member -> the orchestrators that run it, in trading-day order.
+
+    Membership is many-to-many: data_collector runs in all four pipelines (via
+    three different coroutines), alerts and outcome_evaluator in two each.
+    """
+    out: dict[str, list[str]] = {}
+    for pipeline, steps in PIPELINE_STEPS.items():
+        for step, _ in steps:
+            bucket = out.setdefault(step, [])
+            if pipeline not in bucket:
+                bucket.append(pipeline)
+    return {member: tuple(pipelines) for member, pipelines in out.items()}
+
+
+PIPELINES_BY_MEMBER: dict[str, tuple[str, ...]] = _pipelines_by_member()
+
+
+# ---------------------------------------------------------------------------
+# Job identity
+# ---------------------------------------------------------------------------
+
+# Orchestrators, in trading-day order.
+PIPELINE_JOBS: tuple[str, ...] = tuple(PIPELINE_STEPS)
+
+# Own timer, independent of any pipeline.
+SCHEDULED_JOBS: tuple[str, ...] = (
+    "dolt_earnings_import",
+    "sec_fundamentals_import",
+    "ticker_universe_sync",
+    "backtest",
+)
+
+# Registered but never auto-fired; run only when a human asks.
+MANUAL_JOBS: tuple[str, ...] = ("event_study", "data_backfill")
+
+# Steps in the order an operator meets them across the trading day, so the UI
+# reads as a sequence rather than an alphabetical jumble.
+PIPELINE_STEP_JOBS: tuple[str, ...] = tuple(
+    dict.fromkeys(step for steps in PIPELINE_STEPS.values() for step, _ in steps)
+)
+
+VALID_JOB_NAMES: frozenset[str] = frozenset(
+    PIPELINE_JOBS + PIPELINE_STEP_JOBS + SCHEDULED_JOBS + MANUAL_JOBS
+)
+
+JOB_LABELS: dict[str, str] = {
+    "data_collector": "Data Collector (OHLCV)",
+    "data_backfill": "Data Backfill (deep history)",
+    "benchmark_collector": "Benchmark Collector",
+    "sentiment_collector": "Sentiment Collector",
+    "dolt_earnings_import": "Dolt Earnings Import",
+    "sec_fundamentals_import": "SEC Fundamentals Import",
+    "rr_scanner": "R:R Scanner",
+    "ticker_universe_sync": "Ticker Universe Sync",
+    "outcome_evaluator": "Outcome Evaluator",
+    "alerts": "Alerts Dispatcher",
+    # Keys are persisted job ids and must not change; these are display only.
+    "market_regime": "Market Trend (SPY)",
+    "regime_monitor": "AI/Tech Risk Monitor",
+    "event_study": "Event Study",
+    "backtest": "Backtest",
+    "daily_pipeline": "Morning Pipeline",
+    "near_close_pipeline": "Near-Close Pipeline (scan+alert)",
+    "after_close_pipeline": "After-Close Pipeline (outcome)",
+    "intraday_pipeline": "Intraday Pipeline",
+    "shadow_book": "Shadow Book (auto-traded strategy)",
+}
+
+CATEGORY_PIPELINE = "pipeline"
+CATEGORY_STEP = "pipeline_step"
+CATEGORY_SCHEDULED = "scheduled"
+CATEGORY_MANUAL = "manual"
+
+# Order the sections appear in.
+CATEGORY_ORDER: tuple[str, ...] = (
+    CATEGORY_PIPELINE,
+    CATEGORY_STEP,
+    CATEGORY_SCHEDULED,
+    CATEGORY_MANUAL,
+)
+
+CATEGORY_LABELS: dict[str, str] = {
+    CATEGORY_PIPELINE: "Pipelines",
+    CATEGORY_STEP: "Pipeline steps",
+    CATEGORY_SCHEDULED: "Standalone scheduled",
+    CATEGORY_MANUAL: "Manual only",
+}
+
+_CATEGORY_MEMBERS: dict[str, tuple[str, ...]] = {
+    CATEGORY_PIPELINE: PIPELINE_JOBS,
+    CATEGORY_STEP: PIPELINE_STEP_JOBS,
+    CATEGORY_SCHEDULED: SCHEDULED_JOBS,
+    CATEGORY_MANUAL: MANUAL_JOBS,
+}
+
+JOB_CATEGORY: dict[str, str] = {
+    name: category
+    for category, names in _CATEGORY_MEMBERS.items()
+    for name in names
+}
+
+# Registered and triggerable through the API, but kept out of Admin → Jobs.
+# data_backfill's only capability beyond collect_ohlcv (which already backfills
+# full history for *new* tickers) is re-deepening *existing* ones after
+# ohlcv_history_days is raised -- a rare one-off, not something to scan past
+# every time you open the page.
+HIDDEN_JOBS: frozenset[str] = frozenset({"data_backfill"})
+
+_SORT_INDEX: dict[str, tuple[int, int]] = {
+    name: (CATEGORY_ORDER.index(category), position)
+    for category, names in _CATEGORY_MEMBERS.items()
+    for position, name in enumerate(names)
+}
+
+
+def sort_order(job_name: str) -> tuple[int, int]:
+    """(category rank, position within category). Unknown jobs sort last."""
+    return _SORT_INDEX.get(job_name, (len(CATEGORY_ORDER), 0))

app/main.py

@@ -21,7 +21,12 @@ from app.config import settings
 from app.database import async_session_factory, engine
 from app.middleware import register_exception_handlers
 from app.models.user import User
-from app.scheduler import configure_scheduler, load_schedule_config, scheduler
+from app.scheduler import (
+    configure_scheduler,
+    flush_job_run_persists,
+    load_schedule_config,
+    scheduler,
+)
 from app.routers.admin import router as admin_router
 from app.routers.auth import router as auth_router
 from app.routers.health import router as health_router
@@ -91,6 +96,9 @@ async def lifespan(_app: FastAPI) -> AsyncGenerator[None, None]:
 
     scheduler.shutdown(wait=False)
     logger.info("Scheduler stopped")
+    # Drain detached last-run writes before the engine goes away, or a job that
+    # finished during shutdown loses the row it just wrote.
+    await flush_job_run_persists()
     await engine.dispose()
     logger.info("Shutting down")
 

app/models/__init__.py

@@ -18,6 +18,7 @@ from app.models.benchmark_price import BenchmarkPrice
 from app.models.signal_context_snapshot import SignalContextSnapshot
 from app.models.system_event import SystemEvent
 from app.models.sec_filing_gap import SecFilingGap
+from app.models.job_run_state import JobRunState
 
 __all__ = [
     "Ticker",
@@ -42,4 +43,5 @@ __all__ = [
     "SignalContextSnapshot",
     "SystemEvent",
     "SecFilingGap",
+    "JobRunState",
 ]

app/models/job_run_state.py

@@ -0,0 +1,37 @@
+from datetime import datetime
+
+from sqlalchemy import DateTime, Integer, String, Text
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.database import Base
+
+
+class JobRunState(Base):
+    """How each scheduled job last finished. One row per job, overwritten.
+
+    The scheduler's ``_job_runtime`` dict is the live view and is deliberately
+    in-memory, but it is also wiped by every process restart -- so after a deploy
+    Admin → Jobs could only say "Active" with no indication of whether a job had
+    ever run. This is the durable half.
+
+    Deliberately not history: ``system_events`` already grows without a reaper,
+    and a second append-only operational table would repeat that. Rows are
+    upserted on ``job_name``; adding history later is purely additive.
+    """
+
+    __tablename__ = "job_run_state"
+
+    id: Mapped[int] = mapped_column(primary_key=True)
+    job_name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False)
+    # Scheduler vocabulary: completed | skipped | error | rate_limited | deferred.
+    # Distinct from data_import_runs' statuses, which is one reason this is its
+    # own table rather than a widened column there.
+    status: Mapped[str] = mapped_column(String(32), nullable=False)
+    started_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True), nullable=True)
+    finished_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), nullable=False)
+    processed: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    total: Mapped[int | None] = mapped_column(Integer, nullable=True)
+    message: Mapped[str | None] = mapped_column(Text, nullable=True)
+    updated_at: Mapped[datetime] = mapped_column(
+        DateTime(timezone=True), default=datetime.utcnow, onupdate=datetime.utcnow, nullable=False
+    )

app/scheduler.py

@@ -18,11 +18,13 @@ import logging
 import asyncio
 from datetime import date, datetime, timedelta, timezone
 
+from apscheduler.events import EVENT_JOB_ERROR, EVENT_JOB_EXECUTED
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
 from sqlalchemy import and_, case, func, or_, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app import job_catalog
 from app.config import settings
 from app.database import async_session_factory
 from app.models.ohlcv import OHLCVRecord
@@ -31,6 +33,7 @@ from app.models.ticker import Ticker
 from app.exceptions import ProviderError
 from app.providers.alpaca import AlpacaOHLCVProvider
 from app.providers.protocol import SentimentData
+from app.services import job_run_store
 from app.services import (
     ingestion_service,
     pipeline_run,
@@ -84,6 +87,47 @@ scheduler = AsyncIOScheduler(
     }
 )
 
+
+def _on_job_finished(event: object) -> None:
+    """Persist the run, then re-pause the job if it only runs on demand.
+
+    Covers every job APScheduler fires itself, including manual triggers.
+    Pipeline *steps* are invoked as plain coroutines and emit no events, so
+    ``_run_pipeline`` persists those directly.
+    """
+    job_id = getattr(event, "job_id", None)
+    if job_id:
+        _schedule_persist(job_id)
+    _repause_after_manual_run(event)
+
+
+def _repause_after_manual_run(event: object) -> None:
+    """Re-pause a job that only ever runs on demand, once its run finishes.
+
+    Pipeline steps and manual jobs are registered with a 520-week interval and
+    ``next_run_time=None`` as a backstop.  Triggering one sets next_run_time=now,
+    and APScheduler then re-arms that backstop -- so Admin → Jobs would show a
+    "next run" ten years out.  Guarding on category means the six cron jobs and
+    the real interval jobs are never touched.
+
+    Registered at module level, not inside ``configure_scheduler``: that function
+    is called more than once (idempotency test) and ``add_listener`` does not
+    deduplicate.
+    """
+    job_id = getattr(event, "job_id", None)
+    if job_catalog.JOB_CATEGORY.get(job_id) not in (
+        job_catalog.CATEGORY_STEP,
+        job_catalog.CATEGORY_MANUAL,
+    ):
+        return
+    try:
+        scheduler.modify_job(job_id, next_run_time=None)
+    except Exception:  # job gone, scheduler stopped — nothing to re-pause
+        logger.debug("Could not re-pause %s after its run", job_id, exc_info=True)
+
+
+scheduler.add_listener(_on_job_finished, EVENT_JOB_EXECUTED | EVENT_JOB_ERROR)
+
 # Track last successful ticker per job for rate-limit resume
 _last_successful: dict[str, str | None] = {
     "data_collector": None,
@@ -91,26 +135,10 @@ _last_successful: dict[str, str | None] = {
     "sentiment_collector": None,
 }
 
-# Jobs whose per-run progress is surfaced to Admin → Jobs. (outcome_evaluator is
+# Seeded from the catalog rather than a private list. The old literal held 16 of
-# created lazily on first run via _runtime_start.)
+# the 19 jobs -- benchmark_collector, outcome_evaluator and shadow_book were
-_JOB_NAMES = [
+# missing, so they had no runtime row (and so no "last run" line in Admin → Jobs)
-    "data_collector",
+# until their first run in a given process.
-    "data_backfill",
-    "sentiment_collector",
-    "dolt_earnings_import",
-    "sec_fundamentals_import",
-    "rr_scanner",
-    "ticker_universe_sync",
-    "alerts",
-    "market_regime",
-    "regime_monitor",
-    "event_study",
-    "backtest",
-    "daily_pipeline",  # morning: OHLCV/sentiment/regime — no qualifying scan
-    "near_close_pipeline",  # OHLCV fetch → R:R scan → Telegram alerts
-    "after_close_pipeline",  # OHLCV fetch → outcome eval (final bar)
-    "intraday_pipeline",
-]
 
 
 def _idle_runtime() -> dict[str, object]:
@@ -127,7 +155,9 @@ def _idle_runtime() -> dict[str, object]:
     }
 
 
-_job_runtime: dict[str, dict[str, object]] = {name: _idle_runtime() for name in _JOB_NAMES}
+_job_runtime: dict[str, dict[str, object]] = {
+    name: _idle_runtime() for name in sorted(job_catalog.VALID_JOB_NAMES)
+}
 _next_backtest_target_model = PRODUCTION_GTL_TARGET_MODEL
 _next_backtest_cadence = DEFAULT_BACKTEST_CADENCE
 
@@ -292,6 +322,67 @@ def _runtime_finish(
             pass
 
 
+async def _persist_job_run(job_name: str) -> None:
+    """Write a job's finished runtime row to the durable last-run table.
+
+    Never raises: a persistence failure must not break the pipeline that was
+    otherwise successful. The in-memory row stays authoritative for live state.
+    """
+    runtime = _job_runtime.get(job_name)
+    if not runtime or runtime.get("running") or not runtime.get("finished_at"):
+        return
+    try:
+        async with async_session_factory() as db:
+            await job_run_store.record_finish(db, job_name, runtime)
+            await db.commit()
+    except Exception:
+        logger.exception("Could not persist last-run state for %s", job_name)
+
+
+# Detached persists are kept referenced: a bare create_task result can be
+# garbage-collected mid-flight, and the shutdown drain needs something to await.
+_persist_tasks: set[asyncio.Task] = set()
+
+
+def _schedule_persist(job_name: str) -> None:
+    try:
+        task = asyncio.get_running_loop().create_task(_persist_job_run(job_name))
+    except RuntimeError:  # no loop (sync context / tests) — nothing to persist
+        return
+    _persist_tasks.add(task)
+    task.add_done_callback(_persist_tasks.discard)
+
+
+async def flush_job_run_persists(timeout: float = 5.0, settle: float = 0.05) -> None:
+    """Drain last-run writes, including ones queued while we are draining.
+
+    ``scheduler.shutdown(wait=False)`` returns before APScheduler has dispatched
+    its job-completion events, and those events are what create persist tasks. A
+    single snapshot of the set therefore misses writes still to be queued, and
+    ``engine.dispose()`` could then close the pool underneath them. So: give the
+    loop a moment for pending callbacks to land, then keep draining until the
+    set stays empty or the deadline passes.
+    """
+    loop = asyncio.get_running_loop()
+    deadline = loop.time() + timeout
+    # Bounded settle so callbacks dispatched by shutdown get to queue their work
+    # before the first emptiness check decides there is nothing to wait for.
+    await asyncio.sleep(min(settle, timeout))
+    while True:
+        pending = {task for task in _persist_tasks if not task.done()}
+        if not pending:
+            return
+        remaining = deadline - loop.time()
+        if remaining <= 0:
+            logger.warning(
+                "Timed out draining %d last-run write(s); some may be lost", len(pending)
+            )
+            return
+        await asyncio.wait(pending, timeout=remaining)
+        # Loop rather than return: a completion callback may have queued another.
+        await asyncio.sleep(0)
+
+
 def get_job_runtime_snapshot(job_name: str | None = None) -> dict[str, dict[str, object]] | dict[str, object]:
     if job_name is not None:
         return dict(_job_runtime.get(job_name, {}))
@@ -1300,54 +1391,14 @@ async def sync_ticker_universe() -> None:
 # the intraday partial one (covers a long weekend / holiday gap).
 _FINAL_REFETCH_DAYS = 5
 
-_DAILY_PIPELINE_STEPS = [
+# Step lists live in app.job_catalog so the runner, the admin API's pipeline
-    ("data_collector", "collect_ohlcv"),
+# membership and the UI's grouping all read one definition. Re-exported here
-    ("benchmark_collector", "collect_benchmark"),
+# under their original names: _run_pipeline and the scheduler_configured log
-    ("sentiment_collector", "collect_sentiment"),
+# payload refer to them directly.
-    ("market_regime", "compute_market_regime"),
+_DAILY_PIPELINE_STEPS = job_catalog._DAILY_PIPELINE_STEPS
-    # Observational only — display/alerts; not trade selection.
+_NEAR_CLOSE_PIPELINE_STEPS = job_catalog._NEAR_CLOSE_PIPELINE_STEPS
-    ("regime_monitor", "compute_regime_monitor"),
+_AFTER_CLOSE_PIPELINE_STEPS = job_catalog._AFTER_CLOSE_PIPELINE_STEPS
-    # Alerts after regime so quadrant changes reach Telegram in the morning.
+_INTRADAY_PIPELINE_STEPS = job_catalog._INTRADAY_PIPELINE_STEPS
-    # Dispatcher is change-driven; quiet days stay quiet. Setup alerts still
-    # fire on the near-close pipeline after the qualifying scan.
-    ("alerts", "dispatch_alerts_job"),
-]
-
-# Near-close (~15:30 ET Mon–Fri): refresh in-progress day-t bars (incremental
-# ingestion overlaps the latest stored session), then the only daily
-# qualifying R:R scan, then Telegram immediately so manual fills can still hit
-# MOC cutoffs (~15:50/15:55). Under a 15-minute delayed SIP feed a 15:30 scan
-# may see ~15:15 prices — immaterial for a 12-1 momentum signal.
-#
-# US early-close days (~3/year, 13:00 ET close): this job runs post-close and
-# entries behave like stale_close (still acceptable per execution-recovery matrix).
-# No exchange calendar dependency.
-_NEAR_CLOSE_PIPELINE_STEPS = [
-    # Must land today's in-progress bar (~20 min behind live), or the scan falls
-    # back to the previous close and execution degrades to the stale_close floor.
-    ("data_collector", "collect_ohlcv_for_scan"),
-    ("rr_scanner", "scan_rr"),
-    # Straight after the scan so shadow entries mark at the same near-close
-    # prices the discretionary book is looking at.
-    ("shadow_book", "run_shadow_book"),
-    ("alerts", "dispatch_alerts_job"),
-]
-
-# After close (~16:45 ET Mon–Fri): fresh OHLCV fetch so outcomes resolve on the
-# final bar, not the near-close partial bar, then outcome/paper close.
-_AFTER_CLOSE_PIPELINE_STEPS = [
-    ("data_collector", "collect_ohlcv_final"),
-    ("outcome_evaluator", "evaluate_outcomes"),
-]
-
-# Intraday (light): keep prices current and resolve outcomes through the day,
-# without the expensive scan/sentiment. The dashboard recomputes live R:R from
-# the latest price, so refreshing OHLCV is enough to stop prices lagging; the
-# outcome step also closes paper trades that hit their stop/target intraday.
-_INTRADAY_PIPELINE_STEPS = [
-    ("data_collector", "collect_ohlcv"),
-    ("outcome_evaluator", "evaluate_outcomes"),
-]
 
 # Warn if near-close fetch+scan+alert drifts past this — entries leave the close
 # and the stale_close floor quietly becomes the ceiling.
@@ -1370,6 +1421,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
         if not await _is_job_enabled(db, job_name):
             _log_event(logging.INFO, "job_skipped", job=job_name, reason="disabled")
             _runtime_finish(job_name, "skipped", processed=0, total=0, message="Disabled")
+            await _persist_job_run(job_name)
             return
 
     total = len(steps)
@@ -1385,6 +1437,11 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
                 await funcs[func_name]()
             except Exception:
                 logger.exception("%s step %s failed", job_name, step_name)
+            # Outside the except on purpose: the step's own _runtime_finish has
+            # already recorded its outcome, so persisting here captures failures
+            # too. Steps are plain coroutine calls and fire no scheduler events,
+            # so the listener cannot see them -- this is their only write path.
+            await _persist_job_run(step_name)
             done += 1
         _runtime_finish(job_name, "completed", processed=done, total=total, message="Pipeline complete")
         _log_event(logging.INFO, "job_complete", job=job_name)
@@ -1393,6 +1450,7 @@ async def _run_pipeline(job_name: str, steps: list[tuple[str, str]]) -> None:
         _log_event(logging.ERROR, "job_error", job=job_name, error_type=type(exc).__name__, message=str(exc))
     finally:
         pipeline_run.release(token)
+        await _persist_job_run(job_name)
 
 
 async def run_daily_pipeline() -> None:
@@ -1486,6 +1544,12 @@ SCHEDULE_DEFAULTS: dict[str, str] = {
     "schedule_after_close_pipeline_cron": "45 16 * * mon-fri",
     # Hourly mid-session price + outcome (10:00–15:00 ET Mon–Fri).
     "schedule_intraday_pipeline_cron": "0 10-15 * * mon-fri",
+    # Both were interval jobs until 2026-08-08 and hit exactly the pitfall
+    # described above: configure_scheduler calls remove_all_jobs() on every
+    # startup, so an interval countdown restarts from zero each deploy. A 168h
+    # backtest needed a week of uninterrupted uptime to fire even once.
+    "schedule_backtest_cron": "0 3 * * sun",
+    "schedule_ticker_universe_cron": "0 1 * * *",
 }
 
 # job id -> schedule setting key
@@ -1496,6 +1560,8 @@ _CRON_JOBS: dict[str, str] = {
     "near_close_pipeline": "schedule_near_close_pipeline_cron",
     "after_close_pipeline": "schedule_after_close_pipeline_cron",
     "intraday_pipeline": "schedule_intraday_pipeline_cron",
+    "backtest": "schedule_backtest_cron",
+    "ticker_universe_sync": "schedule_ticker_universe_cron",
 }
 
 
@@ -1630,9 +1696,12 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         id="intraday_pipeline", name="Intraday Pipeline", replace_existing=True,
     )
 
-    # Independent interval jobs (own cadence, no ordering dependency)
+    # Independent jobs (own cadence, no ordering dependency). Cron, not interval,
+    # for the reason documented at SCHEDULE_DEFAULTS: an interval countdown
+    # restarts on every deploy, so these could be deferred indefinitely.
     scheduler.add_job(
-        sync_ticker_universe, "interval", hours=24,
+        sync_ticker_universe,
+        _cron_trigger(cfg["schedule_ticker_universe_cron"], tz, "schedule_ticker_universe_cron"),
         id="ticker_universe_sync", name="Ticker Universe Sync", replace_existing=True,
     )
     # Alerts auto-fire only via near_close_pipeline (scan → alert before MOC).
@@ -1643,7 +1712,8 @@ def configure_scheduler(schedule_config: dict[str, str] | None = None) -> None:
         replace_existing=True, next_run_time=None,
     )
     scheduler.add_job(
-        run_backtest_job, "interval", hours=168,
+        run_backtest_job,
+        _cron_trigger(cfg["schedule_backtest_cron"], tz, "schedule_backtest_cron"),
         id="backtest", name="Backtest", replace_existing=True,
     )
     # Deep history backfill: manual only (never auto-fires); triggered from

app/schemas/admin.py

@@ -83,6 +83,8 @@ class ScheduleConfigUpdate(BaseModel):
     schedule_near_close_pipeline_cron: str | None = Field(default=None, max_length=120)
     schedule_after_close_pipeline_cron: str | None = Field(default=None, max_length=120)
     schedule_intraday_pipeline_cron: str | None = Field(default=None, max_length=120)
+    schedule_backtest_cron: str | None = Field(default=None, max_length=120)
+    schedule_ticker_universe_cron: str | None = Field(default=None, max_length=120)
 
 
 class PerformanceConfigUpdate(BaseModel):

app/services/admin_service.py

@@ -7,6 +7,7 @@ from passlib.hash import bcrypt
 from sqlalchemy import delete, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
+from app import job_catalog
 from app.exceptions import DuplicateError, NotFoundError, ValidationError
 from app.models.fundamental import FundamentalData
 from app.models.ohlcv import OHLCVRecord
@@ -17,7 +18,7 @@ from app.models.settings import SystemSetting
 from app.models.ticker import Ticker
 from app.models.trade_setup import TradeSetup
 from app.models.user import User
-from app.services import settings_store
+from app.services import job_run_store, settings_store
 
 logger = logging.getLogger(__name__)
 
@@ -606,91 +607,110 @@ async def get_pipeline_readiness(db: AsyncSession) -> list[dict]:
 # Job control (placeholder — scheduler is Task 12.1)
 # ---------------------------------------------------------------------------
 
-VALID_JOB_NAMES = {
+# Job identity, labels and pipeline membership now live in app.job_catalog, which
-    "data_collector",
+# derives PIPELINE_MEMBERS from the pipeline step lists instead of restating them.
-    "data_backfill",
+# Re-exported here because callers (routers, tests) import them from this module.
-    "benchmark_collector",
+VALID_JOB_NAMES = job_catalog.VALID_JOB_NAMES
-    "sentiment_collector",
+JOB_LABELS = job_catalog.JOB_LABELS
-    "dolt_earnings_import",
+PIPELINE_MEMBERS = job_catalog.PIPELINE_MEMBERS
-    "sec_fundamentals_import",
-    "rr_scanner",
-    "ticker_universe_sync",
-    "outcome_evaluator",
-    "alerts",
-    "market_regime",
-    "regime_monitor",
-    "event_study",
-    "backtest",
-    "daily_pipeline",
-    "near_close_pipeline",
-    "after_close_pipeline",
-    "intraday_pipeline",
-    "shadow_book",
-}
 
-JOB_LABELS = {
+# Anything further out than this is a parked backstop, not a schedule: pipeline
-    "data_collector": "Data Collector (OHLCV)",
+# steps and manual jobs are registered on a 520-week interval, and triggering one
-    "data_backfill": "Data Backfill (deep history)",
+# re-arms it. Belt-and-braces behind the category rule in _next_run_fields.
-    "benchmark_collector": "Benchmark Collector",
+_NEXT_RUN_HORIZON_DAYS = 365
-    "sentiment_collector": "Sentiment Collector",
-    "dolt_earnings_import": "Dolt Earnings Import",
-    "sec_fundamentals_import": "SEC Fundamentals Import",
-    "rr_scanner": "R:R Scanner",
-    "ticker_universe_sync": "Ticker Universe Sync",
-    "outcome_evaluator": "Outcome Evaluator",
-    "alerts": "Alerts Dispatcher",
-    # Keys are persisted job ids and must not change; these are display only.
-    "market_regime": "Market Trend (SPY)",
-    "regime_monitor": "AI/Tech Risk Monitor",
-    "event_study": "Event Study",
-    "backtest": "Backtest",
-    "daily_pipeline": "Morning Pipeline",
-    "near_close_pipeline": "Near-Close Pipeline (scan+alert)",
-    "after_close_pipeline": "After-Close Pipeline (outcome)",
-    "intraday_pipeline": "Intraday Pipeline",
-    "shadow_book": "Shadow Book (auto-traded strategy)",
-}
 
-# Jobs driven by a pipeline (in order) rather than their own auto timer.
+
-PIPELINE_MEMBERS = {
+def _visible_next_run(next_run: datetime | None) -> datetime | None:
-    "data_collector",
+    """Drop a next-run that is really the parked backstop."""
-    "benchmark_collector",
+    if next_run is None:
-    "sentiment_collector",
+        return None
-    "rr_scanner",
+    horizon = datetime.now(next_run.tzinfo) + timedelta(days=_NEXT_RUN_HORIZON_DAYS)
-    "outcome_evaluator",
+    return None if next_run > horizon else next_run
-    "alerts",
+
-    "market_regime",
+
-    "regime_monitor",
+def _own_next_run(scheduler, name: str) -> datetime | None:
-    "shadow_book",
+    # getattr: APScheduler only sets next_run_time once the scheduler is running,
+    # so a job registered but not yet started has no such attribute at all.
+    job = scheduler.get_job(name)
+    return _visible_next_run(getattr(job, "next_run_time", None)) if job else None
+
+
+def _next_run_fields(scheduler, name: str, enabled_map: dict[str, bool]) -> dict:
+    """Where this job's next run comes from, decided by category not by clock.
+
+    A pipeline step has no meaningful schedule of its own, so reporting one is
+    the bug: its parent's timer is the answer. Manual jobs have no answer at all,
+    and saying so beats rendering a parked backstop as a date.
+    """
+    category = job_catalog.JOB_CATEGORY.get(name)
+    if category == job_catalog.CATEGORY_STEP:
+        parents = job_catalog.PIPELINES_BY_MEMBER.get(name, ())
+        soonest: datetime | None = None
+        via: str | None = None
+        for parent in parents:
+            if not enabled_map.get(parent, True):
+                continue
+            candidate = _own_next_run(scheduler, parent)
+            if candidate is not None and (soonest is None or candidate < soonest):
+                soonest, via = candidate, parent
+        return {
+            "next_run_at": None,
+            "next_run_source": "via_pipeline",
+            "via_next_run_at": soonest.isoformat() if soonest else None,
+            "via_next_run_job": via,
+        }
+    if category == job_catalog.CATEGORY_MANUAL:
+        return {
+            "next_run_at": None,
+            "next_run_source": "manual_only",
+            "via_next_run_at": None,
+            "via_next_run_job": None,
+        }
+    own = _own_next_run(scheduler, name)
+    return {
+        "next_run_at": own.isoformat() if own else None,
+        "next_run_source": "own_schedule",
+        "via_next_run_at": None,
+        "via_next_run_job": None,
     }
 
 
 async def list_jobs(db: AsyncSession) -> list[dict]:
-    """Return status of all scheduled jobs."""
+    """Return status of all scheduled jobs, grouped and ordered by category."""
     from app.scheduler import get_job_runtime_snapshot, scheduler
 
+    visible = sorted(VALID_JOB_NAMES - job_catalog.HIDDEN_JOBS, key=job_catalog.sort_order)
+    # One query for every flag instead of one per job. Parents are read too, since
+    # a step reports its parent's next run only while that parent is enabled.
+    flags = await settings_store.get_map(
+        db, [f"job_{name}_enabled" for name in VALID_JOB_NAMES]
+    )
+    enabled_map = {
+        name: flags.get(f"job_{name}_enabled", "true") == "true"
+        for name in VALID_JOB_NAMES
+    }
+    last_runs = await job_run_store.get_map(db, visible)
+
     jobs_out = []
-    for name in sorted(VALID_JOB_NAMES):
+    for name in visible:
-        # Check enabled setting
-        setting = await settings_store.get_setting(db, f"job_{name}_enabled")
-        enabled = setting.value == "true" if setting else True  # default enabled
-
-        # Get scheduler job info
         job = scheduler.get_job(name)
-        next_run = None
-        if job and job.next_run_time:
-            next_run = job.next_run_time.isoformat()
-
         runtime = get_job_runtime_snapshot(name)
+        last = last_runs.get(name)
 
         jobs_out.append({
             "name": name,
             "label": JOB_LABELS.get(name, name),
-            "enabled": enabled,
+            "enabled": enabled_map.get(name, True),
-            "next_run_at": next_run,
+            "category": job_catalog.JOB_CATEGORY.get(name),
-            "via_pipeline": name in PIPELINE_MEMBERS,
+            "sort_order": job_catalog.sort_order(name),
+            # Parent pipelines for a step; the steps themselves for a pipeline.
+            "pipelines": list(job_catalog.PIPELINES_BY_MEMBER.get(name, ())),
+            "steps": [step for step, _ in job_catalog.PIPELINE_STEPS.get(name, ())],
             "registered": job is not None,
             "running": bool(runtime.get("running", False)),
+            # runtime_* are strictly live in-memory state. Persisted history is
+            # reported separately as last_run_*, so a stale error cannot pin the
+            # status chip or the rate-limit banner.
             "runtime_status": runtime.get("status"),
             "runtime_processed": runtime.get("processed"),
             "runtime_total": runtime.get("total"),
@@ -699,6 +719,15 @@ async def list_jobs(db: AsyncSession) -> list[dict]:
             "runtime_started_at": runtime.get("started_at"),
             "runtime_finished_at": runtime.get("finished_at"),
             "runtime_message": runtime.get("message"),
+            # Survives restarts, unlike runtime_*. Reported separately so the
+            # status chip keeps meaning "state now" rather than "last outcome,
+            # forever" -- an error a week ago must not read as Inactive today.
+            "last_run_at": last.finished_at.isoformat() if last else None,
+            "last_run_status": last.status if last else None,
+            "last_run_message": last.message if last else None,
+            "last_run_processed": last.processed if last else None,
+            "last_run_total": last.total if last else None,
+            **_next_run_fields(scheduler, name, enabled_map),
         })
 
     return jobs_out

app/services/job_run_store.py

@@ -0,0 +1,91 @@
+"""Single source for JobRunState reads/writes.
+
+Mirrors ``settings_store``: ``record_finish`` never commits — the caller owns
+the transaction — and reads are batched so the admin listing stays one query.
+"""
+
+from __future__ import annotations
+
+import logging
+from collections.abc import Iterable
+from datetime import datetime, timezone
+
+from sqlalchemy import select
+from sqlalchemy.dialects.postgresql import insert as pg_insert
+from sqlalchemy.dialects.sqlite import insert as sqlite_insert
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.models.job_run_state import JobRunState
+
+logger = logging.getLogger(__name__)
+
+
+def _as_datetime(value: object) -> datetime | None:
+    """Runtime snapshots carry ISO strings; the column wants a datetime."""
+    if isinstance(value, datetime):
+        return value
+    if isinstance(value, str) and value:
+        try:
+            return datetime.fromisoformat(value)
+        except ValueError:
+            return None
+    return None
+
+
+async def get_map(db: AsyncSession, job_names: Iterable[str]) -> dict[str, JobRunState]:
+    """Return {job_name: row} for the given jobs that have ever finished.
+
+    ``populate_existing`` because rows are written by core upserts, which leave
+    any previously-loaded ORM instance in the identity map stale.
+    """
+    result = await db.execute(
+        select(JobRunState)
+        .where(JobRunState.job_name.in_(list(job_names)))
+        .execution_options(populate_existing=True)
+    )
+    return {row.job_name: row for row in result.scalars().all()}
+
+
+def _insert_for(db: AsyncSession):
+    """ON CONFLICT is dialect-specific; prod is Postgres, tests are SQLite."""
+    dialect = db.get_bind().dialect.name
+    return pg_insert if dialect == "postgresql" else sqlite_insert
+
+
+async def record_finish(db: AsyncSession, job_name: str, runtime: dict) -> None:
+    """Upsert the last-run row from a scheduler runtime snapshot.
+
+    Atomic, and newer-wins. Select-then-insert loses races that really happen
+    here: pipelines are separate scheduler jobs that can overlap, and they share
+    step ids -- data_collector belongs to all four. Two of them finishing that
+    step together would both see no row and both insert, and the loser's
+    IntegrityError is swallowed by the caller, so the run silently vanishes.
+
+    The ``where`` guard is the other half: without it a slower pipeline
+    finishing an *older* run last would rewind finished_at and the status with
+    it, so the panel would report a stale outcome as the latest one.
+    """
+    finished_at = _as_datetime(runtime.get("finished_at")) or datetime.now(timezone.utc)
+    message = runtime.get("message")
+    now = datetime.now(timezone.utc)
+    values = {
+        "job_name": job_name,
+        "status": str(runtime.get("status") or "completed"),
+        "started_at": _as_datetime(runtime.get("started_at")),
+        "finished_at": finished_at,
+        "processed": runtime.get("processed"),
+        "total": runtime.get("total"),
+        "message": str(message)[:4000] if message else None,
+        # Set explicitly: the model's onupdate hook does not fire for a core
+        # INSERT ... ON CONFLICT DO UPDATE.
+        "updated_at": now,
+    }
+
+    statement = _insert_for(db)(JobRunState).values(**values)
+    await db.execute(
+        statement.on_conflict_do_update(
+            index_elements=[JobRunState.job_name],
+            set_={key: statement.excluded[key] for key in values if key != "job_name"},
+            where=JobRunState.finished_at < statement.excluded.finished_at,
+        )
+    )

frontend/src/api/admin.ts

@@ -207,14 +207,28 @@ export function backfillTickerNames() {
 }
 
 // Jobs
+export type JobCategory = 'pipeline' | 'pipeline_step' | 'scheduled' | 'manual';
+export type NextRunSource = 'own_schedule' | 'via_pipeline' | 'manual_only';
+
 export interface JobStatus {
   name: string;
   label: string;
   enabled: boolean;
-  next_run_at: string | null;
-  via_pipeline?: boolean;
   registered: boolean;
+  category?: JobCategory;
+  /** Server-assigned ordering; the payload already arrives grouped by it. */
+  sort_order?: [number, number];
+  /** Parent pipelines for a step. Many-to-many: data_collector runs in all four. */
+  pipelines?: string[];
+  /** Step names, for a pipeline row. */
+  steps?: string[];
+  next_run_at: string | null;
+  next_run_source?: NextRunSource;
+  /** For a step: the soonest enabled parent's next run, and which parent. */
+  via_next_run_at?: string | null;
+  via_next_run_job?: string | null;
   running?: boolean;
+  /** runtime_* is live, in-memory state only — it resets when the app restarts. */
   runtime_status?: string | null;
   runtime_processed?: number | null;
   runtime_total?: number | null;
@@ -223,6 +237,13 @@ export interface JobStatus {
   runtime_started_at?: string | null;
   runtime_finished_at?: string | null;
   runtime_message?: string | null;
+  /** last_run_* is persisted and survives restarts. Kept separate from
+   *  runtime_* so a stale error cannot pin the status chip or the banner. */
+  last_run_at?: string | null;
+  last_run_status?: string | null;
+  last_run_message?: string | null;
+  last_run_processed?: number | null;
+  last_run_total?: number | null;
 }
 
 export interface TriggerJobResponse {

frontend/src/components/admin/JobControls.tsx

@@ -1,4 +1,5 @@
 import { useJobs, useToggleJob, useTriggerJob } from '../../hooks/useAdmin';
+import type { JobCategory, JobStatus } from '../../api/admin';
 import { SkeletonTable } from '../ui/Skeleton';
 
 function formatNextRun(iso: string | null): string {
@@ -10,7 +11,8 @@ function formatNextRun(iso: string | null): string {
   const mins = Math.round(diffMs / 60_000);
   if (mins < 60) return `in ${mins}m`;
   const hrs = Math.round(mins / 60);
-  return `in ${hrs}h`;
+  if (hrs < 48) return `in ${hrs}h`;
+  return `in ${Math.round(hrs / 24)}d`;
 }
 
 function formatAgo(iso: string | null | undefined): string {
@@ -29,85 +31,95 @@ function lastRunColor(status: string | null | undefined): string {
   return 'text-gray-500';
 }
 
-export function JobControls() {
+/** The four kinds of job, in the order the API already sorts them. A job whose
-  const { data: jobs, isLoading } = useJobs();
+ *  category the client does not recognise still renders, under "Other" — better
-  const toggleJob = useToggleJob();
+ *  a stray section than a job that silently vanishes from the admin page. */
-  const triggerJob = useTriggerJob();
+const SECTIONS: { key: JobCategory; title: string; hint: string }[] = [
-  const anyJobRunning = (jobs ?? []).some((job) => job.running);
+  {
-  const runningJob = jobs?.find((job) => job.running);
+    key: 'pipeline',
-  const pausedJob = jobs?.find((job) => !job.running && job.runtime_status === 'rate_limited');
+    title: 'Pipelines',
-  const runningJobLabel = runningJob?.label;
+    hint: 'own schedule · run their steps in order',
-
+  },
-  if (isLoading) return <SkeletonTable rows={4} cols={3} />;
+  {
+    key: 'pipeline_step',
+    title: 'Pipeline steps',
+    hint: 'no timer of their own · still triggerable individually',
+  },
+  {
+    key: 'scheduled',
+    title: 'Standalone scheduled',
+    hint: 'own schedule · independent of any pipeline',
+  },
+  { key: 'manual', title: 'Manual only', hint: 'never fires on its own' },
+];
 
+/** One consistent answer per job: its own timer, its parent's, or "manual only".
+ *  A step has no schedule of its own, so reporting one was the original bug. */
+function NextRun({ job, labels }: { job: JobStatus; labels: Record<string, string> }) {
+  const muted = 'text-[11px] text-gray-500';
+  if (job.next_run_source === 'manual_only') {
+    return <span className={muted}>manual only</span>;
+  }
+  if (job.next_run_source === 'via_pipeline') {
+    if (!job.via_next_run_at || !job.via_next_run_job) {
+      return <span className={muted}>runs via pipeline</span>;
+    }
     return (
-    <div className="space-y-3">
+      <span className={muted}>
-      {runningJob && (
+        Next via {labels[job.via_next_run_job] ?? job.via_next_run_job}{' '}
-        <div className="rounded-xl border border-blue-400/30 bg-blue-500/10 px-4 py-3">
+        {formatNextRun(job.via_next_run_at)}
-          <div className="flex flex-wrap items-center justify-between gap-3">
+      </span>
-            <div>
+    );
-              <div className="text-xs font-semibold text-blue-300">
+  }
-                Active job: {runningJob.label}
+  if (!job.next_run_at) return null;
-              </div>
+  return <span className={muted}>Next run {formatNextRun(job.next_run_at)}</span>;
-              <div className="mt-0.5 text-[11px] text-blue-100/80">
+}
-                Manual triggers are blocked until this run finishes.
-              </div>
-            </div>
-            <div className="text-[11px] text-blue-200">
-              {runningJob.runtime_processed ?? 0}
-              {typeof runningJob.runtime_total === 'number'
-                ? ` / ${runningJob.runtime_total}`
-                : ''}
-            </div>
-          </div>
-          <div className="mt-2 h-1.5 w-full rounded-full bg-slate-700/80 overflow-hidden">
-            <div
-              className="h-full bg-blue-400 transition-all duration-500"
-              style={{
-                width: `${
-                  typeof runningJob.runtime_progress_pct === 'number'
-                    ? Math.max(5, Math.min(100, runningJob.runtime_progress_pct))
-                    : 30
-                }%`,
-              }}
-            />
-          </div>
-          {runningJob.runtime_current_ticker && (
-            <div className="mt-1 text-[11px] text-blue-100/80">
-              Current: {runningJob.runtime_current_ticker}
-            </div>
-          )}
-          {runningJob.runtime_message && (
-            <div className="mt-1 text-[11px] text-blue-100/80">
-              {runningJob.runtime_message}
-            </div>
-          )}
-        </div>
-      )}
 
-      {!runningJob && pausedJob && (
+/** Membership, shown rather than nested: a step can belong to several pipelines
-        <div className="rounded-xl border border-amber-400/30 bg-amber-500/10 px-4 py-3">
+ *  (data_collector is in all four), so duplicating rows under each parent would
-          <div className="flex flex-wrap items-center justify-between gap-3">
+ *  render Trigger buttons that are not distinct actions. */
-            <div>
+function Membership({ job, labels }: { job: JobStatus; labels: Record<string, string> }) {
-              <div className="text-xs font-semibold text-amber-300">
+  const name = (id: string) => labels[id] ?? id;
-                Last run paused: {pausedJob.label}
+  if (job.category === 'pipeline' && job.steps?.length) {
+    return (
+      <div className="mt-1 text-[11px] leading-relaxed text-gray-600">
+        {job.steps.map(name).join(' → ')}
       </div>
-              <div className="mt-0.5 text-[11px] text-amber-100/90">
+    );
-                {pausedJob.runtime_message || 'Rate limit hit. The collector stopped early and will resume from last progress on the next run.'}
+  }
+  if (job.category === 'pipeline_step' && job.pipelines?.length) {
+    return (
+      <div className="mt-1 text-[11px] leading-relaxed text-gray-600">
+        runs in: {job.pipelines.map(name).join(', ')}
       </div>
-            </div>
+    );
-            <div className="text-[11px] text-amber-200">
+  }
-              {pausedJob.runtime_processed ?? 0}
+  return null;
-              {typeof pausedJob.runtime_total === 'number'
+}
-                ? ` / ${pausedJob.runtime_total}`
-                : ''}
-            </div>
-          </div>
-        </div>
-      )}
 
-      {jobs?.map((job) => (
+interface JobCardProps {
-        <div key={job.name} className="glass p-4 glass-hover">
+  job: JobStatus;
+  labels: Record<string, string>;
+  anyJobRunning: boolean;
+  runningJobLabel?: string;
+  onToggle: (job: JobStatus) => void;
+  onTrigger: (job: JobStatus) => void;
+  togglePending: boolean;
+  triggerPending: boolean;
+}
+
+function JobCard({
+  job,
+  labels,
+  anyJobRunning,
+  runningJobLabel,
+  onToggle,
+  onTrigger,
+  togglePending,
+  triggerPending,
+}: JobCardProps) {
+  return (
+    <div className="glass p-4 glass-hover">
       <div className="flex flex-wrap items-center justify-between gap-4">
         <div className="flex items-center gap-3">
           {/* Status dot */}
@@ -122,7 +134,9 @@ export function JobControls() {
           />
           <div>
             <span className="text-sm font-medium text-gray-200">{job.label}</span>
-                <div className="flex items-center gap-3 mt-0.5">
+            <div className="mt-0.5 flex flex-wrap items-center gap-3">
+              {/* Live state only — a persisted error must not read as the
+                  current status forever, so this never consults last_run_*. */}
               <span
                 className={`text-[11px] font-medium ${
                   job.running
@@ -148,26 +162,23 @@ export function JobControls() {
                         ? 'Active'
                         : 'Inactive'}
               </span>
-                  {job.via_pipeline ? (
+              {job.enabled && <NextRun job={job} labels={labels} />}
-                    <span className="text-[11px] text-gray-500">runs via pipeline</span>
-                  ) : (
-                    job.enabled && job.next_run_at && (
-                      <span className="text-[11px] text-gray-500">
-                        Next run {formatNextRun(job.next_run_at)}
-                      </span>
-                    )
-                  )}
               {!job.registered && (
                 <span className="text-[11px] text-red-400">Not registered</span>
               )}
             </div>
-                {!job.running && job.runtime_finished_at && (
+            <Membership job={job} labels={labels} />
-                  <div className={`mt-1 text-[11px] ${lastRunColor(job.runtime_status)}`}>
+            {/* Persisted, so this survives a deploy — unlike runtime_* above. */}
-                    Last run {formatAgo(job.runtime_finished_at)}
+            {!job.running && job.last_run_at && (
-                    {job.runtime_status ? ` · ${job.runtime_status}` : ''}
+              <div className={`mt-1 text-[11px] ${lastRunColor(job.last_run_status)}`}>
-                    {job.runtime_message ? ` — ${job.runtime_message}` : ''}
+                Last run {formatAgo(job.last_run_at)}
+                {job.last_run_status ? ` · ${job.last_run_status}` : ''}
+                {job.last_run_message ? ` — ${job.last_run_message}` : ''}
               </div>
             )}
+            {!job.running && !job.last_run_at && (
+              <div className="mt-1 text-[11px] text-gray-600">No run recorded yet</div>
+            )}
             {job.running && (
               <div className="mt-2 space-y-1.5">
                 <div className="flex items-center justify-between text-[11px] text-gray-400">
@@ -180,7 +191,7 @@ export function JobControls() {
                     <span>{Math.max(0, Math.min(100, job.runtime_progress_pct)).toFixed(0)}%</span>
                   )}
                 </div>
-                    <div className="h-1.5 w-56 rounded-full bg-slate-700/80 overflow-hidden">
+                <div className="h-1.5 w-56 overflow-hidden rounded-full bg-slate-700/80">
                   <div
                     className="h-full bg-blue-400 transition-all duration-500"
                     style={{
@@ -203,8 +214,8 @@ export function JobControls() {
         <div className="flex items-center gap-2">
           <button
             type="button"
-                onClick={() => toggleJob.mutate({ jobName: job.name, enabled: !job.enabled })}
+            onClick={() => onToggle(job)}
-                disabled={toggleJob.isPending}
+            disabled={togglePending}
             className={`rounded-lg border px-3 py-1.5 text-xs transition-all duration-200 disabled:opacity-50 ${
               job.enabled
                 ? 'border-red-500/20 bg-red-500/10 text-red-400 hover:bg-red-500/20'
@@ -215,14 +226,14 @@ export function JobControls() {
           </button>
           <button
             type="button"
-                onClick={() => triggerJob.mutate(job.name)}
+            onClick={() => onTrigger(job)}
-                disabled={triggerJob.isPending || !job.enabled || anyJobRunning}
+            disabled={triggerPending || !job.enabled || anyJobRunning}
-                className="btn-primary px-3 py-1.5 text-xs disabled:opacity-50 disabled:cursor-not-allowed"
+            className="btn-primary px-3 py-1.5 text-xs disabled:cursor-not-allowed disabled:opacity-50"
           >
             <span>
               {job.running
                 ? 'Running…'
-                    : triggerJob.isPending
+                : triggerPending
                   ? 'Triggering…'
                   : anyJobRunning
                     ? 'Blocked'
@@ -237,7 +248,128 @@ export function JobControls() {
         </div>
       )}
     </div>
+  );
+}
+
+export function JobControls() {
+  const { data: jobs, isLoading } = useJobs();
+  const toggleJob = useToggleJob();
+  const triggerJob = useTriggerJob();
+  const all = jobs ?? [];
+  // Job id -> display label, so a step can name its parent pipeline.
+  const labels = Object.fromEntries(all.map((job) => [job.name, job.label]));
+  const anyJobRunning = all.some((job) => job.running);
+  const runningJob = all.find((job) => job.running);
+  const pausedJob = all.find((job) => !job.running && job.runtime_status === 'rate_limited');
+
+  if (isLoading) return <SkeletonTable rows={4} cols={3} />;
+
+  const known = new Set<string>(SECTIONS.map((s) => s.key));
+  const groups: { key: string; title: string; hint: string; jobs: JobStatus[] }[] = [
+    ...SECTIONS.map((section) => ({
+      ...section,
+      jobs: all.filter((job) => job.category === section.key),
+    })),
+    {
+      key: 'other',
+      title: 'Other',
+      hint: 'uncategorised',
+      jobs: all.filter((job) => !job.category || !known.has(job.category)),
+    },
+  ];
+
+  const cardProps = {
+    labels,
+    anyJobRunning,
+    runningJobLabel: runningJob?.label,
+    onToggle: (job: JobStatus) =>
+      toggleJob.mutate({ jobName: job.name, enabled: !job.enabled }),
+    onTrigger: (job: JobStatus) => triggerJob.mutate(job.name),
+    togglePending: toggleJob.isPending,
+    triggerPending: triggerJob.isPending,
+  };
+
+  return (
+    <div className="space-y-6">
+      {runningJob && (
+        <div className="rounded-xl border border-blue-400/30 bg-blue-500/10 px-4 py-3">
+          <div className="flex flex-wrap items-center justify-between gap-3">
+            <div>
+              <div className="text-xs font-semibold text-blue-300">
+                Active job: {runningJob.label}
+              </div>
+              <div className="mt-0.5 text-[11px] text-blue-100/80">
+                Manual triggers are blocked until this run finishes.
+              </div>
+            </div>
+            <div className="text-[11px] text-blue-200">
+              {runningJob.runtime_processed ?? 0}
+              {typeof runningJob.runtime_total === 'number'
+                ? ` / ${runningJob.runtime_total}`
+                : ''}
+            </div>
+          </div>
+          <div className="mt-2 h-1.5 w-full overflow-hidden rounded-full bg-slate-700/80">
+            <div
+              className="h-full bg-blue-400 transition-all duration-500"
+              style={{
+                width: `${
+                  typeof runningJob.runtime_progress_pct === 'number'
+                    ? Math.max(5, Math.min(100, runningJob.runtime_progress_pct))
+                    : 30
+                }%`,
+              }}
+            />
+          </div>
+          {runningJob.runtime_current_ticker && (
+            <div className="mt-1 text-[11px] text-blue-100/80">
+              Current: {runningJob.runtime_current_ticker}
+            </div>
+          )}
+          {runningJob.runtime_message && (
+            <div className="mt-1 text-[11px] text-blue-100/80">{runningJob.runtime_message}</div>
+          )}
+        </div>
+      )}
+
+      {!runningJob && pausedJob && (
+        <div className="rounded-xl border border-amber-400/30 bg-amber-500/10 px-4 py-3">
+          <div className="flex flex-wrap items-center justify-between gap-3">
+            <div>
+              <div className="text-xs font-semibold text-amber-300">
+                Last run paused: {pausedJob.label}
+              </div>
+              <div className="mt-0.5 text-[11px] text-amber-100/90">
+                {pausedJob.runtime_message || 'Rate limit hit. The collector stopped early and will resume from last progress on the next run.'}
+              </div>
+            </div>
+            <div className="text-[11px] text-amber-200">
+              {pausedJob.runtime_processed ?? 0}
+              {typeof pausedJob.runtime_total === 'number'
+                ? ` / ${pausedJob.runtime_total}`
+                : ''}
+            </div>
+          </div>
+        </div>
+      )}
+
+      {groups.map(
+        (group) =>
+          group.jobs.length > 0 && (
+            <section key={group.key} className="space-y-3">
+              <h3 className="text-xs font-medium uppercase tracking-widest text-gray-500">
+                {group.title}
+                <span className="ml-2 num text-gray-600">{group.jobs.length}</span>
+                <span className="ml-2 normal-case tracking-normal text-gray-600">
+                  {group.hint}
+                </span>
+              </h3>
+              {group.jobs.map((job) => (
+                <JobCard key={job.name} job={job} {...cardProps} />
               ))}
+            </section>
+          ),
+      )}
     </div>
   );
 }

frontend/src/components/admin/ScheduleSettings.tsx

@@ -11,6 +11,8 @@ const DEFAULTS: ScheduleConfig = {
   schedule_near_close_pipeline_cron: '30 15 * * mon-fri',
   schedule_after_close_pipeline_cron: '45 16 * * mon-fri',
   schedule_intraday_pipeline_cron: '0 10-15 * * mon-fri',
+  schedule_backtest_cron: '0 3 * * sun',
+  schedule_ticker_universe_cron: '0 1 * * *',
 };
 
 const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: boolean }[] = [
@@ -55,6 +57,18 @@ const FIELDS: { key: keyof ScheduleConfig; label: string; hint: string; mono?: b
     hint: 'Refresh prices + resolve outcomes mid-session. Default hourly 10:00–15:00 ET weekdays.',
     mono: true,
   },
+  {
+    key: 'schedule_backtest_cron',
+    label: 'Backtest',
+    hint: 'Replay history and refresh the Track Record report. Default Sunday 03:00 ET. Was a 168h interval, which restarted on every deploy and so could defer indefinitely.',
+    mono: true,
+  },
+  {
+    key: 'schedule_ticker_universe_cron',
+    label: 'Ticker universe sync',
+    hint: 'Refresh the tracked-symbol universe. Default 01:00 ET daily, before the morning pipeline.',
+    mono: true,
+  },
 ];
 
 export function ScheduleSettings() {

frontend/src/lib/types.ts

@@ -196,6 +196,8 @@ export interface ScheduleConfig {
   schedule_near_close_pipeline_cron: string;
   schedule_after_close_pipeline_cron: string;
   schedule_intraday_pipeline_cron: string;
+  schedule_backtest_cron: string;
+  schedule_ticker_universe_cron: string;
 }
 
 // Runtime sentiment LLM configuration

tests/unit/test_admin_jobs.py

@@ -0,0 +1,110 @@
+"""Admin → Jobs listing: categories, ordering, and next-run coherence.
+
+The panel used to render 19 jobs as one alphabetical list in which a pipeline
+step, a cron job and a manual job were indistinguishable, and a triggered job
+could advertise a next run ten years out.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from app import job_catalog
+from app.scheduler import configure_scheduler, scheduler
+from app.services.admin_service import _visible_next_run, list_jobs
+
+
+@pytest.fixture(autouse=True)
+def _configured_scheduler():
+    scheduler.remove_all_jobs()
+    configure_scheduler()
+    yield
+    scheduler.remove_all_jobs()
+
+
+def _by_name(jobs: list[dict]) -> dict[str, dict]:
+    return {job["name"]: job for job in jobs}
+
+
+class TestVisibleNextRun:
+    def test_parked_backstop_is_not_a_schedule(self):
+        """Paused jobs carry a 520-week interval; triggering one re-arms it."""
+        backstop = datetime.now(timezone.utc) + timedelta(weeks=520)
+        assert _visible_next_run(backstop) is None
+
+    def test_a_real_upcoming_run_passes_through(self):
+        soon = datetime.now(timezone.utc) + timedelta(hours=6)
+        assert _visible_next_run(soon) == soon
+
+    def test_none_stays_none(self):
+        assert _visible_next_run(None) is None
+
+
+class TestListJobs:
+    async def test_hidden_jobs_are_not_listed_but_stay_valid(self, db_session):
+        jobs = _by_name(await list_jobs(db_session))
+        assert "data_backfill" not in jobs
+        # Still triggerable through the API, and still registered.
+        assert "data_backfill" in job_catalog.VALID_JOB_NAMES
+        assert scheduler.get_job("data_backfill") is not None
+
+    async def test_every_visible_job_has_a_category(self, db_session):
+        jobs = await list_jobs(db_session)
+        assert {j["name"] for j in jobs} == set(
+            job_catalog.VALID_JOB_NAMES - job_catalog.HIDDEN_JOBS
+        )
+        assert all(j["category"] in job_catalog.CATEGORY_ORDER for j in jobs)
+
+    async def test_jobs_arrive_grouped_by_category(self, db_session):
+        """The frontend renders sections in payload order, so ordering is the
+        API's job — not something each client re-derives."""
+        categories = [j["category"] for j in await list_jobs(db_session)]
+        ranks = [job_catalog.CATEGORY_ORDER.index(c) for c in categories]
+        assert ranks == sorted(ranks)
+
+    async def test_pipeline_steps_defer_their_schedule_to_the_parent(self, db_session):
+        jobs = _by_name(await list_jobs(db_session))
+        step = jobs["rr_scanner"]
+        assert step["category"] == job_catalog.CATEGORY_STEP
+        assert step["next_run_at"] is None
+        assert step["next_run_source"] == "via_pipeline"
+        assert step["pipelines"] == ["near_close_pipeline"]
+
+    async def test_step_reports_the_soonest_enabled_parent(self, db_session):
+        due = datetime.now(timezone.utc) + timedelta(hours=3)
+        scheduler.modify_job("daily_pipeline", next_run_time=due)
+
+        collector = _by_name(await list_jobs(db_session))["data_collector"]
+        assert collector["via_next_run_job"] == "daily_pipeline"
+        assert collector["via_next_run_at"] == due.isoformat()
+        # Runs in all four pipelines — the reason steps are not nested under one.
+        assert set(collector["pipelines"]) == set(job_catalog.PIPELINE_JOBS)
+
+    async def test_manual_jobs_say_so_instead_of_showing_a_date(self, db_session):
+        study = _by_name(await list_jobs(db_session))["event_study"]
+        assert study["category"] == job_catalog.CATEGORY_MANUAL
+        assert study["next_run_source"] == "manual_only"
+        assert study["next_run_at"] is None
+
+    async def test_a_triggered_manual_job_still_shows_no_next_run(self, db_session):
+        """Regression: triggering re-armed the 520-week backstop, which the panel
+        rendered as a real 'next run in ~87600h'."""
+        scheduler.modify_job("event_study", next_run_time=datetime.now(timezone.utc))
+        scheduler.modify_job("event_study", next_run_time=None)
+
+        study = _by_name(await list_jobs(db_session))["event_study"]
+        assert study["next_run_at"] is None
+
+    async def test_pipelines_report_their_own_schedule_and_steps(self, db_session):
+        pipeline = _by_name(await list_jobs(db_session))["daily_pipeline"]
+        assert pipeline["category"] == job_catalog.CATEGORY_PIPELINE
+        assert pipeline["next_run_source"] == "own_schedule"
+        assert pipeline["steps"] == [
+            step for step, _ in job_catalog.PIPELINE_STEPS["daily_pipeline"]
+        ]
+
+    async def test_standalone_jobs_keep_their_own_schedule(self, db_session):
+        backtest = _by_name(await list_jobs(db_session))["backtest"]
+        assert backtest["category"] == job_catalog.CATEGORY_SCHEDULED
+        assert backtest["next_run_source"] == "own_schedule"
+        assert backtest["pipelines"] == []

tests/unit/test_job_run_persistence.py

@@ -0,0 +1,271 @@
+"""Durable last-run state.
+
+Job outcomes lived only in an in-memory dict, so every deploy wiped them and
+Admin → Jobs could only report "Active" with no indication a job had ever run.
+"""
+
+from datetime import datetime, timezone
+
+import pytest
+from sqlalchemy import select
+
+from app import scheduler as sched
+from app.models.job_run_state import JobRunState
+from app.services import job_run_store
+from tests.conftest import _test_session_factory
+
+
+@pytest.fixture
+def session_factory():
+    """A real, independently committing session.
+
+    _persist_job_run opens its own session and commits, which is what production
+    does; the shared db_session fixture holds an outer transaction that a commit
+    would tear down.
+    """
+    return _test_session_factory
+
+
+async def _rows(session) -> dict[str, JobRunState]:
+    result = await session.execute(select(JobRunState))
+    return {row.job_name: row for row in result.scalars().all()}
+
+
+async def _committed_rows() -> dict[str, JobRunState]:
+    async with _test_session_factory() as session:
+        return await _rows(session)
+
+
+class TestRecordFinish:
+    async def test_inserts_then_updates_one_row_per_job(self, db_session):
+        await job_run_store.record_finish(
+            db_session,
+            "rr_scanner",
+            {"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00", "processed": 5, "total": 5},
+        )
+        await db_session.flush()
+        await job_run_store.record_finish(
+            db_session,
+            "rr_scanner",
+            {"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"},
+        )
+        await db_session.flush()
+
+        rows = await _rows(db_session)
+        assert list(rows) == ["rr_scanner"], "upsert, not append-only history"
+        assert rows["rr_scanner"].status == "error"
+        assert rows["rr_scanner"].message == "boom"
+
+    async def test_missing_finish_time_falls_back_to_now(self, db_session):
+        await job_run_store.record_finish(db_session, "alerts", {"status": "completed"})
+        await db_session.flush()
+        assert (await _rows(db_session))["alerts"].finished_at is not None
+
+
+class TestPipelinePersistence:
+    """_run_pipeline is the only write path for steps: they are plain coroutine
+    calls, so they emit no scheduler events for the listener to catch."""
+
+    @pytest.fixture(autouse=True)
+    def _enabled(self, monkeypatch, session_factory):
+        async def enabled(db, job_name):
+            return True
+
+        monkeypatch.setattr("app.scheduler.async_session_factory", session_factory)
+        monkeypatch.setattr("app.scheduler._is_job_enabled", enabled)
+
+    async def test_persists_both_the_step_and_the_orchestrator(self, monkeypatch):
+        async def ok_step():
+            sched._runtime_finish("rr_scanner", "completed", processed=3, total=3)
+
+        monkeypatch.setattr(sched, "ok_step", ok_step, raising=False)
+        await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "ok_step")])
+
+        rows = await _committed_rows()
+        assert rows["rr_scanner"].status == "completed"
+        assert rows["near_close_pipeline"].status == "completed"
+
+    async def test_a_failing_step_still_records_its_error(self, monkeypatch):
+        """The persist sits after the except that swallows step errors — inside
+        it, exactly the runs worth seeing would be skipped."""
+
+        async def boom():
+            sched._runtime_finish("rr_scanner", "error", processed=0, total=1, message="kaboom")
+            raise RuntimeError("kaboom")
+
+        monkeypatch.setattr(sched, "boom", boom, raising=False)
+        await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "boom")])
+
+        rows = await _committed_rows()
+        assert rows["rr_scanner"].status == "error"
+        assert rows["rr_scanner"].message == "kaboom"
+        # The pipeline itself survives a failing step.
+        assert rows["near_close_pipeline"].status == "completed"
+
+    async def test_disabled_pipeline_records_skipped(self, monkeypatch):
+        async def disabled(db, job_name):
+            return False
+
+        monkeypatch.setattr("app.scheduler._is_job_enabled", disabled)
+        await sched._run_pipeline("daily_pipeline", [])
+
+        assert (await _committed_rows())["daily_pipeline"].status == "skipped"
+
+    async def test_persistence_failure_never_breaks_the_pipeline(self, monkeypatch):
+        calls: list[str] = []
+
+        async def exploding_record(db, job_name, runtime):
+            calls.append(job_name)
+            raise RuntimeError("db down")
+
+        async def ok_step():
+            sched._runtime_finish("rr_scanner", "completed", processed=1, total=1)
+
+        monkeypatch.setattr(sched.job_run_store, "record_finish", exploding_record)
+        monkeypatch.setattr(sched, "ok_step", ok_step, raising=False)
+
+        await sched._run_pipeline("near_close_pipeline", [("rr_scanner", "ok_step")])
+
+        assert calls, "persistence was attempted"
+        assert sched.get_job_runtime_snapshot("near_close_pipeline")["status"] == "completed"
+
+    async def test_a_job_that_never_finished_writes_nothing(self):
+        sched._runtime_start("event_study", total=1)
+        await sched._persist_job_run("event_study")
+        assert "event_study" not in await _committed_rows()
+
+
+class TestListJobsSplitsLiveFromPersisted:
+    async def test_a_stale_error_does_not_pin_the_status_chip(self, db_session):
+        """runtime_* must stay live-only: the chip and the rate-limit banner read
+        it, so a week-old error there would read as the current state forever."""
+        from app.scheduler import configure_scheduler, scheduler
+        from app.services.admin_service import list_jobs
+
+        scheduler.remove_all_jobs()
+        configure_scheduler()
+        # _job_runtime is module-global and survives across tests; pin the live
+        # row to idle so the assertion is about the split, not about ordering.
+        sched._job_runtime["rr_scanner"] = sched._idle_runtime()
+        await job_run_store.record_finish(
+            db_session,
+            "rr_scanner",
+            {
+                "status": "error",
+                "finished_at": datetime(2026, 8, 1, tzinfo=timezone.utc).isoformat(),
+                "message": "old failure",
+            },
+        )
+        await db_session.flush()
+
+        job = {j["name"]: j for j in await list_jobs(db_session)}["rr_scanner"]
+        assert job["last_run_status"] == "error"
+        assert job["last_run_message"] == "old failure"
+        assert job["runtime_status"] == "idle"
+        assert job["running"] is False
+        scheduler.remove_all_jobs()
+
+
+class TestConcurrentWrites:
+    """Pipelines are separate scheduler jobs that can overlap, and they share
+    step ids — data_collector belongs to all four."""
+
+    async def test_interleaved_first_writes_do_not_collide(self):
+        """Both sessions SELECT before either INSERTs: select-then-insert lost
+        this race with an IntegrityError, and the caller swallows it."""
+        async with _test_session_factory() as a, _test_session_factory() as b:
+            await job_run_store.record_finish(
+                a, "data_collector",
+                {"status": "completed", "finished_at": "2026-08-08T10:00:00+00:00"},
+            )
+            await job_run_store.record_finish(
+                b, "data_collector",
+                {"status": "completed", "finished_at": "2026-08-08T10:00:01+00:00"},
+            )
+            await a.commit()
+            await b.commit()  # must not raise
+
+        rows = await _committed_rows()
+        assert rows["data_collector"].status == "completed"
+
+    async def test_an_older_finish_never_rewinds_the_row(self):
+        """A slower pipeline finishing an older run last must not overwrite a
+        newer outcome with a stale one."""
+        async with _test_session_factory() as s:
+            await job_run_store.record_finish(
+                s, "alerts",
+                {"status": "completed", "finished_at": "2026-08-08T12:00:00+00:00"},
+            )
+            await s.commit()
+            await job_run_store.record_finish(
+                s, "alerts",
+                {"status": "error", "finished_at": "2026-08-08T09:00:00+00:00", "message": "stale"},
+            )
+            await s.commit()
+
+        row = (await _committed_rows())["alerts"]
+        assert row.finished_at.isoformat().startswith("2026-08-08T12:00")
+        assert row.status == "completed"
+        assert row.message is None
+
+    async def test_a_newer_finish_still_wins(self):
+        async with _test_session_factory() as s:
+            await job_run_store.record_finish(
+                s, "rr_scanner",
+                {"status": "completed", "finished_at": "2026-08-08T09:00:00+00:00"},
+            )
+            await s.commit()
+            await job_run_store.record_finish(
+                s, "rr_scanner",
+                {"status": "error", "finished_at": "2026-08-08T12:00:00+00:00", "message": "boom"},
+            )
+            await s.commit()
+
+        row = (await _committed_rows())["rr_scanner"]
+        assert row.status == "error"
+        assert row.message == "boom"
+
+
+class TestShutdownDrain:
+    async def test_drains_a_task_queued_after_the_flush_starts(self):
+        """scheduler.shutdown(wait=False) returns before APScheduler dispatches
+        its completion events, so writes can appear mid-drain. Snapshotting the
+        task set once would miss them and dispose the engine underneath."""
+        import asyncio
+
+        done: list[str] = []
+
+        async def slow_first():
+            await asyncio.sleep(0.02)
+            done.append("first")
+            # Queued only once the first write is already finishing.
+            sched._persist_tasks.add(asyncio.get_running_loop().create_task(late()))
+
+        async def late():
+            await asyncio.sleep(0.02)
+            done.append("late")
+
+        sched._persist_tasks.clear()
+        sched._persist_tasks.add(asyncio.get_running_loop().create_task(slow_first()))
+
+        await sched.flush_job_run_persists(timeout=2.0)
+
+        assert done == ["first", "late"]
+        sched._persist_tasks.clear()
+
+    async def test_returns_promptly_when_there_is_nothing_to_drain(self):
+        sched._persist_tasks.clear()
+        await sched.flush_job_run_persists(timeout=2.0)
+
+    async def test_gives_up_rather_than_hanging_shutdown(self):
+        import asyncio
+
+        async def never():
+            await asyncio.sleep(30)
+
+        sched._persist_tasks.clear()
+        task = asyncio.get_running_loop().create_task(never())
+        sched._persist_tasks.add(task)
+        await sched.flush_job_run_persists(timeout=0.15)  # returns, does not hang
+        task.cancel()
+        sched._persist_tasks.clear()

tests/unit/test_scheduler.py

@@ -1,16 +1,19 @@
 """Unit tests for app.scheduler module."""
 
 import asyncio
+from datetime import datetime, timezone
 from types import SimpleNamespace
 
 import pytest
 
+from app import job_catalog
 from app.scheduler import (
     _DAILY_PIPELINE_STEPS,
     _NEAR_CLOSE_PIPELINE_STEPS,
     _consume_backtest_options,
     _consume_backtest_target_model,
     _parse_frequency,
+    _repause_after_manual_run,
     _resume_tickers,
     _last_successful,
     _run_source_import,
@@ -112,60 +115,119 @@ class TestResumeTickers:
 
 class TestConfigureScheduler:
     def test_configure_adds_all_jobs(self):
-        # Remove any existing jobs first
+        # Derived from the catalog, not a fourth hand-maintained copy of the
+        # job list: a job added to the catalog but never registered now fails
+        # here instead of silently rendering "Not registered" in the admin UI.
         scheduler.remove_all_jobs()
         configure_scheduler()
-        jobs = scheduler.get_jobs()
+        assert {j.id for j in scheduler.get_jobs()} == set(job_catalog.VALID_JOB_NAMES)
-        job_ids = {j.id for j in jobs}
-        assert job_ids == {
-            "data_collector",
-            "data_backfill",
-            "benchmark_collector",
-            "sentiment_collector",
-            "dolt_earnings_import",
-            "sec_fundamentals_import",
-            "rr_scanner",
-            "shadow_book",
-            "ticker_universe_sync",
-            "outcome_evaluator",
-            "alerts",
-            "market_regime",
-            "regime_monitor",
-            "event_study",
-            "backtest",
-            "daily_pipeline",
-            "near_close_pipeline",
-            "after_close_pipeline",
-            "intraday_pipeline",
-        }
 
     def test_configure_is_idempotent(self):
         scheduler.remove_all_jobs()
         configure_scheduler()
         configure_scheduler()  # Should replace, not duplicate
         job_ids = [j.id for j in scheduler.get_jobs()]
-        # Each ID should appear exactly once
+        assert sorted(job_ids) == sorted(job_catalog.VALID_JOB_NAMES)
-        assert sorted(job_ids) == sorted([
+
-            "after_close_pipeline",
+    def test_independent_jobs_use_cron_not_interval(self):
-            "alerts",
+        """Interval countdowns restart on every deploy, so a weekly interval on a
-            "backtest",
+        frequently-redeployed box can defer forever. Both standalone jobs were
-            "benchmark_collector",
+        migrated to cron; this pins them there."""
-            "daily_pipeline",
+        scheduler.remove_all_jobs()
-            "intraday_pipeline",
+        configure_scheduler()
+        for job_id in ("backtest", "ticker_universe_sync"):
+            trigger = type(scheduler.get_job(job_id).trigger).__name__
+            assert trigger == "CronTrigger", f"{job_id} regressed to {trigger}"
+
+
+class TestJobCatalog:
+    def test_pipeline_members_are_derived_from_step_lists(self):
+        derived = {
+            step
+            for steps in job_catalog.PIPELINE_STEPS.values()
+            for step, _ in steps
+        }
+        assert job_catalog.PIPELINE_MEMBERS == derived
+        # ...and reproduces the set that used to be maintained by hand, so the
+        # derivation is behaviour-preserving rather than merely self-consistent.
+        assert job_catalog.PIPELINE_MEMBERS == {
             "data_collector",
-            "data_backfill",
+            "benchmark_collector",
-            "dolt_earnings_import",
-            "sec_fundamentals_import",
-            "market_regime",
-            "near_close_pipeline",
-            "regime_monitor",
-            "event_study",
-            "outcome_evaluator",
-            "rr_scanner",
             "sentiment_collector",
+            "rr_scanner",
             "shadow_book",
-            "ticker_universe_sync",
+            "outcome_evaluator",
-        ])
+            "alerts",
+            "market_regime",
+            "regime_monitor",
+        }
+
+    def test_categories_partition_every_job_exactly_once(self):
+        buckets = [
+            job_catalog.PIPELINE_JOBS,
+            job_catalog.PIPELINE_STEP_JOBS,
+            job_catalog.SCHEDULED_JOBS,
+            job_catalog.MANUAL_JOBS,
+        ]
+        flat = [name for bucket in buckets for name in bucket]
+        assert len(flat) == len(set(flat)), "a job is in two categories"
+        assert set(flat) == set(job_catalog.VALID_JOB_NAMES)
+        assert all(name in job_catalog.JOB_CATEGORY for name in flat)
+
+    def test_every_job_has_a_label_and_a_unique_sort_order(self):
+        names = job_catalog.VALID_JOB_NAMES
+        assert set(job_catalog.JOB_LABELS) == set(names)
+        assert len({job_catalog.sort_order(n) for n in names}) == len(names)
+
+    def test_multi_pipeline_members_report_every_parent(self):
+        """Membership is many-to-many — the reason the UI groups into sections
+        rather than nesting steps under one parent."""
+        by_member = job_catalog.PIPELINES_BY_MEMBER
+        assert set(by_member["data_collector"]) == set(job_catalog.PIPELINE_JOBS)
+        assert set(by_member["alerts"]) == {"daily_pipeline", "near_close_pipeline"}
+        assert set(by_member["outcome_evaluator"]) == {
+            "intraday_pipeline",
+            "after_close_pipeline",
+        }
+        assert "backtest" not in by_member
+
+    def test_every_job_has_a_runtime_row_before_it_first_runs(self):
+        """The old private _JOB_NAMES list held 16 of 19, so three jobs showed no
+        last-run line until their first run in a given process."""
+        assert set(get_job_runtime_snapshot()) == set(job_catalog.VALID_JOB_NAMES)
+
+
+class TestRepauseListener:
+    def _configured(self):
+        scheduler.remove_all_jobs()
+        configure_scheduler()
+
+    def test_manual_job_is_repaused_after_running(self):
+        """Triggering a paused job re-arms its 520-week backstop, which used to
+        surface as a "next run in ~87600h"."""
+        self._configured()
+        scheduler.modify_job("event_study", next_run_time=datetime.now(timezone.utc))
+        _repause_after_manual_run(SimpleNamespace(job_id="event_study"))
+        assert scheduler.get_job("event_study").next_run_time is None
+
+    def test_pipeline_step_is_repaused_after_running(self):
+        self._configured()
+        scheduler.modify_job("rr_scanner", next_run_time=datetime.now(timezone.utc))
+        _repause_after_manual_run(SimpleNamespace(job_id="rr_scanner"))
+        assert scheduler.get_job("rr_scanner").next_run_time is None
+
+    def test_cron_jobs_are_left_alone(self):
+        # Set an explicit next run first: an unstarted scheduler leaves the
+        # attribute unset, so comparing None to None would prove nothing.
+        self._configured()
+        due = datetime.now(timezone.utc)
+        scheduler.modify_job("daily_pipeline", next_run_time=due)
+        _repause_after_manual_run(SimpleNamespace(job_id="daily_pipeline"))
+        assert scheduler.get_job("daily_pipeline").next_run_time == due
+
+    def test_unknown_job_is_ignored(self):
+        self._configured()
+        _repause_after_manual_run(SimpleNamespace(job_id="not_a_job"))
 
 
 class _SessionContext: