code/scripts/validate_saved.py

#!/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()