feat(scheduler): admin run-now/run-all triggers + live progress + completion notices

Add per-job "Run now" buttons and a "Start all now" button to the admin Scheduler
dashboard (admin-gated endpoints). Triggers run the job in a background thread
independent of its interval, refusing a concurrent run (409). While running, the
long jobs (maintenance, enrich, backfill, shorts) report live progress through a
decoupled contextvar sink, shown as a progress bar on the job row via the existing
4s poll. A manually-triggered run posts a completion notification to the triggering
admin's inbox (scheduled runs stay silent to avoid spam); the inbox renders the
"scheduler" type trilingually from type+data. While here, give the maintenance job
its missing dashboard label/description in all three languages.

backend/app/progress.py

@@ -0,0 +1,40 @@
+"""Lightweight, decoupled progress reporting for long-running scheduler jobs.
+
+A job's work function calls `report(current, total, phase)`; the scheduler binds a sink (via
+`bind()`) that forwards it into the in-memory activity the admin dashboard polls. A contextvar
+holds the sink, so the sync code stays unaware of the scheduler and concurrent jobs (each in
+its own thread/context) never cross wires. When no sink is bound (e.g. a manual sync endpoint),
+`report` is a cheap no-op.
+"""
+import contextvars
+from typing import Callable, Optional
+
+Sink = Callable[[int, Optional[int], Optional[str]], None]
+
+_sink: contextvars.ContextVar[Optional[Sink]] = contextvars.ContextVar(
+    "progress_sink", default=None
+)
+
+
+def report(current: int, total: int | None = None, phase: str | None = None) -> None:
+    """Report job progress to the bound sink, if any. `total=None` means indeterminate."""
+    sink = _sink.get()
+    if sink is not None:
+        sink(current, total, phase)
+
+
+class bind:
+    """Context manager that binds a progress sink for the duration of a job run."""
+
+    def __init__(self, sink: Sink):
+        self._sink = sink
+        self._token = None
+
+    def __enter__(self) -> "bind":
+        self._token = _sink.set(self._sink)
+        return self
+
+    def __exit__(self, *exc) -> bool:
+        if self._token is not None:
+            _sink.reset(self._token)
+        return False

backend/app/routes/scheduler.py

@@ -10,7 +10,15 @@ 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 JOB_INTERVALS, MAX_INTERVAL, MIN_INTERVAL, apply_interval, scheduler_snapshot
+from app.scheduler import (
+    JOB_INTERVALS,
+    MAX_INTERVAL,
+    MIN_INTERVAL,
+    apply_interval,
+    scheduler_snapshot,
+    trigger_all,
+    trigger_job,
+)
 from app.sync.runner import estimate_deep_backfill
 
 router = APIRouter(prefix="/api/admin/scheduler", tags=["admin"])
