backlot phase 3 review fixes: SSE filtering, thumb race, replay robustness

- ChangeHub subscriptions filtered per project: unrelated-project bursts
  can no longer flood a board's queue and starve its own change signal
- thumbnail temp files unique per request (concurrent-miss race on the
  shared .tmp path corrupted the cache)
- watcher change-mapping is pure string work (no per-path resolve() in
  thousand-file change batches); 'C:'-style project ids rejected
- replay: tz-naive timestamps treated as UTC; final render/script no
  longer leak at t0 on storyboard-less projects; single tick chain on
  rapid pause/play; drag-safe scrubber (label tracks input, board renders
  on release); render-video playback survives SSE re-renders
- state: scene id 0 joins correctly, nested depth>0 events don't corrupt
  generating state, out-of-project asset paths honestly unserveable,
  negative durations clamped, tolerant manifest stage parse
- UI: decisions alt filter precedence fixed, activity counts parallel
  same-tool runs, el() html sink removed, NaN-safe formatters

backlot/server.py

@@ -30,25 +30,34 @@ SSE_HEARTBEAT_SECONDS = 15
 
 
 class ChangeHub:
-    """Fan-out of project-change notifications to SSE subscribers."""
+    """Fan-out of project-change notifications to SSE subscribers.
+
+    Subscriptions are filtered: a board subscribed to one project only ever
+    receives that project's ids, so unrelated-project bursts can't flood its
+    queue and starve out the one notification it actually needs.
+    """
 
     def __init__(self) -> None:
-        self._subscribers: set[asyncio.Queue] = set()
+        self._subscribers: dict[asyncio.Queue, Optional[str]] = {}
 
-    def subscribe(self) -> asyncio.Queue:
+    def subscribe(self, project_id: Optional[str] = None) -> asyncio.Queue:
         q: asyncio.Queue = asyncio.Queue(maxsize=64)
-        self._subscribers.add(q)
+        self._subscribers[q] = project_id
         return q
 
     def unsubscribe(self, q: asyncio.Queue) -> None:
-        self._subscribers.discard(q)
+        self._subscribers.pop(q, None)
 
     def publish(self, project_id: str) -> None:
-        for q in list(self._subscribers):
+        for q, only in list(self._subscribers.items()):
+            if only is not None and only != project_id:
+                continue
             try:
                 q.put_nowait(project_id)
             except asyncio.QueueFull:
-                pass  # subscriber is behind; it will refetch on next event
+                # Queue holds only THIS subscriber's relevant ids, so a full
+                # queue already guarantees a pending wake-up → safe to drop.
+                pass
 
 
 hub = ChangeHub()
@@ -88,17 +97,25 @@ def _cached_summaries() -> list[dict]:
     return summaries
 
 
+# Watch-loop hot path: pure string comparison, no per-path filesystem calls
+# (change batches can be thousands of paths during a render).
+import os as _os
+
+_PROJECTS_ROOT_STR = _os.path.normcase(str(PROJECTS_DIR.resolve()))
+
+
 def _project_of_change(path_str: str) -> Optional[str]:
     """Map a changed filesystem path to a project id (None = irrelevant)."""
-    try:
-        rel = Path(path_str).resolve().relative_to(PROJECTS_DIR.resolve())
-    except (ValueError, OSError):
+    norm = _os.path.normcase(_os.path.normpath(path_str))
+    if not norm.startswith(_PROJECTS_ROOT_STR):
         return None
-    if not rel.parts:
+    rel = norm[len(_PROJECTS_ROOT_STR):].lstrip("\\/")
+    if not rel:
         return None
-    if _IGNORE_PARTS.intersection(rel.parts):
+    parts = rel.replace("\\", "/").split("/")
+    if _IGNORE_PARTS.intersection(parts):
         return None
-    return rel.parts[0]
+    return parts[0]
 
 
 async def _watch_projects() -> None:
