feat(scheduler): every background run is a durable row

What a job last did lived in the API process's memory and was gone at
every restart; the audit log kept a run only when it was manual, failed
or changed something. Now each run is a job_runs row written at its
start and its end in its own session: who started it, its outcome as
data, what it spent of the YouTube quota (quota_events.job_run_id) and
the lines it logged (log_events.run_id, filterable in the log console).

- a run a stopped process left open is marked interrupted at startup
- "run now" is a one-shot job on the scheduler, which now always starts
  (interval jobs only with SCHEDULER_ENABLED); the manual actor is reset
  after the run, since pool threads are reused
- the dashboard's last run comes from job_runs, so it survives a deploy
- runs leave the audit log: migration 0109 moves the old job_run rows
  over (and back on downgrade); the audit page's run folding goes
- job_run_retention_days (default 30) prunes old runs in audit_gc

backend/app/README.md

@@ -21,7 +21,8 @@ Three layers, by *who* runs the code — not by feature:
    `issues/bundle.py` is the one definition of a bug report's files, for the zip export and the tracker.
 
 3. **Flat `app/*.py` core modules — cross-cutting primitives.** `auth` (with `apitokens`, a machine's
-   folder-scoped way into the Files library), `sysconfig`, `quota`, `audit`,
+   folder-scoped way into the Files library), `sysconfig`, `quota`, `audit`, `jobruns` (every
+   background run's durable row, its words in `runstatus`),
    `notifications`, `userscope`, `ratelimit`, `progress`, `logs` (the one logging
    configuration), `logsink` (its `log_events` table), `diagnostics` (a user's consented log
    level), `debugsession` (a recording that tags its user's lines, E16) and `collation` (A→Z in the
@@ -36,7 +37,7 @@ pre-abstract for a single caller; don't leave shared logic duplicated once there
 ## Transaction ownership
 
 This follows CLAUDE.md's canonical rule verbatim — **"the handler/job-wrapper commits; helpers flush
-(only `quota` owns its session)"** — so read it that way, not as an exception to it. The commit belongs
+(only `quota` and `jobruns` own their session)"** — so read it that way, not as an exception to it. The commit belongs
 to the **operation owner**; anything called *inside* another transaction only stages its change
 (`db.add` / `flush`) and lets the owner commit, so the whole operation is one transaction.
 `audit.record()` is the reference: it adds the audit row and the caller commits it together with the
@@ -45,7 +46,7 @@ mutation it records, so an audit entry can never exist without its change (or vi
 - The **operation owner** is usually the route handler. A `service.*` operation the handler delegates a
   whole transition to (e.g. `service.cancel`) *is* that owner and commits its one transition — it isn't
   a mid-operation helper. A helper that owns its OWN short-lived session (the worker's `jobdb` writes,
-  the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes.
+  `jobruns`' run rows — the same carve-out CLAUDE.md grants `quota`) commits that session. Everything else flushes.
 - `SessionLocal` is `expire_on_commit=False`: after a raw `UPDATE` + commit, an ORM object you already
   loaded is **stale**. Read the fresh value from the row (RETURNING, or a re-`SELECT`), not the ORM
   attribute — see `service.decr_ref_count`.

backend/app/audit.py

@@ -48,8 +48,6 @@ class AuditAction:
     MAINTENANCE_TUNE = "maintenance_tune"
     SYNC_PAUSE = "sync_pause"
     SYNC_RESUME = "sync_resume"
-    # Scheduler job outcome / run-summary (actor_id set = a manual "run now"; NULL = automatic)
-    JOB_RUN = "job_run"
     # Per-user download cookie/auth credential set or removed (the value is NEVER recorded)
     DOWNLOAD_CREDENTIAL_SET = "download_credential_set"
     DOWNLOAD_CREDENTIAL_DELETE = "download_credential_delete"

backend/app/config.py

@@ -268,6 +268,9 @@ class Settings(BaseSettings):
     # YouTube quota history (quota_events): days kept (0 = forever). A year, so the admin usage view's
     # longest window (365 days) stays whole; the same daily cleanup job prunes it.
     quota_retention_days: int = 365
+    # Background-run history (job_runs): days kept (0 = forever). Every run is a row (~1–1.5k a day
+    # at the default intervals), so a month; the same daily cleanup job prunes it.
+    job_run_retention_days: int = 30
     # On-disk layout: "plex" (Channel/Season YYYY/… [id].ext + .nfo/poster) or "flat".
     download_layout: DownloadLayout = "plex"
     # Default SponsorBlock (skip sponsor segments) for new custom profiles.

backend/app/jobruns.py

@@ -0,0 +1,104 @@
+"""The durable history of background runs — the `job_runs` table.
+
+A run is written twice: a `running` row when it starts, its outcome when it ends. Each write is its
+OWN short session, committed at once, for the reason `quota.record_usage` gives: the run's own work
+may roll back (its `except: db.rollback()`), and the history of what happened must not go with it —
+nor may this bookkeeping commit the job's half-done session.
+
+Only the API process runs jobs (the worker never starts the scheduler), so at its startup nothing can
+still be running: `interrupt_leftovers` closes what a stopped process left open.
+"""
+
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import delete, func, select, update
+from sqlalchemy.engine import CursorResult
+from sqlalchemy.orm import Session
+
+from app.db import SessionLocal
+from app.models import JobRun, QuotaEvent
+from app.runstatus import RunStatus, RunTrigger
+
+
+def start(job: str, trigger: RunTrigger, actor_id: int | None, target: str | None = None) -> int:
+    """Open a run's row; returns its id."""
+    with SessionLocal() as s:
+        row = JobRun(
+            job=job,
+            trigger=trigger,
+            actor_id=actor_id,
+            target=target,
+            status=RunStatus.RUNNING,
+            started_at=datetime.now(timezone.utc),
+        )
+        s.add(row)
+        s.commit()
+        return row.id
+
+
+def finish(
+    run_id: int,
+    status: RunStatus,
+    *,
+    summary: str | None = None,
+    result: dict | None = None,
+    error: str | None = None,
+    failure: dict | None = None,
+) -> None:
+    """Close a run with its outcome and what it spent. Conditional on the row still running: a row
+    the startup sweep already called interrupted is not rewritten by a thread that outlived it."""
+    with SessionLocal() as s:
+        units = s.scalar(
+            select(func.coalesce(func.sum(QuotaEvent.units), 0)).where(
+                QuotaEvent.job_run_id == run_id
+            )
+        )
+        s.execute(
+            update(JobRun)
+            .where(JobRun.id == run_id, JobRun.status == RunStatus.RUNNING)
+            .values(
+                status=status,
+                finished_at=datetime.now(timezone.utc),
+                summary=summary,
+                result=result,
+                error=error,
+                failure=failure,
+                quota_units=units or 0,
+            )
+        )
+        s.commit()
+
+
+def interrupt_leftovers() -> int:
+    """Every row a previous process left `running` → `interrupted`. Called once at startup, before
+    the scheduler can open a new run. Returns how many."""
+    with SessionLocal() as s:
+        res = s.execute(
+            update(JobRun)
+            .where(JobRun.status == RunStatus.RUNNING)
+            .values(status=RunStatus.INTERRUPTED, finished_at=datetime.now(timezone.utc))
+        )
+        s.commit()
+        return res.rowcount if isinstance(res, CursorResult) else 0
+
+
+def last_runs(db: Session) -> dict[str, JobRun]:
+    """Each job's most recent run (any trigger), by job id."""
+    rows = db.execute(
+        select(JobRun)
+        .distinct(JobRun.job)
+        .order_by(JobRun.job, JobRun.started_at.desc(), JobRun.id.desc())
+    ).scalars()
+    return {r.job: r for r in rows}
+
+
+def gc(db: Session, retention_days: int) -> int:
+    """Delete finished runs older than `retention_days` (0 = keep forever). A run still going is
+    never reaped. The caller commits."""
+    if retention_days <= 0:
+        return 0
+    cutoff = datetime.now(timezone.utc) - timedelta(days=retention_days)
+    res = db.execute(
+        delete(JobRun).where(JobRun.started_at < cutoff, JobRun.status != RunStatus.RUNNING)
+    )
+    return res.rowcount if isinstance(res, CursorResult) else 0

backend/app/logs.py

@@ -10,8 +10,8 @@ the logs were both unjoinable and leaky. `configure_logging()` replaces all four
 Event shape — the OpenTelemetry log data model, without the SDK; one JSON object per line:
     ts · level · sev · source · logger · event · msg · attrs
 plus whatever correlation the context carries (`request_id`, `user_id`, `debug_session_id`,
-`job_id`) and the app version. That shape is what the admin log console filters on and what a Gitea
-issue attachment will carry, so it is deliberately flat and stable.
+`job_id`, `run_id`) and the app version. That shape is what the admin log console filters on and
+what a Gitea issue attachment will carry, so it is deliberately flat and stable.
 
 Call sites stay ordinary stdlib logging:
 
@@ -81,15 +81,23 @@ _debug_session: contextvars.ContextVar[str | None] = contextvars.ContextVar(
     "log_debug_session", default=None
 )
 _job_id: contextvars.ContextVar[int | None] = contextvars.ContextVar("log_job_id", default=None)
+# The background run (app/jobruns.py) this context is doing; quota's spend is tagged with it too.
+_run_id: contextvars.ContextVar[int | None] = contextvars.ContextVar("log_run_id", default=None)
 
 _CONTEXT_VARS = {
     "request_id": _request_id,
     "user_id": _user_id,
     "debug_session_id": _debug_session,
     "job_id": _job_id,
+    "run_id": _run_id,
 }
 
 
+def current_run_id() -> int | None:
+    """The background run this context is doing, if any — `bind(run_id=…)` sets it."""
+    return _run_id.get()
+
+
 class _RequestScope:
     """What the auth dependency learns about a request, for every line logged after it.
 

backend/app/logsink.py

@@ -362,6 +362,7 @@ def serialize(row: LogEvent) -> dict:
         "request_id": row.request_id,
         "user_id": row.user_id,
         "job_id": row.job_id,
+        "run_id": row.run_id,
         "debug_session_id": row.debug_session_id,
         "version": row.version,
     }

backend/app/models.py

@@ -24,6 +24,7 @@ from sqlalchemy.dialects.postgresql import JSONB, TSVECTOR
 from sqlalchemy.orm import Mapped, mapped_column, relationship, validates
 
 from app.db import Base
+from app.runstatus import RunStatus, RunTrigger, sql_in
 from app.utils import canonical_email
 
 # Every user-supplied display string in this schema shares one width, and it lives HERE so the
@@ -563,6 +564,11 @@ class QuotaEvent(Base):
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True), server_default=func.now(), index=True
     )
+    # The background run that spent it (app/jobruns.py); NULL = a request's own spend. SET NULL: a
+    # reaped run leaves the spend in the budget's history.
+    job_run_id: Mapped[int | None] = mapped_column(
+        ForeignKey("job_runs.id", ondelete="SET NULL"), index=True
+    )
 
 
 class AuditLog(Base):
@@ -609,6 +615,8 @@ class LogEvent(Base):
             "debug_session_id",
             postgresql_where="debug_session_id IS NOT NULL",
         ),
+        # A run's lines are `WHERE run_id = …`; request lines carry none.
+        Index("ix_log_events_run", "run_id", postgresql_where="run_id IS NOT NULL"),
     )
 
     id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
@@ -624,7 +632,11 @@ class LogEvent(Base):
     request_id: Mapped[str | None] = mapped_column(String(64), index=True)
     user_id: Mapped[int | None] = mapped_column(Integer)
     debug_session_id: Mapped[str | None] = mapped_column(String(64))
+    # A DOWNLOAD job's id (download_jobs) — not a background run, which is `run_id`.
     job_id: Mapped[int | None] = mapped_column(Integer)
+    # The background run (job_runs) that logged it. No foreign key, as `user_id`: the writer runs
+    # behind the event, and a run reaped meanwhile must not fail the batch its lines travel in.
+    run_id: Mapped[int | None] = mapped_column(BigInteger)
     version: Mapped[str | None] = mapped_column(String(64))
 
 
@@ -700,6 +712,46 @@ class SchedulerSetting(Base):
     interval_minutes: Mapped[int] = mapped_column(Integer)
 
 
+class JobRun(Base):
+    """One run of a background job — scheduled, an admin's "run now", or a user's ad-hoc work on one
+    target — written when it starts and again when it ends (app/jobruns.py), each in its own
+    committed session, so the history survives the job's own rollback and a restart. A row left
+    `running` by a process that died is set `interrupted` at the next startup.
+
+    `quota_units` is what the run spent of the YouTube budget: the `quota_events` rows carrying its
+    id, summed when it ends. `summary` is the English one-line fallback; `result` / `failure` are the
+    same outcome as data, which the client names in the reader's language."""
+
+    __tablename__ = "job_runs"
+    __table_args__ = (
+        Index("ix_job_runs_job_started", "job", "started_at"),
+        CheckConstraint(f"status IN ({sql_in(RunStatus)})", name="ck_job_runs_status"),
+        CheckConstraint(f"trigger IN ({sql_in(RunTrigger)})", name="ck_job_runs_trigger"),
+        # Only a run still going has no end.
+        CheckConstraint(
+            "(status = 'running') = (finished_at IS NULL)", name="ck_job_runs_finished"
+        ),
+    )
+
+    id: Mapped[int] = mapped_column(BigInteger, primary_key=True)
+    job: Mapped[str] = mapped_column(String(40))
+    trigger: Mapped[str] = mapped_column(String(16))
+    # Who started it: the admin of a manual run, the user of an ad-hoc one; NULL = the schedule.
+    actor_id: Mapped[int | None] = mapped_column(
+        ForeignKey("users.id", ondelete="SET NULL"), index=True
+    )
+    # What an ad-hoc run worked on (a channel id); NULL for the recurring jobs.
+    target: Mapped[str | None] = mapped_column(String(128))
+    status: Mapped[str] = mapped_column(String(16))
+    started_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), index=True)
+    finished_at: Mapped[datetime | None] = mapped_column(DateTime(timezone=True))
+    summary: Mapped[str | None] = mapped_column(Text)
+    result: Mapped[dict | None] = mapped_column(JSONB)
+    error: Mapped[str | None] = mapped_column(Text)
+    failure: Mapped[dict | None] = mapped_column(JSONB)
+    quota_units: Mapped[int] = mapped_column(Integer, default=0, server_default="0")
+
+
 class SystemConfig(Base):
     """Admin-set overrides for operational config that otherwise comes from env/config
     defaults. Generic key/value store (one row per key); absent = use the env/config

backend/app/quota.py

@@ -16,7 +16,7 @@ from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.engine import CursorResult
 from sqlalchemy.orm import Session
 
-from app import sysconfig
+from app import logs, sysconfig
 from app.db import SessionLocal
 from app.models import ApiQuotaUsage, QuotaEvent
 
@@ -204,6 +204,14 @@ def record_usage(units: int) -> None:
     )
     with SessionLocal() as qs:
         qs.execute(stmt)
-        # Audit detail: who/what spent it (per-user attribution; NULL actor = system).
-        qs.add(QuotaEvent(user_id=_actor_id.get(), action=_action.get(), units=units))
+        # Audit detail: who/what spent it (per-user attribution; NULL actor = system), and the
+        # background run it belongs to, so the run's row can say what it cost.
+        qs.add(
+            QuotaEvent(
+                user_id=_actor_id.get(),
+                action=_action.get(),
+                units=units,
+                job_run_id=logs.current_run_id(),
+            )
+        )
         qs.commit()

backend/app/routes/admin_logs.py

@@ -86,6 +86,7 @@ class LogFilters:
     event: str | None
     request_id: str | None
     job_id: int | None
+    run_id: int | None
     q: str | None
     since: datetime | None
     until: datetime | None
@@ -113,6 +114,8 @@ class LogFilters:
             out.append(LogEvent.request_id == self.request_id)
         if self.job_id is not None:
             out.append(LogEvent.job_id == self.job_id)
+        if self.run_id is not None:
+            out.append(LogEvent.run_id == self.run_id)
         if self.since is not None:
             out.append(LogEvent.ts >= _utc(self.since))
         if self.until is not None:
@@ -140,11 +143,14 @@ def log_filters(
     ),
     request_id: str | None = Query(default=None, pattern=r"^[A-Za-z0-9._-]{8,64}$"),
     job_id: int | None = Query(default=None, ge=1),
+    run_id: int | None = Query(default=None, ge=1),
     q: str | None = Query(default=None, max_length=200),
     since: datetime | None = None,
     until: datetime | None = None,
 ) -> LogFilters:
-    return LogFilters(list(levels), list(sources), user, event, request_id, job_id, q, since, until)
+    return LogFilters(
+        list(levels), list(sources), user, event, request_id, job_id, run_id, q, since, until
+    )
 
 
 @router.get("")

backend/app/routes/scheduler.py

@@ -117,7 +117,7 @@ def set_maintenance_settings(
 
 @router.get("")
 def get_scheduler(_: User = Depends(admin_user), db: Session = Depends(get_db)) -> dict:
-    snap = scheduler_snapshot()
+    snap = scheduler_snapshot(db)
 
     # Work still queued (shared catalog, so these are instance-wide, not per-user).
     deep_requested = (

backend/app/runstatus.py

@@ -0,0 +1,35 @@
+"""The words a background run's row is written with — the `job_runs` table's status and trigger.
+
+One source for the literals: the scheduler writes them, the startup sweep rewrites `running`, the
+model's CHECK constraints are built from these sets, and the dashboard reads them back. A run's
+status is never typed as a string anywhere else (the download pipeline's `cancelled`/`canceled`
+drift, `downloads/status.py`, is why).
+"""
+
+from enum import StrEnum
+
+
+class RunStatus(StrEnum):
+    RUNNING = "running"
+    OK = "ok"
+    ERROR = "error"
+    # Sync was paused: the run was due and deliberately did nothing.
+    SKIPPED = "skipped"
+    # The process stopped while the run was going (a deploy, a crash): set at the next startup, so a
+    # row never claims to be running in a process that no longer exists.
+    INTERRUPTED = "interrupted"
+
+
+class RunTrigger(StrEnum):
+    # The run's interval came round.
+    SCHEDULED = "scheduled"
+    # An admin pressed "run now" (or "run all").
+    MANUAL = "manual"
+    # A user's own action started work on one target (a channel's full history), outside the
+    # recurring jobs.
+    AD_HOC = "ad_hoc"
+
+
+def sql_in(values: type[StrEnum]) -> str:
+    """`'a', 'b'` — the IN-list a CHECK constraint names these values with."""
+    return ", ".join(f"'{v.value}'" for v in values)

backend/app/scheduler.py

@@ -6,11 +6,12 @@ import threading
 from datetime import datetime, timezone
 from typing import Callable
 
+from apscheduler.executors.pool import ThreadPoolExecutor
+from apscheduler.jobstores.base import ConflictingIdError
 from apscheduler.schedulers.background import BackgroundScheduler
 from sqlalchemy import select
 
-from app import audit, debugsession, logs, logsink, progress, quota
-from app.audit import AuditAction
+from app import audit, debugsession, jobruns, logs, logsink, progress, quota
 from app.config import settings
 from app.db import SessionLocal
 from app.downloads.gc import run_download_gc
@@ -20,6 +21,7 @@ from app.plex.tmdb import run_enrich as run_tmdb_enrich
 from app.plex.tmdb import run_people_enrich
 from app.plex.watch_sync import run_plex_watch_reconcile, run_plex_watch_sync
 from app.notifications import create_notification
+from app.runstatus import RunStatus, RunTrigger
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
 from app.sync.explore import purge_ephemeral
@@ -37,10 +39,9 @@ from app.sync.runner import (
 log = logs.get_logger(__name__)
 _scheduler: BackgroundScheduler | None = None
 
-# Per-job run activity, kept in-memory for the admin Scheduler dashboard to read (the
-# scheduler lives in the same single-worker process as the API, so a request can read it
-# directly). Ephemeral by design — it reflects "what the scheduler is doing right now and
-# how the last runs went" since this process started; durable history isn't the goal here.
+# What is running RIGHT NOW, per job: the single-run claim and the live progress. In memory on
+# purpose — the scheduler lives in the API's one process, and progress changes every second. What a
+# run did is durable instead: its `job_runs` row (app/jobruns.py), which outlives a restart.
 _activity: dict[str, dict] = {}
 _activity_lock = threading.Lock()
 
@@ -130,49 +131,6 @@ def _notify_done(
         log.exception("failed to write completion notification for job %s", job_id)
 
 
-def _run_changed(result) -> bool:
-    """Did an automatic job run actually do something worth an audit row? (No-op polls don't.)
-    A "change" = a truthy NUMERIC count; status-string dicts like {"skipped": "disabled"} (Plex
-    off — the default) are no-ops, not changes, so they don't flood the trail every interval."""
-    if result is None:
-        return False
-    if isinstance(result, dict):
-        # A phased job (backfill's recent/deep, people_enrich's bios/portraits) nests a result per
-        # phase: a count inside a phase is a change too.
-        return any(
-            _run_changed(v) if isinstance(v, dict) else isinstance(v, (int, float)) and bool(v)
-            for v in result.values()
-        )
-    if isinstance(result, (int, float)):
-        return result != 0
-    return bool(result)
-
-
-def _audit_run(db, actor_id, job_id, status, summary, result) -> None:
-    """Write ONE run-summary audit row per job run (never per-item). Manual runs (an admin
-    clicked "run now" → actor_id set) and errors are always logged; automatic runs only when
-    they changed something, so routine no-op polls don't spam the trail. Best-effort."""
-    if actor_id is None and status == "ok" and not _run_changed(result):
-        return
-    # The result as data too: the audit log names it in the reader's language, as the inbox
-    # does; `summary` stays the English line of the row's own summary column.
-    data = _result_data(result)
-    try:
-        audit.record(
-            db,
-            actor_id,
-            AuditAction.JOB_RUN,
-            target_type="job",
-            target_id=job_id,
-            summary=f"{job_id}: {status}" + (f" — {summary}" if summary else ""),
-            after={"status": status, "summary": summary, **({"result": data} if data else {})},
-        )
-        db.commit()
-    except Exception:
-        db.rollback()
-        log.exception("failed to write audit row for job %s", job_id)
-
-
 def _job(name: str, fn) -> None:
     # The one place both run paths converge (manual trigger AND scheduled fire), so claim the run
     # HERE, atomically: whichever reaches this first wins, the other skips. This is the real
@@ -180,11 +138,29 @@ def _job(name: str, fn) -> None:
     if not _claim_running(name):
         log.info("job %s already running — skipping this trigger", name)
         return
-    db = SessionLocal()
     actor_id = _manual_actor.get()  # set only for a manual "run now" trigger
-    _record(
-        name, last_started=datetime.now(timezone.utc).isoformat(), last_error=None, progress=None
-    )  # running already claimed
+    trigger = RunTrigger.SCHEDULED if actor_id is None else RunTrigger.MANUAL
+    _record(name, progress=None)  # running already claimed
+    db = SessionLocal()
+    try:
+        try:
+            run_id = jobruns.start(name, trigger, actor_id)
+        except Exception:
+            # No row to tell it in: the database is what failed. The log is the record.
+            log.exception("job %s not started: its run could not be recorded", name)
+            return
+        with logs.bind(run_id=run_id):
+            _run(db, name, fn, run_id, actor_id)
+    finally:
+        # ALWAYS release the claim, even if the body raised before its own status update — so a
+        # failure can never wedge the job as permanently "running".
+        _record(name, running=False, progress=None)
+        db.close()
+
+
+def _run(db, name: str, fn, run_id: int, actor_id: int | None) -> None:
+    """The run itself, inside its `job_runs` row: the pause-skip, the work, and its outcome — on
+    the row, and in the triggering admin's inbox for a manual run."""
 
     def sink(current, total, phase):
         _record(name, progress={"current": current, "total": total, "phase": phase})
@@ -192,34 +168,18 @@ def _job(name: str, fn) -> None:
     try:
         if is_sync_paused(db):
             log.info("job %s skipped (sync paused)", name)
-            _record(
-                name,
-                status="skipped",
-                last_finished=datetime.now(timezone.utc).isoformat(),
-                last_result="paused",
-                result=_PAUSED,
-                progress=None,
-            )
+            jobruns.finish(run_id, RunStatus.SKIPPED, summary="sync paused", result=_PAUSED)
             if actor_id is not None:
                 _notify_done(db, actor_id, name, "skipped", "sync paused", _PAUSED)
-                _audit_run(db, actor_id, name, "skipped", "sync paused", _PAUSED)
             return
         with progress.bind(sink):
             result = fn(db)
         summary = _summarize(result)
         data = _result_data(result)
         log.info("job %s -> %s", name, result)
-        _record(
-            name,
-            status="ok",
-            last_finished=datetime.now(timezone.utc).isoformat(),
-            last_result=summary,
-            result=data,
-            progress=None,
-        )
+        jobruns.finish(run_id, RunStatus.OK, summary=summary, result=data)
         if actor_id is not None:
             _notify_done(db, actor_id, name, "ok", summary, data)
-        _audit_run(db, actor_id, name, "ok", summary, result)
     except Exception as exc:
         db.rollback()
         log.exception("job %s failed", name, extra=logs.event("scheduler.job.failed", job=name))
@@ -229,22 +189,9 @@ def _job(name: str, fn) -> None:
         failure = (
             {"failure": exc.reason, **exc.params} if isinstance(exc, progress.JobFailed) else None
         )
-        _record(
-            name,
-            status="error",
-            last_finished=datetime.now(timezone.utc).isoformat(),
-            last_error=err,
-            result=failure,
-            progress=None,
-        )
+        jobruns.finish(run_id, RunStatus.ERROR, error=err, failure=failure)
         if actor_id is not None:
             _notify_done(db, actor_id, name, "error", err, failure)
-        _audit_run(db, actor_id, name, "error", err, failure)
-    finally:
-        # ALWAYS release the claim, even if the body raised before its own status update — so a
-        # failure can never wedge the job as permanently "running".
-        _record(name, running=False)
-        db.close()
 
 
 # A skipped run's result, in the same shape as a job's own `{"skipped": …}`.
@@ -271,11 +218,17 @@ def _summarize(result) -> str | None:
     return str(result)
 
 
-def scheduler_snapshot() -> dict:
-    """What the scheduler is doing, for the admin dashboard. `running_here` is False in
-    environments where the scheduler is disabled (e.g. local dev shares the prod-ish DB but
-    runs no scheduler), so the UI can say so while still showing DB-derived queue/quota."""
-    running_here = _scheduler is not None
+def _iso(ts: datetime | None) -> str | None:
+    return ts.astimezone(timezone.utc).isoformat() if ts else None
+
+
+def scheduler_snapshot(db) -> dict:
+    """What the scheduler is doing, for the admin dashboard: each job's live state from this process
+    and its last run from `job_runs`, so a restart no longer blanks "when did it last run, and how
+    did it go". `running_here` is False where the scheduler runs no interval jobs
+    (SCHEDULER_ENABLED=false), so the UI can say so while still showing the DB-derived history,
+    queue and quota."""
+    running_here = _scheduler is not None and settings.scheduler_enabled
     next_runs: dict[str, str | None] = {}
     live_intervals: dict[str, int] = {}
     if _scheduler is not None:
@@ -287,27 +240,41 @@ def scheduler_snapshot() -> dict:
                 live_intervals[job.id] = round(iv.total_seconds() / 60)
     with _activity_lock:
         acts = {k: dict(v) for k, v in _activity.items()}
+    last = jobruns.last_runs(db)
     jobs = []
     for job_id, default_interval in JOB_INTERVALS.items():
         a = acts.get(job_id, {})
+        run = last.get(job_id)
+        # A run still going has no outcome yet: only an ended run says how it went.
+        ended = run if run is not None and run.status != RunStatus.RUNNING else None
         jobs.append(
             {
                 "id": job_id,
                 "interval_minutes": live_intervals.get(job_id, default_interval),
                 "next_run": next_runs.get(job_id),
                 "running": a.get("running", False),
-                "status": a.get("status"),
-                "last_started": a.get("last_started"),
-                "last_finished": a.get("last_finished"),
-                "last_result": a.get("last_result"),
-                "result": a.get("result"),
-                "last_error": a.get("last_error"),
+                "status": ended.status if ended else None,
+                "last_started": _iso(run.started_at) if run else None,
+                "last_finished": _iso(run.finished_at) if run else None,
+                "last_result": ended.summary if ended else None,
+                # The outcome as data: the result, or on a failed run the failure the job named.
+                "result": _outcome_data(ended),
+                "last_error": ended.error if ended else None,
+                "last_trigger": run.trigger if run else None,
+                "last_run_id": run.id if run else None,
+                "quota_units": ended.quota_units if ended else None,
                 "progress": a.get("progress"),
             }
         )
     return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
 
 
+def _outcome_data(run) -> dict | None:
+    if run is None:
+        return None
+    return run.failure if run.status == RunStatus.ERROR else run.result
+
+
 def _rss_job() -> None:
     _job("rss_poll", run_rss_poll)
 
@@ -408,15 +375,17 @@ def _people_enrich_job() -> None:
 
 
 def _audit_gc_job() -> None:
-    # The history cleanup: audit-log rows past audit_retention_days and YouTube quota history past
-    # quota_retention_days (0 = keep forever, each). Pure DB, no quota.
+    # The history cleanup: audit-log rows past audit_retention_days, YouTube quota history past
+    # quota_retention_days and finished runs past job_run_retention_days (0 = keep forever, each).
+    # Pure DB, no quota.
     def work(db):
         from app import sysconfig
 
         reaped = audit.gc(db, sysconfig.get_int(db, "audit_retention_days"))
         quota_reaped = quota.gc(db, sysconfig.get_int(db, "quota_retention_days"))
+        runs_reaped = jobruns.gc(db, sysconfig.get_int(db, "job_run_retention_days"))
         db.commit()
-        return {"reaped": reaped, "quota_reaped": quota_reaped}
+        return {"reaped": reaped, "quota_reaped": quota_reaped, "runs_reaped": runs_reaped}
 
     _job("audit_gc", work)
 
@@ -485,15 +454,29 @@ def running_job_ids() -> set[str]:
         return {jid for jid, a in _activity.items() if a.get("running")}
 
 
-def trigger_job(job_id: str, actor_id: int | None = None) -> str:
-    """Run a job immediately in a background thread, independent of its interval schedule.
-    The wrapper handles its own DB session, activity tracking and pause-skip, so the live
-    dashboard reflects the run. Returns "started", "already_running", or raises KeyError for
-    an unknown job. Refusing a concurrent run keeps a manual trigger from overlapping a
-    scheduled run (APScheduler enforces max_instances=1 for the scheduled side).
+def _as_actor(fn: Callable[[], None], actor_id: int | None) -> Callable[[], None]:
+    """`fn` run as `actor_id`'s manual run. The actor is RESET afterwards: the executor's pool
+    threads are reused and keep their context from task to task, so an actor left set would make
+    the next scheduled run that thread picks up look like this admin's manual run."""
 
-    `actor_id` (the admin who clicked "run now") is stashed in a contextvar inside the new
-    thread so the run posts a completion notification to that admin's inbox."""
+    def run() -> None:
+        token = _manual_actor.set(actor_id)
+        try:
+            fn()
+        finally:
+            _manual_actor.reset(token)
+
+    return run
+
+
+def trigger_job(job_id: str, actor_id: int | None = None) -> str:
+    """Run a job now, independent of its interval schedule: a one-shot job on the same scheduler,
+    so a manual run shares the scheduled runs' executor and shutdown — no thread of its own. The
+    wrapper handles its own DB session, run row and pause-skip. Returns "started",
+    "already_running", or raises KeyError for an unknown job.
+
+    `actor_id` (the admin who clicked "run now") is set in a contextvar inside the run's thread, so
+    the run is recorded as theirs and posts a completion notification to their inbox."""
     fn = JOB_FUNCS.get(job_id)
     if fn is None:
         raise KeyError(job_id)
@@ -502,13 +485,18 @@ def trigger_job(job_id: str, actor_id: int | None = None) -> str:
     # _job claims, sees it's taken, and skips — no double run.
     if _is_running(job_id):
         return "already_running"
+    sched = _scheduler
+    if sched is None:
+        # Only between shutdown and exit: start_scheduler runs at every API startup.
+        raise RuntimeError("the scheduler is not running")
 
-    def run() -> None:
-        if actor_id is not None:
-            _manual_actor.set(actor_id)
-        fn()
-
-    threading.Thread(target=run, name=f"trigger-{job_id}", daemon=True).start()
+    try:
+        # One id per job: a second click before the first one-shot has even started is the same
+        # run, not a queued second one. misfire_grace_time=None: a busy pool delays it, never drops it.
+        sched.add_job(_as_actor(fn, actor_id), "date", id=f"{job_id}@now", misfire_grace_time=None)
+    # silent-ok: not a failure — the answer is the 409 the admin sees ("already running").
+    except ConflictingIdError:
+        return "already_running"
     return "started"
 
 
@@ -521,22 +509,36 @@ def trigger_all(actor_id: int | None = None) -> list[str]:
     return started
 
 
+# Room for every job at once (a "run all") beside a few ad-hoc runs: APScheduler's default pool of
+# 10 would queue a scheduled fire behind long manual runs.
+_POOL_SIZE = len(JOB_FUNCS) + 8
+
+
 def start_scheduler() -> None:
+    """Start this process's scheduler. It always starts, so "run now" works on every install; only
+    with SCHEDULER_ENABLED does it also register the recurring interval jobs. Before anything can
+    run, every run a previous process left open is closed as interrupted."""
     global _scheduler
-    if not settings.scheduler_enabled or _scheduler is not None:
+    if _scheduler is not None:
         return
-    # Effective intervals = env defaults overlaid with any admin overrides from the DB.
-    db = SessionLocal()
-    try:
-        iv = load_intervals(db)
-    finally:
-        db.close()
-    scheduler = BackgroundScheduler(timezone="UTC")
-    for job_id, fn in JOB_FUNCS.items():
-        scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
+    left = jobruns.interrupt_leftovers()
+    if left:
+        log.warning("%s run(s) left running by the previous process marked interrupted", left)
+    scheduler = BackgroundScheduler(
+        timezone="UTC", executors={"default": ThreadPoolExecutor(_POOL_SIZE)}
+    )
+    if settings.scheduler_enabled:
+        # Effective intervals = env defaults overlaid with any admin overrides from the DB.
+        db = SessionLocal()
+        try:
+            iv = load_intervals(db)
+        finally:
+            db.close()
+        for job_id, fn in JOB_FUNCS.items():
+            scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
     scheduler.start()
     _scheduler = scheduler
-    log.info("scheduler started")
+    log.info("scheduler started (interval jobs %s)", "on" if settings.scheduler_enabled else "off")
 
 
 def shutdown_scheduler() -> None:

backend/app/sysconfig.py

@@ -181,6 +181,8 @@ SPECS: tuple[ConfigSpec, ...] = (
     # --- Audit log ---
     # Days to keep admin audit-log rows before the audit_gc job prunes them (0 = keep forever).
     ConfigSpec("audit_retention_days", "int", "audit", "audit_retention_days", min=0, max=3650),
+    # Days the job_runs table keeps a finished background run before the audit_gc job prunes it.
+    ConfigSpec("job_run_retention_days", "int", "audit", "job_run_retention_days", min=0, max=3650),
     # --- Diagnostics ---
     # Runtime log level for the API process. Applied without a restart — see routes/config.py and
     # main.py's lifespan.