Live channels: background scheduler + authoring/control UI (slices 3-4)

Completes the live-channel feature on top of the engine primitives from the prior
commits: a background scheduler that auto-updates channels marked "live", and the
UI to author and control it. Ships OFF (recipes_enabled defaults false).

Backend — scheduler (backend/scheduler.py, new):
  - A single in-process asyncio loop started from main.py's lifespan. Wakes every
    60s and runs a cycle when recipes_enabled is true, not paused, and
    recipe_interval_hours has elapsed (re-reads config each tick, so toggles apply
    within a minute — mirrors how the auth middleware re-reads config per request).
  - Each cycle (under deploy_lock, blocking I/O offloaded to a worker thread):
    build the Tunarr index once, then for every "live": true channel, re-resolve
    its content and diff the fresh program-id set against read_channel_programming
    (the channel's currently-scheduled set). Patch in place via
    update_channel_in_place ONLY on a difference. No state file — Tunarr is the
    source of truth, so the loop survives restarts and an unchanged channel is a
    cheap no-op (no Plex guide churn). 404 / unreadable / resolves-to-empty are
    skipped and logged, never fatal.
  - Writes a rolling diff log to data/logs/recipes.log; keeps last_cycle in memory.
  - deploy_lock (asyncio.Lock) is shared with pipeline_router: probe/deploy/
    deploy-selective now stream through _locked_stream, so a manual deploy and a
    scheduler cycle never touch Tunarr concurrently.

Backend — endpoints (recipes_router.py):
  - GET  /api/recipes/status — enabled, paused, running, interval, live_count, last_cycle
  - POST /api/recipes/run?apply= — run one cycle on demand; apply=false is a dry run
    (detect + log, no Tunarr writes). Same code path the loop uses — the test/refresh trigger.
  - POST /api/recipes/pause?paused= — runtime kill switch, no restart needed.
  - POST /api/recipes/config — merge-writes recipes_enabled / recipe_interval_hours
    into config.json. Deliberately separate from the strict /config ConfigModel,
    which rewrites the whole file and would drop unknown keys.

Frontend (slice 4):
  - Channels page: the edit modal gains a "Live recipe" section — a Switch to mark
    the channel live, and a franchise auto-match builder. The builder calls
    /api/recipes/preview and shows matched titles as a checklist; unchecking a title
    adds it to the rule's exclude (the false-positive escape hatch). Saving stores a
    {"match":"title_contains", value, order, exclude} entry in the channel's content.
    A "live" badge marks live channels.
  - Settings: a "Live Channels" card — master enable Switch + check-interval input,
    saved via /api/recipes/config.
  - Dashboard: an "Auto-Updates" card (shown when enabled or any channel is live) —
    on/paused state, live count + interval, last-cycle changes, Pause/Resume toggle,
    and a "Check now" button.

Verified locally in Docker against live Plex/Tunarr (no mutations):
  - Scheduler: idempotent no-op cycle = changed:0; positive detection (synthetic
    live channel) correctly reports adds with human labels (e.g. "Martin x132");
    dry-run mutates nothing; loop auto-runs on boot when enabled, idle when disabled.
  - UI/endpoints: SPA serves; POST /api/recipes/config persists enabled+interval and
    leaves all connection/auth keys intact (clobber check); status reflects changes;
    frontend builds clean (tsc + vite).

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

Author: AlpineArchitecture

backend/main.py

@@ -18,6 +18,7 @@
 from fastapi.responses import HTMLResponse, Response
 from fastapi.staticfiles import StaticFiles
 
+import scheduler
 from routers import channels_router, config_router, logs_router, pipeline_router, recipes_router, status_router
 
 DATA_DIR = Path(os.environ.get("PROGRAMMARR_DATA", Path(__file__).parent.parent))
@@ -28,7 +29,16 @@
 async def lifespan(app: FastAPI):
     # Ensure data directories exist before serving any requests
     (DATA_DIR / "logs").mkdir(parents=True, exist_ok=True)
-    yield
+    # Start the live-channel scheduler loop (no-op unless recipes_enabled in config)
+    scheduler_task = asyncio.create_task(scheduler.scheduler_loop())
+    try:
+        yield
+    finally:
+        scheduler_task.cancel()
+        try:
+            await scheduler_task
+        except asyncio.CancelledError:
+            pass
 
 
 app = FastAPI(title="Programmarr", docs_url="/api/docs", redoc_url=None, lifespan=lifespan)

backend/routers/pipeline_router.py

@@ -12,6 +12,8 @@
 from fastapi.responses import FileResponse, StreamingResponse
 from pydantic import BaseModel
 
+import scheduler  # noqa: E402  (backend/ on sys.path) — shared deploy_lock
+
 router = APIRouter()
 DATA_DIR = Path(os.environ.get("PROGRAMMARR_DATA", Path(__file__).parent.parent.parent))
 SCRIPTS_DIR = Path(os.environ.get("PROGRAMMARR_SCRIPTS", Path(__file__).parent.parent.parent))
@@ -91,6 +93,17 @@ def _sse(gen: AsyncGenerator[str, None]) -> StreamingResponse:
     )
 
 
