""" manifest_writer.py — Per-shot manifest (camera report) writer and reader. Writes a JSON sidecar alongside each generated output recording everything about the generation: refs used, weights, ordering, prompt, model, seed, cost. Manifest lifecycle: preview -> starsend preview compiles the manifest (zero cost) generated -> starsend generate fills in output path, seed, cost, timing rejected -> filmmaker marks output as rejected in Dailies/Inspector The manifest is the digital equivalent of a film set's camera report. """ import json import logging from datetime import datetime, timezone from pathlib import Path from typing import Optional from recoil.pipeline.core.cost import read_cost_from_record_safe logger = logging.getLogger(__name__) MANIFEST_VERSION = "1.0" MANIFEST_WRITER_VALID_STATUSES = ("preview", "generated", "rejected") def build_manifest( shot_id: str, episode: str, prompt: str, model: str, refs: list[dict], *, negative_prompt: str = "", dimensions: str = "576x1024", status: str = "preview", seed: Optional[int] = None, cost_usd: Optional[float] = None, generation_time_sec: Optional[float] = None, output_path: Optional[str] = None, ) -> dict: """Build a manifest dict for a single shot. Args: shot_id: e.g. "ep01_sh14" episode: e.g. "tartarus_s01e01" prompt: The full prompt text model: Model ID used for generation refs: List of ref dicts, each with: - path (str): relative path from project root - slot (str): pipeline slot name - type (str): asset type (identity, turn, etc.) - weight (int): integer weight 1-10 - auto (bool): True if pipeline auto-resolved, False if manual - reason (str): why this ref was selected negative_prompt: Negative prompt text dimensions: Output dimensions e.g. "576x1024" status: One of "preview", "generated", "rejected" seed: Generation seed (None for preview) cost_usd: Generation cost estimate generation_time_sec: Wall-clock generation time output_path: Relative path to output file (None for preview) """ if status not in MANIFEST_WRITER_VALID_STATUSES: raise ValueError(f"Invalid status: {status!r}. Must be one of {MANIFEST_WRITER_VALID_STATUSES}") now = datetime.now(timezone.utc).isoformat() manifest = { "manifest_version": MANIFEST_VERSION, "shot_id": shot_id, "episode": episode, "status": status, "created_at": now, "updated_at": now, "seed": seed, "prompt": prompt, "negative_prompt": negative_prompt, "model": model, "dimensions": dimensions, "cost_usd": cost_usd, "generation_time_sec": generation_time_sec, "refs": refs, "output": { "path": output_path, }, } return manifest def build_ref_entry( path: str, slot: str, asset_type: str, weight: int, auto: bool, reason: str, ) -> dict: """Build a single ref entry for the manifest refs array.""" return { "path": path, "slot": slot, "type": asset_type, "weight": weight, "auto": auto, "reason": reason, } def write_manifest(manifest: dict, output_dir: Path, shot_id: str) -> Path: """Write a manifest JSON file to disk. File is written to: {output_dir}/{shot_id}.manifest.json If the file already exists, preserves created_at from the existing manifest and updates updated_at. Returns the path to the written manifest. """ manifest_path = output_dir / f"{shot_id}.manifest.json" # Preserve created_at from existing manifest if present if manifest_path.exists(): try: existing = json.loads(manifest_path.read_text(encoding="utf-8")) manifest["created_at"] = existing.get("created_at", manifest["created_at"]) except (json.JSONDecodeError, IOError): pass manifest["updated_at"] = datetime.now(timezone.utc).isoformat() output_dir.mkdir(parents=True, exist_ok=True) manifest_path.write_text( json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) logger.info("Manifest written: %s (status=%s)", manifest_path, manifest.get("status")) return manifest_path def read_manifest(manifest_path: Path) -> dict: """Read and return a manifest dict from disk. Raises FileNotFoundError if missing.""" return json.loads(manifest_path.read_text(encoding="utf-8")) def update_manifest_status( manifest_path: Path, status: str, *, seed: Optional[int] = None, cost_usd: Optional[float] = None, generation_time_sec: Optional[float] = None, output_path: Optional[str] = None, note: Optional[str] = None, ) -> dict: """Update the status of an existing manifest and optionally fill in generation fields. Returns the updated manifest dict. """ if status not in MANIFEST_WRITER_VALID_STATUSES: raise ValueError(f"Invalid status: {status!r}. Must be one of {MANIFEST_WRITER_VALID_STATUSES}") manifest = read_manifest(manifest_path) manifest["status"] = status manifest["updated_at"] = datetime.now(timezone.utc).isoformat() if seed is not None: manifest["seed"] = seed if cost_usd is not None: manifest["cost_usd"] = cost_usd if generation_time_sec is not None: manifest["generation_time_sec"] = generation_time_sec if output_path is not None: manifest["output"]["path"] = output_path manifest_path.write_text( json.dumps(manifest, indent=2, ensure_ascii=False) + "\n", encoding="utf-8", ) return manifest def generate_episode_rollup(episode_dir: Path) -> dict: """Scan all shot manifests in an episode directory, return aggregated stats. This is computed on-demand — never written to disk as a source of truth. 30-40 JSON files at ~2KB each completes in <50ms. """ from collections import Counter manifests = sorted(episode_dir.rglob("*.manifest.json")) ref_usage: Counter = Counter() type_usage: Counter = Counter() model_usage: Counter = Counter() total_cost = 0.0 shots_by_status: dict[str, int] = {} shot_details = [] for mpath in manifests: try: m = json.loads(mpath.read_text(encoding="utf-8")) except (json.JSONDecodeError, IOError): continue status = m.get("status", "unknown") shots_by_status[status] = shots_by_status.get(status, 0) + 1 total_cost += read_cost_from_record_safe(m) model_usage[m.get("model", "unknown")] += 1 for ref in m.get("refs", []): ref_path = ref.get("path", "unknown") ref_usage[ref_path] += 1 ref_type = ref.get("type", "unknown") type_usage[ref_type] += 1 shot_details.append({ "shot_id": m.get("shot_id"), "status": status, "ref_count": len(m.get("refs", [])), "model": m.get("model"), "cost": m.get("cost_usd"), }) return { "episode_dir": str(episode_dir), "total_shots": len(manifests), "shots_by_status": shots_by_status, "total_estimated_cost": round(total_cost, 4), "ref_usage_heatmap": dict(ref_usage.most_common()), "type_breakdown": dict(type_usage.most_common()), "model_breakdown": dict(model_usage), "shots": shot_details, }