Add scheduler_service and pipeline_runner

pipeline_runner: subprocess execution wrapper replacing flock -n with a
single asyncio.Lock (deliberately one global lock, not per-resource --
matches the old flock's all-jobs-share-one-lock behavior rather than
over-engineering it). Records one job_runs row per invocation
(started/finished/status/exit_code/summary/log_path/triggered_by).
If the lock is already held, records status='skipped_lock' immediately
instead of silently dropping the run.

Found and fixed a real concurrency bug via testing: the obvious
asyncio.wait_for(lock.acquire(), timeout=0) idiom for a non-blocking
try-acquire is broken in asyncio -- the wrapping Task's first iteration and
the timeout-0 callback race with no guaranteed ordering, so it timed out on
literally every call, including the very first uncontended one. Fixed using
lock.locked() + acquire(), relying on acquire()'s fast path never
suspending when uncontended.

scheduler_service: AsyncIOScheduler with the default in-memory job store
(NOT SQLAlchemyJobStore -- it pickles job functions to persist them, and
the _lib()/_beet() factory closures here aren't picklable; MemoryJobStore
avoids this since register_all_jobs() rebuilds the whole schedule from code
+ DB on every startup anyway). All 19 maintenance jobs ported 1:1 from
/etc/cron.d/sldl-maintenance with their exact cron expressions; dedup
deliberately has no --apply (dry-run only on schedule, per the
false-negative-biased dedup preference). Playlist jobs are synced live from
the playlists table (sync_playlist_jobs), so add/remove-in-the-UI takes
effect with no redeploy -- wired into playlist_service's create/update/
delete. Maintenance job enable/disable persists to the scheduled_jobs table
so a paused job stays paused across a restart despite the in-memory job
store. trigger_now() supports "Run now" by invoking a job's function
immediately with triggered_by='manual', bypassing its schedule.

Wired into main.py's lifespan (start on boot, shutdown on exit).

Verified via integration tests: 36 jobs register correctly (17 seeded
playlists + 19 maintenance) with correct cron triggers; live playlist
create/delete correctly adds/removes its scheduler job; maintenance
enable/disable persists to the DB and takes effect live; trigger_now
correctly bypasses the schedule; pipeline_runner correctly handles
success/failure/timeout/concurrent-lock-contention with real subprocesses;
and the full app boots with the scheduler running and shuts down cleanly.

app/main.py

@@ -6,6 +6,7 @@ from starlette.middleware.sessions import SessionMiddleware
 
 from app.db import enable_beets_db_wal, init_db
 from app.routers import auth, dashboard
+from app.services import scheduler_service
 from app.settings import settings
 
 
@@ -13,10 +14,15 @@ from app.settings import settings
 async def lifespan(_app: FastAPI):
     init_db()
     enable_beets_db_wal()
-    # Task 5 (scheduler_service) hooks APScheduler start/stop in here once
+
-    # the job registry and pipeline_runner exist.
+    scheduler = scheduler_service.create_scheduler()
+    scheduler_service.register_all_jobs(scheduler)
+    scheduler.start()
+
     yield
 
+    scheduler.shutdown(wait=False)
+
 
 def create_app() -> FastAPI:
     app = FastAPI(title="alembic", lifespan=lifespan)

app/services/pipeline_runner.py

