diff --git a/backlot/__init__.py b/backlot/__init__.py new file mode 100644 index 00000000..1d0c582a --- /dev/null +++ b/backlot/__init__.py @@ -0,0 +1,16 @@ +"""Backlot — the living storyboard. + +A read-only, disk-derived production board for OpenMontage. A small local web +server watches ``projects/`` and renders each production's pipeline stages, +script, scene plan, generated assets, decisions, cost, and activity — live. + +Design contract (see internal/design/LIVING_STORYBOARD.md): +- Observation, not reporting: all state derives from files the pipeline + already writes. Agents never update the UI. +- Never block, never break: malformed or missing state degrades gracefully. +- The agent's only duty: ``python -m backlot open `` at pipeline init. +""" + +__version__ = "0.1.0" + +DEFAULT_PORT = 4750 diff --git a/backlot/__main__.py b/backlot/__main__.py new file mode 100644 index 00000000..6004ef3d --- /dev/null +++ b/backlot/__main__.py @@ -0,0 +1,109 @@ +"""Backlot CLI. + + python -m backlot open [project-id] # start server if needed, open browser + python -m backlot serve [--port N] # run the server in the foreground + +``open`` is idempotent and non-fatal by design: agents call it at pipeline +initialization and must continue the production even if it fails. +""" + +from __future__ import annotations + +import argparse +import os +import subprocess +import sys +import time +import urllib.request +import webbrowser + +from backlot import DEFAULT_PORT + + +def _port() -> int: + try: + return int(os.environ.get("BACKLOT_PORT", DEFAULT_PORT)) + except ValueError: + return DEFAULT_PORT + + +def _server_alive(port: int) -> bool: + try: + with urllib.request.urlopen(f"http://127.0.0.1:{port}/api/health", timeout=1.5) as resp: + return resp.status == 200 + except Exception: + return False + + +def _spawn_server(port: int) -> None: + """Start the server as a detached background process.""" + cmd = [sys.executable, "-m", "backlot", "serve", "--port", str(port)] + kwargs: dict = { + "stdout": subprocess.DEVNULL, + "stderr": subprocess.DEVNULL, + "stdin": subprocess.DEVNULL, + } + if os.name == "nt": + kwargs["creationflags"] = ( + subprocess.CREATE_NEW_PROCESS_GROUP | getattr(subprocess, "DETACHED_PROCESS", 0x00000008) + ) + else: + kwargs["start_new_session"] = True + subprocess.Popen(cmd, **kwargs) + + +def cmd_open(project_id: str | None) -> int: + port = _port() + if not _server_alive(port): + try: + _spawn_server(port) + except Exception as exc: + print(f"backlot: could not start server ({exc}) — continuing without the board") + return 1 + deadline = time.time() + 15 + while time.time() < deadline: + if _server_alive(port): + break + time.sleep(0.4) + else: + print("backlot: server did not come up in time — continuing without the board") + return 1 + url = f"http://127.0.0.1:{port}/" + if project_id: + url = f"http://127.0.0.1:{port}/p/{project_id}" + try: + webbrowser.open(url) + except Exception: + pass + print(f"backlot: {url}") + return 0 + + +def cmd_serve(port: int) -> int: + import uvicorn + + uvicorn.run("backlot.server:app", host="127.0.0.1", port=port, log_level="warning") + return 0 + + +def main(argv: list[str] | None = None) -> int: + parser = argparse.ArgumentParser(prog="backlot", description=__doc__) + sub = parser.add_subparsers(dest="command") + + p_open = sub.add_parser("open", help="open the board in the browser (starts server if needed)") + p_open.add_argument("project_id", nargs="?", default=None) + + p_serve = sub.add_parser("serve", help="run the Backlot server in the foreground") + p_serve.add_argument("--port", type=int, default=_port()) + + args = parser.parse_args(argv) + if args.command == "open": + return cmd_open(args.project_id) + if args.command == "serve": + return cmd_serve(args.port) + parser.print_help() + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/backlot/server.py b/backlot/server.py new file mode 100644 index 00000000..e24e5b91 --- /dev/null +++ b/backlot/server.py @@ -0,0 +1,217 @@ +"""Backlot server — FastAPI app: board state API, SSE change feed, media. + +The watcher observes ``projects/`` with watchfiles; on any change it bumps a +per-project version and wakes SSE subscribers, who tell the browser to +refetch state. The server never writes to project directories. +""" + +from __future__ import annotations + +import asyncio +import json +import time +from pathlib import Path +from typing import Optional + +from fastapi import FastAPI, HTTPException, Request +from fastapi.responses import FileResponse, StreamingResponse +from fastapi.staticfiles import StaticFiles + +from backlot.state import PROJECTS_DIR, REPO_ROOT, list_projects, load_board_state + +UI_DIR = Path(__file__).resolve().parent / "ui" + +# Paths inside a project whose changes are pure noise for the board. +_IGNORE_PARTS = {"node_modules", ".git", "__pycache__", ".cache"} + +SSE_HEARTBEAT_SECONDS = 15 + + +class ChangeHub: + """Fan-out of project-change notifications to SSE subscribers.""" + + def __init__(self) -> None: + self._subscribers: set[asyncio.Queue] = set() + + def subscribe(self) -> asyncio.Queue: + q: asyncio.Queue = asyncio.Queue(maxsize=64) + self._subscribers.add(q) + return q + + def unsubscribe(self, q: asyncio.Queue) -> None: + self._subscribers.discard(q) + + def publish(self, project_id: str) -> None: + for q in list(self._subscribers): + try: + q.put_nowait(project_id) + except asyncio.QueueFull: + pass # subscriber is behind; it will refetch on next event + + +hub = ChangeHub() + + +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): + return None + if not rel.parts: + return None + if _IGNORE_PARTS.intersection(rel.parts): + return None + return rel.parts[0] + + +async def _watch_projects() -> None: + """Background task: watch projects/ and publish debounced changes.""" + try: + from watchfiles import awatch + except ImportError: + return # watcher unavailable → board still works via manual refresh + if not PROJECTS_DIR.is_dir(): + return + async for changes in awatch(PROJECTS_DIR, recursive=True, step=400): + touched: set[str] = set() + for _change, path_str in changes: + pid = _project_of_change(path_str) + if pid: + touched.add(pid) + for pid in touched: + hub.publish(pid) + + +def create_app() -> FastAPI: + app = FastAPI(title="Backlot", docs_url=None, redoc_url=None) + + @app.on_event("startup") + async def _startup() -> None: + app.state.watch_task = asyncio.create_task(_watch_projects()) + + @app.on_event("shutdown") + async def _shutdown() -> None: + task = getattr(app.state, "watch_task", None) + if task: + task.cancel() + + # ---- API ---------------------------------------------------------- + + @app.get("/api/health") + async def health() -> dict: + return {"ok": True, "app": "backlot"} + + @app.get("/api/projects") + async def projects() -> list: + return await asyncio.to_thread(list_projects) + + @app.get("/api/project/{project_id}/state") + async def project_state(project_id: str) -> dict: + project_dir = _safe_project_dir(project_id) + return await asyncio.to_thread(load_board_state, project_dir) + + @app.get("/api/project/{project_id}/events") + async def project_events(project_id: str, request: Request) -> StreamingResponse: + _safe_project_dir(project_id) # 404 early for unknown projects + + async def stream(): + q = hub.subscribe() + 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) + 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}) + finally: + hub.unsubscribe(q) + + return StreamingResponse(stream(), media_type="text/event-stream", headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }) + + @app.get("/api/library/events") + async def library_events(request: Request) -> StreamingResponse: + async def stream(): + q = hub.subscribe() + try: + yield _sse({"type": "hello"}) + while True: + if await request.is_disconnected(): + return + try: + changed = await asyncio.wait_for(q.get(), timeout=SSE_HEARTBEAT_SECONDS) + except asyncio.TimeoutError: + yield _sse({"type": "heartbeat", "ts": time.time()}) + continue + while not q.empty(): + try: + q.get_nowait() + except asyncio.QueueEmpty: + break + yield _sse({"type": "change", "project_id": changed}) + finally: + hub.unsubscribe(q) + + return StreamingResponse(stream(), media_type="text/event-stream", headers={ + "Cache-Control": "no-cache", + "X-Accel-Buffering": "no", + }) + + # ---- Media (range requests handled by FileResponse) --------------- + + @app.get("/media/{project_id}/{file_path:path}") + async def media(project_id: str, file_path: str) -> FileResponse: + project_dir = _safe_project_dir(project_id) + target = (project_dir / file_path).resolve() + try: + target.relative_to(project_dir.resolve()) + except ValueError: + raise HTTPException(status_code=403, detail="path escapes project") + if not target.is_file(): + raise HTTPException(status_code=404, detail="media not found") + return FileResponse(target) + + # ---- UI ------------------------------------------------------------ + + @app.get("/p/{project_id}") + async def board_page(project_id: str) -> FileResponse: + return FileResponse(UI_DIR / "board.html") + + @app.get("/") + async def library_page() -> FileResponse: + return FileResponse(UI_DIR / "index.html") + + if UI_DIR.is_dir(): + app.mount("/ui", StaticFiles(directory=UI_DIR), name="ui") + + return app + + +def _safe_project_dir(project_id: str) -> Path: + if "/" in project_id or "\\" in project_id 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(): + raise HTTPException(status_code=404, detail=f"unknown project: {project_id}") + return project_dir + + +def _sse(payload: dict) -> str: + return f"data: {json.dumps(payload)}\n\n" + + +app = create_app() diff --git a/backlot/state.py b/backlot/state.py new file mode 100644 index 00000000..3c93dcb0 --- /dev/null +++ b/backlot/state.py @@ -0,0 +1,559 @@ +"""BoardState derivation — turn a project directory into renderable state. + +Everything here is read-only and defensive: a malformed JSON file, a missing +artifact, or a half-written checkpoint must degrade the board, never crash it +(design principle: "never block, never break"). +""" + +from __future__ import annotations + +import json +import re +from pathlib import Path +from typing import Any, Optional + +from lib.events import read_events + +REPO_ROOT = Path(__file__).resolve().parent.parent +PROJECTS_DIR = REPO_ROOT / "projects" + +MEDIA_IMAGE_EXT = {".png", ".jpg", ".jpeg", ".webp", ".gif"} +MEDIA_VIDEO_EXT = {".mp4", ".webm", ".mov"} +MEDIA_AUDIO_EXT = {".mp3", ".wav", ".m4a", ".ogg"} + +# Directories inside a project we never scan for media (build noise). +SCAN_EXCLUDE = {"node_modules", ".git", "__pycache__", "history", ".cache"} + +# Stages every pipeline shares (fallback rail when the manifest is unknown). +FALLBACK_STAGES = [ + "research", "proposal", "idea", "script", "scene_plan", + "assets", "edit", "compose", "publish", +] + +# How long (seconds) without filesystem activity before a board reads "idle". +LIVE_WINDOW_SECONDS = 5 * 60 + + +def _read_json(path: Path) -> Optional[dict]: + """Read a JSON file, returning None on any failure.""" + try: + with open(path, encoding="utf-8", errors="replace") as f: + data = json.load(f) + return data if isinstance(data, dict) else None + except (OSError, json.JSONDecodeError, UnicodeError): + return None + + +def _rel(project_dir: Path, path: Path) -> str: + """Project-relative POSIX path for media URLs.""" + try: + return path.relative_to(project_dir).as_posix() + except ValueError: + return path.name + + +# --------------------------------------------------------------------------- +# Pipeline / stages +# --------------------------------------------------------------------------- + +def _load_pipeline_meta(pipeline_type: Optional[str]) -> dict[str, Any]: + """Stage order + gate flags from the manifest; graceful fallback.""" + if pipeline_type and pipeline_type != "unknown": + try: + from lib.pipeline_loader import load_pipeline + manifest = load_pipeline(pipeline_type) + stages = [ + { + "name": s["name"], + "gated": bool(s.get("human_approval_default", False)), + } + for s in manifest.get("stages", []) + ] + if stages: + return { + "pipeline_type": pipeline_type, + "stages": stages, + "known": True, + } + except Exception: + pass + return { + "pipeline_type": pipeline_type or "unknown", + "stages": [{"name": s, "gated": False} for s in FALLBACK_STAGES], + "known": False, + } + + +def _resolve_artifact(project_dir: Path, value: Any) -> Optional[dict]: + """Checkpoint artifacts may be inline dicts or path strings — resolve both.""" + if isinstance(value, dict): + return value + if isinstance(value, str) and value: + p = Path(value) + if not p.is_absolute(): + p = project_dir / value + return _read_json(p) + return None + + +def _collect_checkpoints(project_dir: Path) -> dict[str, dict]: + """Current checkpoint per stage (raw dicts, unvalidated by design).""" + out: dict[str, dict] = {} + for path in sorted(project_dir.glob("checkpoint_*.json")): + stage = path.stem[len("checkpoint_"):] + data = _read_json(path) + if data is not None: + data["_mtime"] = path.stat().st_mtime + out[stage] = data + return out + + +def _collect_history(project_dir: Path) -> dict[str, list[dict]]: + """Archived checkpoint versions per stage (oldest first).""" + history_dir = project_dir / "history" + out: dict[str, list[dict]] = {} + if not history_dir.is_dir(): + return out + for path in sorted(history_dir.glob("checkpoint_*.json")): + m = re.match(r"checkpoint_(.+?)_\d", path.stem) + stage = m.group(1) if m else path.stem[len("checkpoint_"):] + data = _read_json(path) + if data is not None: + out.setdefault(stage, []).append(data) + return out + + +def _build_stage_rail( + pipeline_meta: dict, + checkpoints: dict[str, dict], + history: dict[str, list[dict]], +) -> list[dict]: + """One entry per manifest stage with derived status + gate audit.""" + rail = [] + manifest_stage_names = {s["name"] for s in pipeline_meta["stages"]} + for stage_def in pipeline_meta["stages"]: + name = stage_def["name"] + cp = checkpoints.get(name) + versions = history.get(name, []) + status = cp.get("status") if cp else "pending" + entry: dict[str, Any] = { + "name": name, + "gated": stage_def["gated"], + "status": status or "pending", + "timestamp": cp.get("timestamp") if cp else None, + "review": cp.get("review") if cp else None, + "cost_snapshot": cp.get("cost_snapshot") if cp else None, + "error": cp.get("error") if cp else None, + "human_approved": cp.get("human_approved") if cp else None, + "partial_progress": (cp.get("metadata") or {}).get("partial_progress") if cp else None, + "versions": len(versions) + (1 if cp else 0), + } + # Gate audit: a gated stage that completed without ever passing + # through awaiting_human (current or archived) was gate-skipped. + if ( + stage_def["gated"] + and cp is not None + and cp.get("status") == "completed" + ): + saw_wait = any(v.get("status") == "awaiting_human" for v in versions) + approved = bool(cp.get("human_approved")) + entry["gate_skipped"] = not (saw_wait or approved) + rail.append(entry) + + # Checkpoints for stages the manifest doesn't declare (legacy runs, + # pipeline mismatch) still deserve a slot at the end. + for name, cp in checkpoints.items(): + if name not in manifest_stage_names: + rail.append({ + "name": name, + "gated": False, + "status": cp.get("status") or "unknown", + "timestamp": cp.get("timestamp"), + "review": cp.get("review"), + "cost_snapshot": cp.get("cost_snapshot"), + "error": cp.get("error"), + "human_approved": cp.get("human_approved"), + "partial_progress": None, + "versions": 1 + len(history.get(name, [])), + "undeclared": True, + }) + return rail + + +# --------------------------------------------------------------------------- +# Artifacts +# --------------------------------------------------------------------------- + +ARTIFACT_FILES = { + "research_brief": "research_brief.json", + "brief": "brief.json", + "proposal_packet": "proposal_packet.json", + "script": "script.json", + "scene_plan": "scene_plan.json", + "asset_manifest": "asset_manifest.json", + "edit_decisions": "edit_decisions.json", + "render_report": "render_report.json", + "final_review": "final_review.json", + "publish_log": "publish_log.json", + "decision_log": "decision_log.json", +} + + +def _collect_artifacts(project_dir: Path, checkpoints: dict[str, dict]) -> dict[str, dict]: + """Artifacts from artifacts/*.json, backfilled from checkpoint payloads.""" + artifacts: dict[str, dict] = {} + art_dir = project_dir / "artifacts" + for name, filename in ARTIFACT_FILES.items(): + data = _read_json(art_dir / filename) + if data is not None: + artifacts[name] = data + # decision_log historically also lives at project root + if "decision_log" not in artifacts: + data = _read_json(project_dir / "decision_log.json") + if data is not None: + artifacts["decision_log"] = data + # Backfill from checkpoint-embedded artifacts. + for cp in checkpoints.values(): + for name, value in (cp.get("artifacts") or {}).items(): + if name not in artifacts: + resolved = _resolve_artifact(project_dir, value) + if resolved is not None: + artifacts[name] = resolved + return artifacts + + +# --------------------------------------------------------------------------- +# Storyboard join +# --------------------------------------------------------------------------- + +def _asset_entry(project_dir: Path, asset: dict) -> dict: + """Normalize a manifest asset entry + resolve file existence.""" + raw_path = asset.get("path") or "" + file_path = Path(raw_path) + if not file_path.is_absolute(): + file_path = project_dir / raw_path + exists = file_path.is_file() + kind = asset.get("type") or "" + if not kind and file_path.suffix: + ext = file_path.suffix.lower() + if ext in MEDIA_IMAGE_EXT: + kind = "image" + elif ext in MEDIA_VIDEO_EXT: + kind = "video" + elif ext in MEDIA_AUDIO_EXT: + kind = "audio" + return { + "id": asset.get("id"), + "type": kind, + "scene_id": asset.get("scene_id"), + "path": _rel(project_dir, file_path) if exists else raw_path, + "exists": exists, + "prompt": asset.get("prompt"), + "model": asset.get("model"), + "source_tool": asset.get("source_tool"), + "provider": asset.get("provider"), + "cost_usd": asset.get("cost_usd"), + "quality_score": asset.get("quality_score"), + "duration_seconds": asset.get("duration_seconds"), + "resolution": asset.get("resolution"), + } + + +def _find_script_section(scene: dict, sections: list[dict]) -> Optional[dict]: + """Join scene → script section by id, falling back to timing overlap.""" + sid = scene.get("script_section_id") + if sid: + for s in sections: + if s.get("id") == sid: + return s + start = scene.get("start_seconds") + end = scene.get("end_seconds") + if start is None or end is None: + return None + best, best_overlap = None, 0.0 + for s in sections: + s0, s1 = s.get("start_seconds"), s.get("end_seconds") + if s0 is None or s1 is None: + continue + overlap = min(end, s1) - max(start, s0) + if overlap > best_overlap: + best, best_overlap = s, overlap + return best + + +def _build_storyboard( + project_dir: Path, + artifacts: dict[str, dict], + events: list[dict], +) -> Optional[dict]: + """Scene cards: scene_plan × script × asset_manifest (+ live events).""" + scene_plan = artifacts.get("scene_plan") + if not scene_plan or not isinstance(scene_plan.get("scenes"), list): + return None + sections = (artifacts.get("script") or {}).get("sections") or [] + manifest_assets = (artifacts.get("asset_manifest") or {}).get("assets") or [] + + 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) + + # A scene is "generating" if its most recent event is an unfinished start. + generating: dict[str, dict] = {} + for ev in events: + sid = ev.get("scene_id") + if not sid: + continue + if ev.get("event") == "start": + generating[sid] = ev + elif ev.get("event") in ("finish", "error"): + generating.pop(sid, None) + + cards = [] + for scene in scene_plan["scenes"]: + if not isinstance(scene, dict): + continue + sid = str(scene.get("id") or "") + 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")] + audio = [a for a in scene_assets if a["type"] in ("audio", "narration", "music", "sfx")] + # Takes: multiple visual assets for the same slot, ordered as listed. + active_visual = visuals[-1] if visuals else None + cards.append({ + "id": sid, + "type": scene.get("type"), + "description": scene.get("description"), + "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) + if scene.get("end_seconds") is not None and scene.get("start_seconds") is not None + else None + ), + "hero_moment": bool(scene.get("hero_moment")), + "shot_language": scene.get("shot_language"), + "shot_intent": scene.get("shot_intent"), + "framing": scene.get("framing"), + "movement": scene.get("movement"), + "narration": (section or {}).get("text"), + "section_label": (section or {}).get("label"), + "required_assets": scene.get("required_assets") or [], + "visual": active_visual, + "takes": visuals, + "audio": audio, + "generating": generating.get(sid) is not None, + "generating_tool": (generating.get(sid) or {}).get("tool"), + }) + + total = scene_plan.get("metadata", {}).get("total_duration_seconds") + if total is None and cards: + ends = [c["end_seconds"] for c in cards if c["end_seconds"] is not None] + total = max(ends) if ends else None + return { + "scenes": cards, + "total_duration_seconds": total, + "style_playbook": scene_plan.get("style_playbook"), + } + + +# --------------------------------------------------------------------------- +# Media discovery +# --------------------------------------------------------------------------- + +def _scan_media(project_dir: Path) -> dict[str, list[dict]]: + """Discovered media files (renders, loose assets, snapshots).""" + renders: list[dict] = [] + snapshots: list[dict] = [] + music: list[dict] = [] + + renders_dir = project_dir / "renders" + if renders_dir.is_dir(): + for f in sorted(renders_dir.iterdir()): + if f.suffix.lower() in MEDIA_VIDEO_EXT and f.is_file(): + renders.append({"path": _rel(project_dir, f), "size": f.stat().st_size, + "mtime": f.stat().st_mtime}) + # Atelier heuristic: deliverables at project root. + for f in sorted(project_dir.glob("*.mp4")): + renders.append({"path": _rel(project_dir, f), "size": f.stat().st_size, + "mtime": f.stat().st_mtime, "at_root": True}) + for f in sorted(project_dir.glob("*.mp3")): + music.append({"path": _rel(project_dir, f), "at_root": True}) + music_dir = project_dir / "assets" / "music" + if music_dir.is_dir(): + for f in sorted(music_dir.iterdir()): + if f.suffix.lower() in MEDIA_AUDIO_EXT: + music.append({"path": _rel(project_dir, f)}) + + for dirname in ("snapshots", "verify"): + d = project_dir / dirname + if d.is_dir(): + for f in sorted(d.iterdir()): + if f.suffix.lower() in MEDIA_IMAGE_EXT and f.is_file(): + snapshots.append({"path": _rel(project_dir, f)}) + + renders.sort(key=lambda r: r.get("mtime", 0), reverse=True) + return {"renders": renders, "snapshots": snapshots, "music": music} + + +def _find_poster(project_dir: Path, state: dict) -> Optional[str]: + """Best poster image for the library card.""" + board = state.get("storyboard") or {} + for card in board.get("scenes", []): + visual = card.get("visual") + if visual and visual.get("exists") and visual.get("type") == "image": + return visual["path"] + for snap in (state.get("media") or {}).get("snapshots", []): + return snap["path"] + images_dir = project_dir / "assets" / "images" + if images_dir.is_dir(): + for f in sorted(images_dir.iterdir()): + if f.suffix.lower() in MEDIA_IMAGE_EXT: + return _rel(project_dir, f) + return None + + +def _last_activity(project_dir: Path) -> float: + """Most recent mtime among state-bearing files (bounded scan).""" + latest = 0.0 + try: + candidates = list(project_dir.glob("checkpoint_*.json")) + candidates.append(project_dir / "events.jsonl") + art = project_dir / "artifacts" + if art.is_dir(): + candidates.extend(art.glob("*.json")) + for p in candidates: + try: + latest = max(latest, p.stat().st_mtime) + except OSError: + continue + except OSError: + pass + return latest + + +# --------------------------------------------------------------------------- +# Public API +# --------------------------------------------------------------------------- + +def load_board_state(project_dir: Path) -> dict[str, Any]: + """Full BoardState for one project. Never raises.""" + project_dir = Path(project_dir) + project_id = project_dir.name + + marker = _read_json(project_dir / "project.json") or {} + meta_json = _read_json(project_dir / "meta.json") or {} + + checkpoints = _collect_checkpoints(project_dir) + history = _collect_history(project_dir) + + pipeline_type = marker.get("pipeline_type") + if not pipeline_type: + for cp in checkpoints.values(): + pt = cp.get("pipeline_type") + if pt and pt != "unknown": + pipeline_type = pt + break + pipeline_meta = _load_pipeline_meta(pipeline_type) + + artifacts = _collect_artifacts(project_dir, checkpoints) + events = read_events(project_dir, limit=250) + storyboard = _build_storyboard(project_dir, artifacts, events) + media = _scan_media(project_dir) + + stages = _build_stage_rail(pipeline_meta, checkpoints, history) + + # Cost: latest checkpoint snapshot wins; fall back to manifest total. + cost = None + for cp in sorted(checkpoints.values(), key=lambda c: c.get("_mtime", 0), reverse=True): + if cp.get("cost_snapshot"): + cost = cp["cost_snapshot"] + break + if cost is None: + total = (artifacts.get("asset_manifest") or {}).get("total_cost_usd") + if total is not None: + cost = {"total_spent_usd": total} + + import time + last_activity = _last_activity(project_dir) + now = time.time() + + state: dict[str, Any] = { + "project_id": project_id, + "title": marker.get("title") or meta_json.get("name") or project_id.replace("-", " ").title(), + "pipeline": pipeline_meta, + "style_playbook": marker.get("style_playbook"), + "created_at": marker.get("created_at"), + "has_marker": bool(marker), + "has_pipeline_state": bool(checkpoints), + "stages": stages, + "artifacts": artifacts, + "storyboard": storyboard, + "media": media, + "events": events, + "cost": cost, + "last_activity": last_activity, + "live": bool(last_activity and (now - last_activity) < LIVE_WINDOW_SECONDS), + } + state["poster"] = _find_poster(project_dir, state) + return state + + +def summarize_project(project_dir: Path) -> dict[str, Any]: + """Cheap library-card summary (no full artifact parse of big files).""" + state = load_board_state(project_dir) + active = next((s for s in state["stages"] if s["status"] in ("in_progress", "awaiting_human")), None) + done = [s for s in state["stages"] if s["status"] == "completed"] + return { + "project_id": state["project_id"], + "title": state["title"], + "pipeline_type": state["pipeline"]["pipeline_type"], + "has_pipeline_state": state["has_pipeline_state"], + "poster": state["poster"], + "live": state["live"], + "last_activity": state["last_activity"], + "active_stage": active["name"] if active else None, + "awaiting_human": bool(active and active["status"] == "awaiting_human"), + "stage_states": [ + {"name": s["name"], "status": s["status"]} + for s in state["stages"] if not s.get("undeclared") + ], + "completed_count": len(done), + "render_count": len(state["media"]["renders"]), + "scene_count": len((state["storyboard"] or {}).get("scenes", [])), + } + + +def list_projects(projects_dir: Optional[Path] = None) -> list[dict[str, Any]]: + """Library view: every project directory, live-first then recency.""" + root = Path(projects_dir) if projects_dir else PROJECTS_DIR + if not root.is_dir(): + return [] + summaries = [] + for entry in sorted(root.iterdir()): + if not entry.is_dir() or entry.name.startswith(("_", ".")): + continue + try: + summaries.append(summarize_project(entry)) + except Exception: + summaries.append({ + "project_id": entry.name, + "title": entry.name.replace("-", " ").title(), + "pipeline_type": "unknown", + "has_pipeline_state": False, + "poster": None, + "live": False, + "last_activity": 0, + "active_stage": None, + "awaiting_human": False, + "stage_states": [], + "completed_count": 0, + "render_count": 0, + "scene_count": 0, + "error": "unreadable", + }) + summaries.sort(key=lambda s: (not s["live"], -(s["last_activity"] or 0))) + return summaries diff --git a/requirements.txt b/requirements.txt index b0ac36ad..e7f2b802 100644 --- a/requirements.txt +++ b/requirements.txt @@ -7,3 +7,8 @@ Pillow>=10.0 numpy>=1.24 requests>=2.31 google-auth>=2.0 # service-account auth for Google TTS + Imagen (Vertex AI) + +# Backlot — the living storyboard (local board server) +fastapi>=0.110 +uvicorn>=0.29 +watchfiles>=0.21 diff --git a/tests/backlot/test_state.py b/tests/backlot/test_state.py new file mode 100644 index 00000000..a5382cf9 --- /dev/null +++ b/tests/backlot/test_state.py @@ -0,0 +1,190 @@ +"""Unit tests for Backlot BoardState derivation (backlot/state.py).""" + +import json +import time +from pathlib import Path + +import pytest + +from backlot import state as state_mod +from backlot.state import list_projects, load_board_state, summarize_project + + +@pytest.fixture +def projects_root(tmp_path, monkeypatch): + root = tmp_path / "projects" + root.mkdir() + monkeypatch.setattr(state_mod, "PROJECTS_DIR", root) + return root + + +def _make_project(root: Path, pid: str) -> Path: + p = root / pid + (p / "artifacts").mkdir(parents=True) + (p / "assets" / "images").mkdir(parents=True) + (p / "renders").mkdir() + return p + + +def _write(p: Path, data: dict) -> None: + p.parent.mkdir(parents=True, exist_ok=True) + p.write_text(json.dumps(data), encoding="utf-8") + + +SCENE_PLAN = { + "version": "1.0", + "scenes": [ + {"id": "sc1", "type": "generated", "description": "opening", + "start_seconds": 0, "end_seconds": 4, "script_section_id": "s1", + "hero_moment": False}, + {"id": "sc2", "type": "generated", "description": "climax", + "start_seconds": 4, "end_seconds": 10, "hero_moment": True}, + ], +} + +SCRIPT = { + "version": "1.0", "title": "Test Film", "total_duration_seconds": 10, + "sections": [ + {"id": "s1", "text": "It begins.", "start_seconds": 0, "end_seconds": 4}, + {"id": "s2", "text": "It ends.", "start_seconds": 4, "end_seconds": 10}, + ], +} + + +class TestBoardState: + def test_full_project(self, projects_root): + p = _make_project(projects_root, "film") + _write(p / "project.json", {"project_id": "film", "title": "My Film", + "pipeline_type": "cinematic", "created_at": "2026-01-01T00:00:00Z"}) + _write(p / "artifacts" / "scene_plan.json", SCENE_PLAN) + _write(p / "artifacts" / "script.json", SCRIPT) + img = p / "assets" / "images" / "sc1.png" + img.write_bytes(b"fake") + _write(p / "artifacts" / "asset_manifest.json", { + "version": "1.0", + "assets": [ + {"id": "a1", "type": "image", "path": "assets/images/sc1.png", + "scene_id": "sc1", "source_tool": "t", "cost_usd": 0.1}, + {"id": "a2", "type": "image", "path": "assets/images/missing.png", + "scene_id": "sc2", "source_tool": "t"}, + ], + "total_cost_usd": 0.1, + }) + _write(p / "checkpoint_script.json", { + "version": "1.0", "project_id": "film", "pipeline_type": "cinematic", + "stage": "script", "status": "completed", "timestamp": "2026-01-01T01:00:00Z", + "human_approved": True, "artifacts": {}, + }) + + s = load_board_state(p) + assert s["title"] == "My Film" + assert s["pipeline"]["pipeline_type"] == "cinematic" + assert s["pipeline"]["known"] is True + board = s["storyboard"] + assert len(board["scenes"]) == 2 + sc1, sc2 = board["scenes"] + assert sc1["narration"] == "It begins." + assert sc1["visual"]["exists"] is True + # sc2 has no script_section_id -> joined by timing overlap + assert sc2["narration"] == "It ends." + assert sc2["hero_moment"] is True + assert sc2["visual"]["exists"] is False # missing file flagged + script_stage = next(x for x in s["stages"] if x["name"] == "script") + assert script_stage["status"] == "completed" + + def test_gate_skip_detection(self, projects_root): + p = _make_project(projects_root, "sneaky") + # completed on a gated stage with no awaiting_human history and no + # human_approved -> gate_skipped flag + _write(p / "checkpoint_script.json", { + "version": "1.0", "project_id": "sneaky", "pipeline_type": "cinematic", + "stage": "script", "status": "completed", + "timestamp": "2026-01-01T01:00:00Z", "artifacts": {}, + }) + s = load_board_state(p) + script_stage = next(x for x in s["stages"] if x["name"] == "script") + assert script_stage["gate_skipped"] is True + + # with an archived awaiting_human version, the gate was honored + _write(p / "history" / "checkpoint_script_20260101.json", { + "stage": "script", "status": "awaiting_human", + }) + s2 = load_board_state(p) + script_stage2 = next(x for x in s2["stages"] if x["name"] == "script") + assert script_stage2["gate_skipped"] is False + + def test_generating_state_from_events(self, projects_root): + p = _make_project(projects_root, "live") + _write(p / "artifacts" / "scene_plan.json", SCENE_PLAN) + events = [ + {"ts": "t1", "tool": "img", "event": "start", "scene_id": "sc1"}, + {"ts": "t2", "tool": "img", "event": "finish", "scene_id": "sc1"}, + {"ts": "t3", "tool": "img", "event": "start", "scene_id": "sc2"}, + ] + (p / "events.jsonl").write_text( + "\n".join(json.dumps(e) for e in events) + "\n", encoding="utf-8") + s = load_board_state(p) + cards = {c["id"]: c for c in s["storyboard"]["scenes"]} + assert cards["sc1"]["generating"] is False + assert cards["sc2"]["generating"] is True + assert cards["sc2"]["generating_tool"] == "img" + + def test_degraded_project_never_crashes(self, projects_root): + p = projects_root / "bare" + p.mkdir() + (p / "something.mp4").write_bytes(b"x") + (p / "artifacts").mkdir() + (p / "artifacts" / "script.json").write_text("NOT JSON", encoding="utf-8") + s = load_board_state(p) + assert s["has_pipeline_state"] is False + assert s["storyboard"] is None + assert s["media"]["renders"][0]["path"] == "something.mp4" + assert s["media"]["renders"][0]["at_root"] is True + + def test_undeclared_stage_surfaces(self, projects_root): + p = _make_project(projects_root, "legacy") + _write(p / "checkpoint_idea.json", { + "version": "1.0", "project_id": "legacy", "pipeline_type": "cinematic", + "stage": "idea", "status": "completed", + "timestamp": "2026-01-01T01:00:00Z", "artifacts": {}, + }) + s = load_board_state(p) + idea = next(x for x in s["stages"] if x["name"] == "idea") + assert idea.get("undeclared") is True + + +class TestLibrary: + def test_list_projects_sorts_live_first(self, projects_root): + old = _make_project(projects_root, "old-film") + _write(old / "checkpoint_script.json", {"stage": "script", "status": "completed"}) + # backdate everything in old-film + import os + past = time.time() - 60 * 60 * 24 * 30 + for f in old.rglob("*"): + if f.is_file(): + os.utime(f, (past, past)) + + fresh = _make_project(projects_root, "fresh-film") + _write(fresh / "checkpoint_script.json", {"stage": "script", "status": "in_progress"}) + + projects = list_projects(projects_root) + assert [p["project_id"] for p in projects][0] == "fresh-film" + assert projects[0]["live"] is True + assert projects[1]["live"] is False + + def test_underscore_dirs_skipped(self, projects_root): + (projects_root / "_analysis").mkdir() + _make_project(projects_root, "real") + ids = [p["project_id"] for p in list_projects(projects_root)] + assert ids == ["real"] + + def test_summary_shape(self, projects_root): + p = _make_project(projects_root, "sum") + _write(p / "project.json", {"title": "Sum", "pipeline_type": "cinematic"}) + _write(p / "checkpoint_script.json", { + "stage": "script", "status": "awaiting_human", + "timestamp": "2026-01-01T01:00:00Z", "artifacts": {}, + }) + summary = summarize_project(p) + assert summary["awaiting_human"] is True + assert summary["active_stage"] == "script"