backlot phase 1: board server (state derivation, watcher, SSE, media, CLI)

- backlot/state.py: BoardState from disk — stage rail with gate audit
  (gate_skipped detection from history/), scene_plan x script x
  asset_manifest storyboard join, takes, generating-state from events,
  media discovery incl. atelier root-render heuristic, degradation ladder,
  library summaries
- backlot/server.py: FastAPI on 4750 — /api/projects, /api/project/{id}/state,
  SSE change feeds (project + library) fed by a watchfiles watcher,
  /media with range support and traversal protection, UI mounts
- backlot/__main__.py: 'python -m backlot open [project]' idempotent
  launcher (spawns detached server, opens browser); 'serve' foreground
- verified against real projects: 73 listed, full state for
  signal-from-tomorrow, 206 range responses, SSE change push on
  filesystem write

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"