| |
| """Record a predeclared evaluation sample with Godot Movie Maker and local ffmpeg. |
| |
| Linux only. Requires Xvfb, bubblewrap, Mesa software rendering and ffmpeg. This |
| records the original untouched generation with no input, retries or fixes. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| import json |
| import os |
| from pathlib import Path |
| import platform |
| import shutil |
| import signal |
| import subprocess |
| import sys |
| import tempfile |
| import time |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) |
| from vibegame.project import parse_project, project_sha256, write_project |
| from vibegame.validate import ERROR_PATTERN, sandbox_command, with_resource_limits |
| from generate import file_hash |
|
|
|
|
| def record(files: dict, godot: Path, output: Path, seconds: float = 15, fps: int = 30, *, generation_metadata: dict | None = None): |
| if platform.system() != "Linux": |
| raise RuntimeError("Recording requires the Linux runtime; local macOS Godot is disabled due to sandbox loader crashes") |
| if not 1 <= seconds <= 60 or not 1 <= fps <= 60: |
| raise ValueError("Recording requires 1–60 seconds and 1–60 fps") |
| ffmpeg, xvfb, ffprobe = shutil.which("ffmpeg"), shutil.which("Xvfb"), shutil.which("ffprobe") |
| if not ffmpeg or not xvfb or not ffprobe: |
| raise RuntimeError("Install ffmpeg, xvfb, libgl1-mesa-dri and bubblewrap first") |
| output = output.resolve() |
| if output.exists() or output.with_suffix(".recording.json").exists(): |
| raise FileExistsError("Refusing to overwrite recording or evidence") |
| output.parent.mkdir(parents=True, exist_ok=True) |
| staging = output.with_suffix(".recording") |
| staging.mkdir(exist_ok=False) |
| project = write_project(files, staging / "project") |
| movie = project / "capture.avi" |
| |
| display = None |
| xprocess = None |
| for number in range(101, 151): |
| if not Path(f"/tmp/.X11-unix/X{number}").exists(): |
| display = f":{number}" |
| xlog = (staging / "xvfb.log").open("wb") |
| xprocess = subprocess.Popen([xvfb, display, "-screen", "0", "1280x720x24", "-nolisten", "tcp", "-ac"], stdout=xlog, stderr=subprocess.STDOUT, start_new_session=True) |
| break |
| if display is None: |
| raise RuntimeError("No private X display available") |
| evidence = {"source": generation_metadata, "seconds": seconds, "fps": fps, |
| "mode": "fixed-timestep movie; no human or simulated input; unmodified generation", |
| "godot_binary_sha256": file_hash(godot), "success": False, "capture_success": False, |
| "project_sha256": project_sha256(files), "duration_seconds": seconds, "fixed_timestep": True, "player_input": False, **(generation_metadata or {})} |
| try: |
| socket = Path(f"/tmp/.X11-unix/X{display[1:]}") |
| deadline = time.monotonic() + 10 |
| while not socket.exists() and xprocess.poll() is None and time.monotonic() < deadline: |
| time.sleep(0.1) |
| if not socket.exists(): |
| raise RuntimeError("Private Xvfb display failed to start") |
| commands = [ |
| [str(godot), "--headless", "--editor", "--import", "--path", str(project)], |
| [str(godot), "--path", str(project), "--display-driver", "x11", "--rendering-method", "gl_compatibility", "--audio-driver", "Dummy", "--write-movie", str(movie), "--fixed-fps", str(fps), "--quit-after", str(round(seconds * fps)), "--disable-vsync"], |
| ] |
| for index, command in enumerate(commands): |
| wrapped, sandbox = sandbox_command(command, project) |
| separator = wrapped.index("--") |
| |
| wrapped[separator:separator] = ["--ro-bind", str(socket), str(socket)] |
| env = {"PATH": "/usr/bin:/bin", "HOME": str(project / ".sandbox-home"), "DISPLAY": display, |
| "LIBGL_ALWAYS_SOFTWARE": "1", "GODOT_SILENCE_ROOT_WARNING": "1", "LANG": "C.UTF-8"} |
| log_path = staging / ("import.log" if index == 0 else "movie.log") |
| with log_path.open("wb") as log: |
| process = subprocess.Popen(with_resource_limits(wrapped, file_bytes=2 * 1024**3), cwd=project, env=env, stdout=log, stderr=subprocess.STDOUT, start_new_session=True) |
| try: |
| returncode = process.wait(timeout=180 if index == 0 else 600) |
| except subprocess.TimeoutExpired: |
| os.killpg(process.pid, signal.SIGKILL) |
| process.wait() |
| raise RuntimeError("Godot movie render timed out") |
| text = log_path.read_text(errors="replace") |
| if returncode != 0 or ERROR_PATTERN.search(text): |
| raise RuntimeError(f"Godot failed; inspect {log_path}") |
| subprocess.run([ffmpeg, "-nostdin", "-v", "error", "-i", str(movie), "-c:v", "libx264", "-pix_fmt", "yuv420p", "-c:a", "aac", "-movflags", "+faststart", "-n", str(output)], check=True, timeout=180) |
| probe = subprocess.run([ffprobe, "-v", "error", "-select_streams", "v:0", "-count_frames", |
| "-show_entries", "stream=nb_read_frames,avg_frame_rate:format=duration", "-of", "json", str(output)], |
| check=True, capture_output=True, text=True, timeout=60) |
| video = json.loads(probe.stdout) |
| duration = float(video["format"]["duration"]) |
| frames = int(video["streams"][0]["nb_read_frames"]) |
| expected_frames = round(seconds * fps) |
| if duration < seconds - 2 / fps or abs(frames - expected_frames) > 2: |
| raise RuntimeError(f"Recording ended early: expected {expected_frames} frames, found {frames}, duration {duration}s") |
| evidence.update({"success": True, "capture_success": True, "output": str(output), "output_sha256": file_hash(output), |
| "duration_seconds": duration, "frames": frames, "requested_duration_seconds": seconds, "sandbox": sandbox}) |
| except Exception as exc: |
| evidence["error"] = f"{type(exc).__name__}: {exc}" |
| raise |
| finally: |
| if xprocess and xprocess.poll() is None: |
| os.killpg(xprocess.pid, signal.SIGTERM) |
| xprocess.wait(timeout=5) |
| if xprocess: |
| xlog.close() |
| output.with_suffix(".recording.json").write_text(json.dumps(evidence, indent=2) + "\n") |
| return evidence |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| sources = parser.add_mutually_exclusive_group(required=True) |
| sources.add_argument("--evaluation", type=Path) |
| sources.add_argument("--project-json", type=Path, help="Separate smoke/demo project; never counted as an eval sample") |
| parser.add_argument("--index", type=int, default=0) |
| parser.add_argument("--godot", type=Path, required=True) |
| parser.add_argument("--output", type=Path, required=True) |
| parser.add_argument("--seconds", type=float, default=15) |
| parser.add_argument("--fps", type=int, default=30) |
| parser.add_argument("--showcase-selection", type=Path, help="Explicit success-only showcase receipt; original fixed-sample mode is unchanged") |
| args = parser.parse_args() |
| selection = None |
| if args.showcase_selection: |
| if not args.evaluation: |
| parser.error("A success showcase requires --evaluation, never a scaffold project") |
| from showcase import load_selection |
| selection = load_selection(args.evaluation, args.showcase_selection) |
| if args.index not in selection['showcase_indices']: |
| parser.error("Only the exact first three clean results may be captured; no substitutions") |
| expected_output = args.evaluation / 'showcase' / 'recordings' / f'sample-{args.index:03d}.mp4' |
| if args.output.resolve() != expected_output.resolve(): |
| parser.error("Showcase recordings must use this evaluation's showcase/recordings/sample-NNN.mp4") |
| manifest = json.loads((args.evaluation / 'manifest.json').read_text()) |
| if file_hash(args.godot) != manifest.get('runtime_sha256'): |
| parser.error("Showcase capture must use the evaluated Godot binary") |
| if args.evaluation: |
| if selection is None and args.index not in (0, 200, 400): |
| parser.error("Published samples are predeclared indices 0,200,400; do not cherry-pick") |
| item = args.evaluation / "items" / f"{args.index:04d}" |
| generation = json.loads((item / "generation.json").read_text()) |
| result = json.loads((item / "result.json").read_text()) |
| if not result.get("clean_run_60s"): |
| parser.error("Predeclared sample failed the 60s check; report that failure instead of substituting a different sample") |
| files = parse_project(generation["output"]) |
| metadata = {"evaluation_manifest_sha256": file_hash(args.evaluation / "manifest.json"), "index": args.index, "sample_index": args.index, "prompt_id": result.get("prompt_id"), "generation_sha256": file_hash(item / "generation.json")} |
| if selection is not None: |
| metadata.update({'kind': 'model-success-showcase', 'showcase_policy': selection['policy'], |
| 'showcase_selection_sha256': file_hash(args.showcase_selection), |
| 'selection_disclosure': selection['disclosure']}) |
| else: |
| files = parse_project(args.project_json.read_text()) |
| metadata = {"kind": "scaffold-preview", "project_json_sha256": file_hash(args.project_json)} |
| print(json.dumps(record(files, args.godot.resolve(), args.output, args.seconds, args.fps, generation_metadata=metadata), indent=2)) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
