feat: the Backups screen says whether backups are keeping up, and the System view warns when they aren't

- A Backups row on Settings → System and the nav dot turn amber when the
  newest database dump is more than a day old or failed, or the weekly
  bucket check found files missing that no media run has sent since.
  No row when backups aren't set up.
- The weekly bucket check is kept as a run (kind "reconcile") with its
  counts, trimmed to the last 12, and listed on the Backups card.
- The Backups card leads with the same verdict and the newest dump's size,
  shows bytes beside the running media job's file count, and lets each run
  open its list of failed files with why.

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

backend/app/api/system_backup.py

@@ -20,7 +20,7 @@ system_backup_bp = Blueprint(
     "system_backup", __name__, url_prefix="/api/system/backup",
 )
 
-_KINDS = frozenset({"db", "images", "media", "restore"})
+_KINDS = frozenset({"db", "images", "media", "restore", "reconcile"})
 _TAG_MAX_LEN = 64
 _BACKUP_SETTINGS_FIELDS = (
     "backup_db_nightly_enabled",

backend/app/api/system_health.py

@@ -35,6 +35,7 @@ from sqlalchemy import select, text
 from ..config import get_config
 from ..extensions import get_session
 from ..models import ServiceSeen
+from ..services import backup_health
 from ..services.worker_lanes import SWEEP_PERIOD_SECONDS
 
 system_health_bp = Blueprint("system_health", __name__, url_prefix="/api/system")
@@ -227,6 +228,17 @@ async def system_health():
                     **{k: v for k, v in (row.details or {}).items() if k != "agent_id"},
                 })
 
+            # Whether the backups are keeping up. Read like the rest, and a
+            # failure to read it is reported as unknown rather than raised.
+            try:
+                backup = await session.run_sync(lambda s: backup_health.backup_part(s, now))
+            except Exception as exc:  # noqa: BLE001 — a probe reports, it never raises
+                backup = {"key": "backup", "kind": "backup", "name": "Backups",
+                          "state": _UNKNOWN,
+                          "detail": f"could not be checked: {type(exc).__name__}"}
+            if backup is not None:
+                parts.append(backup)
+
     parts.append(await _probe_redis())
 
     overall = max((p["state"] for p in parts), key=lambda s: _SEVERITY[s], default=_UNKNOWN)

backend/app/services/backup_health.py

@@ -0,0 +1,120 @@
+"""Are the backups keeping up? One row on the System view, and the nav dot.
+
+The backups run on their own at night, so the only way to notice one has
+stopped is to be told. The System view's health endpoint asks `backup_part`,
+which answers with a part like any other: `ok`, or `stale` with the sentence
+that says what is wrong. It warns when
+
+- the newest database backup is more than a day old, or there is none;
+- the newest database backup failed;
+- the weekly check of the off-site bucket found files missing from it, and no
+  media run has finished since to send them again.
+
+No part at all when backups aren't set up (no off-site destination and the
+nightly backup off): someone who chose not to back up is not told so every
+fifteen seconds.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from datetime import datetime
+
+from sqlalchemy import select
+
+from ..models import BackupRun, ImportSettings
+from . import backup_destination
+
+# A nightly backup is due every 24 hours. The hour of slack covers the dump
+# itself and a beat tick that lands a little late.
+DUMP_DUE_SECONDS = 25 * 3600
+
+
+@dataclass(frozen=True)
+class BackupFacts:
+    expected: bool
+    offsite: bool
+    last_dump_at: datetime | None = None
+    last_dump_error: str | None = None
+    missing: int = 0
+
+
+def _age(seconds: float) -> str:
+    hours = int(seconds // 3600)
+    if hours >= 48:
+        return f"{hours // 24} days"
+    if hours >= 1:
+        return f"{hours} h"
+    return f"{max(1, int(seconds // 60))} min"
+
+
+def verdict(facts: BackupFacts, now: datetime) -> dict | None:
+    """The part for these facts, or None when backups aren't set up."""
+    if not facts.expected:
+        return None
+    where = "to the off-site bucket" if facts.offsite else "on this server"
+    problems = []
+    if facts.last_dump_error:
+        problems.append(f"The last database backup failed: {facts.last_dump_error}")
+    if facts.last_dump_at is None:
+        if not facts.last_dump_error:
+            problems.append(f"No database backup has been taken {where} yet")
+    else:
+        age = (now - facts.last_dump_at).total_seconds()
+        if age >= DUMP_DUE_SECONDS:
+            problems.append(f"The newest database backup {where} is {_age(age)} old")
+    if facts.missing:
+        problems.append(
+            f"The bucket check found {facts.missing:,} file(s) missing from the bucket; "
+            "the next media run sends them again"
+        )
+    part = {"key": "backup", "kind": "backup", "name": "Backups"}
+    if problems:
+        return {**part, "state": "stale", "detail": ". ".join(problems)}
+    age = (now - facts.last_dump_at).total_seconds()
+    return {**part, "state": "ok",
+            "detail": f"The newest database backup {where} is {_age(age)} old"}
+
+
+def gather(session) -> BackupFacts:
+    """Read the facts `verdict` judges, from a sync session."""
+    settings = ImportSettings.load_sync(session)
+    offsite = backup_destination.describe(settings)["configured"]
+    expected = offsite or bool(settings.backup_db_nightly_enabled)
+    if not expected:
+        return BackupFacts(expected=False, offsite=False)
+
+    def newest(kind: str, statuses: tuple[str, ...]):
+        return session.execute(
+            select(BackupRun).where(BackupRun.kind == kind)
+            .where(BackupRun.status.in_(statuses))
+            .order_by(BackupRun.started_at.desc()).limit(1)
+        ).scalar_one_or_none()
+
+    last_ok = newest("db", ("ok",))
+    last_done = newest("db", ("ok", "error"))
+    failed = None
+    if last_done is not None and last_done.status == "error":
+        failed = last_done.error or "no reason recorded"
+
+    missing = 0
+    if offsite:
+        check = newest("reconcile", ("ok",))
+        dropped = int((check.manifest or {}).get("dropped") or 0) if check else 0
+        if dropped:
+            sent_since = session.execute(
+                select(BackupRun.id).where(BackupRun.kind == "media")
+                .where(BackupRun.status == "ok")
+                .where(BackupRun.finished_at > check.finished_at).limit(1)
+            ).scalar_one_or_none()
+            missing = 0 if sent_since is not None else dropped
+
+    return BackupFacts(
+        expected=True, offsite=offsite,
+        last_dump_at=(last_ok.finished_at or last_ok.started_at) if last_ok else None,
+        last_dump_error=failed, missing=missing,
+    )
+
+
+def backup_part(session, now: datetime) -> dict | None:
+    return verdict(gather(session), now)