@@ -153,25 +170,24 @@ def create_app() -> FastAPI:
         _safe_project_dir(project_id)  # 404 early for unknown projects
 
         async def stream():
-            q = hub.subscribe()
+            q = hub.subscribe(project_id)
             try:
                 yield _sse({"type": "hello", "project_id": project_id})
                 while True:
                     if await request.is_disconnected():
                         return
                     try:
-                        changed = await asyncio.wait_for(q.get(), timeout=SSE_HEARTBEAT_SECONDS)
+                        await asyncio.wait_for(q.get(), timeout=SSE_HEARTBEAT_SECONDS)
                     except asyncio.TimeoutError:
                         yield _sse({"type": "heartbeat", "ts": time.time()})
                         continue
-                    if changed == project_id:
-                        # Coalesce bursts: drain anything queued for this project.
-                        while not q.empty():
-                            try:
-                                q.get_nowait()
-                            except asyncio.QueueEmpty:
-                                break
-                        yield _sse({"type": "change", "project_id": project_id})
+                    # Coalesce bursts: drain anything else queued.
+                    while not q.empty():
+                        try:
+                            q.get_nowait()
+                        except asyncio.QueueEmpty:
+                            break
+                    yield _sse({"type": "change", "project_id": project_id})
             finally:
                 hub.unsubscribe(q)
 
@@ -257,7 +273,9 @@ def create_app() -> FastAPI:
 
 
 def _safe_project_dir(project_id: str) -> Path:
-    if "/" in project_id or "\\" in project_id or project_id in (".", ".."):
+    # ':' rejects Windows drive-relative ids like "C:" (PROJECTS_DIR / "C:"
+    # collapses back to PROJECTS_DIR itself).
+    if any(c in project_id for c in "/\\:") or project_id in (".", ".."):
         raise HTTPException(status_code=400, detail="invalid project id")
     project_dir = PROJECTS_DIR / project_id
     if not project_dir.is_dir():
@@ -286,7 +304,10 @@ def _thumbnail_for(source: Path, width: int) -> Optional[Path]:
         if cached.is_file():
             return cached
         THUMB_CACHE_DIR.mkdir(parents=True, exist_ok=True)
-        tmp = cached.with_suffix(".tmp.jpg")
+        # Unique temp per request — concurrent misses for the same source
+        # must not write (and replace from) the same temp file.
+        import uuid
+        tmp = THUMB_CACHE_DIR / f"{key}.{uuid.uuid4().hex[:8]}.tmp.jpg"
         if is_video:
             import subprocess
             result = subprocess.run(

backlot/state.py

@@ -68,6 +68,7 @@ def _load_pipeline_meta(pipeline_type: Optional[str]) -> dict[str, Any]:
                     "gated": bool(s.get("human_approval_default", False)),
                 }
                 for s in manifest.get("stages", [])