@@ -40,6 +48,29 @@ def set_job_interval(
     return {"id": job_id, "interval_minutes": applied}
 
 
+@router.post("/jobs/{job_id}/run")
+def run_job_now(
+    job_id: str,
+    user: User = Depends(admin_user),
+) -> dict:
+    """Trigger a job to run immediately (in a background thread), independent of its
+    interval. 409 if it's already running, 404 for an unknown job. On completion the
+    triggering admin gets an inbox notification with the result."""
+    try:
+        result = trigger_job(job_id, actor_id=user.id)
+    except KeyError:
+        raise HTTPException(status_code=404, detail="Unknown job")
+    if result == "already_running":
+        raise HTTPException(status_code=409, detail="Job is already running")
+    return {"id": job_id, "started": True}
+
+
+@router.post("/run-all")
+def run_all_now(user: User = Depends(admin_user)) -> dict:
+    """Trigger every job that isn't already running. Returns the ids actually started."""
+    return {"started": trigger_all(actor_id=user.id)}
+
+
 @router.get("")
 def get_scheduler(
     _: User = Depends(admin_user), db: Session = Depends(get_db)

backend/app/scheduler.py

@@ -1,16 +1,19 @@
 """Background scheduler: free RSS detection, enrichment of new videos, and quota-aware
 recent-first / deep backfill. Runs inside the API process (single worker)."""
+import contextvars
 import logging
 import threading
 from datetime import datetime, timezone
+from typing import Callable
 
 from apscheduler.schedulers.background import BackgroundScheduler
 from sqlalchemy import select
 
-from app import quota
+from app import progress, quota
 from app.config import settings
 from app.db import SessionLocal
 from app.models import SchedulerSetting
+from app.notifications import create_notification
 from app.state import is_sync_paused
 from app.sync.autotag import run_autotag_all
 from app.sync.maintenance import run_maintenance
@@ -35,6 +38,13 @@ _scheduler: BackgroundScheduler | None = None
 _activity: dict[str, dict] = {}
 _activity_lock = threading.Lock()
 
+# Set (per-thread) by a manual "run now" trigger to the id of the admin who clicked, so that
+# run — and only that run, not the recurring scheduled ones — posts a completion notification
+# to their inbox. Unset (None) for scheduled runs.
+_manual_actor: contextvars.ContextVar[int | None] = contextvars.ContextVar(
+    "manual_actor", default=None
+)
+
 # Each interval job's env/config default period (minutes). The admin can override any of
 # these at runtime (stored in scheduler_settings, applied live); see load_intervals.
 JOB_INTERVALS: dict[str, int] = {
@@ -90,24 +100,55 @@ def _record(name: str, **fields) -> None:
         _activity.setdefault(name, {}).update(fields)
 
 
+def _notify_done(db, actor_id: int, job_id: str, status: str, summary: str | None) -> None:
+    """Post a completion notice to the triggering admin's inbox (manual runs only). Best
+    effort — a notification failure must never mask the job's own outcome."""
+    try:
+        create_notification(
+            db,
+            user_id=actor_id,
+            type="scheduler",
+            title=f"{job_id}: {status}",  # English fallback; UI renders translated from data
+            body=summary,
+            data={"job_id": job_id, "status": status, "summary": summary},
+        )
+    except Exception:
+        db.rollback()
+        logger.exception("failed to write completion notification for job %s", job_id)
+
+
 def _job(name: str, fn) -> None:
     db = SessionLocal()
-    _record(name, running=True, last_started=_now_iso(), last_error=None)
+    actor_id = _manual_actor.get()  # set only for a manual "run now" trigger
+    _record(name, running=True, last_started=_now_iso(), last_error=None, progress=None)
+
+    def sink(current, total, phase):
+        _record(name, progress={"current": current, "total": total, "phase": phase})
+
     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")
+                     last_result="paused", progress=None)
+            if actor_id is not None:
+                _notify_done(db, actor_id, name, "skipped", "sync paused")
             return
-        result = fn(db)
+        with progress.bind(sink):
+            result = fn(db)
+        summary = _summarize(result)
         logger.info("job %s -> %s", name, result)
         _record(name, running=False, status="ok", last_finished=_now_iso(),
-                last_result=_summarize(result))
+                last_result=summary, progress=None)
+        if actor_id is not None:
+            _notify_done(db, actor_id, name, "ok", summary)
     except Exception as exc:
         db.rollback()
         logger.exception("job %s failed", name)
+        err = str(exc) or exc.__class__.__name__
         _record(name, running=False, status="error", last_finished=_now_iso(),
-                last_error=str(exc) or exc.__class__.__name__)
+                last_error=err, progress=None)
+        if actor_id is not None:
+            _notify_done(db, actor_id, name, "error", err)
     finally:
         db.close()
 
@@ -151,6 +192,7 @@ def scheduler_snapshot() -> dict:
                 "last_finished": a.get("last_finished"),
                 "last_result": a.get("last_result"),
                 "last_error": a.get("last_error"),
+                "progress": a.get("progress"),
             }
         )
     return {"running_here": running_here, "enabled": settings.scheduler_enabled, "jobs": jobs}
@@ -211,6 +253,58 @@ def _maintenance_job() -> None:
     _job("maintenance", work)
 
 