@@ -0,0 +1,168 @@
+import asyncio
+import time
+from pathlib import Path
+
+from app.db import SessionLocal
+from app.models import JobRun
+from app.settings import settings
+
+# Single global lock replacing /var/lock/sldl-pipeline.lock + flock -n. One
+# process (this one) now owns every pipeline invocation, so one asyncio.Lock
+# is enough -- deliberately not per-resource, matching the original flock's
+# all-jobs-share-one-lock behavior exactly rather than over-engineering it.
+_lock = asyncio.Lock()
+
+_ENV_PASSTHROUGH_KEYS = ("PATH", "HOME", "LANG", "LC_ALL", "TZ")
+
+
+async def _try_acquire_nowait() -> bool:
+    """asyncio.Lock has no acquire_nowait(). asyncio.wait_for(lock.acquire(),
+    timeout=0) looks like the obvious idiom but is broken in practice: it
+    wraps acquire() in a Task, and the Task's first iteration and the
+    timeout-0 callback are both scheduled via the event loop with no
+    guaranteed ordering, so it times out almost every time even when the
+    lock is completely uncontended (confirmed empirically -- the very first,
+    uncontended call failed every time). The correct pattern relies on
+    Lock.acquire()'s fast path never suspending when the lock is free: the
+    .locked() check and the subsequent acquire() happen with no `await`
+    point in between, so nothing else can interleave in this single-
+    threaded event loop."""
+    if _lock.locked():
+        return False
+    await _lock.acquire()
+    return True
+
+
+def _subprocess_env() -> dict:
+    import os
+
+    env = {k: v for k, v in os.environ.items() if k in _ENV_PASSTHROUGH_KEYS}
+    env.setdefault("PATH", "/opt/venv/bin:/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin")
+    env["ALEMBIC_CONFIG_DIR"] = str(settings.alembic_config_dir)
+    env["MUSIC_DATA_DIR"] = str(settings.music_data_dir)
+    env["PIPELINE_DIR"] = str(settings.pipeline_dir)
+    env["BEETSDIR"] = str(settings.beets_dir)
+    return env
+
+
+def _summarize_log(log_path: Path, max_len: int = 200) -> str:
+    """Best-effort one-liner for the jobs list: the log's last non-empty
+    line. Scripts here already end their own runs with a human-readable
+    '=== ... done ===' / summary line (see pipeline-status.sh's convention),
+    so this is usually meaningful without any per-script special-casing."""
+    try:
+        lines = [line for line in log_path.read_text(errors="replace").splitlines() if line.strip()]
+        return lines[-1][:max_len] if lines else ""
+    except OSError:
+        return ""
+
+
+async def run_job(
+    job_key: str,
+    argv: list[str],
+    triggered_by: str = "schedule",
+    timeout: float | None = None,
+) -> JobRun:
+    """Run a pipeline command under the shared mutual-exclusion lock,
+    recording one job_runs row start-to-finish. If the lock is already
+    held, records status='skipped_lock' immediately and returns without
+    running anything -- the old flock -n behavior, now visible in the UI
+    instead of silently skipping."""
+    started_at = time.time()
+
+    if not await _try_acquire_nowait():
+        db = SessionLocal()
+        run = JobRun(
+            job_key=job_key,
+            started_at=started_at,
+            finished_at=started_at,
+            status="skipped_lock",
+            triggered_by=triggered_by,
+        )
+        db.add(run)
+        db.commit()
+        db.refresh(run)
+        db.close()
+        return run
+
+    try:
+        log_dir = settings.logs_dir / job_key.replace(":", "_")
+        log_dir.mkdir(parents=True, exist_ok=True)
+        log_path = log_dir / f"{time.strftime('%Y%m%d-%H%M%S')}.log"
+
+        db = SessionLocal()
+        run = JobRun(
+            job_key=job_key,
+            started_at=started_at,
+            status="running",
+            triggered_by=triggered_by,
+            log_path=str(log_path),
+        )
+        db.add(run)
+        db.commit()
+        db.refresh(run)
+        run_id = run.id
+        db.close()
+
+        exit_code: int | None
+        try:
+            with open(log_path, "wb") as log_file:
+                proc = await asyncio.create_subprocess_exec(
+                    *argv,
+                    stdout=log_file,
+                    stderr=asyncio.subprocess.STDOUT,
+                    env=_subprocess_env(),
+                )
+                try:
+                    exit_code = await asyncio.wait_for(proc.wait(), timeout=timeout)
+                except asyncio.TimeoutError:
+                    proc.kill()
+                    await proc.wait()
+                    exit_code = -1
+                    with open(log_path, "a") as f:
+                        f.write(f"\n[pipeline_runner] TIMEOUT after {timeout}s -- process killed\n")
+            status = "success" if exit_code == 0 else "failed"
+        except Exception as exc:
+            exit_code = None
+            status = "failed"
+            with open(log_path, "a") as f:
+                f.write(f"\n[pipeline_runner] exception before/while running: {exc!r}\n")
+
+        finished_at = time.time()
+        db = SessionLocal()
+        run = db.get(JobRun, run_id)
+        run.finished_at = finished_at
+        run.status = status
+        run.exit_code = exit_code
+        run.summary = _summarize_log(log_path)
+        db.commit()
+        db.refresh(run)
+        db.close()
+        return run
+    finally:
+        _lock.release()
+
+
+async def run_playlist(playlist_name: str, no_m3u: bool = False, triggered_by: str = "schedule") -> JobRun:
+    script = settings.pipeline_dir / "bin" / "run-playlist.sh"
+    argv = [str(script)]
+    if no_m3u:
+        argv.append("--no-m3u")
+    argv.append(playlist_name)
+    # 45 min SLDL_TIMEOUT (run-playlist.sh) + generous slack for the
+    # tagging/beets-import/M3U steps that follow it in the same script.
+    return await run_job(f"playlist:{playlist_name}", argv, triggered_by=triggered_by, timeout=3300)
+
+
+async def run_import_track(args: list[str], triggered_by: str = "manual") -> JobRun:
+    script = settings.pipeline_dir / "bin" / "import-track.sh"
+    return await run_job("manual:import", [str(script)] + args, triggered_by=triggered_by)
+
+
+async def run_lib_script(job_key: str, script_name: str, args: list[str] | None = None, triggered_by: str = "schedule") -> JobRun:
+    script = settings.pipeline_dir / "lib" / script_name
+    return await run_job(job_key, [str(script)] + (args or []), triggered_by=triggered_by)
+
+
+async def run_beet(job_key: str, args: list[str], triggered_by: str = "schedule") -> JobRun:
+    return await run_job(job_key, ["beet"] + args, triggered_by=triggered_by)

