| |
| """Build a self-contained, output-free Colab notebook with embedded audited source.""" |
| from __future__ import annotations |
|
|
| import argparse |
| import ast |
| import base64 |
| import hashlib |
| import io |
| import json |
| import textwrap |
| import zipfile |
| from pathlib import Path |
|
|
| ROOT = Path(__file__).resolve().parents[1] |
|
|
|
|
| def make_bundle(): |
| paths = [] |
| for pattern in ["src/vibegame/*.py", "scripts/*.py", "requirements-colab.txt", "pyproject.toml", "LICENSE", "NOTICE", "README.md", |
| "data/*.json", "data/*.jsonl", "data/licenses/**/*", "examples/**/*.gd", "examples/**/*.tscn", "examples/**/project.godot", "docs/*.md"]: |
| paths.extend(p for p in ROOT.glob(pattern) if p.is_file()) |
| buffer = io.BytesIO() |
| with zipfile.ZipFile(buffer, "w", zipfile.ZIP_DEFLATED, compresslevel=9) as archive: |
| for path in sorted(set(paths)): |
| info = zipfile.ZipInfo(path.relative_to(ROOT).as_posix(), date_time=(1980, 1, 1, 0, 0, 0)) |
| info.create_system = 3 |
| info.external_attr = 0o100644 << 16 |
| info.compress_type = zipfile.ZIP_DEFLATED |
| archive.writestr(info, path.read_bytes(), compress_type=zipfile.ZIP_DEFLATED, compresslevel=9) |
| raw = buffer.getvalue() |
| return base64.b64encode(raw).decode(), hashlib.sha256(raw).hexdigest(), len(set(paths)) |
|
|
|
|
| def build(): |
| cells = [] |
| def md(text): |
| cells.append({"cell_type": "markdown", "metadata": {}, "source": textwrap.dedent(text).strip() + "\n"}) |
| def code(text): |
| source = textwrap.dedent(text).strip() + "\n" |
| ast.parse(source) |
| cells.append({"cell_type": "code", "execution_count": None, "outputs": [], "metadata": {}, "source": source}) |
| encoded, checksum, count = make_bundle() |
| md(''' |
| # VibeGame-8b |
| Project brand: **VibeGame-8B**. Actual base: **Qwen2.5-Coder-7B-Instruct (7.61B parameters)**. |
| This notebook contains its own source and original seed examples. It does not depend on an unpublished GitHub repository. |
| |
| In **Runtime → Change runtime type**, choose a **T4 GPU**. Training fits the free tier; paid compute is optional. |
| No cell provisions cloud machines, calls paid inference, or attaches billing. Python cannot inspect your Colab subscription tier. |
| A free GPU is not guaranteed, and sessions can end before 12 hours. Checkpoints and the final adapter go to Drive. |
| This run completed training on a free T4. After the free GPU quota ran out, its owner purchased Colab Pro. |
| A separate full 500-prompt evaluation ran on an A100; the earlier partial T4 results remain separate. |
| The A100 pass had **6/500 compile successes (1.2%) and 5/500 clean 60-second runs (1.0%)**. |
| Its median prompt-to-launch time among successful launches was **282.8 seconds**, and **none launched within 60 seconds**. |
| Those measurements apply to the 4-bit base plus adapter; merged-fp16 and GGUF were not separately benchmarked. |
| |
| The included original teacher candidates are a bootstrap experiment, not evidence of a good game-generation model. |
| Training only begins after Godot accepts examples. The 500 frozen prompts are never used for training or loss. |
| A development split measures loss. Blind evaluation allows up to 500 minutes of runtime checks, spread across four CPU workers with a measured GPU batch size, and can resume across sessions. |
| **The original goal of reliably generating playable games in 60 seconds was not achieved.** This is a research prototype, and clean headless execution does not prove playable controls or correct game mechanics. |
| ''') |
| md("## 1. Mount Drive and choose persistent paths") |
| code(''' |
| from google.colab import drive |
| drive.mount("/content/drive") |
| from pathlib import Path |
| import os, subprocess, sys, json, shutil |
| DRIVE = Path("/content/drive/MyDrive/VibeGame-8B") |
| RUN = DRIVE / "runs" / "teacher-v2" |
| DATA = DRIVE / "data" |
| EVAL = RUN / "evaluation" |
| EXPORT = Path("/content/vibegame-exports") |
| PROJECT = Path("/content/vibegame-8b") |
| for path in (DRIVE, RUN, DATA): |
| path.mkdir(parents=True, exist_ok=True) |
| # Optional: a public Hugging Face dataset repository containing these files. |
| # Leave empty to use the original seed candidates embedded in this notebook. |
| DATASET_REPO = "" |
| # The embedded Codex-teacher candidates are already generated. Optional |
| # expansion uses the local 14B teacher; zero avoids repeating its weak pilot. |
| LOCAL_TEACHER_ATTEMPTS = 0 |
| os.environ["HF_HUB_DISABLE_TELEMETRY"] = "1" |
| os.environ["WANDB_DISABLED"] = "true" |
| ''') |
| md(f"## 2. Restore the embedded project\n\nThis bundle contains {count} files and is checked before extraction.") |
| |
| code(f''' |
| import base64, hashlib, io, zipfile |
| BUNDLE = {encoded!r} |
| raw = base64.b64decode(BUNDLE) |
| assert hashlib.sha256(raw).hexdigest() == {checksum!r}, "Embedded project checksum mismatch" |
| PROJECT.mkdir(parents=True, exist_ok=True) |
| with zipfile.ZipFile(io.BytesIO(raw)) as archive: |
| for info in archive.infolist(): |
| destination = (PROJECT / info.filename).resolve() |
| if not destination.is_relative_to(PROJECT.resolve()): |
| raise ValueError("Unsafe archive path") |
| archive.extractall(PROJECT) |
| del BUNDLE, raw |
| os.chdir(PROJECT) |
| os.environ["PYTHONPATH"] = str(PROJECT / "src") + os.pathsep + os.environ.get("PYTHONPATH", "") |
| print("Restored", PROJECT) |
| ''') |
| md("""## 3. Install dependencies and the pinned Godot editor |
| |
| On a resumed run, setup restores selected package versions from saved evidence before installing. The original package snapshot is never overwritten; later setups create dated session snapshots. Recorded evaluation versions take priority after training has completed. CUDA wheels come from PyTorch's matching official index. |
| This restores the recorded software stack, not Google's GPU driver, OS, or every transitive library. Existing training/evaluation guards still refuse incompatible resumes. All model work runs in fresh subprocesses. |
| Godot's editor binary includes headless mode. The installer checks the official download's SHA-256. |
| """) |
| code(r''' |
| import re |
| from datetime import datetime, timezone |
| from packaging.version import Version |
| |
| def run(*args, env=None): |
| subprocess.run([str(a) for a in args], cwd=PROJECT, env=env, check=True) |
| |
| def resume_package_pins(run_dir, evaluation_dir): |
| # Restore a small runtime stack, never an arbitrary full pip-freeze file. |
| allowed = {"unsloth", "unsloth-zoo", "transformers", "trl", "datasets", "peft", |
| "bitsandbytes", "accelerate", "torch", "torchvision", "torchaudio", |
| "triton", "xformers", "numpy", "tokenizers", "huggingface-hub", "safetensors"} |
| pins, evidence = {}, {} |
| def add(values, source): |
| if not isinstance(values, dict): |
| raise RuntimeError("Invalid saved package-version mapping: " + source) |
| for raw_name, value in values.items(): |
| name = re.sub(r"[-_.]+", "-", raw_name).lower() |
| if name not in allowed: |
| continue |
| if not isinstance(value, str) or not re.fullmatch(r"[A-Za-z0-9.!+_-]+", value): |
| raise RuntimeError("Invalid saved package version for " + name) |
| Version(value) |
| pins[name], evidence[name] = value, source |
| original = next((run_dir / name for name in ("packages-original.txt", "packages.txt") |
| if (run_dir / name).is_file()), None) |
| if original is not None: |
| frozen = {} |
| for line in original.read_text().splitlines(): |
| match = re.fullmatch(r"([A-Za-z0-9_.-]+)==([^;\s]+)", line.strip()) |
| if match: |
| frozen[match[1]] = match[2] |
| add(frozen, original.name) |
| runtime_path = run_dir / "runtime-package-versions.json" |
| if runtime_path.is_file(): |
| recorded = json.loads(runtime_path.read_text()) |
| add(recorded.get("versions", recorded), runtime_path.name) |
| training_path = run_dir / "run_manifest.json" |
| if not training_path.is_file(): |
| training_path = run_dir / "adapter/run_manifest.json" |
| training_versions = {} |
| if training_path.is_file(): |
| training_versions = json.loads(training_path.read_text())["versions"] |
| add(training_versions, str(training_path.relative_to(run_dir))) |
| evaluation_path = evaluation_dir / "manifest.json" |
| if evaluation_path.is_file(): |
| generation = json.loads(evaluation_path.read_text())["generation"] |
| if generation.get("backend") != "unsloth": |
| raise RuntimeError("This Colab resume setup expects the saved Unsloth evaluation runtime") |
| evaluated = dict(generation["packages"]) |
| evaluated["torch"] = generation["torch_version"] |
| marker = run_dir / "adapter/TRAINING_COMPLETE.json" |
| completed = json.loads(marker.read_text()) if marker.is_file() else {} |
| if completed.get("status") != "complete": |
| conflicts = [name for name in evaluated if name in training_versions |
| and training_versions[name] != evaluated[name]] |
| if conflicts: |
| raise RuntimeError("Saved training and evaluation require conflicting versions: " + ", ".join(conflicts)) |
| # Finished training is reused, so the in-progress evaluation runtime |
| # takes priority for the versions that its own resume guard records. |
| add(evaluated, "evaluation/manifest.json") |
| extra_index = None |
| torch_version = pins.get("torch", "") |
| if "+" in torch_version: |
| build = torch_version.split("+", 1)[1] |
| if re.fullmatch(r"cu[0-9]{3}", build): |
| extra_index = "https://download.pytorch.org/whl/" + build |
| else: |
| raise RuntimeError("Saved Torch build needs a reviewed CUDA wheel source: " + torch_version) |
| return pins, evidence, extra_index |
| |
| def preserve_package_snapshot(run_dir, contents): |
| candidate = run_dir / "packages.txt" |
| stamp = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") |
| number = 0 |
| while True: |
| try: |
| with candidate.open("x") as handle: |
| handle.write(contents) |
| handle.flush() |
| os.fsync(handle.fileno()) |
| return candidate |
| except FileExistsError: |
| number += 1 |
| candidate = run_dir / f"packages-session-{stamp}-{number:03d}.txt" |
| |
| saved_pins, package_evidence, torch_index = resume_package_pins(RUN, EVAL) |
| install_args = [sys.executable, "-m", "pip", "install", "-r", "requirements-colab.txt"] |
| if saved_pins: |
| constraints = PROJECT / "resume-runtime-constraints.txt" |
| constraints.write_text("".join(name + "===" + value + "\n" for name, value in sorted(saved_pins.items()))) |
| install_args += ["--constraint", constraints] |
| # Constraints alone do not request absent optional stack packages. |
| install_args += [name + "===" + value for name, value in sorted(saved_pins.items())] |
| if torch_index: |
| install_args += ["--extra-index-url", torch_index] |
| print("Restoring", len(saved_pins), "saved runtime package versions.") |
| run(*install_args) |
| run(sys.executable, "-m", "pip", "install", "-e", ".") |
| snapshot = preserve_package_snapshot(RUN, subprocess.check_output([sys.executable, "-m", "pip", "freeze"], text=True)) |
| print("Preserved package snapshot:", snapshot.name) |
| if saved_pins: |
| # Check in a fresh process; this notebook kernel may retain old imports. |
| probe = "import importlib.metadata as m,json,sys; print(json.dumps({n:m.version(n) for n in json.loads(sys.argv[1])}))" |
| restored = json.loads(subprocess.check_output([sys.executable, "-c", probe, json.dumps(sorted(saved_pins))], text=True)) |
| mismatch = {name: (version, restored.get(name)) for name, version in saved_pins.items() if restored.get(name) != version} |
| if mismatch: |
| raise RuntimeError("Installed package versions do not match saved evidence: " + str(mismatch)) |
| run("apt-get", "update", "-qq") |
| run("apt-get", "install", "-y", "-qq", "bubblewrap", "xvfb", "ffmpeg", "fonts-dejavu-core", "libgl1", "libgl1-mesa-dri", "libxkbcommon0", "cmake", "build-essential", "git") |
| run(sys.executable, "scripts/install_godot.py", "--destination", "/content/vibegame-godot") |
| GODOT = Path("/content/vibegame-godot/Godot_v4.5.1-stable_linux.x86_64") |
| assert GODOT.exists() |
| ''') |
| md("## 4. Check the GPU, Drive, sandbox, and disk before training") |
| code(''' |
| import torch |
| assert torch.cuda.is_available(), "Select a T4 GPU runtime" |
| assert torch.cuda.device_count() == 1 and "T4" in torch.cuda.get_device_name(0), "This notebook only permits one Tesla T4" |
| assert RUN.resolve().is_relative_to(Path("/content/drive/MyDrive").resolve()) |
| probe = RUN / ".drive-check" |
| probe.write_text("ok") |
| probe.unlink() |
| from vibegame.validate import sandbox_command |
| preflight = EVAL / ".sandbox-preflight" |
| preflight.mkdir(parents=True, exist_ok=True) |
| wrapped, sandbox = sandbox_command([str(GODOT)], preflight) |
| wrapped[-1] = "/usr/bin/true" |
| subprocess.run(wrapped, check=True, timeout=15) |
| print("Sandbox:", sandbox) |
| print("GPU:", torch.cuda.get_device_name(0)) |
| print("Drive checkpoints:", RUN) |
| print("Local free GiB:", round(shutil.disk_usage("/content").free / 1024**3, 1)) |
| print("No paid runtime or inference service is created by this notebook.") |
| ''') |
| md(''' |
| ## 5. Generate with a stronger local teacher, then validate data |
| Supply a larger reviewed candidate file at `MyDrive/VibeGame-8B/data/synthetic_candidates.jsonl` before this cell to expand the bootstrap set. |
| Existing Drive files take priority. References from permissive repositories remain a separate corpus; they are not falsely presented as complete projects. |
| The default uses the embedded original candidates authored by the in-session Codex teacher. Their provenance is preserved. The initial 14B pilot made basic Godot syntax errors; those rejected attempts do not become training examples. |
| Set `LOCAL_TEACHER_ATTEMPTS` above to expand with Qwen2.5-Coder-14B-Instruct in 4-bit on this free T4. It never calls a paid teacher API. |
| The 14B teacher writes gameplay scripts; original fixed project and scene scaffolding complete each candidate. Additional in-session Codex teacher batches are embedded with their provenance. |
| Teacher attempts, raw responses, license references, and accepted pairs persist to Drive. Before training starts, rerunning resumes unattempted prompts. Once a training manifest exists, this cell checks the exact train/dev/blind/admission hashes and skips all downloads, generation, and rebuilding. A mismatch stops without changing evidence. Optional pilot outputs are included only when `LOCAL_TEACHER_ATTEMPTS > 0`. |
| Every candidate must import and run cleanly for 60 seconds in the sandbox. Failed candidates are excluded. No seed-only final training is silently substituted when the teacher fails. |
| The default teacher sample is an initial research dataset; published metrics decide what claims it supports. |
| ''') |
| code(''' |
| def file_sha256(path): |
| digest = hashlib.sha256() |
| with Path(path).open("rb") as handle: |
| for block in iter(lambda: handle.read(8 * 1024 * 1024), b""): |
| digest.update(block) |
| return digest.hexdigest() |
| |
| def verify_run_dataset(): |
| manifest_path = RUN / "run_manifest.json" |
| manifest = json.loads(manifest_path.read_text()) |
| signature = manifest["signature"] |
| required = {"train_sha256": "train.jsonl", "dev_sha256": "dev.jsonl", |
| "blind_sha256": "eval_prompts.jsonl", |
| "validation_manifest_sha256": "validation_manifest.jsonl"} |
| for field, filename in required.items(): |
| path = DATA / filename |
| if not path.is_file() or signature.get(field) != file_sha256(path): |
| raise RuntimeError("Saved training data changed or is missing: " + filename + |
| ". Restore the original Drive data before resuming this run.") |
| return manifest |
| |
| if (RUN / "run_manifest.json").exists(): |
| verify_run_dataset() |
| print("Reusing the exact admitted dataset from this training run; admission evidence is unchanged.") |
| else: |
| if DATASET_REPO: |
| from huggingface_hub import snapshot_download |
| snapshot_download(DATASET_REPO, repo_type="dataset", local_dir=DATA, |
| allow_patterns=["*.jsonl", "*.json", "licenses/**", "data/*.jsonl", "data/*.json", "data/licenses/**"]) |
| if (DATA / "data" / "synthetic_candidates.jsonl").exists(): |
| for p in (DATA / "data").iterdir(): |
| target = DATA / p.name |
| if p.is_file(): |
| shutil.copy2(p, target) |
| for p in (PROJECT / "data").iterdir(): |
| target = DATA / p.name |
| if p.is_file() and p.suffix in {".json", ".jsonl"} and not target.exists(): |
| shutil.copy2(p, target) |
| elif p.is_dir() and p.name == "licenses" and not target.exists(): |
| shutil.copytree(p, target) |
| TEACHER = DRIVE / "teacher-v2" |
| admitted = TEACHER / "validated_candidates.jsonl" |
| if LOCAL_TEACHER_ATTEMPTS: |
| run(sys.executable, "scripts/synthesize.py", "--prompts", DATA / "teacher_prompts.jsonl", |
| "--blind", DATA / "eval_prompts.jsonl", "--corpus", DATA / "reference_gd.jsonl", |
| "--output", TEACHER, "--godot", GODOT, "--limit", LOCAL_TEACHER_ATTEMPTS, "--session-hours", "10.5") |
| attempts = [json.loads(line) for line in (TEACHER / "attempts.jsonl").read_text().splitlines() if line.strip()] |
| if len(attempts) < LOCAL_TEACHER_ATTEMPTS: |
| raise RuntimeError("Teacher session budget reached. Reconnect a free T4 and rerun to finish the configured attempts.") |
| shutil.copy2(TEACHER / "manifest.json", DATA / "teacher_run_manifest.json") |
| shutil.copy2(TEACHER / "attempts.jsonl", DATA / "teacher_attempts.jsonl") |
| authored_batches = sorted(DATA.glob("codex_teacher_batch_*.jsonl")) |
| if not (LOCAL_TEACHER_ATTEMPTS > 0 and admitted.exists() and admitted.read_text().strip()) and not authored_batches: |
| raise RuntimeError("No teacher projects passed validation. Inspect retained teacher attempts; no seed-only final training.") |
| inputs = [DATA / "synthetic_candidates.jsonl"] + authored_batches |
| if LOCAL_TEACHER_ATTEMPTS > 0 and admitted.exists() and admitted.read_text().strip(): |
| inputs.append(admitted) |
| input_args = [item for source in inputs for item in ("--input", source)] |
| run(sys.executable, "scripts/build_dataset.py", *input_args, |
| "--output", DATA, "--godot", GODOT, "--workers", "4", "--run-seconds", "60") |
| manifest = json.loads((DATA / "dataset_manifest.json").read_text()) |
| print(json.dumps(manifest, indent=2)) |
| assert manifest["train"] > 0 and manifest["dev"] > 0 |
| accepted = [json.loads(line) for line in (DATA / "validation_manifest.jsonl").read_text().splitlines() if line.strip()] |
| assert any(row.get("accepted") and ("codex_teacher_batch_" in row["source"] or "validated_candidates.jsonl" in row["source"]) for row in accepted), "No expansion candidates passed; refusing a silently seed-only final run" |
| ''') |
| md(''' |
| ## 6. Train or resume QLoRA |
| NF4, fp16 compute, rank 32, alpha 64, attention + MLP adapters, 4096 tokens, batch 1 × accumulation 16, LR 2e-4, cosine schedule, two epochs. |
| Whole projects longer than the context are rejected and logged; they are never silently truncated. |
| Assistant-only loss is measured on the development split. All optimizer, scheduler, RNG, adapter, and manifest checkpoint files are saved to Drive every 100 updates and at each epoch. |
| The latest checksum-verified complete checkpoint resumes automatically. A completed final adapter is checked against its training manifest, unchanged dataset, and final checkpoint weights, then reused without retraining. A 10.5-hour timer stops at an optimizer step so a new free session can continue. |
| ''') |
| code(''' |
| ADAPTER = RUN / "adapter" |
| |
| def verify_completed_training(): |
| manifest = verify_run_dataset() |
| saved = json.loads((ADAPTER / "run_manifest.json").read_text()) |
| if saved["signature"] != manifest["signature"]: |
| raise RuntimeError("Final adapter does not belong to this training/data manifest") |
| from vibegame import BASE_MODEL, SYSTEM_PROMPT |
| signature = saved["signature"] |
| if signature["base_model"] != BASE_MODEL or signature["system_prompt_sha256"] != hashlib.sha256(SYSTEM_PROMPT.encode()).hexdigest(): |
| raise RuntimeError("Current base or system prompt differs from the completed training run") |
| complete = json.loads((ADAPTER / "TRAINING_COMPLETE.json").read_text()) |
| step, target = complete.get("global_step", 0), complete.get("target_steps", 0) |
| if complete.get("status") != "complete" or target <= 0 or step != target: |
| raise RuntimeError("Final adapter training marker is incomplete") |
| checkpoint = RUN / f"checkpoint-{step}" |
| receipt = json.loads((checkpoint / "COMPLETE.json").read_text()) |
| weight = ADAPTER / "adapter_model.safetensors" |
| checkpoint_weight = checkpoint / weight.name |
| expected = receipt["sha256"].get(weight.name) |
| if receipt.get("global_step") != step or not expected or not checkpoint_weight.is_file() or file_sha256(checkpoint_weight) != expected: |
| raise RuntimeError("Final training checkpoint weight checksum is missing or invalid") |
| if not weight.is_file(): |
| raise RuntimeError("Final adapter weights are missing") |
| if file_sha256(weight) != expected: |
| # Serializers may change header ordering while preserving all tensors. |
| # Compare actual saved values against the intact completed checkpoint. |
| from safetensors import safe_open |
| with safe_open(weight, framework="pt", device="cpu") as final, safe_open(checkpoint_weight, framework="pt", device="cpu") as prior: |
| if not final.keys() or set(final.keys()) != set(prior.keys()): |
| raise RuntimeError("Final adapter tensors differ from its completed checkpoint") |
| for name in final.keys(): |
| actual, original = final.get_tensor(name), prior.get_tensor(name) |
| if actual.dtype != original.dtype or actual.shape != original.shape or not torch.equal(actual, original): |
| raise RuntimeError("Final adapter weight differs from its completed checkpoint: " + name) |
| return complete |
| |
| if (ADAPTER / "TRAINING_COMPLETE.json").exists(): |
| metrics = verify_completed_training() |
| print("Verified completed training; reusing the final adapter without retraining.") |
| else: |
| run(sys.executable, "scripts/train.py", "--train", DATA / "train.jsonl", "--dev", DATA / "dev.jsonl", |
| "--blind-prompts", DATA / "eval_prompts.jsonl", "--output", RUN, |
| "--max-seq-length", "4096", "--epochs", "2", "--save-steps", "100", "--session-hours", "10.5") |
| metrics = verify_completed_training() |
| print(json.dumps(metrics, indent=2)) |
| ''') |
| md(''' |
| ## 7. Measure this adapter's safe generation batch |
| This capacity check uses four teacher prompts, never blind prompts. It tries batch 32, then 16, 8, and 4 as needed, with the same 3072-token generation cap used for evaluation. |
| Only a completed report for the exact adapter, runtime packages, and training prompts is reused. Incomplete or mismatched reports remain on Drive, and new attempts receive numbered filenames. |
| Every size must pass both the short and full-cap checks below the 12.5GB memory threshold, including the conservative full-context KV-cache estimate with 2GB GPU reserve. EOS may end earlier, so this is a measured capacity check with a memory estimate rather than a forced worst-case allocation test. |
| Once blind evaluation has started, its original batch size remains fixed across disconnects. Capacity throughput is not a gameplay result or a single-prompt latency claim. |
| ''') |
| code(''' |
| import importlib.metadata |
| sys.path.insert(0, str(PROJECT / "scripts")) |
| from generate import directory_fingerprint |
| verify_completed_training() |
| adapter_fingerprint = directory_fingerprint(ADAPTER) |
| adapter_signature = json.loads((ADAPTER / "run_manifest.json").read_text())["signature"] |
| capacity_versions = {name: importlib.metadata.version(name) for name in ["torch", "unsloth", "transformers", "bitsandbytes"]} |
| teacher_rows = [json.loads(line) for line in (PROJECT / "data/teacher_prompts.jsonl").read_text().splitlines() if line.strip()][:4] |
| capacity_prompt_hash = hashlib.sha256(json.dumps([row["instruction"] for row in teacher_rows], ensure_ascii=False).encode()).hexdigest() |
| EVAL_BATCH_SIZE = 4 |
| |
| def capacity_report_matches(report, size): |
| generation = report.get("generation", {}) |
| return (report.get("status") == "complete" |
| and report.get("memory_estimate_policy") == "kv_full_context_v1" |
| and report.get("batch_sizes") == [size] |
| and report.get("kind", "").startswith("trained_adapter_") |
| and report.get("blind_prompts_read") is False |
| and report.get("full_max_new_tokens") == 3072 |
| and 0 < report.get("memory_limit_bytes", 0) <= 12_500_000_000 |
| and report.get("training_prompts_sha256") == capacity_prompt_hash |
| and report.get("versions") == capacity_versions |
| and report.get("gpu") == torch.cuda.get_device_name(0) |
| and report.get("base_model") == adapter_signature["base_model"] |
| and report.get("base_revision") == adapter_signature["base_revision"] |
| and generation.get("system_prompt_sha256") == adapter_signature["system_prompt_sha256"] |
| and generation.get("adapter", {}).get("sha256") == adapter_fingerprint["sha256"]) |
| |
| def safe_capacity_recommendation(report, size): |
| full = report.get("full_check") or {} |
| short = next((row for row in report.get("measurements", []) if row.get("batch_size") == size), {}) |
| limit = report.get("memory_limit_bytes", 0) |
| def safe_estimate(row): |
| estimate = row.get("full_context_memory_estimate") or {} |
| effective = estimate.get("effective_limit_bytes", 0) |
| return (estimate.get("within_limit") is True |
| and estimate.get("assessed_batch_size") == size |
| and estimate.get("target_context_tokens") == 4096 |
| and estimate.get("cache_growth_allowance_tokens", 0) >= 512 |
| and estimate.get("reserve_bytes", 0) >= 2_000_000_000 |
| and 0 < estimate.get("estimated_peak_bytes", 0) <= effective |
| and effective <= min(limit, report.get("gpu_memory_bytes", 0) - 2_000_000_000)) |
| return (capacity_report_matches(report, size) |
| and report.get("recommended_batch_size") == size |
| and short.get("status") == "success" |
| and full.get("status") == "success" and full.get("batch_size") == size |
| and full.get("max_new_tokens") == 3072 and full.get("within_memory_limit") is True |
| and safe_estimate(short) and safe_estimate(full) |
| and all(0 < row.get(key, limit) < limit for row in (short, full) |
| for key in ("peak_allocated_bytes", "peak_reserved_bytes"))) |
| |
| def next_capacity_path(): |
| candidate = RUN / "adapter-capacity.json" |
| number = 0 |
| while candidate.exists(): |
| number += 1 |
| candidate = RUN / f"adapter-capacity-{number:03d}.json" |
| return candidate |
| |
| if (EVAL / "manifest.json").exists(): |
| protocol = json.loads((EVAL / "manifest.json").read_text()) |
| if (protocol.get("generation", {}).get("adapter", {}).get("sha256") != adapter_fingerprint["sha256"] |
| or protocol.get("expected_count") != 500 or protocol.get("max_new_tokens") != 3072 |
| or protocol.get("run_seconds") != 60 or protocol.get("smoke_test") is not False |
| or protocol.get("prompts_sha256") != file_sha256(DATA / "eval_prompts.jsonl")): |
| raise RuntimeError("Existing blind evaluation belongs to a different adapter or protocol") |
| EVAL_BATCH_SIZE = protocol["batch_size"] |
| if EVAL_BATCH_SIZE not in (1, 2, 4, 8, 16, 32): |
| raise RuntimeError("Invalid persisted evaluation batch size") |
| print("Keeping the existing blind evaluation batch:", EVAL_BATCH_SIZE) |
| else: |
| if any((EVAL / "items").glob("*/attempt.json")): |
| raise RuntimeError("Blind attempt evidence exists without its protocol manifest; restore that manifest before resuming") |
| for size in (32, 16, 8, 4): |
| matching = [] |
| for path in sorted(RUN.glob("adapter-capacity*.json")): |
| try: |
| report = json.loads(path.read_text()) |
| except (ValueError, OSError): |
| continue |
| if capacity_report_matches(report, size): |
| matching.append((path, report)) |
| reusable = next(((path, report) for path, report in matching if safe_capacity_recommendation(report, size)), None) |
| if reusable is None and matching: |
| reusable = matching[-1] |
| if reusable is not None: |
| capacity_path, report = reusable |
| print("Reusing completed capacity evidence:", capacity_path.name) |
| else: |
| capacity_path = next_capacity_path() |
| try: |
| run(sys.executable, "scripts/benchmark_batch.py", capacity_path, |
| "--adapter", ADAPTER, "--batch-sizes", size) |
| except subprocess.CalledProcessError: |
| print("Capacity process did not finish for batch", size, "; its report is preserved.") |
| continue |
| report = json.loads(capacity_path.read_text()) |
| if safe_capacity_recommendation(report, size): |
| EVAL_BATCH_SIZE = size |
| print("Selected measured adapter batch:", EVAL_BATCH_SIZE, "from", capacity_path.name) |
| break |
| print("Batch", size, "did not pass both memory checks; checking a smaller size.") |
| else: |
| raise RuntimeError("No batch passed the adapter capacity checks. Reports are preserved; blind evaluation has not started.") |
| ''') |
| md(''' |
| ## 8. Evaluate all 500 frozen prompts |
| Each prompt receives one generation. Parse errors, timeouts, sandbox denials, and script errors count as failures. |
| This cell reuses completed evaluation records after a disconnect; it does not regenerate unsuccessful cases. |
| Keep this adapter fixed until evaluation ends. Do not tune against the blind prompts. |
| ''') |
| code(''' |
| ADAPTER = RUN / "adapter" |
| assert (ADAPTER / "TRAINING_COMPLETE.json").exists() |
| run(sys.executable, "scripts/evaluate.py", "--prompts", DATA / "eval_prompts.jsonl", "--output", EVAL, |
| "--backend", "unsloth", "--adapter", ADAPTER, "--base-model", "unsloth/Qwen2.5-Coder-7B-Instruct-bnb-4bit", |
| "--godot", GODOT, "--run-seconds", "60", "--expected-count", "500", "--max-new-tokens", "3072", "--batch-size", EVAL_BATCH_SIZE, "--workers", "4") |
| print((EVAL / "summary.json").read_text()) |
| ''') |
| md(''' |
| ## 9. Record the fixed samples, or opt in to a success showcase |
| The default remains the original sample indices 0, 200, and 400, including failure cards. Set `DEMO_MODE = "showcase"` before running this cell to select the separate display policy: the first three clean results in frozen prompt order, only after all 500 results are complete. That policy was declared after the A100 experiment began; it changes no benchmark scores or fixed sample slots. All three original captures are required; failed captures cannot be replaced with later successes. The showcase labels these selected successes as unrepresentative, and keeps full benchmark counts and actual timing visible. Sixty seconds is the edited video's duration, not generation time. |
| A clean headless run does not prove human playability. Review these recordings and controls before making a gameplay claim. |
| ''') |
| code(''' |
| DEMO_MODE = globals().get("DEMO_MODE", "fixed") |
| assert DEMO_MODE in ("showcase", "fixed") |
| demo_options = [] |
| indices = (0, 200, 400) |
| recording_dir = EVAL / "recordings" |
| if DEMO_MODE == "showcase": |
| run(sys.executable, "scripts/showcase.py", "--evaluation", EVAL) |
| selection_path = EVAL / "showcase" / "selection.json" |
| selection = json.loads(selection_path.read_text()) |
| indices = selection["showcase_indices"] |
| recording_dir = EVAL / "showcase" / "recordings" |
| demo_options = ["--showcase-selection", selection_path] |
| print(selection["disclosure"], indices, selection["metrics"]) |
| for index in indices: |
| result = json.loads((EVAL / "items" / f"{index:04d}" / "result.json").read_text()) |
| if not result.get("clean_run_60s"): |
| print(f"Predeclared sample {index} failed; the video will show its failure card.") |
| continue |
| video = recording_dir / f"sample-{index:03d}.mp4" |
| if not video.exists(): |
| run(sys.executable, "scripts/record_game.py", "--evaluation", EVAL, |
| "--index", index, "--godot", GODOT, "--output", video, "--seconds", "15", "--fps", "30", *demo_options) |
| run(sys.executable, "scripts/build_demo.py", "--evaluation", EVAL, "--output", RUN / "VibeGame-launch-60s.mp4", *demo_options) |
| from IPython.display import Video, display |
| display(Video(str(RUN / "VibeGame-launch-60s.mp4"), embed=True)) |
| ''') |
| md(''' |
| ## 10. Export fp16 and GGUF Q4_K_M |
| The training/evaluation subprocesses have exited, freeing GPU and CPU memory. |
| Merge reads one original-base tensor at a time and computes LoRA deltas in small row chunks. GGUF conversion uses lazy loading and temporary files; quantization runs on CPU. |
| The fp16 model alone is about 15.2GB and does not fit in a default empty 15GB Drive alongside checkpoints. |
| Exports therefore use ephemeral VM disk, with resource checks. The disposable original-base cache is removed after merge. Keep about 34GiB free for each export stage. |
| **Publish or download exports before the VM disconnects.** Adapter/checkpoints and evaluation remain on Drive, allowing reconstruction. |
| This path is implemented but must be measured on your allocated free runtime; a successful small test does not establish full-model memory use. |
| ''') |
| code(''' |
| run(sys.executable, "scripts/export_model.py", "--adapter", RUN / "adapter", "--output", EXPORT, |
| "--cache", "/content/vibegame-base-cache", "--workspace", "/content/vibegame-export-work") |
| print("Merged model:", EXPORT / "merged-fp16") |
| print("Ollama GGUF:", EXPORT / "gguf") |
| ''') |
| md(''' |
| ## 11. Publish artifacts, with inference services off |
| This final cell asks for a Hugging Face write token using a hidden input. It never writes the token to a file or notebook and never logs in to a persistent credential store. |
| It uploads only after actual training, export, and complete evaluation evidence exist. No paid endpoint, billing attachment, or inference request is created. |
| The publisher checks repository provider mappings. Account-wide payment guarantees require inspecting actual account settings; an empty provider mapping alone cannot prove an account can never be billed. |
| ''') |
| code(''' |
| from getpass import getpass |
| NAMESPACE = input("Hugging Face username or organization: ").strip() |
| assert NAMESPACE and "/" not in NAMESPACE |
| token = getpass("Hugging Face write token (hidden, not saved): ") |
| assert token |
| env = dict(os.environ) |
| env["HF_TOKEN"] = token |
| try: |
| run(sys.executable, "scripts/publish.py", "--adapter", RUN / "adapter", |
| "--merged", EXPORT / "merged-fp16", "--gguf", EXPORT / "gguf", "--data", DATA, |
| "--evaluation", EVAL / "summary.json", "--namespace", NAMESPACE, |
| "--output", RUN / "publish-report.json", env=env) |
| finally: |
| env.pop("HF_TOKEN", None) |
| token = None |
| print((RUN / "publish-report.json").read_text()) |
| ''') |
| md(''' |
| ## Run locally after publication |
| Download the GGUF and its `Modelfile`, then run `ollama create vibegame-8b -f Modelfile` and the repository's `scripts/generate.py --backend ollama` command. |
| Use the supplied extraction/validation harness before opening a generated project in Godot. A published GGUF needs its own evaluation; adapter scores must not be presented as GGUF scores. |
| |
| Primary references: [Unsloth Qwen model](https://huggingface.co/unsloth/Qwen2.5-Coder-7B-Instruct-bnb-4bit), |
| [Unsloth Colab install](https://unsloth.ai/docs/get-started/install/google-colab), |
| [Colab resource limits](https://research.google.com/colaboratory/faq.html), |
| [llama.cpp conversion](https://github.com/ggml-org/llama.cpp/blob/master/convert_hf_to_gguf.py). |
| ''') |
| for i, cell in enumerate(cells): |
| cell["id"] = f"vibegame-{i:03d}" |
| return {"nbformat": 4, "nbformat_minor": 5, "metadata": {"accelerator": "GPU", "colab": {"name": "VibeGame-8b.ipynb", "provenance": [], "gpuType": "T4"}, |
| "kernelspec": {"display_name": "Python 3", "language": "python", "name": "python3"}, "language_info": {"name": "python", "version": "3.11.0"}}, "cells": cells} |
|
|
|
|
| def main(): |
| parser = argparse.ArgumentParser(description=__doc__) |
| parser.add_argument("--output", type=Path, default=ROOT / "notebooks/VibeGame-8b.ipynb") |
| args = parser.parse_args() |
| notebook = build() |
| args.output.parent.mkdir(parents=True, exist_ok=True) |
| args.output.write_text(json.dumps(notebook, indent=1) + "\n") |
| try: |
| import nbformat |
| nbformat.validate(nbformat.read(args.output, as_version=4)) |
| except ImportError: |
| pass |
| print(f"Wrote {args.output}: {len(notebook['cells'])} cells, every code cell parses as Python") |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