+# job_id -> wrapper. The single source of truth for which jobs exist and how to run one,
+# shared by start_scheduler (recurring registration) and trigger_job (manual "run now").
+JOB_FUNCS: dict[str, Callable[[], None]] = {
+    "rss_poll": _rss_job,
+    "enrich": _enrich_job,
+    "backfill": _backfill_job,
+    "autotag": _autotag_job,
+    "shorts": _shorts_job,
+    "subscriptions": _subscriptions_job,
+    "playlist_sync": _playlist_sync_job,
+    "maintenance": _maintenance_job,
+}
+
+
+def _is_running(job_id: str) -> bool:
+    with _activity_lock:
+        return bool(_activity.get(job_id, {}).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).
+
+    `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."""
+    fn = JOB_FUNCS.get(job_id)
+    if fn is None:
+        raise KeyError(job_id)
+    if _is_running(job_id):
+        return "already_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()
+    return "started"
+
+
+def trigger_all(actor_id: int | None = None) -> list[str]:
+    """Trigger every job that isn't already running; returns the ids actually started."""
+    started = []
+    for job_id in JOB_FUNCS:
+        if trigger_job(job_id, actor_id) == "started":
+            started.append(job_id)
+    return started
+
+
 def start_scheduler() -> None:
     global _scheduler
     if not settings.scheduler_enabled or _scheduler is not None:
@@ -221,18 +315,8 @@ def start_scheduler() -> None:
         iv = load_intervals(db)
     finally:
         db.close()
-    fns = {
-        "rss_poll": _rss_job,
-        "enrich": _enrich_job,
-        "backfill": _backfill_job,
-        "autotag": _autotag_job,
-        "shorts": _shorts_job,
-        "subscriptions": _subscriptions_job,
-        "playlist_sync": _playlist_sync_job,
-        "maintenance": _maintenance_job,
-    }
     scheduler = BackgroundScheduler(timezone="UTC")
-    for job_id, fn in fns.items():
+    for job_id, fn in JOB_FUNCS.items():
         scheduler.add_job(fn, "interval", minutes=iv[job_id], id=job_id)
     scheduler.start()
     _scheduler = scheduler

backend/app/sync/maintenance.py

@@ -25,7 +25,7 @@ from datetime import datetime, timedelta, timezone
 from sqlalchemy import or_, select
 from sqlalchemy.orm import Session
 
-from app import quota
+from app import progress, quota
 from app.config import settings
 from app.models import Playlist, PlaylistItem, Video, VideoState
 from app.notifications import create_notification
@@ -147,6 +147,7 @@ def _recheck_flagged(db: Session, yt: YouTubeClient) -> dict:
     )
     if not flagged:
         return {"checked": 0, "recovered": 0, "deleted": 0, "notified": 0}
+    progress.report(0, len(flagged), "recheck")
     items = {it["id"]: it for it in yt.get_videos([v.id for v in flagged])}
     now = _now()
     recovered = 0
@@ -221,6 +222,7 @@ def _revalidate_rolling(db: Session, yt: YouTubeClient) -> dict:
                 flagged_new += 1
         db.commit()
         checked += len(page)
+        progress.report(checked, batch, "revalidate")
     return {"checked": checked, "flagged": flagged_new}
 
 

backend/app/sync/runner.py

@@ -9,7 +9,7 @@ import logging
 from sqlalchemy import select
 from sqlalchemy.orm import Session
 
-from app import quota
+from app import progress, quota
 from app.config import settings
 
 log = logging.getLogger("subfeed.sync")
@@ -64,6 +64,7 @@ def run_enrich(db: Session, max_batches: int = 200, floor: int = 200) -> int:
                 break
             n = enrich_pending(db, yt)
             total += n
+            progress.report(total, None, "enrich")  # total unknown (drains until empty)
             if n == 0:
                 break
         # Revisit transient live/upcoming videos (enrich_pending is one-shot) so an ended
@@ -121,6 +122,7 @@ def run_recent_backfill(
             try:
                 videos_added += backfill_channel_recent(db, yt, channel)
                 processed += 1
+                progress.report(processed, len(channels), "backfill_recent")
             except Exception:
                 db.rollback()
                 log.exception("Recent backfill failed for channel %s", channel.id)
@@ -166,6 +168,7 @@ def run_deep_backfill(db: Session, max_channels: int = 10, max_pages: int = 5) -
             try:
                 videos_added += backfill_channel_deep(db, yt, channel, max_pages)
                 processed += 1
+                progress.report(processed, len(channels), "backfill_deep")
             except Exception:
                 db.rollback()
                 log.exception("Deep backfill failed for channel %s", channel.id)

backend/app/sync/videos.py

@@ -9,6 +9,7 @@ from sqlalchemy import and_, func, or_, select, update
 from sqlalchemy.dialects.postgresql import insert as pg_insert
 from sqlalchemy.orm import Session
 
+from app import progress
 from app.config import settings
 from app.models import Channel, Video
 from app.youtube.client import YouTubeClient, best_thumbnail
@@ -365,5 +366,6 @@ def run_shorts_classification(db: Session, limit: int | None = None) -> dict:
                 probed += 1
                 if result:
                     shorts += 1
+                progress.report(probed, len(candidates), "probe")
     db.commit()
     return {"probed": probed, "shorts": shorts}