#!/usr/bin/env python3 """Build an honest 60-second edited walkthrough from immutable evaluation evidence. Default: complete 500-prompt evaluation and fixed sample indices 0,200,400 only. Separate scaffold mode permanently labels every frame as a teacher seed preview. Requires Pillow, ffmpeg, and ffprobe; no API, GPU, or downloaded visual assets. """ from __future__ import annotations import argparse import hashlib import json import math from pathlib import Path import shutil import statistics import subprocess import sys ROOT = Path(__file__).resolve().parents[1] sys.path.insert(0, str(ROOT / "src")) sys.path.insert(0, str(ROOT / "scripts")) from vibegame.project import parse_project, project_sha256 WIDTH, HEIGHT, FPS = 1280, 720, 30 DURATIONS = [8, 6, 14, 14, 14, 4] SAMPLE_INDICES = [0, 200, 400] PREVIEW_WATERMARK = "TEACHER SEED PREVIEW — NOT TRAINED MODEL" BG, PANEL, INK, MUTED, ACCENT = "#101724", "#182234", "#f3f6fb", "#afbdd0", "#63e3c0" def file_hash(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as stream: for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""): digest.update(chunk) return digest.hexdigest() def read_json(path: Path) -> dict: value = json.loads(path.read_text(encoding="utf-8")) if not isinstance(value, dict): raise ValueError(f"Expected JSON object: {path}") return value def finite_nonnegative(value) -> bool: return isinstance(value, (float, int)) and not isinstance(value, bool) and math.isfinite(value) and value >= 0 def probe_video(path: Path) -> dict: executable = shutil.which("ffprobe") if not executable: raise RuntimeError("ffprobe is required; install ffmpeg (which includes ffprobe)") result = subprocess.run([executable, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=width,height,duration,nb_frames,r_frame_rate:format=duration", "-of", "json", str(path)], check=True, capture_output=True, text=True, timeout=30) value = json.loads(result.stdout) if len(value.get("streams", [])) != 1: raise ValueError(f"Expected one video stream in {path}") stream = value["streams"][0] duration = float(stream.get("duration", value.get("format", {}).get("duration", 0))) if not math.isfinite(duration) or duration <= 0 or stream.get("width", 0) <= 0 or stream.get("height", 0) <= 0: raise ValueError(f"Invalid video dimensions or duration: {path}") return {**stream, "duration_seconds": duration} def verify_capture(path: Path, *, manifest_sha: str | None = None, generation_sha: str | None = None, index: int | None = None, prompt_id: str | None = None, project_sha: str | None = None) -> dict: metadata_path = path.with_suffix(".recording.json") metadata = read_json(metadata_path) if metadata.get("capture_success") is not True: raise ValueError(f"Recording did not succeed: {metadata_path}") if metadata.get("output_sha256") != file_hash(path): raise ValueError(f"Recording content hash mismatch: {path}") if metadata.get("fixed_timestep") is not True or metadata.get("player_input") is not False: raise ValueError("Expected disclosed fixed-timestep capture without player input") if not finite_nonnegative(metadata.get("duration_seconds")) or metadata["duration_seconds"] < 14: raise ValueError("Each capture must contain at least 14 unmodified seconds") if manifest_sha is not None: for key, expected in {"evaluation_manifest_sha256": manifest_sha, "generation_sha256": generation_sha, "sample_index": index, "prompt_id": prompt_id}.items(): if metadata.get(key) != expected: raise ValueError(f"Recording source mismatch for {key}: {path}") if project_sha is not None and metadata.get("project_sha256") != project_sha: raise ValueError(f"Recording project hash mismatch: {path}") video = probe_video(path) if video["duration_seconds"] < 14 - 1 / FPS: raise ValueError(f"Recording is shorter than the 14-second segment: {path}") return {"path": str(path.resolve()), "metadata": metadata, "metadata_sha256": file_hash(metadata_path), "probe": video} def evaluation_plan(directory: Path, *, showcase_selection: Path | None = None) -> dict: """Read all 500 results and recompute rates; never promote a partial summary.""" directory = directory.resolve() selection = None if showcase_selection is not None: from showcase import load_selection selection = load_selection(directory, showcase_selection) manifest_path = directory / "manifest.json" manifest = read_json(manifest_path) if (manifest.get("expected_count") != 500 or manifest.get("run_seconds") != 60 or manifest.get("smoke_test") is not False or manifest.get("sample_indices") != SAMPLE_INDICES): raise ValueError("Launch walkthrough requires the complete fixed 500-prompt / 60-second evaluation, not a smoke test") actual_results = sorted((directory / "items").glob("*/result.json")) expected_results = [directory / "items" / f"{index:04d}" / "result.json" for index in range(500)] if actual_results != expected_results or any(not path.is_file() for path in expected_results): raise ValueError("Evaluation must contain exactly results 0000 through 0499; partial runs cannot produce a launch asset") manifest_sha = file_hash(manifest_path) results, attempts, hashes = [], [], {} for index, path in enumerate(expected_results): result = read_json(path) attempt_path = path.parent / "attempt.json" attempt = read_json(attempt_path) if result.get("index") != index or attempt.get("index") != index: raise ValueError(f"Evaluation index mismatch: {path}") if any(type(result.get(key)) is not bool for key in ["compile_success", "clean_run_60s"]): raise ValueError(f"Result success fields must be explicit booleans: {path}") if not isinstance(attempt.get("instruction"), str) or not attempt["instruction"].strip() or not attempt.get("prompt_id"): raise ValueError(f"Missing actual prompt evidence: {attempt_path}") if result.get("prompt_id", attempt["prompt_id"]) != attempt["prompt_id"]: raise ValueError(f"Result prompt does not match attempt: {path}") if result.get("infrastructure_failure"): raise ValueError("Infrastructure failures make this evaluation incomplete; a launch asset cannot claim final rates") if result.get("clean_run_60s") and not (result.get("compile_success") and result.get("run_seconds", 0) >= 60): raise ValueError(f"Inconsistent clean-run evidence: {path}") results.append(result); attempts.append(attempt) hashes[str(path.relative_to(directory))] = file_hash(path) hashes[str(attempt_path.relative_to(directory))] = file_hash(attempt_path) if len({item["prompt_id"] for item in attempts}) != 500 or len({item["instruction"] for item in attempts}) != 500: raise ValueError("Evaluation contains duplicate prompt IDs or instructions") compiled = sum(bool(row.get("compile_success")) for row in results) clean = sum(bool(row.get("clean_run_60s")) for row in results) latency = [row["prompt_to_launch_seconds"] for row in results if row.get("compile_success") and finite_nonnegative(row.get("prompt_to_launch_seconds"))] metrics = {"denominator": 500, "compile_success_count": compiled, "clean_run_60s_count": clean, "compile_success_rate": compiled / 500, "clean_run_60s_rate": clean / 500, "prompt_to_launch_median_seconds": statistics.median(latency) if latency else None, "timed_successful_launches": len(latency), "under_60s_launch_count": sum(value <= 60 for value in latency)} if (directory / "summary.json").exists(): summary = read_json(directory / "summary.json") if (summary.get("status") != "complete" or summary.get("recorded_prompts") != 500 or summary.get("compile_success_count") != compiled or summary.get("clean_run_60s_count") != clean): raise ValueError("Evaluation summary disagrees with its 500 item results") hashes["summary.json"] = file_hash(directory / "summary.json") samples = [] for index in (selection['showcase_indices'] if selection else SAMPLE_INDICES): result, attempt = results[index], attempts[index] item = directory / "items" / f"{index:04d}" generation_path = item / "generation.json" generation = read_json(generation_path) if generation_path.is_file() else {} generation_sha = file_hash(generation_path) if generation_path.is_file() else None output = generation.get("output") files, parse_error = None, None if isinstance(output, str): if generation.get("output_sha256") != hashlib.sha256(output.encode()).hexdigest(): raise ValueError(f"Generation output hash mismatch: {generation_path}") try: files = parse_project(output) except ValueError as exc: parse_error = str(exc) elif result.get("compile_success"): raise ValueError(f"Successful result is missing its original output: {generation_path}") sample = {"index": index, "prompt_id": attempt["prompt_id"], "instruction": attempt["instruction"], "generation_seconds": generation.get("generation_seconds"), "prompt_to_launch_seconds": result.get("prompt_to_launch_seconds"), "generation_sha256": generation_sha, "files": files, "compile_success": bool(result.get("compile_success")), "clean_run_60s": bool(result.get("clean_run_60s")), "capture": None, "diagnostic": None} if selection: sample['generation_seconds'] = generation.get('generation_call_seconds', generation.get('generation_seconds')) sample['generation_timing_label'] = ('Batch generation call' if generation.get('generation_call_seconds') is not None else 'Generation (legacy timer)') if result.get("compile_success") and (files is None or result.get("project_sha256") != project_sha256(files)): raise ValueError(f"Compiled project does not match generation: {generation_path}") recording = directory / ("showcase/recordings" if selection else "recordings") / f"sample-{index:03d}.mp4" if not result.get("clean_run_60s"): sample["diagnostic"] = (result.get("error") or parse_error or ("Project imported, but did not complete a clean 60-second run." if result.get("compile_success") else "Project did not import successfully.")) elif not recording.is_file(): sample["diagnostic"] = "The project passed its 60-second check, but a gameplay recording is unavailable." sidecar = recording.with_suffix(".recording.json") if sidecar.is_file(): report = read_json(sidecar) sample["diagnostic"] += " " + str(report.get("error", "Capture did not complete.")) else: sample["capture"] = verify_capture(recording, manifest_sha=manifest_sha, generation_sha=generation_sha, index=index, prompt_id=attempt["prompt_id"], project_sha=project_sha256(files)) if selection: metadata = sample['capture']['metadata'] if (metadata.get('kind') != 'model-success-showcase' or metadata.get('showcase_policy') != selection['policy'] or metadata.get('showcase_selection_sha256') != file_hash(showcase_selection) or metadata.get('godot_binary_sha256') != manifest.get('runtime_sha256')): raise ValueError('Showcase capture belongs to different selection or engine evidence') samples.append(sample) if generation_sha: hashes[str(generation_path.relative_to(directory))] = generation_sha if selection and not all(sample['capture'] for sample in samples): raise ValueError('All three selected successes require real captures; failed captures cannot be replaced') if not any(sample["capture"] for sample in samples): raise ValueError("At least one real recording from predeclared samples 0,200,400 is required; diagnostic cards alone are not a gameplay demo") plan = {"kind": "model-success-showcase" if selection else "model-evaluation", "scaffold_preview": False, "evaluation_directory": str(directory), "evaluation_manifest_sha256": manifest_sha, "model": manifest.get("generation"), "metrics": metrics, "samples": samples, "evidence_hashes": hashes, "includes_failed_or_unrecorded_samples": any(sample["diagnostic"] for sample in samples)} if selection: plan.update({'showcase_selection': selection, 'showcase_selection_sha256': file_hash(showcase_selection), 'selection_disclosure': selection['disclosure'], 'benchmark_sample_indices': SAMPLE_INDICES}) return plan def scaffold_plan(examples: list[Path], recordings: list[Path], candidates: Path) -> dict: if len(examples) != 3 or len(recordings) != 3: raise ValueError("Scaffold preview requires exactly three --example and three --recording arguments, in display order") seed_prompts = {} if candidates.is_file(): for line in candidates.read_text(encoding="utf-8").splitlines(): if line.strip(): row = json.loads(line) seed_prompts[project_sha256(parse_project(row["output"]))] = row["instruction"] samples = [] for index, (example, recording) in enumerate(zip(examples, recordings)): if example.is_dir(): files = {str(path.relative_to(example)): path.read_text(encoding="utf-8") for path in sorted(example.rglob("*")) if path.is_file() and path.suffix in {".godot", ".tscn", ".gd"} and ".godot" not in path.relative_to(example).parts} files = parse_project(json.dumps({"files": files})) else: files = parse_project(example.read_text(encoding="utf-8")) digest = project_sha256(files) capture = verify_capture(recording, project_sha=digest) if recording.is_file() else None if capture and capture["metadata"].get("kind") != "scaffold-preview": raise ValueError("Seed preview recording must come from record_game.py --project-json, not a mislabeled model evaluation") samples.append({"index": index, "prompt_id": example.stem, "files": files, "instruction": seed_prompts.get(digest, "Original seed prompt is unavailable. This is a source-code seed preview, not a model generation."), "generation_seconds": None, "prompt_to_launch_seconds": None, "capture": capture, "project_sha256": digest, "compile_success": None, "clean_run_60s": None, "diagnostic": None if capture else "No recording was provided for this teacher seed."}) if not any(sample["capture"] for sample in samples): raise ValueError("A seed preview still requires at least one genuine captured game") return {"kind": "scaffold-preview", "scaffold_preview": True, "metrics": None, "samples": samples, "watermark": PREVIEW_WATERMARK, "includes_failed_or_unrecorded_samples": any(sample["diagnostic"] for sample in samples)} def font(size: int, mono: bool = False): from PIL import ImageFont options = (["/usr/share/fonts/truetype/dejavu/DejaVuSansMono.ttf", "/System/Library/Fonts/Menlo.ttc"] if mono else ["/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", "/System/Library/Fonts/Supplemental/Arial.ttf", "/System/Library/Fonts/Helvetica.ttc"]) for path in options: if Path(path).is_file(): return ImageFont.truetype(path, size=size) return ImageFont.load_default(size=size) def wrapped_lines(draw, text: str, selected_font, max_width: int) -> list[str]: lines = [] for paragraph in str(text).splitlines() or [""]: current = "" for word in paragraph.split(): candidate = f"{current} {word}".strip() if draw.textlength(candidate, font=selected_font) <= max_width: current = candidate else: if current: lines.append(current) current = "" for character in word: if draw.textlength(current + character, font=selected_font) > max_width and current: lines.append(current); current = character else: current += character lines.append(current) return lines def block(draw, text: str, box: tuple[int, int, int, int], size: int = 28, minimum: int = 18, fill: str = INK, mono: bool = False, max_lines: int | None = None, truncate: bool = False) -> None: x, y, width, height = box for selected in range(size, minimum - 1, -1): selected_font = font(selected, mono) lines = str(text).expandtabs(4).splitlines() if mono else wrapped_lines(draw, text, selected_font, width) spacing = math.ceil(selected * 1.35) if (len(lines) * spacing <= height and (max_lines is None or len(lines) <= max_lines) and all(draw.textlength(line, font=selected_font) <= width for line in lines)): break else: if not truncate: raise ValueError("Text does not fit card; enlarge its area or explicitly design more pages, never silently omit the prompt") selected = minimum; selected_font = font(selected, mono); spacing = math.ceil(selected * 1.35) all_lines = str(text).expandtabs(4).splitlines() if mono else wrapped_lines(draw, text, selected_font, width) lines = all_lines[:max(1, height // spacing)] if len(all_lines) > len(lines): lines[-1] = lines[-1][:-3] + "..." for index, line in enumerate(lines): if draw.textlength(line, font=selected_font) > width: while line and draw.textlength(line + "...", font=selected_font) > width: line = line[:-1] lines[index] = line + "..." for line in lines: draw.text((x, y), line, font=selected_font, fill=fill) y += spacing def base_card(preview: bool, section: str): from PIL import Image, ImageDraw image = Image.new("RGB", (WIDTH, HEIGHT), BG) draw = ImageDraw.Draw(image) draw.rectangle((0, 0, WIDTH, 9), fill=ACCENT) draw.text((54, 36), "VibeGame-8B", font=font(31), fill=INK) block(draw, section, (430, 42, 796, 40), size=23, minimum=20, fill=MUTED) draw.line((54, 94, WIDTH - 54, 94), fill="#334155", width=2) if preview: draw.rectangle((0, HEIGHT - 70, WIDTH, HEIGHT), fill="#802c22") block(draw, PREVIEW_WATERMARK, (38, HEIGHT - 57, WIDTH - 76, 42), size=28, minimum=24) else: draw.text((54, HEIGHT - 48), "Edited 60-second sequence • actual measurements shown • no simulated typing", font=font(19), fill=MUTED) return image, draw def latency_text(sample: dict, preview: bool = False) -> str: if preview: return "Teacher seed • no model inference or latency claim" generation, launch = sample.get("generation_seconds"), sample.get("prompt_to_launch_seconds") label = sample.get('generation_timing_label', 'Generation') first = f"{label}: {generation:.1f}s" if finite_nonnegative(generation) else "Generation timing unavailable" second = f"prompt → launch: {launch:.1f}s" if finite_nonnegative(launch) else "no measured successful launch" return first + " | " + second def render_assets(plan: dict, directory: Path) -> list[dict]: from PIL import Image, ImageDraw directory.mkdir(parents=True, exist_ok=False) preview, first = plan["scaffold_preview"], plan["samples"][0] showcase = plan.get('kind') == 'model-success-showcase' def disclosure(draw): if showcase: metrics = plan['metrics'] draw.rectangle((0, 628, WIDTH, HEIGHT), fill=BG) block(draw, 'SELECTED SUCCESSES — NOT REPRESENTATIVE', (32, 633, 1216, 29), size=21, minimum=19, fill=ACCENT) block(draw, f"All 500: compile {metrics['compile_success_count']}/500 ({metrics['compile_success_rate']:.1%}); " f"clean 60s {metrics['clean_run_60s_count']}/500 ({metrics['clean_run_60s_rate']:.1%})", (32, 663, 1216, 25), size=20, minimum=18, fill=INK) block(draw, '60-second edited video, not generation time • fixed timestep • no player input', (32, 691, 1216, 24), size=19, minimum=17, fill=MUTED) assets = [] intro, draw = base_card(preview, "01 / the actual prompt") block(draw, first["instruction"], (54, 130, 1172, 424), size=32, minimum=21) block(draw, latency_text(first, preview), (54, 582, 1172, 52), size=24, minimum=20, fill=ACCENT) disclosure(draw) target = directory / "01-prompt.png"; intro.save(target) assets.append({"type": "card", "path": str(target), "seconds": 8}) source, draw = base_card(preview, "02 / the resulting project files") files = first.get("files") if files: listing = "\n".join(f"{name} {len(text.splitlines())} lines" for name, text in sorted(files.items())) block(draw, listing, (54, 135, 465, 440), size=29, minimum=20, mono=True) draw.rounded_rectangle((540, 123, 1226, 589), radius=16, fill=PANEL) draw.text((566, 143), "main.gd • original source excerpt", font=font(20), fill=ACCENT) excerpt = "\n".join(files.get("main.gd", "").splitlines()[:14]) block(draw, excerpt, (566, 185, 630, 368), size=21, minimum=16, mono=True, truncate=True) else: block(draw, "This sample did not produce a valid project bundle.", (54, 166, 1140, 170), size=43, minimum=30) block(draw, first.get("diagnostic") or "No complete output was saved.", (54, 360, 1140, 200), size=27, minimum=20, fill=MUTED, truncate=True) target = directory / "02-project.png"; source.save(target) if showcase: disclosure(draw) source.save(target) assets.append({"type": "card", "path": str(target), "seconds": 6}) for sample in plan["samples"]: index = sample["index"] if sample["capture"]: overlay = Image.new("RGBA", (WIDTH, HEIGHT), (0, 0, 0, 0)); overlay_draw = ImageDraw.Draw(overlay) overlay_draw.rectangle((0, 0, WIDTH, 91), fill=BG) block(overlay_draw, f"SAMPLE {index + 1:03d} / {sample['prompt_id']}", (34, 13, 1212, 34), size=23, minimum=18) block(overlay_draw, latency_text(sample, preview), (34, 52, 1212, 28), size=19, minimum=16, fill=ACCENT) overlay_draw.rectangle((0, 640, WIDTH, HEIGHT), fill="#802c22" if preview else BG) if preview: block(overlay_draw, PREVIEW_WATERMARK, (28, 648, 1224, 31), size=23, minimum=21) footer = "Fixed-timestep capture • no player input • edited preview" else: footer = "Fixed-timestep capture • no player input • unmodified model output" block(overlay_draw, footer, (28, 686 if preview else 662, 1224, 30), size=20, minimum=18, fill=INK if preview else MUTED) disclosure(overlay_draw) target = directory / f"sample-{index:03d}-overlay.png"; overlay.save(target) assets.append({"type": "capture", "path": sample["capture"]["path"], "overlay": str(target), "seconds": 14}) else: diagnostic, draw = base_card(preview, f"Sample {index + 1:03d} / original slot retained") title = "Recording unavailable" if sample.get("clean_run_60s") else "This sample failed" if preview: title = "Teacher seed recording unavailable" block(draw, title, (54, 142, 1140, 112), size=47, minimum=32) block(draw, sample["diagnostic"], (54, 283, 1140, 220), size=29, minimum=19, fill=MUTED, truncate=True) block(draw, "No replacement sample. No repaired output. No invented footage.", (54, 540, 1140, 57), size=24, minimum=20, fill=ACCENT) block(draw, latency_text(sample, preview), (54, 605, 1140, 32), size=20, minimum=18, fill=MUTED) target = directory / f"sample-{index:03d}-diagnostic.png"; diagnostic.save(target) assets.append({"type": "card", "path": str(target), "seconds": 14}) outro, draw = base_card(preview, "03 / what the evidence says") if preview: block(draw, "This is a teacher seed preview.", (54, 145, 1160, 115), size=48, minimum=35) block(draw, "No trained-model evaluation or 60-second generation claim is being made.", (54, 300, 1140, 180), size=35, minimum=26, fill=MUTED) block(draw, "Original scaffold projects • actual Godot recordings", (54, 555, 1140, 50), size=27, minimum=21, fill=ACCENT) else: metrics = plan["metrics"] block(draw, "All 500 prompts. One attempt each.", (54, 127, 1160, 85), size=42, minimum=32) block(draw, f"Compile: {metrics['compile_success_count']} / 500 ({100 * metrics['compile_success_rate']:.1f}%)\n" f"Clean 60s run: {metrics['clean_run_60s_count']} / 500 ({100 * metrics['clean_run_60s_rate']:.1f}%)", (54, 250, 1160, 160), size=42, minimum=32, fill=ACCENT) median = metrics["prompt_to_launch_median_seconds"] timing = (f"Median measured launch: {median:.1f}s across {metrics['timed_successful_launches']} successful launches." if median is not None else "No successful launch latency was measured.") block(draw, timing + "\nHeadless checks do not prove gameplay quality. No player input was tested.", (54, 466, 1160, 158), size=26, minimum=20, fill=MUTED) target = directory / "06-results.png"; outro.save(target) if showcase: disclosure(draw) outro.save(target) assets.append({"type": "card", "path": str(target), "seconds": 4}) assert [asset["seconds"] for asset in assets] == DURATIONS return assets def encode_video(plan: dict, output: Path) -> dict: executable = shutil.which("ffmpeg") if not executable or not shutil.which("ffprobe"): raise RuntimeError("Install ffmpeg and ffprobe before building a video") output = output.resolve() if output.suffix.lower() != ".mp4": raise ValueError("Output must use .mp4") evidence_path = output.with_suffix(".demo.json") assets_dir = output.with_suffix(".assets") if output.exists() or evidence_path.exists() or assets_dir.exists(): raise FileExistsError("Refusing to overwrite the video, its evidence, or build assets; choose a new output name") output.parent.mkdir(parents=True, exist_ok=True) assets = render_assets(plan, assets_dir) segments = [] for index, asset in enumerate(assets): target = assets_dir / f"segment-{index:02d}.mp4" command = [executable, "-nostdin", "-v", "error"] if asset["type"] == "card": command += ["-loop", "1", "-framerate", str(FPS), "-i", asset["path"], "-vf", "setsar=1"] else: viewport = 536 if plan.get('kind') == 'model-success-showcase' else 548 command += ["-i", asset["path"], "-loop", "1", "-framerate", str(FPS), "-i", asset["overlay"], "-filter_complex", (f"[0:v]scale=1280:{viewport}:force_original_aspect_ratio=decrease," f"pad=1280:720:(ow-iw)/2:92+({viewport}-ih)/2:color=0x101724,fps=30,setsar=1[game];" "[game][1:v]overlay=0:0:shortest=1,format=yuv420p[out]"), "-map", "[out]"] command += ["-frames:v", str(asset["seconds"] * FPS), "-r", str(FPS), "-an", "-c:v", "libx264", "-preset", "veryfast", "-crf", "20", "-pix_fmt", "yuv420p", "-movflags", "+faststart", "-n", str(target)] subprocess.run(command, check=True, timeout=300) probe = probe_video(target) if abs(probe["duration_seconds"] - asset["seconds"]) > 1 / FPS + .005: raise ValueError(f"Segment {index} duration mismatch: {probe['duration_seconds']}") segments.append(target) concat = assets_dir / "concat.txt" concat.write_text("".join(f"file '{path.name}'\n" for path in segments), encoding="utf-8") temporary = assets_dir / "complete.mp4" subprocess.run([executable, "-nostdin", "-v", "error", "-f", "concat", "-safe", "1", "-i", str(concat), "-c", "copy", "-an", "-movflags", "+faststart", "-n", str(temporary)], check=True, timeout=300) probe = probe_video(temporary) if (abs(probe["duration_seconds"] - 60) > 1 / FPS + .005 or probe["width"] != WIDTH or probe["height"] != HEIGHT or probe.get("r_frame_rate") != "30/1"): raise ValueError(f"Final video failed 60-second / 1280x720 / 30fps verification: {probe}") temporary.rename(output) evidence = {"schema_version": 1, "kind": plan["kind"], "scaffold_preview": plan["scaffold_preview"], "output": str(output), "output_sha256": file_hash(output), "probe": probe, "timeline_seconds": DURATIONS, "total_seconds": sum(DURATIONS), "source_plan": plan, "assets": assets, "editing": "Hard-cut edited sequence. Actual latencies displayed, no typing simulation or realtime generation claim. Recordings show fixed-timestep simulation with no player input. No sample substitution or repairs.", "watermark": PREVIEW_WATERMARK if plan["scaffold_preview"] else plan.get('selection_disclosure')} evidence_path.write_text(json.dumps(evidence, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") return {"video": str(output), "evidence": str(evidence_path), "seconds": probe["duration_seconds"], "scaffold_preview": plan["scaffold_preview"], "includes_diagnostic_cards": plan["includes_failed_or_unrecorded_samples"]} def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--evaluation", type=Path) parser.add_argument("--output", type=Path, required=True) parser.add_argument("--scaffold-preview", action="store_true") parser.add_argument("--showcase-selection", type=Path, help="Explicit first-three-clean success showcase, never representative samples") parser.add_argument("--example", type=Path, action="append", default=[], help="Scaffold only: source directory or strict project JSON, exactly three") parser.add_argument("--recording", type=Path, action="append", default=[], help="Scaffold only: matching capture MP4, exactly three; missing files become diagnostics") parser.add_argument("--seed-candidates", type=Path, default=ROOT / "data/synthetic_candidates.jsonl") args = parser.parse_args() if args.scaffold_preview: if args.evaluation or args.showcase_selection: parser.error("--scaffold-preview cannot consume an evaluation; keep seed previews and model evidence separate") plan = scaffold_plan(args.example, args.recording, args.seed_candidates) else: if not args.evaluation or args.example or args.recording: parser.error("Default mode requires --evaluation; --example/--recording require the explicit --scaffold-preview flag") plan = evaluation_plan(args.evaluation, showcase_selection=args.showcase_selection) print(json.dumps(encode_video(plan, args.output), indent=2)) if __name__ == "__main__": main()