VibeGame-8B / code /scripts /build_demo.py
khursheed's picture
Release pipeline source, tests, notebook, and licenses
ae36cfb verified
Raw History Blame Contribute Delete
32 kB
#!/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()