#!/usr/bin/env python3 """Finish validation of persisted generations on CPU; never load or call a model. The original 500-prompt manifest and all completed results remain immutable. Only generation.json files already present at startup can receive new results. """ from __future__ import annotations import argparse import hashlib import json import multiprocessing import re import subprocess import sys import tempfile from concurrent.futures import ProcessPoolExecutor, as_completed from contextlib import contextmanager from datetime import datetime, timezone from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parent)) import evaluate from generate import file_hash from vibegame.project import ProjectError, parse_project, project_sha256 def read_json(path): return json.loads(Path(path).read_text()) def safe_path(root: Path, path: Path): if not path.resolve().is_relative_to(root.resolve()): raise ValueError("Evidence path escapes its evaluation directory") for part in (path, *path.parents): if part.is_symlink(): raise ValueError("Symlink evaluation evidence is not allowed") if part == root: break return path def check_completed_evidence(item, files, runtime_hash, seconds): """A finished worker check is reusable only with matching source and logs.""" checks = sorted(path for path in item.glob("check*") if path.is_dir()) finished = [] for check in checks: safe_path(item, check) if not re.fullmatch(r"check(?:-resume-[0-9]{3,})?", check.name): raise ValueError("Unexpected validation evidence directory") path = check / "validation.json" if path.exists(): safe_path(item, path) finished.append(check) if len(finished) > 1: raise ValueError("Multiple completed validation checks are ambiguous") if not finished: return check = finished[0] result = read_json(check / "validation.json") if (result.get("project_sha256") != project_sha256(files) or result.get("godot_binary_sha256") != runtime_hash or result.get("run_seconds") != seconds): raise ValueError("Completed validation belongs to different sources or runtime") for name, source in files.items(): path = safe_path(item, check / "project" / name) if not path.is_file() or path.read_bytes() != source.encode(): raise ValueError("Completed validation project differs from the saved generation") for stage in ("import", "runtime"): record = result.get(stage) if record is not None: path = safe_path(item, check / (stage + ".log")) if not path.is_file() or record.get("log_sha256") != file_hash(path): raise ValueError("Completed validation log fingerprint does not match") if not isinstance(result.get("import"), dict): raise TypeError("Completed validation has no import evidence") if result.get("compile_success") != result["import"].get("success"): raise ValueError("Completed compile result contradicts its import evidence") if result.get("compile_success") and not isinstance(result.get("runtime"), dict): raise ValueError("Completed successful import lacks its runtime evidence") if result.get("clean_run_60s") and (not result.get("compile_success") or not result.get("runtime", {}).get("success") or result["runtime"].get("required_seconds") != seconds or result["runtime"].get("wall_seconds", 0) < seconds): raise ValueError("Completed clean result lacks a full runtime window") def prepare(evaluation, prompts_path, godot): root = Path(evaluation).resolve(strict=True) manifest_path = safe_path(root, root / "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") != [0, 200, 400] or manifest.get("workers") not in (1, 2, 4)): raise ValueError("An existing frozen 500-prompt, 60-second evaluation is required") if file_hash(prompts_path) != manifest.get("prompts_sha256"): raise ValueError("Prompt file differs from the original evaluation manifest") runtime_hash = file_hash(godot) if runtime_hash != manifest.get("runtime_sha256"): raise ValueError("Godot binary differs from the original evaluation runtime") rows = [json.loads(line) for line in Path(prompts_path).read_text().splitlines() if line.strip()] prompts = [row.get("instruction", row.get("prompt")) for row in rows] ids = [str(row.get("id", index)) for index, row in enumerate(rows)] if (len(rows) != 500 or any(not isinstance(value, str) or not value.strip() for value in prompts) or len(set(prompts)) != 500 or len(set(ids)) != 500): raise ValueError("Frozen prompt identities are incomplete or duplicated") existing, candidates = {}, [] items_root = safe_path(root, root / "items") for item in sorted(items_root.iterdir()): safe_path(root, item) if not item.is_dir() or not re.fullmatch(r"[0-9]{4}", item.name) or int(item.name) >= 500: raise ValueError("Unexpected evaluation item") index = int(item.name) result_path = safe_path(root, item / "result.json") if result_path.exists(): existing[str(result_path)] = file_hash(result_path) continue # Completed failures and successes are never retried. generation_path = safe_path(root, item / "generation.json") if not generation_path.exists(): continue # No scoring or regeneration of interrupted/unseen outputs. attempt_path = safe_path(root, item / "attempt.json") attempt = read_json(attempt_path) if (attempt.get("index") != index or attempt.get("prompt_id") != ids[index] or attempt.get("instruction") != prompts[index]): raise ValueError("Saved generation does not match its original prompt identity") try: generation = read_json(generation_path) except (json.JSONDecodeError, UnicodeError): generation = {"output": None, "error": "Generation persistence interrupted; retained partial file and did not resample"} if not isinstance(generation, dict): raise TypeError("Malformed saved generation record") if generation.get("output") is not None and (not isinstance(generation["output"], str) or generation.get("output_sha256") != hashlib.sha256(generation["output"].encode()).hexdigest()): raise ValueError("Persisted generation hash mismatch") result = {"index": index, "prompt_id": ids[index], "compile_success": False, "clean_run_60s": False, "generation_seconds": generation.get("generation_seconds"), "generation_batch_size": generation.get("generation_batch_size", 1), "generation_call_seconds": generation.get("generation_call_seconds"), "prompt_to_launch_seconds": None} files = None if generation.get("error"): result["error"] = generation["error"] else: try: files = parse_project(generation.get("output")) except (ProjectError, ValueError) as exc: result["error"] = f"{type(exc).__name__}: {exc}" if files is not None: check_completed_evidence(item, files, runtime_hash, 60) elif any(item.glob("check*/validation.json")): raise ValueError("Completed engine evidence conflicts with an invalid saved project") candidates.append({"item": item, "files": files, "result": result, "generation_sha256": file_hash(generation_path), "attempt_sha256": file_hash(attempt_path)}) return {"root": root, "manifest": manifest, "manifest_sha256": file_hash(manifest_path), "runtime_hash": runtime_hash, "existing_results": existing, "candidates": candidates} def assert_unchanged(plan, candidate=None): if file_hash(plan["root"] / "manifest.json") != plan["manifest_sha256"]: raise ValueError("Evaluation manifest changed during recovery") for filename, digest in plan["existing_results"].items(): if file_hash(Path(filename)) != digest: raise ValueError("An existing evaluation result changed during recovery") if candidate is not None: for name, key in (("generation.json", "generation_sha256"), ("attempt.json", "attempt_sha256")): if file_hash(candidate["item"] / name) != candidate[key]: raise ValueError("Saved generation or attempt changed during recovery") if (candidate["item"] / "result.json").exists(): raise ValueError("Another process wrote the pending result; stop concurrent evaluation") def preflight(root, godot): destination = root / ".sandbox-preflight" destination.mkdir(exist_ok=True) wrapped, _ = evaluate.sandbox_command([str(Path(godot).resolve())], destination) wrapped[-1] = "/usr/bin/true" probe = subprocess.run(wrapped, capture_output=True, timeout=15, check=False) if probe.returncode: raise RuntimeError("Sandbox preflight failed; no saved project was run") @contextmanager def recovery_lock(root): import fcntl key = hashlib.sha256(str(root.resolve()).encode()).hexdigest() path = Path(tempfile.gettempdir()) / ("vibegame-validate-saved-" + key + ".lock") with path.open("a+") as handle: try: fcntl.flock(handle, fcntl.LOCK_EX | fcntl.LOCK_NB) except BlockingIOError: raise RuntimeError("A saved-validation recovery is already running") from None yield def recover(plan, godot): """Use the evaluator's original scoring logic without constructing a backend.""" assert_unchanged(plan) preflight(plan["root"], godot) manifest = plan["manifest"] session = {"kind": "cpu-validation-of-saved-generations", "status": "running", "started_utc": datetime.now(timezone.utc).isoformat(), "manifest_sha256": plan["manifest_sha256"], "godot_binary_sha256": plan["runtime_hash"], "expected_prompts": 500, "generation_calls": 0, "resampling": False, "workers": manifest["workers"], "saved_outputs": {row["item"].name: row["generation_sha256"] for row in plan["candidates"]}, "completed_indices": []} number = 1 while (plan["root"] / f"saved-validation-{number:03d}.json").exists(): number += 1 report = plan["root"] / f"saved-validation-{number:03d}.json" evaluate.atomic_json(report, session, immutable=True) def finish(candidate, check): assert_unchanged(plan, candidate) if candidate["files"] is not None: check_completed_evidence(candidate["item"], candidate["files"], plan["runtime_hash"], 60) if check is not None: check = {**check, "validation_queue_seconds": None} evaluate.finish_result(candidate["item"], candidate["result"], check, plan["root"], 500, 60) # A resumed check has no valid original generation-to-launch wall clock. session["completed_indices"].append(candidate["result"]["index"]) evaluate.atomic_json(report, session) try: with ProcessPoolExecutor(max_workers=manifest["workers"], mp_context=multiprocessing.get_context("spawn")) as pool: pending = {} for candidate in plan["candidates"]: assert_unchanged(plan, candidate) if candidate["files"] is None: finish(candidate, None) else: future = pool.submit(evaluate.validate_saved_project, candidate["files"], candidate["item"], str(godot), 60, None) pending[future] = candidate for future in as_completed(pending): finish(pending[future], future.result()) assert_unchanged(plan) summary = evaluate.summarize(plan["root"], 500, 60) session.update(status="saved_outputs_validated", recorded_prompts=summary["recorded_prompts"], evaluation_status=summary["status"]) evaluate.atomic_json(report, session) return summary except BaseException as error: session.update(status="interrupted_or_failed", failure_type=type(error).__name__) evaluate.atomic_json(report, session) raise def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--evaluation", type=Path, required=True) parser.add_argument("--prompts", type=Path, required=True) parser.add_argument("--godot", type=Path, required=True) args = parser.parse_args() # Do not run alongside evaluate.py or another process writing this directory. with recovery_lock(args.evaluation): plan = prepare(args.evaluation, args.prompts, args.godot) print(json.dumps({"saved_outputs_to_validate": len(plan["candidates"]), "generation_calls": 0}), flush=True) print(json.dumps(recover(plan, args.godot), indent=2)) if __name__ == "__main__": main()