code/scripts/record_game.py

#!/usr/bin/env python3
"""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"
    # Private X socket, no TCP listener, no access to the user's desktop.
    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("--")
            # Share only this private Xvfb socket. Network namespace remains isolated.
            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()