backend/app/tasks/backup.py

@@ -280,9 +280,10 @@ def prune_backups() -> dict:
     """Daily Beat. Per-kind retention from ImportSettings.
 
     Returns {"db_deleted": N, "images_deleted": M, "media_deleted": L,
-    "restore_deleted": R, "files_unlinked": K}. Media and restore runs leave
-    no files here; only their history is trimmed, to the latest
-    MEDIA_RUNS_KEPT and RESTORE_RUNS_KEPT.
+    "restore_deleted": R, "reconcile_deleted": C, "files_unlinked": K}. Media,
+    restore and reconcile runs leave no files here; only their history is
+    trimmed, to the latest MEDIA_RUNS_KEPT, RESTORE_RUNS_KEPT and
+    RECONCILE_RUNS_KEPT.
     Tagged rows (tag IS NOT NULL) are never pruned.
     Status='running' / 'restoring' rows are never pruned — the
     recover_stalled_backup_runs sweep flips truly-stuck ones to
@@ -292,7 +293,7 @@ def prune_backups() -> dict:
     """
     SessionLocal = _sync_session_factory()
     counts = {"db_deleted": 0, "images_deleted": 0, "media_deleted": 0,
-              "restore_deleted": 0, "files_unlinked": 0}
+              "restore_deleted": 0, "reconcile_deleted": 0, "files_unlinked": 0}
     with SessionLocal() as session:
         s = ImportSettings.load_sync(session)
         for kind, keep in (
@@ -300,6 +301,7 @@ def prune_backups() -> dict:
             ("images", s.backup_images_keep_last_n),
             ("media", MEDIA_RUNS_KEPT),
             ("restore", RESTORE_RUNS_KEPT),
+            ("reconcile", RECONCILE_RUNS_KEPT),
         ):
             candidates = session.execute(
                 select(BackupRun)
@@ -365,6 +367,8 @@ MEDIA_CHUNK_SECONDS = 45 * 60
 MEDIA_RUNS_KEPT = 60
 # A failed file is listed on the run, up to this many; the count covers all.
 MEDIA_FAILURES_LISTED = 20
+# The weekly bucket check keeps about a quarter's history.
+RECONCILE_RUNS_KEPT = 12
 
 
 def _offsite_manifest(target) -> dict:
@@ -585,7 +589,11 @@ def reconcile_media_backup() -> dict:
     """Weekly Beat. Compare what the off-site bucket holds under `media/` with
     what FabledCurator believes it sent: a missing object is sent again by
     the next run, and an object already there is not sent twice. Skipped
-    while a media run is going, and when no destination is set up."""
+    while a media run is going, and when no destination is set up.
+
+    Each check is kept as a run of kind "reconcile" with its counts
+    ({listed, dropped, adopted}), so the Backups card lists it and the System
+    view can warn when files were missing from the bucket."""
     SessionLocal = _sync_session_factory()
     with SessionLocal() as session:
         target = backup_destination.saved_target(ImportSettings.load_sync(session))
@@ -597,7 +605,27 @@ def reconcile_media_backup() -> dict:
         ).scalar_one_or_none()
         if going is not None:
             return {"skipped": "a media backup is running"}
-        return backup_media.reconcile(session, target)
+        now = datetime.now(UTC)
+        row = BackupRun(kind="reconcile", status="running", triggered_by="weekly",
+                        started_at=now, last_progress_at=now,
+                        manifest={"offsite": _offsite_manifest(target)})
+        session.add(row)
+        session.commit()
+        run_id = row.id
+        try:
+            counts = backup_media.reconcile(session, target)
+        except (SoftTimeLimitExceeded, Exception) as exc:
+            session.rollback()
+            row = session.get(BackupRun, run_id)
+            if row is not None:
+                _mark_failed(session, row, exc)
+            raise
+        row = session.get(BackupRun, run_id)
+        row.status = "ok"
+        row.finished_at = datetime.now(UTC)
+        row.manifest = {**(row.manifest or {}), **counts}
+        session.commit()
+        return {"backup_run_id": run_id, **counts}
 
 
 # ── Restoring from the off-site bucket ───────────────────────────────────────