app/services/playlist_service.py

@@ -10,25 +10,27 @@ from app.settings import settings
 
 # The 17-entry array this project is migrating off of (was hardcoded in
 # /opt/sldl/configs/regen.sh). Used once by seed_legacy() during Stage 1/2
-# of the migration; the DB is the source of truth from then on.
+# of the migration; the DB is the source of truth from then on. cron_expr
+# values are the exact staggered slots from /etc/cron.d/sldl-maintenance,
+# preserved so cutover doesn't change anyone's download schedule.
 LEGACY_PLAYLISTS = {
-    "techno": "https://open.spotify.com/playlist/5GlAkczmLWJFIJO4FvkLEZ",
+    "digicore":  ("https://open.spotify.com/playlist/6tWHtPBECZTkZHPKFA3Fq4", "0 0 * * *", False),
-    "house": "https://open.spotify.com/playlist/2XF8ye6O5xqzT4NSfIXdfJ",
+    "techno":    ("https://open.spotify.com/playlist/5GlAkczmLWJFIJO4FvkLEZ", "0 1 * * *", False),
-    "lofi": "https://open.spotify.com/playlist/0qnyF6TAmuT3U6hFH2fiF1",
+    "house":     ("https://open.spotify.com/playlist/2XF8ye6O5xqzT4NSfIXdfJ", "30 1 * * *", False),
-    "y2k": "https://open.spotify.com/playlist/0iaA7ZJS00aRxhMwZn2GCU",
+    "lofi":      ("https://open.spotify.com/playlist/0qnyF6TAmuT3U6hFH2fiF1", "0 2 * * *", False),
-    "weeb": "https://open.spotify.com/playlist/0599AbpsKRjsp2rlJI8jhf",
+    "y2k":       ("https://open.spotify.com/playlist/0iaA7ZJS00aRxhMwZn2GCU", "30 2 * * *", False),
-    "shoegaze": "https://open.spotify.com/playlist/3R4S3tmbVKjXQ1RPAmryC5",
+    "weeb":      ("https://open.spotify.com/playlist/0599AbpsKRjsp2rlJI8jhf", "0 3 * * *", False),
-    "bass": "https://open.spotify.com/playlist/2zkF4S16efaMmmzdIOe5w2",
+    "shoegaze":  ("https://open.spotify.com/playlist/3R4S3tmbVKjXQ1RPAmryC5", "30 3 * * *", False),
-    "hiphop": "https://open.spotify.com/playlist/42JxTtOLJ7gG1ybfyy7Vmi",
+    "bass":      ("https://open.spotify.com/playlist/2zkF4S16efaMmmzdIOe5w2", "0 4 * * *", False),
-    "modular": "https://open.spotify.com/playlist/2kOfP2EM8dzdMYC6fXvlLb",
+    "hiphop":    ("https://open.spotify.com/playlist/42JxTtOLJ7gG1ybfyy7Vmi", "30 4 * * *", False),
-    "hotdog": "https://open.spotify.com/playlist/53HrOL0qw47G9ywYZz3kgX",
+    "modular":   ("https://open.spotify.com/playlist/2kOfP2EM8dzdMYC6fXvlLb", "0 5 * * *", False),
-    "kpop": "https://open.spotify.com/playlist/31e2V512TIqr5JIfgkyseo",
+    "hotdog":    ("https://open.spotify.com/playlist/53HrOL0qw47G9ywYZz3kgX", "30 5 * * *", False),
-    "jungle": "https://open.spotify.com/playlist/4LjLeXg6ElneQiTaIcAHhE",
+    "kpop":      ("https://open.spotify.com/playlist/31e2V512TIqr5JIfgkyseo", "0 6 * * *", False),
-    "goldenera": "https://open.spotify.com/playlist/57vArhigysJfgwB6CY4VXR",
+    "jungle":    ("https://open.spotify.com/playlist/4LjLeXg6ElneQiTaIcAHhE", "30 6 * * *", False),
-    "botanica": "https://open.spotify.com/playlist/5KgQT9YWz3EqhIN4Jh69IU",
+    "goldenera": ("https://open.spotify.com/playlist/57vArhigysJfgwB6CY4VXR", "0 7 * * *", False),
-    "digicore": "https://open.spotify.com/playlist/6tWHtPBECZTkZHPKFA3Fq4",
+    "botanica":  ("https://open.spotify.com/playlist/5KgQT9YWz3EqhIN4Jh69IU", "30 7 * * *", False),
-    "hardcore": "https://open.spotify.com/playlist/1J1lbzlQKligzr2LWaq9Ex",
+    "hardcore":  ("https://open.spotify.com/playlist/1J1lbzlQKligzr2LWaq9Ex", "0 23 * * *", False),
-    "liked": "https://open.spotify.com/playlist/4Z3qCYuU1sjNeNYO3Amzeo",
+    "liked":     ("https://open.spotify.com/playlist/4Z3qCYuU1sjNeNYO3Amzeo", "30 23 * * *", True),
 }
 
 
