feat(jobs): persist each job's last run so it survives a restart

Job outcomes lived only in scheduler._job_runtime, an in-memory dict. Every
deploy wiped it, so Admin -> Jobs could report "Active" with no indication a job
had ever run or how it ended -- which is the main thing that page is for.

New job_run_state table (migration 031): one row per job, upserted on job_name.
Deliberately not history -- system_events already grows unbounded with no
retention job, and a second append-only operational table would repeat that
debt. Adding history later is purely additive.

Written from two hooks, NOT from _runtime_finish. That looked cheapest (one
function, ~40 call sites) but unit tests invoke job coroutines directly, so it
would fire detached DB writes at the real session factory throughout the suite,
and there is no testing flag to guard on.

  - An APScheduler EVENT_JOB_EXECUTED/ERROR listener covers everything the
    scheduler fires, including manual triggers. Its detached task is held in a
    module-level set (a bare create_task result can be collected mid-flight) and
    drained in the app lifespan before engine.dispose().
  - _run_pipeline persists directly, and must: pipeline steps are plain
    coroutine calls that emit no scheduler events, so the listener cannot see
    them. The step persist sits AFTER the except that swallows step errors --
    inside it, exactly the failed runs worth seeing would be skipped. The
    orchestrator persists in the finally, and the disabled early-return persists
    too, or "skipped" is silently dropped.

_persist_job_run never raises: a persistence failure must not break an otherwise
successful pipeline.

The API reports this as last_run_* and leaves runtime_* meaning strictly live
in-memory state. Reusing runtime_status would have been a regression, not a
no-op: JobControls drives the status chip from it (a job that errored eight days
ago would read "Last run error" forever instead of "Active") and picks the
rate-limit banner from it (a week-old rate limit would pin the banner
permanently). Tests pin the split.

The table starts empty; each job fills its row the next time it finishes. No
backfill from system_events, which records only warning/error outcomes under a
different status vocabulary and would invent successes that never happened.

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/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

@@ -33,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,
@@ -87,6 +88,19 @@ 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.
 
@@ -112,7 +126,7 @@ def _repause_after_manual_run(event: object) -> None:
         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)
+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] = {
@@ -308,6 +322,44 @@ 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) -> None:
+    """Await in-flight last-run writes. Called from the app's shutdown path."""
+    if not _persist_tasks:
+        return
+    await asyncio.wait(set(_persist_tasks), timeout=timeout)
+
+
 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, {}))
@@ -1346,6 +1398,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)
@@ -1361,6 +1414,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)
@@ -1369,6 +1427,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:

app/services/admin_service.py

@@ -18,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__)
 
@@ -689,11 +689,13 @@ async def list_jobs(db: AsyncSession) -> list[dict]:
         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 visible:
         job = scheduler.get_job(name)
         runtime = get_job_runtime_snapshot(name)
+        last = last_runs.get(name)
 
         jobs_out.append({
             "name": name,
@@ -717,6 +719,14 @@ 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),
         })
 

app/services/job_run_store.py

@@ -0,0 +1,70 @@
+"""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.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."""
+    result = await db.execute(
+        select(JobRunState).where(JobRunState.job_name.in_(list(job_names)))
+    )
+    return {row.job_name: row for row in result.scalars().all()}
+
+
+async def record_finish(db: AsyncSession, job_name: str, runtime: dict) -> JobRunState:
+    """Upsert the last-run row from a scheduler runtime snapshot.
+
+    Select-then-update-or-insert rather than a dialect-specific upsert, matching
+    ``settings_store.upsert_setting`` — tests run on SQLite, prod on Postgres.
+    """
+    existing = await db.execute(
+        select(JobRunState).where(JobRunState.job_name == job_name)
+    )
+    row = existing.scalar_one_or_none()
+
+    finished_at = _as_datetime(runtime.get("finished_at")) or datetime.now(timezone.utc)
+    status = str(runtime.get("status") or "completed")
+    message = runtime.get("message")
+    values = {
+        "status": status,
+        "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,
+    }
+
+    if row is None:
+        row = JobRunState(job_name=job_name, **values)
+        db.add(row)
+    else:
+        for field, value in values.items():
+            setattr(row, field, value)
+    return row

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 { JobStatus } from '../../api/admin';
 import { SkeletonTable } from '../ui/Skeleton';
 
 function formatNextRun(iso: string | null): string {
@@ -29,10 +30,34 @@ function lastRunColor(status: string | null | undefined): string {
   return 'text-gray-500';
 }
 
+/** 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 (
+      <span className={muted}>
+        Next via {labels[job.via_next_run_job] ?? job.via_next_run_job}{' '}
+        {formatNextRun(job.via_next_run_at)}
+      </span>
+    );
+  }
+  if (!job.next_run_at) return null;
+  return <span className={muted}>Next run {formatNextRun(job.next_run_at)}</span>;
+}
+
 export function JobControls() {
   const { data: jobs, isLoading } = useJobs();
   const toggleJob = useToggleJob();
   const triggerJob = useTriggerJob();
+  // Job id -> display label, so a step can name its parent pipeline.
+  const labels = Object.fromEntries((jobs ?? []).map((job) => [job.name, job.label]));
   const anyJobRunning = (jobs ?? []).some((job) => job.running);
   const runningJob = jobs?.find((job) => job.running);
   const pausedJob = jobs?.find((job) => !job.running && job.runtime_status === 'rate_limited');
@@ -148,24 +173,18 @@ export function JobControls() {
                             ? 'Active'
                             : 'Inactive'}
                   </span>
-                  {job.via_pipeline ? (
-                    <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.enabled && <NextRun job={job} labels={labels} />}
                   {!job.registered && (
                     <span className="text-[11px] text-red-400">Not registered</span>
                   )}
                 </div>
-                {!job.running && job.runtime_finished_at && (
-                  <div className={`mt-1 text-[11px] ${lastRunColor(job.runtime_status)}`}>
-                    Last run {formatAgo(job.runtime_finished_at)}
-                    {job.runtime_status ? ` · ${job.runtime_status}` : ''}
-                    {job.runtime_message ? ` — ${job.runtime_message}` : ''}
+                {/* Persisted, so this survives a deploy — unlike runtime_*,
+                    which the status chip above still reads for live state. */}
+                {!job.running && job.last_run_at && (
+                  <div className={`mt-1 text-[11px] ${lastRunColor(job.last_run_status)}`}>
+                    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 && (

tests/unit/test_job_run_persistence.py

@@ -0,0 +1,166 @@
+"""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()