+async def _locked_stream(script: str, args: list[str], tag: str) -> AsyncGenerator[str, None]:
+    """Stream a subprocess while holding the shared deploy_lock for its full duration.
+
+    Used for create.py runs (probe/deploy) so the live-channel scheduler can't patch
+    a channel in the middle of a deploy deleting/recreating it (and vice versa).
+    """
+    async with scheduler.deploy_lock:
+        async for chunk in _stream(script, args, tag):
+            yield chunk
+
+
 class ExportOptions(BaseModel):
     no_crossref: bool = False
     movie_sections: Optional[list[str]] = None  # None = auto-detect; [] = skip type entirely
@@ -272,7 +285,7 @@ async def run_probe(from_channel: Optional[str] = Query(None)):
     args = ["--probe"]
     if from_channel:
         args += ["--from", from_channel]
-    return _sse(_stream("create.py", args, "probe"))
+    return _sse(_locked_stream("create.py", args, "probe"))
 
 
 @router.post("/pipeline/deploy")
@@ -284,7 +297,7 @@ async def run_deploy(from_channel: Optional[str] = Query(None), protected: str =
         args += ["--from", from_channel]
     if protected:
         args += ["--protect", protected]
-    return _sse(_stream("create.py", args, "deploy"))
+    return _sse(_locked_stream("create.py", args, "deploy"))
 
 
 class DeployRequest(BaseModel):
@@ -319,7 +332,7 @@ async def run_deploy_selective(req: DeployRequest):
         args.append("--no-delete")
     if req.protected_numbers:
         args += ["--protect", ",".join(str(n) for n in req.protected_numbers)]
-    return _sse(_stream("create.py", args, "deploy"))
+    return _sse(_locked_stream("create.py", args, "deploy"))
 
 
 @router.post("/pipeline/images")

backend/routers/recipes_router.py

@@ -12,7 +12,7 @@
 from pathlib import Path
 from typing import Optional
 
-from fastapi import APIRouter, HTTPException
+from fastapi import APIRouter, HTTPException, Query
 from pydantic import BaseModel
 
 DATA_DIR = Path(os.environ.get("PROGRAMMARR_DATA", Path(__file__).parent.parent.parent))
@@ -24,6 +24,7 @@
     sys.path.insert(0, str(SCRIPTS_DIR))
 
 import channel_engine  # noqa: E402
+import scheduler  # noqa: E402  (backend/ is on sys.path)
 
 router = APIRouter()
 
@@ -66,3 +67,47 @@ def preview_recipe(req: PreviewRequest):
         req.value, movie_map, show_map, order=req.order, exclude=req.exclude
     )
     return {"value": req.value, "order": req.order, "count": len(preview), "matches": preview}
+
+
+# ── Scheduler control / status ─────────────────────────────────────────────────
+
+@router.get("/recipes/status")
+def recipes_status():
+    """Current scheduler state: enabled flag, pause, interval, live count, last cycle."""
+    return scheduler.get_status()
+
+
+@router.post("/recipes/run")
+async def recipes_run(apply: bool = Query(True)):
+    """Run one cycle on demand. apply=false is a dry run (detect + log, no patch).
+
+    This is the manual-refresh / test trigger — it runs the exact same code path
+    the background loop does, so you don't have to wait for the interval.
+    """
+    return await scheduler.run_cycle(apply=apply)
+
+
+@router.post("/recipes/pause")
+def recipes_pause(paused: bool = Query(True)):
+    """Runtime kill switch — halt (or resume) auto-updates without a restart."""
+    return scheduler.set_paused(paused)
+
+
+class RecipeConfigRequest(BaseModel):
+    enabled: bool
+    interval_hours: float = 12
+
+
+@router.post("/recipes/config")
+def set_recipe_config(req: RecipeConfigRequest):
+    """Persist the master enable flag + interval. Merge-writes config.json so it
+    never clobbers connection/auth settings (unlike the strict /config model)."""
+    cfg = _load_config()
+    cfg["recipes_enabled"] = req.enabled
+    cfg["recipe_interval_hours"] = req.interval_hours
+    try:
+        with open(DATA_DIR / "config.json", "w") as f:
+            json.dump(cfg, f, indent=4)
+    except Exception as e:
+        raise HTTPException(500, f"Could not save config: {e}")
+    return scheduler.get_status()