From cf3edafff08e427e29cb2d029c09b87c810d3788 Mon Sep 17 00:00:00 2001 From: calesthio Date: Thu, 2 Jul 2026 00:05:13 -0700 Subject: [PATCH] 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 | 71 ++++++++++++++++++++------------ backlot/state.py | 31 ++++++++++---- backlot/ui/board.js | 99 ++++++++++++++++++++++++++++++++------------- backlot/ui/lib.js | 11 ++--- 4 files changed, 148 insertions(+), 64 deletions(-) diff --git a/backlot/server.py b/backlot/server.py index d140cbb7..b957c2ac 100644 --- a/backlot/server.py +++ b/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( diff --git a/backlot/state.py b/backlot/state.py index a4dec019..d2790191 100644 --- a/backlot/state.py +++ b/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 ), diff --git a/backlot/ui/board.js b/backlot/ui/board.js index 0b9101c0..d12e9031 100644 --- a/backlot/ui/board.js +++ b/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(); } // --------------------------------------------------------------------------- diff --git a/backlot/ui/lib.js b/backlot/ui/lib.js index d99649b3..5a93d4e9 100644 --- a/backlot/ui/lib.js +++ b/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) {