@@ -51,6 +53,7 @@ def create(
     active: bool = True,
     no_m3u: bool = False,
     notes: str | None = None,
+    cron_expr: str | None = None,
 ) -> Playlist:
     now = time.time()
     playlist = Playlist(
@@ -59,6 +62,7 @@ def create(
         active=active,
         no_m3u=no_m3u,
         notes=notes,
+        cron_expr=cron_expr,
         created_at=now,
         updated_at=now,
     )
@@ -96,16 +100,21 @@ def delete(db: Session, playlist_id: int) -> None:
     conf_path.unlink(missing_ok=True)
     _write_playlists_json(db)
 
+    from app.services import scheduler_service
+
+    scheduler = scheduler_service.get_scheduler()
+    if scheduler is not None:
+        scheduler_service.sync_playlist_jobs(scheduler)
+
 
 def seed_legacy(db: Session) -> int:
     """One-time import of the legacy hardcoded array (migration Stage 1/2).
     Skips any name that already exists. Returns the number of rows created."""
     created = 0
-    for name, url in LEGACY_PLAYLISTS.items():
+    for name, (url, cron_expr, no_m3u) in LEGACY_PLAYLISTS.items():
         if get_by_name(db, name) is not None:
             continue
-        no_m3u = name == "liked"  # matches --no-m3u liked in the old crontab
+        create(db, name=name, spotify_url=url, no_m3u=no_m3u, cron_expr=cron_expr)
-        create(db, name=name, spotify_url=url, no_m3u=no_m3u)
         created += 1
     return created
 
@@ -152,14 +161,20 @@ def regenerate_confs(db: Session) -> subprocess.CompletedProcess:
 
 
 def _sync_to_disk(db: Session) -> None:
-    """Convenience wrapper used by create()/update(): regenerate confs and
+    """Convenience wrapper used by create()/update(): regenerate confs,
-    immediately re-patch credentials into any newly-rendered file."""
+    immediately re-patch credentials into any newly-rendered file, and push
+    the change to the live scheduler if one is running -- this is what
+    makes add/remove-in-the-UI take effect without a redeploy."""
     regenerate_confs(db)
-    # Imported lazily to avoid a circular import (credential_service doesn't
+    # Imported lazily to avoid a circular import (credential_service and
-    # depend on playlist_service, but importing at module load time would
+    # scheduler_service don't depend on playlist_service, but importing at
-    # still work here — done lazily anyway to keep the dependency direction
+    # module load time would still work here — done lazily anyway to keep
-    # obvious as both services grow).
+    # the dependency direction obvious as all three services grow).
-    from app.services import credential_service
+    from app.services import credential_service, scheduler_service
 
     credential_service.render_scope(db, "soulseek")
     credential_service.render_scope(db, "spotify")
+
+    scheduler = scheduler_service.get_scheduler()
+    if scheduler is not None:
+        scheduler_service.sync_playlist_jobs(scheduler)

app/services/scheduler_service.py

@@ -0,0 +1,302 @@
+import time
+
+from apscheduler.schedulers.asyncio import AsyncIOScheduler
+from apscheduler.triggers.cron import CronTrigger
+from sqlalchemy import select
+
+from app.db import SessionLocal
+from app.models import Playlist, ScheduledJob
+from app.services import pipeline_runner
+from app.settings import settings
+
+TIMEZONE = "America/Edmonton"
+
+# ---------------------------------------------------------------------------
+# Maintenance jobs, ported 1:1 from /etc/cron.d/sldl-maintenance. Each entry
+# is (job_key, cron kwargs for CronTrigger, callable). Dedup deliberately has
+# no --apply here (dry-run only on schedule) -- a deliberate behavior change
+# from the old blind `--apply` cron, matching the false-negative-biased
+# dedup preference; applying deletions requires a reviewed confirm through
+# the dedup UI (services/dedup_review_service.py).
+#
+# NOTE on job store choice: APScheduler's SQLAlchemyJobStore pickles job
+# functions to persist them, and closures returned from a factory (like
+# _lib()/_beet() below) are NOT picklable -- `add_job` would fail at
+# runtime. MemoryJobStore (the default) keeps live objects in-process
+# instead, which is fine here because register_all_jobs() rebuilds the
+# entire schedule from code + the DB on every startup anyway. Persisted
+# enable/disable state (so a paused job stays paused across a restart)
+# lives in the app's own `scheduled_jobs` table, not in APScheduler's store.
+# ---------------------------------------------------------------------------
+
+
+def _lib(job_key: str, script_name: str, args: list[str] | None = None, timeout: float | None = None):
+    async def _run(triggered_by: str = "schedule"):
+        return await pipeline_runner.run_lib_script(job_key, script_name, args, triggered_by=triggered_by)
+
+    return _run
+
+
+def _beet(job_key: str, args: list[str]):
+    async def _run(triggered_by: str = "schedule"):
+        return await pipeline_runner.run_beet(job_key, args, triggered_by=triggered_by)
+
+    return _run
+
+
+async def _log_rotation(triggered_by: str = "schedule"):
+    """Replaces `find /var/log/sldl -name '*.log' -mtime +30 -delete`."""
+    cutoff = time.time() - 30 * 86400
+    removed = 0
+    if settings.logs_dir.exists():
+        for path in settings.logs_dir.rglob("*.log"):
+            if path.is_file() and path.stat().st_mtime < cutoff:
+                path.unlink(missing_ok=True)
+                removed += 1
+    return removed
+
+
+MAINTENANCE_JOBS: dict[str, tuple[dict, callable]] = {
+    # ==== Daily ====
+    "maintenance:sync_bandcamp": (
+        dict(minute=0, hour=8),
+        _lib("maintenance:sync_bandcamp", "sync-bandcamp.sh"),
+    ),
+    "maintenance:normalize_casing": (
+        dict(minute=52, hour=8),
+        _lib("maintenance:normalize_casing", "normalize-artist-casing.py", ["--apply"]),
+    ),
+    "maintenance:dedup": (
+        dict(minute=55, hour=8),
+        _lib("maintenance:dedup", "dedup-library.sh"),  # no --apply: dry-run only
+    ),
+    "maintenance:gen_djmix_playlist": (
+        dict(minute=57, hour=8),
+        _lib("maintenance:gen_djmix_playlist", "gen-djmix-playlist.sh"),
+    ),
+    "maintenance:gen_vgm_playlist": (
+        dict(minute=58, hour=8),
+        _lib("maintenance:gen_vgm_playlist", "gen-vgm-playlist.sh"),
+    ),
+    "maintenance:navidrome_scan": (
+        dict(minute=0, hour=9),
+        _lib("maintenance:navidrome_scan", "navidrome-scan.sh"),
+    ),
+    "maintenance:export_laptop_playlists": (
+        dict(minute=5, hour=9),
+        _lib("maintenance:export_laptop_playlists", "export-laptop-playlists.py"),
+    ),
+    "maintenance:enrich_buy_url": (
+        dict(minute=10, hour=9),
+        _lib("maintenance:enrich_buy_url", "enrich-buy-url.py", ["--apply"]),
+    ),
+    "maintenance:build_fingerprint_index": (
+        dict(minute=25, hour=9),
+        _lib("maintenance:build_fingerprint_index", "build-fingerprint-index.py", ["--workers", "8"]),
+    ),
+    "maintenance:pipeline_status_report": (
+        dict(minute=30, hour=9),
+        _lib("maintenance:pipeline_status_report", "pipeline-status.sh"),
+    ),
+    "maintenance:log_rotation": (
+        dict(minute=0, hour=0),
+        _log_rotation,
+    ),
+    # ==== Weekly (Sunday) ====
+    "maintenance:strip_mb_tags": (
+        dict(minute=30, hour=8, day_of_week="sun"),
+        _lib("maintenance:strip_mb_tags", "strip-mb-tags.sh"),
+    ),
+    "maintenance:strip_watermark_art": (
+        dict(minute=35, hour=8, day_of_week="sun"),
+        _lib("maintenance:strip_watermark_art", "strip-watermark-art.py", ["--apply"]),
+    ),
+    "maintenance:scrub_watermark_text": (
+        dict(minute=40, hour=8, day_of_week="sun"),
+        _lib("maintenance:scrub_watermark_text", "scrub-watermark-text.py", ["--apply"]),
+    ),
+    "maintenance:clean_sldl_index": (
+        dict(minute=45, hour=8, day_of_week="sun"),
+        _lib("maintenance:clean_sldl_index", "clean-sldl-index.py", ["--apply"]),
+    ),
+    "maintenance:clear_bad_genres": (
+        dict(minute=48, hour=8, day_of_week="sun"),
+        _lib("maintenance:clear_bad_genres", "clear-bad-genres.py", ["--apply"]),
+    ),
+    "maintenance:spotify_genre": (
+        dict(minute=50, hour=8, day_of_week="sun"),
+        _lib("maintenance:spotify_genre", "spotify-genre.py", ["--apply", "--force"]),
+    ),
+    "maintenance:beets_update_sync": (
+        dict(minute=53, hour=8, day_of_week="sun"),
+        _beet("maintenance:beets_update_sync", ["update"]),
+    ),
+    # ==== Monthly ====
+    "maintenance:upgrade_mp3_to_flac": (
+        dict(minute=0, hour=10, day=1),
+        _lib("maintenance:upgrade_mp3_to_flac", "upgrade-mp3-to-flac.sh"),
+    ),
+}
+
+
+# Module-level singleton so other services (playlist_service, future
+# routers) can reach the live scheduler without main.py threading it through
+# every function call. None until main.py's lifespan starts it.
+_instance: AsyncIOScheduler | None = None
+
+
+def get_scheduler() -> AsyncIOScheduler | None:
+    return _instance
+
+
+def create_scheduler() -> AsyncIOScheduler:
+    global _instance
+    # Default job store (MemoryJobStore) -- see module docstring for why
+    # SQLAlchemyJobStore doesn't work here.
+    _instance = AsyncIOScheduler(timezone=TIMEZONE)
+    return _instance
+
+
+def _ensure_scheduled_job_rows(db) -> None:
+    """Make sure every MAINTENANCE_JOBS entry has a scheduled_jobs row,
+    seeding cron_expr from the hardcoded default and enabled=True the first
+    time. Subsequent runs never overwrite an existing row -- this table is
+    the persisted source of truth for enable/disable once it exists."""
+    existing = {row.job_key for row in db.execute(select(ScheduledJob)).scalars()}
+    now = time.time()
+    for job_key, (cron_kwargs, _runner) in MAINTENANCE_JOBS.items():
+        if job_key in existing:
+            continue
+        cron_expr = "{minute} {hour} {day} {month} {day_of_week}".format(
+            minute=cron_kwargs.get("minute", "*"),
+            hour=cron_kwargs.get("hour", "*"),
+            day=cron_kwargs.get("day", "*"),
+            month=cron_kwargs.get("month", "*"),
+            day_of_week=cron_kwargs.get("day_of_week", "*"),
+        )
+        db.add(
+            ScheduledJob(
+                job_key=job_key,
+                job_type="maintenance",
+                playlist_id=None,
+                cron_expr=cron_expr,
+                enabled=True,
+            )
+        )
+    db.commit()
+    _ = now  # unused, kept for symmetry with playlists' created_at/updated_at style
+
+
+def _playlist_job_key(name: str) -> str:
+    return f"playlist:{name}"
+
+
+async def _run_playlist_job(name: str, no_m3u: bool, triggered_by: str = "schedule"):
+    return await pipeline_runner.run_playlist(name, no_m3u=no_m3u, triggered_by=triggered_by)
+
+
+def sync_playlist_jobs(scheduler: AsyncIOScheduler) -> None:
+    """Converge the scheduler's registered playlist:<name> jobs with the
+    playlists table. Call after any playlist create/update/delete -- this
+    is what makes add/remove-in-the-UI take effect without a redeploy."""
+    db = SessionLocal()
+    try:
+        playlists = list(db.execute(select(Playlist)).scalars())
+    finally:
+        db.close()
+
+    desired_ids = {
+        _playlist_job_key(p.name) for p in playlists if p.cron_expr and p.active
+    }
+    existing_ids = {j.id for j in scheduler.get_jobs() if j.id.startswith("playlist:")}
+
+    for job_id in existing_ids - desired_ids:
+        scheduler.remove_job(job_id)
+
+    for playlist in playlists:
+        job_id = _playlist_job_key(playlist.name)
+        if job_id not in desired_ids:
+            continue
+        trigger = CronTrigger.from_crontab(playlist.cron_expr, timezone=TIMEZONE)
+        scheduler.add_job(
+            _run_playlist_job,
+            trigger=trigger,
+            id=job_id,
+            args=[playlist.name, playlist.no_m3u],
+            replace_existing=True,
+            misfire_grace_time=3600,
+        )
+
+
+def register_all_jobs(scheduler: AsyncIOScheduler) -> None:
+    """Rebuild the entire in-memory schedule from code (MAINTENANCE_JOBS) +
+    the DB (scheduled_jobs.enabled, playlists). Safe to call at every
+    startup -- MemoryJobStore has nothing to reconcile against."""
+    db = SessionLocal()
+    try:
+        _ensure_scheduled_job_rows(db)
+        enabled_by_key = {
+            row.job_key: row.enabled for row in db.execute(select(ScheduledJob)).scalars()
+        }
+    finally:
+        db.close()
+
+    for job_key, (cron_kwargs, runner) in MAINTENANCE_JOBS.items():
+        if not enabled_by_key.get(job_key, True):
+            continue
+        scheduler.add_job(
+            runner,
+            trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE),
+            id=job_key,
+            replace_existing=True,
+            misfire_grace_time=3600,
+        )
+
+    sync_playlist_jobs(scheduler)
+
+
+async def trigger_now(scheduler: AsyncIOScheduler, job_key: str) -> None:
+    """'Run now' support: bypass the schedule and run this job's function
+    immediately, still through pipeline_runner's lock/logging/job_runs
+    recording (triggered_by='manual'). Works even for a currently-disabled
+    job (disabled only means "don't run on schedule")."""
+    job = scheduler.get_job(job_key)
+    if job is None:
+        raise ValueError(f"no such registered job: {job_key}")
+    func = job.func
+    args = list(job.args)
+    kwargs = dict(job.kwargs)
+    kwargs["triggered_by"] = "manual"
+    await func(*args, **kwargs)
+
+
+def set_maintenance_enabled(scheduler: AsyncIOScheduler, job_key: str, enabled: bool) -> None:
+    """Pause/resume a maintenance job AND persist the choice to
+    scheduled_jobs.enabled so it survives a restart -- this is the actual
+    'pause an individual job without a redeploy' mechanism."""
+    if job_key not in MAINTENANCE_JOBS:
+        raise ValueError(f"not a maintenance job: {job_key}")
+
+    db = SessionLocal()
+    try:
+        row = db.execute(
+            select(ScheduledJob).where(ScheduledJob.job_key == job_key)
+        ).scalar_one_or_none()
+        if row is not None:
+            row.enabled = enabled
+            db.commit()
+    finally:
+        db.close()
+
+    if enabled:
+        cron_kwargs, runner = MAINTENANCE_JOBS[job_key]
+        scheduler.add_job(
+            runner,
+            trigger=CronTrigger(**cron_kwargs, timezone=TIMEZONE),
+            id=job_key,
+            replace_existing=True,
+            misfire_grace_time=3600,
+        )
+    else:
+        if scheduler.get_job(job_key) is not None:
+            scheduler.remove_job(job_key)