"""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 functools import lru_cache
from pathlib import Path
from typing import Any, Optional
from lib.events import read_events
from lib.paths import PROJECTS_DIR, REPO_ROOT
MEDIA_IMAGE_EXT = {".png", ".jpg", ".jpeg", ".webp", ".gif", ".svg"}
MEDIA_VIDEO_EXT = {".mp4", ".webm", ".mov"}
MEDIA_AUDIO_EXT = {".mp3", ".wav", ".m4a", ".ogg"}
SCAN_EXCLUDE = {"node_modules", ".git", "__pycache__", "history", ".cache"}
FALLBACK_STAGES = [
"research", "proposal", "idea", "script", "scene_plan",
"assets", "edit", "compose", "publish",
]
LIVE_WINDOW_SECONDS = 5 * 60
STALL_WINDOW_SECONDS = 10 * 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.resolve().relative_to(Path(project_dir).resolve()).as_posix()
except (ValueError, OSError):
return path.name
@lru_cache(maxsize=32)
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)),
"produces": [
str(name) for name in (s.get("produces") or [])
if isinstance(name, str) and name
],
}
for s in manifest.get("stages", [])
if isinstance(s, dict) and s.get("name")
]
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, "produces": []} 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.
Path references are only followed INSIDE the project directory: a
checkpoint must not be able to pull arbitrary JSON from elsewhere on
disk onto the board (F-04).
"""
if isinstance(value, dict):
return value
if isinstance(value, str) and value:
p = Path(value)
if not p.is_absolute():
p = project_dir / value
try:
p.resolve().relative_to(Path(project_dir).resolve())
except (ValueError, OSError):
return None
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"],
"produces": list(stage_def.get("produces") or []),
"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),
"history_entries": (
[{"status": v.get("status"), "timestamp": v.get("timestamp")} for v in versions]
+ ([{"status": cp.get("status"), "timestamp": cp.get("timestamp")}] if cp else [])
),
}
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)
canon = {name: i for i, name in enumerate(FALLBACK_STAGES)}
for name, cp in checkpoints.items():
if name in manifest_stage_names:
continue
entry = {
"name": name,
"gated": False,
"produces": [
str(artifact_name)
for artifact_name in (cp.get("artifacts") or {})
if isinstance(artifact_name, str) and artifact_name
],
"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,
}
pos = canon.get(name)
if pos is None:
rail.append(entry)
continue
insert_at = len(rail)
for i, existing in enumerate(rail):
existing_pos = canon.get(existing["name"])
if existing_pos is not None and existing_pos > pos:
insert_at = i
break
rail.insert(insert_at, entry)
return rail
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
if "decision_log" not in artifacts:
data = _read_json(project_dir / "decision_log.json")
if data is not None:
artifacts["decision_log"] = data
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
def _resolve_asset_path(project_dir: Path, raw_path: str) -> Optional[Path]:
"""Manifest paths appear in several real-world flavors — try them all.
Observed on disk: project-relative ("assets/images/x.png"),
repo-relative ("projects/<id>/assets/images/x.png"), and absolute.
"""
if not raw_path:
return None
p = Path(raw_path)
candidates = []
if p.is_absolute():
candidates.append(p)
else:
candidates.append(project_dir / raw_path)
candidates.append(REPO_ROOT / raw_path)
parts = p.parts
if len(parts) > 2 and parts[0] == "projects":
candidates.append(project_dir.parent / Path(*parts[1:]))
for c in candidates:
try:
if c.is_file():
return c
except OSError:
continue
return None
def _asset_entry(project_dir: Path, asset: dict) -> dict:
"""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 ""
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"
ext = file_path.suffix.lower()
renderable = exists and ext in (MEDIA_IMAGE_EXT | MEDIA_VIDEO_EXT)
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,
"renderable": renderable,
"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_scene_snapshot(project_dir: Path, scene_id: str) -> Optional[dict]:
"""A per-scene review still, if the run wrote one.
Atelier/animation scenes have no thumbnailable asset file, so the
assets-stage snapshot (`snapshots/<scene_id>.png`) is what the filmstrip
shows. Accept exact `<scene_id>.<ext>` and `<scene_id>_*.<ext>` forms.
"""
snap_dir = project_dir / "snapshots"
if not scene_id or not snap_dir.is_dir():
return None
try:
for f in sorted(snap_dir.iterdir()):
if not f.is_file() or f.suffix.lower() not in MEDIA_IMAGE_EXT:
continue
stem = f.stem
if stem == scene_id or stem.startswith(f"{scene_id}_"):
return {
"id": f"snap_{scene_id}",
"type": "image",
"scene_id": scene_id,
"path": _rel(project_dir, f),
"exists": True,
"renderable": True,
"snapshot": True,
}
except OSError:
return None
return None
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 []
def scene_key(value: Any) -> str:
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)
assets_by_scene.setdefault(scene_key(entry.get("scene_id")), []).append(entry)
generating: dict[str, dict] = {}
for ev in events:
sid = ev.get("scene_id")
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"):
generating.pop(sid, None)
cards = []
for scene in scene_plan["scenes"]:
if not isinstance(scene, dict):
continue
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")]
audio = [a for a in scene_assets if a["type"] in ("audio", "narration", "music", "sfx")]
renderable = [a for a in visuals if a.get("renderable")]
missing = [a for a in visuals if not a.get("exists") and a["type"] in ("image", "video", "diagram")]
active_visual = (
renderable[-1] if renderable
else missing[-1] if missing
else _find_scene_snapshot(project_dir, sid)
)
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": (
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
),
"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": renderable,
"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"),
}
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})
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 for the library card (image path, or a video path —
the /thumb endpoint extracts a frame from videos)."""
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"]
for rel_dir in ("assets/images", "assets/frames", "exports", "assets", "."):
d = (project_dir / rel_dir) if rel_dir != "." else project_dir
if not d.is_dir():
continue
try:
for f in sorted(d.iterdir()):
if f.is_file() and f.suffix.lower() in MEDIA_IMAGE_EXT:
return _rel(project_dir, f)
except OSError:
continue
renders = (state.get("media") or {}).get("renders", [])
if renders:
return renders[0]["path"]
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
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 = 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()
for stage_entry in stages:
if (
stage_entry["status"] == "in_progress"
and last_activity
and (now - last_activity) > STALL_WINDOW_SECONDS
):
stage_entry["stalled"] = True
stage_entry["stalled_minutes"] = int((now - last_activity) / 60)
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