| |
| """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 |
| generation_path = safe_path(root, item / "generation.json") |
| if not generation_path.exists(): |
| continue |
| 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) |
| |
| 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() |
| |
| 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() |
|
|