+                if isinstance(s, dict) and s.get("name")
             ]
             if stages:
                 return {
@@ -260,9 +261,19 @@ def _resolve_asset_path(project_dir: Path, raw_path: str) -> Optional[Path]:
 
 
 def _asset_entry(project_dir: Path, asset: dict) -> dict:
-    """Normalize a manifest asset entry + resolve file existence."""
+    """Normalize a manifest asset entry + resolve file existence.
+
+    A file that resolves OUTSIDE the project directory is treated as
+    not-servable (exists=False): /media only serves within the project, and
+    a bare-filename fallback path would 404 or hit the wrong file.
+    """
     raw_path = asset.get("path") or ""
     resolved = _resolve_asset_path(project_dir, raw_path)
+    if resolved is not None:
+        try:
+            resolved.resolve().relative_to(Path(project_dir).resolve())
+        except (ValueError, OSError):
+            resolved = None
     file_path = resolved if resolved is not None else (project_dir / raw_path)
     exists = resolved is not None
     kind = asset.get("type") or ""
@@ -325,20 +336,26 @@ def _build_storyboard(
     sections = (artifacts.get("script") or {}).get("sections") or []
     manifest_assets = (artifacts.get("asset_manifest") or {}).get("assets") or []
 
+    def scene_key(value: Any) -> str:
+        # 0 is a legitimate scene id — only None/absent collapses to "".
+        return str(value) if value is not None else ""
+
     assets_by_scene: dict[str, list[dict]] = {}
     for asset in manifest_assets:
         if not isinstance(asset, dict):
             continue
         entry = _asset_entry(project_dir, asset)
-        key = str(entry.get("scene_id") or "")
-        assets_by_scene.setdefault(key, []).append(entry)
+        assets_by_scene.setdefault(scene_key(entry.get("scene_id")), []).append(entry)
 
-    # A scene is "generating" if its most recent event is an unfinished start.
+    # A scene is "generating" if its most recent top-level event is an
+    # unfinished start. Nested (depth>0) provider events inside a selector
+    # call are skipped — the outer call's finish is the real completion.
     generating: dict[str, dict] = {}
     for ev in events:
         sid = ev.get("scene_id")
-        if not sid:
+        if sid is None or ev.get("depth"):
             continue
+        sid = scene_key(sid)
         if ev.get("event") == "start":
             generating[sid] = ev
         elif ev.get("event") in ("finish", "error"):
@@ -348,7 +365,7 @@ def _build_storyboard(
     for scene in scene_plan["scenes"]:
         if not isinstance(scene, dict):
             continue
-        sid = str(scene.get("id") or "")
+        sid = scene_key(scene.get("id"))
         section = _find_script_section(scene, sections)
         scene_assets = assets_by_scene.get(sid, [])
         visuals = [a for a in scene_assets if a["type"] in ("image", "video", "diagram", "animation")]
@@ -362,7 +379,7 @@ def _build_storyboard(
             "start_seconds": scene.get("start_seconds"),
             "end_seconds": scene.get("end_seconds"),
             "duration_seconds": (
-                (scene.get("end_seconds") or 0) - (scene.get("start_seconds") or 0)
+                max(0, (scene.get("end_seconds") or 0) - (scene.get("start_seconds") or 0))
                 if scene.get("end_seconds") is not None and scene.get("start_seconds") is not None
                 else None
             ),

backlot/ui/board.js

@@ -252,8 +252,7 @@ function renderDecisions(s) {
   const body = el("div", { class: "panel-body" });
   for (const d of decisions.slice(-8).reverse()) {
     const alts = (d.options_considered || [])
-      .filter((o) => (o.option_id || o.label) !== d.selected && o.rejected_because !== undefined || (o.option_id !== d.selected && (o.option_id || o.label)))
-      .filter((o) => o.option_id !== d.selected);
+      .filter((o) => (o.option_id ?? o.label) !== d.selected && (o.option_id || o.label));
     body.append(el("div", { class: "decision" },
       el("div", { class: "d-cat" }, `${d.category || "decision"}${d.confidence ? ` · ${d.confidence}` : ""}`),
       el("div", { class: "d-pick" }, `${d.subject || ""} `, el("span", { class: "arrow" }, "→"), ` ${d.selected || ""}`),
@@ -273,19 +272,27 @@ function renderActivity(s) {
   const body = el("div", { class: "panel-body" });
   // A start is "running" only until a later finish/error for the same
   // tool+scene closes it — closed starts are dropped (the finish row tells
-  // the story), unmatched starts render as live.
-  const open = new Map();
+  // the story), unmatched starts render as live. Counted (not keyed-single)
+  // so parallel runs of the same tool on the same scene stay visible.
+  const open = new Map(); // key -> {count, ev}
   const rows = [];
   for (const ev of events) {
     const key = `${ev.tool}:${ev.scene_id || ""}`;
     if (ev.event === "start") {
-      open.set(key, ev);
+      const slot = open.get(key) || { count: 0, ev };
+      slot.count += 1;
+      slot.ev = ev;
+      open.set(key, slot);
     } else {
-      open.delete(key);
+      const slot = open.get(key);
+      if (slot) {
+        slot.count -= 1;
+        if (slot.count <= 0) open.delete(key);
+      }
       rows.push(ev);
     }
   }
-  rows.push(...open.values());
+  for (const slot of open.values()) rows.push(slot.ev);
   rows.sort((a, b) => String(a.ts).localeCompare(String(b.ts)));
   for (const ev of rows.slice(-10).reverse()) {
     let statusEl;
@@ -445,7 +452,18 @@ function renderRenders(s) {
   if (!renders.length) return null;
   if (activeRender >= renders.length) activeRender = 0;
   const current = renders[activeRender];
-  const video = el("video", { src: mediaURL(s.project_id, current.path), controls: "", preload: "none" });
+  // Full re-renders (every SSE refresh) must not reset an in-progress
+  // watch: carry playback position/state over to the recreated element.
+  const prev = document.querySelector(".render-hero video");
+  const src = mediaURL(s.project_id, current.path);
+  const video = el("video", { src, controls: "", preload: "none" });
+  if (prev && prev.getAttribute("src") === src && (prev.currentTime > 0 || !prev.paused)) {
+    const t = prev.currentTime;
+    const wasPlaying = !prev.paused && !prev.ended;
+    video.addEventListener("loadedmetadata", () => { video.currentTime = t; }, { once: true });
+    video.setAttribute("preload", "metadata");
+    if (wasPlaying) video.autoplay = true;
+  }
   const versions = el("div", { class: "render-meta" },
     renders.map((r, i) => el("span", {
       class: `v${i === activeRender ? " active" : ""}`,
@@ -498,7 +516,16 @@ function renderAwaitingNotice(s) {
 // replay — scrub a completed run from its timestamps
 // ---------------------------------------------------------------------------
 
-const ts = (iso) => { const t = Date.parse(iso); return Number.isFinite(t) ? t : null; };
+// Python writers emit tz-aware UTC isoformat, but treat tz-naive strings as
+// UTC too — mixing local-parsed and UTC-parsed timestamps would skew replay
+// ordering by the user's UTC offset.
+const ts = (iso) => {
+  if (!iso) return null;
+  let s = String(iso);
+  if (!/(Z|[+-]\d{2}:?\d{2})$/.test(s)) s += "Z";
+  const t = Date.parse(s);
+  return Number.isFinite(t) ? t : null;
+};
 
 function replayBounds(s) {
   const moments = [];
@@ -556,14 +583,17 @@ function stateAt(s, T) {
         card.generating_tool = (startedNow.get(card.id) || {}).tool;
       }
     }
-    const scriptStage = view.stages.find((x) => x.name === "script");
-    if (!(scriptStage && ["completed", "awaiting_human"].includes(scriptStage.status))) {
-      delete view.artifacts.script;
-    }
-    const composeStage = view.stages.find((x) => x.name === "compose");
-    if (!(composeStage && composeStage.status === "completed")) {
-      view.media.renders = [];
-    }
+  }
+  // Final artifacts hide until their stage happened — for every project
+  // shape, storyboard or not (a degraded run must not show the finished
+  // movie before its stages ran).
+  const scriptStage = view.stages.find((x) => x.name === "script");
+  if (!(scriptStage && ["completed", "awaiting_human"].includes(scriptStage.status))) {
+    delete view.artifacts.script;
+  }
+  const composeStage = view.stages.find((x) => x.name === "compose");
+  if (!(composeStage && composeStage.status === "completed")) {
+    view.media.renders = [];
   }
   return view;
 }
@@ -578,33 +608,42 @@ function renderReplayBar(s) {
       el("span", { class: "rp-btn", onclick: startReplay }, "▶ REPLAY RUN"));
   }
   const pos = (replay.t - replay.t0) / Math.max(1, replay.t1 - replay.t0);
+  const timeLabel = el("span", { class: "rp-time" },
+    new Date(replay.t).toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", second: "2-digit" }));
+  const setT = (value) => {
+    replay.t = replay.t0 + (Number(value) / 1000) * (replay.t1 - replay.t0);
+    timeLabel.textContent = new Date(replay.t)
+      .toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", second: "2-digit" });
+  };
   return el("div", { class: "replay-bar" },
     el("span", { class: "rp-btn", onclick: toggleReplayPlay }, replay.playing ? "❚❚" : "▶"),
     el("input", {
       type: "range", min: "0", max: "1000", value: String(Math.round(pos * 1000)),
-      oninput: (e) => {
-        replay.t = replay.t0 + (Number(e.target.value) / 1000) * (replay.t1 - replay.t0);
-        replay.playing = false;
-        render();
-      },
+      // A full render() would destroy this slider mid-drag: while dragging,
+      // only pause + track the time label; re-render the board on release.
+      onpointerdown: () => { replay.playing = false; },
+      oninput: (e) => setT(e.target.value),
+      onchange: (e) => { setT(e.target.value); render(); },
     }),
-    el("span", { class: "rp-time" },
-      new Date(replay.t).toLocaleTimeString([], { hour: "2-digit", minute: "2-digit", second: "2-digit" })),
+    timeLabel,
     el("span", { class: "rp-btn", onclick: stopReplay }, "✕ LIVE"),
   );
 }
 
+let replayTimer = null;
+
 function startReplay() {
   const bounds = replayBounds(state);
   if (!bounds) return;
   replay = { ...bounds, t: bounds.t0, playing: true };
   document.body.classList.add("replaying");
-  tickReplay();
+  scheduleTick();
   render();
 }
 
 function stopReplay() {
   replay = null;
+  clearTimeout(replayTimer);
   document.body.classList.remove("replaying");
   render();
 }
@@ -612,10 +651,16 @@ function stopReplay() {
 function toggleReplayPlay() {
   if (!replay) return;
   replay.playing = !replay.playing;
-  if (replay.playing) tickReplay();
+  if (replay.playing) scheduleTick();
   render();
 }
 
+function scheduleTick() {
+  // Single pending tick, ever — rapid pause/play must not stack chains.
+  clearTimeout(replayTimer);
+  replayTimer = setTimeout(tickReplay, 100);
+}
+
 function tickReplay() {
   if (!replay || !replay.playing) return;
   // A full run replays in ~20 seconds regardless of real duration
@@ -624,7 +669,7 @@ function tickReplay() {
   replay.t = Math.min(replay.t1, replay.t + step);
   if (replay.t >= replay.t1) replay.playing = false;
   render();
-  if (replay.playing) setTimeout(tickReplay, 100);
+  if (replay.playing) scheduleTick();
 }
 
 // ---------------------------------------------------------------------------

backlot/ui/lib.js

@@ -11,7 +11,6 @@ export function el(tag, attrs = {}, ...children) {
   for (const [k, v] of Object.entries(attrs)) {
     if (v == null) continue;
     if (k === "class") node.className = v;
-    else if (k === "html") node.innerHTML = v;
     else if (k.startsWith("on")) node.addEventListener(k.slice(2), v);
     else node.setAttribute(k, v);
   }
@@ -23,15 +22,17 @@ export function el(tag, attrs = {}, ...children) {
 }
 
 export function fmtDuration(seconds) {
-  if (seconds == null) return "";
-  const s = Math.round(seconds);
+  const n = Number(seconds);
+  if (seconds == null || !Number.isFinite(n)) return "";
+  const s = Math.max(0, Math.round(n));
   const m = Math.floor(s / 60);
   return `${m}:${String(s % 60).padStart(2, "0")}`;
 }
 
 export function fmtMoney(v) {
-  if (v == null) return "—";
-  return `$${Number(v).toFixed(2)}`;
+  const n = Number(v);
+  if (v == null || !Number.isFinite(n)) return "—";
+  return `$${n.toFixed(2)}`;
 }
 
 export function fmtAgo(epochSeconds) {