refactor(jobs): derive job topology from one catalog, make next-run coherent

Groundwork for the Admin -> Jobs cleanup. Three sources of truth collapse into
app/job_catalog.py, which imports nothing from app so both the scheduler and
admin_service can import it at module level (admin_service otherwise has to
import the scheduler inside functions to dodge a cycle).

PIPELINE_MEMBERS is now DERIVED from the four pipeline step lists instead of
being a literal set in admin_service duplicating four lists in scheduler.py with
nothing asserting they agreed. A test pins that the derivation reproduces the
previous hand-maintained 9 names exactly, so this is behaviour-preserving.

Deletes the private _JOB_NAMES list, which held 16 of the 19 jobs:
benchmark_collector, outcome_evaluator and shadow_book had no runtime row, and
so no "last run" line in the panel, until their first run in a given process.
_job_runtime is now seeded from the catalog, and a test pins the invariant.

Next-run is decided by category rather than by reading a timestamp. A pipeline
step has no schedule of its own, so it reports its parent's ("next via Morning
Pipeline in 3h") instead of nothing; a manual job says manual_only rather than
rendering a date. This also fixes a real bug: triggering a paused job set
next_run_time=now, APScheduler re-armed the 520-week backstop behind it, and the
panel displayed "next run in ~87600h". Two independent guards -- the category
rule, plus _visible_next_run dropping anything past a year -- and an APScheduler
listener that re-pauses steps and manual jobs once their run finishes. The
listener is registered at module level because configure_scheduler is called
more than once and add_listener does not deduplicate.

Migrates backtest and ticker_universe_sync from interval to cron (Sun 03:00 ET
and 01:00 ET). configure_scheduler calls remove_all_jobs() on every startup, so
an interval countdown restarts each deploy -- a 168h backtest needed a week of
uninterrupted uptime to fire even once. The codebase already documented this
pitfall as the reason cron was adopted; these two were never migrated. Both are
now editable in Admin -> Schedule.

Also: list_jobs went from one settings query per job (19) to one for all of
them, and data_backfill is hidden from the listing while staying registered and
API-triggerable.

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

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/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
@@ -84,6 +86,34 @@ scheduler = AsyncIOScheduler(
     }
 )
 
+
+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(_repause_after_manual_run, 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 +121,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 +141,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
 
@@ -1300,54 +1316,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.
@@ -1486,6 +1462,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 +1478,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 +1614,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 +1630,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
@@ -606,91 +607,108 @@ 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
+    }
+
     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)
 
         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 +717,7 @@ 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"),
+            **_next_run_fields(scheduler, name, enabled_map),
         })
 
     return jobs_out

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_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: