feat(dashboards): scheduler health strip, failing-source rollup, 24h activity sparkline, credential staleness nudge

D1 scheduler visibility: AppSetting last-tick stamp on every Beat tick +
GET /api/sources/schedule-status (last_tick_at/next_due_at/due_now/auto_sources)
+ SchedulerStatusBar on the Subscriptions tab (re-polled every 30s).

D2 failing-source rollup: ?failing=true on the sources list + FailingSourcesCard
on Downloads with per-source and bulk "retry" (re-runs the feed via /check).

D3 activity sparkline: GET /api/downloads/activity hourly buckets + CSS bar
chart by the stat chips (failures stacked in error color); refreshes on live poll.

D4 credential staleness: surface last_verified age + "re-verify recommended"
warning past 30d; also fixes the dead last_verified_at field-name mismatch so
the verification row renders at all.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/app/api/downloads.py

@@ -126,6 +126,54 @@ async def downloads_stats():
     return jsonify(out)
 
 
+@downloads_bp.route("/activity", methods=["GET"])
+async def downloads_activity():
+    """Hourly download-event counts over the last `?hours=` (default 24).
+
+    Returns a fixed-length, oldest-first bucket array so the UI can render
+    a sparkline directly. Bucketing is done in Python against UTC to dodge
+    session-timezone ambiguity in SQL date_trunc.
+    """
+    try:
+        hours = int(request.args.get("hours", "24"))
+    except ValueError:
+        return jsonify({"error": "invalid_hours"}), 400
+    hours = max(1, min(168, hours))
+
+    now = datetime.now(UTC)
+    end = now.replace(minute=0, second=0, microsecond=0)
+    start = end - timedelta(hours=hours - 1)
+    buckets = [
+        {"hour": (start + timedelta(hours=i)).isoformat(),
+         "ok": 0, "error": 0, "other": 0, "total": 0}
+        for i in range(hours)
+    ]
+
+    async with get_session() as session:
+        rows = (await session.execute(
+            select(DownloadEvent.started_at, DownloadEvent.status)
+            .where(DownloadEvent.started_at >= start)
+        )).all()
+
+    for started_at, status in rows:
+        if started_at is None:
+            continue
+        sa = started_at if started_at.tzinfo else started_at.replace(tzinfo=UTC)
+        idx = int((sa - start).total_seconds() // 3600)
+        if not (0 <= idx < hours):
+            continue
+        b = buckets[idx]
+        if status == "ok":
+            b["ok"] += 1
+        elif status == "error":
+            b["error"] += 1
+        else:
+            b["other"] += 1
+        b["total"] += 1
+
+    return jsonify({"hours": hours, "buckets": buckets})
+
+
 @downloads_bp.route("/<int:event_id>", methods=["GET"])
 async def get_download(event_id: int):
     async with get_session() as session:

backend/app/api/sources.py

@@ -5,6 +5,7 @@ from sqlalchemy import select
 
 from ..extensions import get_session
 from ..models import DownloadEvent, Source
+from ..services.scheduler_service import scheduler_status
 from ..services.source_service import (
     KNOWN_PLATFORMS,
     ArtistNotFoundError,
@@ -35,11 +36,19 @@ async def list_sources():
             artist_id = int(artist_id_raw)
         except ValueError:
             return _bad("invalid_artist_id", detail="artist_id must be an integer")
+    failing = request.args.get("failing", "").lower() in ("1", "true", "yes")
     async with get_session() as session:
-        records = await SourceService(session).list(artist_id=artist_id)
+        records = await SourceService(session).list(artist_id=artist_id, failing=failing)
     return jsonify([r.to_dict() for r in records])
 
 
+@sources_bp.route("/schedule-status", methods=["GET"])
+async def schedule_status():
+    """FC-dashboards: scheduler health for the Subscriptions hub."""
+    async with get_session() as session:
+        return jsonify(await scheduler_status(session))
+
+
 @sources_bp.route("/<int:source_id>", methods=["GET"])
 async def get_source(source_id: int):
     async with get_session() as session:

backend/app/services/scheduler_service.py

@@ -12,12 +12,17 @@ from sqlalchemy import select
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
-from ..models import Artist, ImportSettings, Source
+from ..models import AppSetting, Artist, ImportSettings, Source
 
 MIN_INTERVAL_SECONDS = 60
 MAX_INTERVAL_SECONDS = 86400
 MAX_BACKOFF_EXPONENT = 6
 
+# AppSetting key stamped every time the Beat tick fires (see scan.py). The
+# tick runs every 60s; the UI flags the scheduler as stalled if the last
+# stamp is older than a few minutes.
+SCHEDULER_LAST_TICK_KEY = "scheduler_last_tick_at"
+
 
 def compute_effective_interval(
     source: Source, artist: Artist, settings: ImportSettings,
@@ -78,3 +83,64 @@ def compute_next_check_at(
         return None
     interval = compute_effective_interval(source, artist, settings)
     return source.last_checked_at + timedelta(seconds=interval)
+
+
+async def record_tick(session: AsyncSession) -> None:
+    """Stamp the current time on the SCHEDULER_LAST_TICK_KEY AppSetting.
+
+    Called once per Beat tick so the UI can prove the scheduler is alive.
+    Commits its own write so the stamp survives even if the rest of the
+    tick errors out.
+    """
+    now_iso = datetime.now(UTC).isoformat()
+    row = (await session.execute(
+        select(AppSetting).where(AppSetting.key == SCHEDULER_LAST_TICK_KEY)
+    )).scalar_one_or_none()
+    if row is None:
+        session.add(AppSetting(key=SCHEDULER_LAST_TICK_KEY, value=now_iso))
+    else:
+        row.value = now_iso
+    await session.commit()
+
+
+async def scheduler_status(session: AsyncSession) -> dict:
+    """Summarise scheduler health for the dashboard.
+
+    Returns last_tick_at (when Beat last fired), next_due_at (earliest
+    upcoming scheduled check across enabled auto-check sources), due_now
+    (how many are due right now), and auto_sources (total under schedule).
+    """
+    last_tick_at = (await session.execute(
+        select(AppSetting.value).where(AppSetting.key == SCHEDULER_LAST_TICK_KEY)
+    )).scalar_one_or_none()
+
+    rows = (await session.execute(
+        select(Source)
+        .options(selectinload(Source.artist))
+        .join(Artist, Source.artist_id == Artist.id)
+        .where(Source.enabled.is_(True))
+        .where(Artist.auto_check.is_(True))
+    )).scalars().all()
+    settings = (await session.execute(
+        select(ImportSettings).where(ImportSettings.id == 1)
+    )).scalar_one()
+
+    now = datetime.now(UTC)
+    due_now = 0
+    next_due_at: datetime | None = None
+    for s in rows:
+        if s.last_checked_at is None:
+            due_now += 1
+            continue
+        nca = compute_next_check_at(s, s.artist, settings)
+        if nca is None or nca <= now:
+            due_now += 1
+        elif next_due_at is None or nca < next_due_at:
+            next_due_at = nca
+
+    return {
+        "last_tick_at": last_tick_at,
+        "next_due_at": next_due_at.isoformat() if next_due_at else None,
+        "due_now": due_now,
+        "auto_sources": len(rows),
+    }

backend/app/services/source_service.py

@@ -151,14 +151,19 @@ class SourceService:
         settings = await self._load_settings()
         return self._build_record(source, artist, settings)
 
-    async def list(self, artist_id: int | None = None) -> list[SourceRecord]:
-        stmt = (
-            select(Source, Artist)
-            .join(Artist, Artist.id == Source.artist_id)
-            .order_by(Artist.name.asc(), Source.id.asc())
-        )
+    async def list(
+        self, artist_id: int | None = None, failing: bool = False,
+    ) -> list[SourceRecord]:
+        stmt = select(Source, Artist).join(Artist, Artist.id == Source.artist_id)
         if artist_id is not None:
             stmt = stmt.where(Source.artist_id == artist_id)
+        if failing:
+            # Worst-first so the rollup card surfaces the loudest failures.
+            stmt = stmt.where(Source.consecutive_failures > 0).order_by(
+                Source.consecutive_failures.desc(), Artist.name.asc(),
+            )
+        else:
+            stmt = stmt.order_by(Artist.name.asc(), Source.id.asc())
         rows = (await self.session.execute(stmt)).all()
         settings = await self._load_settings()
         return [self._build_record(s, a, settings) for s, a in rows]

backend/app/tasks/scan.py

@@ -18,7 +18,7 @@ from ..celery_app import celery
 from ..config import get_config
 from ..models import DownloadEvent, ImportBatch, ImportSettings, ImportTask
 from ..services.archive_extractor import is_archive
-from ..services.scheduler_service import select_due_sources
+from ..services.scheduler_service import record_tick, select_due_sources
 from ._sync_engine import sync_session_factory as _sync_session_factory
 
 
@@ -151,6 +151,8 @@ async def _tick_due_sources_async() -> dict:
     factory, engine = _async_session_factory()
     try:
         async with factory() as session:
+            # Prove the scheduler is alive even on empty ticks (UI reads this).
+            await record_tick(session)
             due = await select_due_sources(session)
             if not due:
                 return {