feat(scheduler): admin dashboard endpoint + per-job activity tracking

Record each scheduler job's run activity in-process (running / last result /
error / timestamps) and expose a snapshot. New admin-only GET
/api/admin/scheduler returns the per-job activity (with APScheduler next-run
times), the DB-derived work still queued (channels/videos pending, deep ETA,
live-refresh backlog) and the shared quota picture.

backend/app/main.py

@@ -26,7 +26,19 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from app import auth
 from app.config import settings
-from app.routes import admin, channels, feed, health, me, playlists, quota, sync, tags, version
+from app.routes import (
+    admin,
+    channels,
+    feed,
+    health,
+    me,
+    playlists,
+    quota,
+    scheduler as scheduler_routes,
+    sync,
+    tags,
+    version,
+)
 from app.scheduler import shutdown_scheduler, start_scheduler
 
 
@@ -68,6 +80,7 @@ app.include_router(me.router)
 app.include_router(channels.router)
 app.include_router(playlists.router)
 app.include_router(admin.router)
+app.include_router(scheduler_routes.router)
 app.include_router(quota.router)
 app.include_router(version.router)
 

backend/app/routes/scheduler.py

@@ -0,0 +1,86 @@
+"""Admin Scheduler dashboard: a live view of what the background scheduler is doing —
+per-job run activity (from the in-process scheduler), the work still queued (DB-derived),
+and the shared quota picture."""
+from fastapi import APIRouter, Depends
+from sqlalchemy import and_, func, or_, select
+from sqlalchemy.orm import Session
+
+from app import quota, state
+from app.config import settings
+from app.db import get_db
+from app.models import Channel, Subscription, User, Video
+from app.routes.admin import admin_user
+from app.scheduler import scheduler_snapshot
+from app.sync.runner import estimate_deep_backfill
+
+router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"])
+
+
+@router.get("")
+def get_scheduler(
+    _: User = Depends(admin_user), db: Session = Depends(get_db)
+) -> dict:
+    snap = scheduler_snapshot()
+
+    # Work still queued (shared catalog, so these are instance-wide, not per-user).
+    deep_requested = (
+        select(Subscription.id)
+        .where(
+            Subscription.channel_id == Channel.id,
+            Subscription.deep_requested.is_(True),
+        )
+        .exists()
+    )
+    deep_channels = (
+        db.execute(
+            select(Channel).where(Channel.backfill_done.is_(False), deep_requested)
+        )
+        .scalars()
+        .all()
+    )
+    deep_eta = estimate_deep_backfill(db, deep_channels)
+
+    queue = {
+        "channels_recent_pending": db.scalar(
+            select(func.count()).select_from(Channel).where(Channel.recent_synced_at.is_(None))
+        )
+        or 0,
+        "channels_deep_pending": deep_eta["channels_pending"],
+        "deep_eta_seconds": deep_eta["eta_seconds"],
+        "videos_pending_enrich": db.scalar(
+            select(func.count()).select_from(Video).where(Video.enriched_at.is_(None))
+        )
+        or 0,
+        "videos_pending_shorts": db.scalar(
+            select(func.count())
+            .select_from(Video)
+            .where(Video.shorts_probed.is_(False), Video.enriched_at.is_not(None))
+        )
+        or 0,
+        "videos_live_refresh": db.scalar(
+            select(func.count())
+            .select_from(Video)
+            .where(
+                or_(
+                    Video.live_status.in_(("live", "upcoming")),
+                    and_(Video.live_status == "was_live", Video.duration_seconds.is_(None)),
+                )
+            )
+        )
+        or 0,
+    }
+
+    used = quota.units_used_today(db)
+    quota_info = {
+        "used_today": used,
+        "remaining_today": quota.remaining_today(db),
+        "daily_budget": settings.quota_daily_budget,
+        "backfill_reserve": settings.backfill_quota_reserve,
+    }
+
+    return {
+        **snap,
+        "paused": state.is_sync_paused(db),
+        "queue": queue,
+        "quota": quota_info,
+    }

backend/app/scheduler.py

@@ -1,6 +1,8 @@
 """Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
 recent-first / deep backfill. Runs inside the API process (single worker)."""
 import logging
+import threading
+from datetime import datetime, timezone
 
 from apscheduler.schedulers.background import BackgroundScheduler
 
@@ -23,22 +25,96 @@ logger = logging.getLogger("subfeed.scheduler")
 
 _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.
+_activity: dict[str, dict] = {}
+_activity_lock = threading.Lock()
+
+# Each interval job's configured period, surfaced so the dashboard can show "every N min".
+JOB_INTERVALS: dict[str, int] = {
+    "rss_poll": settings.rss_poll_minutes,
+    "enrich": settings.enrich_interval_minutes,
+    "backfill": settings.backfill_interval_minutes,
+    "autotag": settings.autotag_interval_minutes,
+    "shorts": settings.shorts_probe_interval_minutes,
+    "subscriptions": settings.subscriptions_resync_minutes,
+    "playlist_sync": settings.playlist_sync_minutes,
+}
+
+
+def _now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def _record(name: str, **fields) -> None:
+    with _activity_lock:
+        _activity.setdefault(name, {}).update(fields)
+
 
 def _job(name: str, fn) -> None:
     db = SessionLocal()
+    _record(name, running=True, last_started=_now_iso(), last_error=None)
     try:
         if is_sync_paused(db):
             logger.info("job %s skipped (sync paused)", name)
+            _record(name, running=False, status="skipped", last_finished=_now_iso(),
+                     last_result="paused")
             return
         result = fn(db)
         logger.info("job %s -> %s", name, result)
-    except Exception:
+        _record(name, running=False, status="ok", last_finished=_now_iso(),
+                last_result=_summarize(result))
+    except Exception as exc:
         db.rollback()
         logger.exception("job %s failed", name)
+        _record(name, running=False, status="error", last_finished=_now_iso(),
+                last_error=str(exc) or exc.__class__.__name__)
     finally:
         db.close()
 
 
+def _summarize(result) -> str | None:
+    """Compact one-line rendering of a job's return value for the dashboard."""
+    if result is None:
+        return None
+    if isinstance(result, dict):
+        return ", ".join(f"{k}={v}" for k, v in result.items()) or 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
+    next_runs: dict[str, str | None] = {}
+    if _scheduler is not None:
+        for job in _scheduler.get_jobs():
+            nr = getattr(job, "next_run_time", None)
+            next_runs[job.id] = nr.astimezone(timezone.utc).isoformat() if nr else None
+    with _activity_lock:
+        acts = {k: dict(v) for k, v in _activity.items()}
+    jobs = []
+    for job_id, interval in JOB_INTERVALS.items():
+        a = acts.get(job_id, {})
+        jobs.append(
+            {
+                "id": job_id,
+                "interval_minutes": 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"),
+                "last_error": a.get("last_error"),
+            }
+        )
+    return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
+
+
 def _rss_job() -> None:
     _job("rss_poll", lambda db: run_rss_poll(db))