| |
| """Freeze blind prompts and admit only validated whole-project SFT examples. |
| |
| Usage: |
| python scripts/build_dataset.py --freeze-eval-only |
| python scripts/build_dataset.py --input data/synthetic_candidates.jsonl --godot /path/to/godot |
| |
| The 500 blind prompts are not supplied to a teacher and are never used for loss. |
| The small development split has complete targets and is used for eval loss. |
| """ |
| from __future__ import annotations |
|
|
| import argparse |
| from concurrent.futures import ThreadPoolExecutor |
| import hashlib |
| import json |
| from pathlib import Path |
| import sys |
|
|
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) |
| from vibegame.data import Deduper, canonical_output, read_jsonl, sha256, write_jsonl |
|
|
| |
| |
| FAMILIES = [ |
| ("snake", "Steer a snake around a square grid with the arrow keys. Eating food extends its body; touching a wall or its own body ends the round."), |
| ("breakout", "Move a paddle horizontally to bounce a ball into a wall of bricks. Remove every brick to win, and lose a life when the ball falls below the paddle."), |
| ("pong", "Play a paddle duel against a simple computer opponent. Use up and down to defend your goal while a bouncing ball speeds up after each paddle hit."), |
| ("asteroids", "Pilot a small triangular spaceship with rotation, thrust, and a fire button. Destroy drifting asteroids that split into smaller pieces; wrap ships and rocks at screen edges."), |
| ("invaders", "Move a defender along the bottom and fire upward at descending rows of invaders. Enemies fire back, and the player loses if an enemy reaches the bottom."), |
| ("flap", "Press space to flap a flying circle upward against gravity. Fly through gaps in pairs of scrolling columns and score once per safely cleared gap."), |
| ("runner", "Build a side-scrolling runner whose character moves automatically. Space jumps over approaching blocks, and distance traveled increases the score."), |
| ("dodge", "Move a character in all four directions inside an arena. Survive falling geometric hazards, with collisions removing health and survival time determining score."), |
| ("maze", "Navigate a grid maze from an entrance to an exit with the arrow keys. Collect three keys before the exit opens, and prevent movement through maze walls."), |
| ("crate_puzzle", "Push crates one cell at a time through a small warehouse grid. Arrow keys move the worker; a crate cannot be pulled, and every marked goal must contain a crate to win."), |
| ("turret", "Rotate a stationary central turret and fire at enemies approaching from the arena boundary. Projectiles destroy enemies, while enemies that reach the turret damage its health."), |
| ("tower_defense", "Let the player place simple towers beside a fixed enemy path using the mouse. Towers automatically shoot nearby enemies; defeated enemies grant coins for more towers."), |
| ("match_three", "Create a grid of colored tiles that the player swaps with adjacent tiles. Remove horizontal or vertical lines of at least three matching colors, then let new tiles fall into gaps."), |
| ("memory", "Display face-down cards in a grid and let the player reveal two at a time with the mouse. Matching symbols stay revealed, while different symbols turn face-down after a short delay."), |
| ("minefield", "Create a small grid with hidden mines and safe tiles showing adjacent mine counts. Left-click reveals a tile, right-click toggles a flag, and revealing a mine loses the round."), |
| ("lights_out", "Create a grid of lights where clicking a cell toggles itself and its orthogonal neighbors. Start from a solvable scrambled arrangement and win when every light is off."), |
| ("sliding_tiles", "Arrange numbered tiles in a square grid with one empty cell. Clicking a tile adjacent to the empty cell slides it, and the player wins by restoring numerical order."), |
| ("rhythm", "Move notes downward in four lanes toward a hit line. The player presses four mapped keys at the right time; accurate hits score points and missed notes break the streak."), |
| ("typing", "Let short words fall from the top toward a danger line. The player types a displayed word and presses Enter to remove it; escaped words remove health."), |
| ("artillery", "Let the player adjust a cannon angle and launch a projectile affected by gravity. Place a target across the screen and award points when a shot lands on it."), |
| ("lander", "Pilot a lunar lander with left/right rotation and upward thrust while gravity pulls it down. Land slowly and upright on a marked platform; hard or tilted contact crashes the ship."), |
| ("pinball", "Bounce a ball around a geometric pinball table with bumpers and two player-controlled flippers. Score on bumper contact and lose a life when the ball exits below the flippers."), |
| ("rally", "Drive a top-down car around a closed track using steering and acceleration keys. Checkpoints must be visited in order before a lap counts, and driving off track slows the car."), |
| ("fishing", "Press a button to cast a line into a pond containing moving fish. When a fish bites, keep a marker inside a moving capture zone to reel it in and earn points."), |
| ("reaction", "Show several target circles and briefly light one at a time. The player clicks the lit target before it expires; clicking another circle or reacting too slowly counts as a miss."), |
| ] |
|
|
| MODIFIERS = [ |
| ("countdown", "The whole round has a visible 90-second countdown; reaching zero ends it and displays the final score."), |
| ("three_attempts", "Give the player exactly three attempts, show the remaining attempts, and restart the active stage after each failed attempt."), |
| ("streak_scoring", "Consecutive successful actions build a score multiplier up to five; any failure resets the multiplier to one and the HUD explains the change."), |
| ("three_stages", "Include three stages with a visible stage number. Each completed stage advances to a harder layout or faster timing, and finishing stage three wins."), |
| ("practice_mode", "Add a title-screen choice between practice and challenge. Practice permits unlimited retries, while challenge counts failures and reports them on completion."), |
| ("slow_motion", "Give the player one slow-motion charge per round. Holding Shift spends its three seconds of duration, and a meter shows the remaining charge."), |
| ("precision_bonus", "Track a precision percentage based on successful versus failed actions. Finishing above 80 percent earns an explicit score bonus shown in the result screen."), |
| ("checkpoint", "Save one in-memory checkpoint halfway through the objective. A failure after that point resumes at the checkpoint while earlier failures restart the round."), |
| ("risk_reward", "Add an optional harder objective worth twice the normal points. Make its active state clearly visible, and let the player toggle it with the B key before a round."), |
| ("speed_choice", "Offer slow, normal, and fast modes on a start screen. All mode-dependent motion or timers use the chosen speed consistently, and fast mode awards extra points."), |
| ("perfect_round", "Track whether the current round is flawless. Any failure removes the flawless badge; completing a round without one displays a perfect-round banner."), |
| ("endurance", "Include an endless mode that increases difficulty every 20 seconds. Display the current difficulty and end the run after five failures."), |
| ("tutorial", "Begin with a short interactive tutorial that waits for each required input before advancing. Allow Enter to skip it, and show the same controls during play."), |
| ("best_score", "Track the best score for the current application session. Show both current and best score during play and update the record immediately when it is exceeded."), |
| ("daily_seed", "Use a visible numeric seed for all randomized gameplay choices. Pressing N before a round changes the seed; restarting the same seed reproduces the initial state."), |
| ("score_attack", "Make score attack a selectable mode with a fixed target score and a countdown. Reaching the target wins immediately, and running out of time loses."), |
| ("limited_actions", "Offer an expert mode with a visible budget of 30 meaningful player actions. Reaching the game objective within that budget wins; exhausting it ends the attempt."), |
| ("comeback", "After two consecutive failures, temporarily give a clearly labeled assist that widens timing windows or reduces difficulty. A success removes the assist."), |
| ("round_series", "Play a series of five short rounds. Preserve cumulative score between rounds, show an intermission after each one, and display a results table after the fifth."), |
| ("accessibility", "Provide an on-screen toggle for a high-contrast palette and use labels or shapes as well as colors. Include a reduced-motion toggle that removes decorative animation."), |
| ] |
|
|
|
|
| def frozen_eval_rows() -> list[dict]: |
| rows = [] |
| for family, description in FAMILIES: |
| for mechanic, modifier in MODIFIERS: |
| instruction = (f"Build a complete playable 2D Godot 4 game. {description} {modifier} " |
| "Use original geometric visuals drawn in code, no external assets. Include visible controls, " |
| "an unambiguous win or loss state, pause on P, and restart on R. Return only a JSON object " |
| 'with a files mapping containing project.godot, main.tscn, and every required .gd file.') |
| rows.append({"id": f"vg8b-eval-v1-{family}-{mechanic}", "family": family, |
| "mechanic": mechanic, "instruction": instruction, "sha256": sha256(instruction)}) |
| assert len(rows) == 500 and len({row["sha256"] for row in rows}) == 500 |
| return rows |
|
|
|
|
| def freeze_eval(output: Path) -> list[dict]: |
| rows = frozen_eval_rows() |
| target = output / "eval_prompts.jsonl" |
| if target.exists() and list(read_jsonl(target)) != rows: |
| raise ValueError("Frozen eval file changed: use a new version/directory, never silently replace a benchmark") |
| write_jsonl(target, rows) |
| digest = hashlib.sha256(target.read_bytes()).hexdigest() |
| manifest = {"version": "vg8b-eval-v1", "count": 500, "sha256": digest, |
| "families": len(FAMILIES), "rule_modifiers": len(MODIFIERS), |
| "design": "25 core game interactions crossed with 20 explicit rule modifiers", |
| "source": "original prompts authored for this repository; templated, not independently human-curated", |
| "policy": "blind generation benchmark; never train on these prompts, generate teacher targets for them, or use them for checkpoint selection", |
| "limitations": "Compositional variants share vocabulary and mechanics. 500 rows do not imply 500 independent task families."} |
| (output / "eval_manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") |
| return rows |
|
|
|
|
| def split_records(rows: list[dict], dev_fraction: float = 0.1) -> tuple[list[dict], list[dict]]: |
| """Stable hash split after global prompt/output dedupe, never row duplication.""" |
| ranked = sorted(rows, key=lambda row: sha256("vibegame-dev-v1:" + row["instruction"])) |
| if len(ranked) < 2: |
| raise ValueError("At least two unique validated projects are required for train + dev") |
| count = max(1, min(len(ranked) - 1, round(len(ranked) * dev_fraction))) |
| return ranked[count:], ranked[:count] |
|
|
|
|
| def successful_cached_validation(directory: Path, files: dict, binary_sha: str, seconds: int) -> dict | None: |
| """Reuse only matching complete evidence whose original logs still hash.""" |
| from vibegame.project import project_sha256 |
| evidence = directory / "validation.json" |
| if not evidence.is_file(): |
| return None |
| try: |
| result = json.loads(evidence.read_text(encoding="utf-8")) |
| if not (result.get("compile_success") and result.get("clean_run_60s") |
| and result.get("godot_binary_sha256") == binary_sha |
| and result.get("project_sha256") == project_sha256(files) |
| and result.get("run_seconds", 0) >= seconds |
| and result.get("runtime", {}).get("wall_seconds", 0) >= seconds): |
| return None |
| for phase in ["import", "runtime"]: |
| item = result[phase] |
| log = directory / ("import.log" if phase == "import" else "runtime.log") |
| if not item.get("success") or hashlib.sha256(log.read_bytes()).hexdigest() != item["log_sha256"]: |
| return None |
| for name, source in files.items(): |
| if (directory / "project" / name).read_text(encoding="utf-8") != source: |
| return None |
| return {**result, "reused_evidence": str(evidence)} |
| except (OSError, ValueError, KeyError, TypeError): |
| return None |
|
|
|
|
| def main() -> None: |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--input", type=Path, action="append", help="Candidate instruction/output JSONL, repeatable") |
| parser.add_argument("--output", type=Path, default=Path("data")) |
| parser.add_argument("--godot", help="Godot 4 executable; required to admit training examples") |
| parser.add_argument("--run-seconds", type=int, default=60) |
| parser.add_argument("--workers", type=int, default=2) |
| parser.add_argument("--dev-fraction", type=float, default=0.1) |
| parser.add_argument("--freeze-eval-only", action="store_true") |
| args = parser.parse_args() |
| if not 0 < args.dev_fraction < 1: |
| parser.error("dev fraction must be between zero and one") |
| if args.run_seconds < 60: |
| parser.error("admitted synthetic pairs require a full 60-second run") |
| if args.workers < 1 or args.workers > 4: |
| parser.error("workers must be 1 to 4") |
| args.output.mkdir(parents=True, exist_ok=True) |
| evaluation = freeze_eval(args.output) |
| if args.freeze_eval_only: |
| print(f"Frozen {len(evaluation)} blind prompts in {args.output / 'eval_prompts.jsonl'}"); return |
| if not args.input or not args.godot: |
| parser.error("--input and --godot are required; no unvalidated training data can be written") |
| from vibegame.validate import validate_project |
| binary_path = Path(args.godot).expanduser().resolve(strict=True) |
| binary_sha = hashlib.sha256(binary_path.read_bytes()).hexdigest() |
| prompts = Deduper(threshold=0.78, width=3) |
| outputs = Deduper(threshold=0.85, width=5) |
| for row in evaluation: |
| prompts.add(row["id"], row["instruction"]) |
| candidates, rejected = [], [] |
| for path in args.input: |
| for number, row in enumerate(read_jsonl(path), 1): |
| identifier = f"{path}:{number}" |
| try: |
| if set(row) != {"instruction", "output"}: |
| raise ValueError("SFT schema must contain exactly instruction and output") |
| if not isinstance(row["instruction"], str) or len(row["instruction"].strip()) < 20: |
| raise ValueError("instruction must be a meaningful string") |
| canonical, files = canonical_output(row["output"]) |
| duplicate = prompts.match(row["instruction"]) or outputs.match(canonical) |
| if duplicate: |
| raise ValueError("exact/near duplicate of " + duplicate) |
| prompts.add(identifier, row["instruction"]); outputs.add(identifier, canonical) |
| candidates.append((identifier, {"instruction": row["instruction"], "output": canonical}, files)) |
| except (ValueError, TypeError) as exc: |
| rejected.append({"source": identifier, "reason": str(exc)}) |
| def check(item): |
| identifier, row, files = item |
| fingerprint = sha256(row["output"]) |
| work_base = args.output / "validation" |
| try: |
| for previous in sorted(work_base.glob(fingerprint[:16] + "*")): |
| cached = successful_cached_validation(previous, files, binary_sha, args.run_seconds) |
| if cached: |
| return identifier, row, cached, True |
| work_dir = work_base / fingerprint[:16] |
| attempt = 1 |
| while work_dir.exists(): |
| work_dir = work_base / f"{fingerprint[:16]}-attempt-{attempt:03d}" |
| attempt += 1 |
| result = validate_project(files, work_dir, str(binary_path), run_seconds=args.run_seconds) |
| passed = bool(result.get("compile_success") and result.get("clean_run_60s")) |
| return identifier, row, result, passed |
| except Exception as exc: |
| return identifier, row, {"error": type(exc).__name__ + ": " + str(exc)}, False |
| accepted, audits = [], [] |
| with ThreadPoolExecutor(max_workers=args.workers) as pool: |
| for identifier, row, result, passed in pool.map(check, candidates): |
| audits.append({"source": identifier, "instruction_sha256": sha256(row["instruction"]), |
| "output_sha256": sha256(row["output"]), "accepted": passed, "validation": result}) |
| print(f"{'PASS' if passed else 'DROP'} {identifier}", flush=True) |
| if passed: |
| accepted.append(row) |
| else: |
| rejected.append({"source": identifier, "reason": "compile_or_60s_run_failed"}) |
| write_jsonl(args.output / "validation_manifest.jsonl", audits) |
| write_jsonl(args.output / "rejected_candidates.jsonl", rejected) |
| train, dev = split_records(accepted, args.dev_fraction) |
| write_jsonl(args.output / "train.jsonl", train) |
| write_jsonl(args.output / "dev.jsonl", dev) |
| manifest = {"schema_version": 1, "train": len(train), "dev": len(dev), "blind_eval": len(evaluation), |
| "candidate_count_after_dedupe": len(candidates), "accepted": len(accepted), "rejected": len(rejected), |
| "training_schema": {"instruction": "string", "output": "JSON string with files mapping"}, |
| "dedupe": {"instruction_token_3gram_jaccard": 0.78, "output_token_5gram_jaccard": 0.85}, |
| "validation_run_seconds": args.run_seconds, "sources": [str(path) for path in args.input], |
| "files_sha256": {name: hashlib.sha256((args.output / name).read_bytes()).hexdigest() |
| for name in ["train.jsonl", "dev.jsonl", "eval_prompts.jsonl"]}, |
| "reference_corpus_trained_on": False, |
| "scope": "bootstrap synthetic data; quantity and diversity are not sufficient to claim a good game-generation model"} |
| (args.output / "dataset_manifest.json").write_text(json.dumps(manifest, indent=2) + "\n", encoding="utf-8") |
| print(json.dumps({"train": len(train), "dev": len(dev), "blind_eval": len(evaluation), "rejected": len(rejected)})) |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
