diff options
Diffstat (limited to 'experiments/transformer_crossover_t2.py')
| -rw-r--r-- | experiments/transformer_crossover_t2.py | 392 |
1 files changed, 392 insertions, 0 deletions
diff --git a/experiments/transformer_crossover_t2.py b/experiments/transformer_crossover_t2.py new file mode 100644 index 0000000..17f61a2 --- /dev/null +++ b/experiments/transformer_crossover_t2.py @@ -0,0 +1,392 @@ +#!/usr/bin/env python3 +"""Immutable driver for the complete 27-cell Transformer T2 crossover.""" +import argparse +import hashlib +import json +import os +import subprocess +import sys +import time + + +ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) +PROTOCOL = os.path.join(ROOT, "TRANSFORMER_CROSSOVER.md") +RESULT_ROOT = os.path.join(ROOT, "results", "transformer_crossover") +DEFAULT_SELECTOR = os.path.join(RESULT_ROOT, "p1_selector.json") +METHODS = ( + "bp", "fa", "dfa", "pepita", "ff", "ep", "dualprop", + "clean_kp", "sdil", +) +DEPTHS = (4, 8, 12) +ORDER = ( + ("ep", 12), + ("dualprop", 12), + ("ff", 12), + ("ep", 8), + ("dualprop", 8), + ("ff", 8), + ("ep", 4), + ("dualprop", 4), + ("ff", 4), + ("pepita", 12), + ("sdil", 12), + ("clean_kp", 12), + ("fa", 12), + ("dfa", 12), + ("bp", 12), + ("pepita", 8), + ("sdil", 8), + ("clean_kp", 8), + ("fa", 8), + ("dfa", 8), + ("bp", 8), + ("pepita", 4), + ("sdil", 4), + ("clean_kp", 4), + ("fa", 4), + ("dfa", 4), + ("bp", 4), +) + + +def sha256(path): + digest = hashlib.sha256() + with open(path, "rb") as handle: + for chunk in iter(lambda: handle.read(1024 * 1024), b""): + digest.update(chunk) + return digest.hexdigest() + + +def text_sha256(text): + return hashlib.sha256(text.encode("utf-8")).hexdigest() + + +def read_json(path): + with open(path, encoding="utf-8") as handle: + return json.load(handle) + + +def git_output(*args): + return subprocess.run( + ["git", *args], + cwd=ROOT, + check=True, + capture_output=True, + text=True, + ).stdout.strip() + + +def rate_tag(rate): + return f"{rate:g}".replace(".", "p") + + +def selector_report(path): + selector = read_json(path) + if not ( + selector.get("gate") == "pass" + and selector.get("stage") == "transformer_crossover_p1" + and selector.get("num_expected_records") == 27 + and selector.get("num_audited_records") == 27 + and selector.get("missing_experiments") == [] + and selector.get("test_policy") == "none" + and set(selector.get("selected", {})) == set(METHODS) + ): + raise RuntimeError( + "Transformer T2 requires the complete passed T1 selector" + ) + rates = {} + for method in METHODS: + rate = float(selector["selected"][method]["rate"]) + if not rate > 0: + raise RuntimeError(f"invalid selected Transformer rate: {method}") + rates[method] = rate + return { + "path": os.path.relpath(os.path.abspath(path), ROOT), + "sha256": sha256(path), + "p1_source": selector["source"], + "selected_rates": rates, + } + + +def source_report(selector_path): + if git_output("status", "--porcelain", "--untracked-files=no"): + raise RuntimeError("Transformer T2 requires clean tracked source") + paths = [ + PROTOCOL, + os.path.abspath(__file__), + os.path.join( + ROOT, "experiments", "analyze_transformer_crossover_t2.py" + ), + os.path.join( + ROOT, "experiments", "transformer_crossover_native.py" + ), + os.path.join( + ROOT, "experiments", "transformer_feedback_smoke.py" + ), + os.path.join(ROOT, "sdil", "transformer.py"), + os.path.abspath(selector_path), + ] + for path in paths: + relative = os.path.relpath(path, ROOT) + tracked = subprocess.run( + ["git", "ls-files", "--error-unmatch", relative], + cwd=ROOT, + capture_output=True, + ).returncode == 0 + if not tracked: + raise RuntimeError(f"untracked Transformer T2 source: {relative}") + freeze = subprocess.run( + [sys.executable, "-m", "pip", "freeze"], + check=True, + capture_output=True, + text=True, + ).stdout + return { + "git_commit": git_output("rev-parse", "HEAD"), + "protocol_path": PROTOCOL, + "protocol_sha256": sha256(PROTOCOL), + "tracked_files": { + os.path.relpath(path, ROOT): sha256(path) for path in paths + }, + "python": sys.executable, + "environment_freeze_sha256": text_sha256(freeze), + "environment_freeze": freeze.splitlines(), + } + + +def t2_command(method, depth, rate): + name = f"transformer-t2-{method}-d{depth}-lr{rate_tag(rate)}" + output = os.path.join(RESULT_ROOT, "t2", name + ".json") + command = [ + sys.executable, + "experiments/transformer_crossover_native.py", + "--method", method, + "--out", output, + "--device", "cuda", + "--depth", str(depth), + "--width", "128", + "--heads", "4", + "--mlp_ratio", "4", + "--context_length", "64", + "--batch_size", "32", + "--eval_batch_size", "32", + "--train_steps", "5000", + "--lr", str(rate), + "--schedule", "cosine", + "--min_lr", str(rate * 0.1), + "--warmup_steps", "100", + "--weight_decay", "0.1", + "--run_seed", "0", + "--model_seed", "2027", + "--loader_seed", "0", + "--negative_seed", "5001", + "--ep_sign_seed", "5002", + "--traffic_ratio", "4.0", + "--ff_threshold", "2.0", + "--ff_score_from_layer", "1", + "--ep_beta", "0.5", + "--ep_dt", "0.5", + "--ep_free_steps", "20", + "--ep_nudge_steps", "4", + "--dp_alpha", "0.0", + "--dp_beta", "0.1", + "--dp_inference_passes", "16", + "--eval_every", "0", + "--log_every", "50", + "--max_val_batches", "0", + ] + return { + "stage": "t2", + "method": method, + "architecture": f"transformer{depth}", + "depth": depth, + "rate": rate, + "experiment_name": name, + "output": output, + "timeout_seconds": 48 * 60 * 60, + "command": command, + } + + +def t2_jobs(selected_rates): + if set(selected_rates) != set(METHODS): + raise ValueError("T2 requires one selected rate for every method") + jobs = [ + t2_command(method, depth, float(selected_rates[method])) + for method, depth in ORDER + ] + identifiers = [job["experiment_name"] for job in jobs] + cells = {(job["method"], job["depth"]) for job in jobs} + if ( + len(jobs) != 27 + or len(set(identifiers)) != 27 + or cells != {(method, depth) for method in METHODS for depth in DEPTHS} + ): + raise AssertionError( + "Transformer T2 registry must contain 27 unique cells" + ) + return jobs + + +def registry_sha256(jobs): + encoded = json.dumps( + jobs, sort_keys=True, separators=(",", ":") + ).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() + + +def ensure_launch(source, selector, jobs): + path = os.path.join(RESULT_ROOT, "t2_launch.json") + expected = { + "stage": "t2", + "source": source, + "selector": selector, + "registry_sha256": registry_sha256(jobs), + "num_jobs": len(jobs), + "allowed_physical_gpus": [5, 7], + } + if os.path.isfile(path): + if read_json(path) != expected: + raise RuntimeError("Transformer T2 launch lock drift") + return path + os.makedirs(os.path.dirname(path), exist_ok=True) + with open(path, "w", encoding="utf-8") as handle: + json.dump(expected, handle, indent=2, sort_keys=True) + handle.write("\n") + return path + + +def assert_source_unchanged(source): + if git_output("status", "--porcelain", "--untracked-files=no"): + raise RuntimeError("tracked source changed after Transformer T2 launch") + if git_output("rev-parse", "HEAD") != source["git_commit"]: + raise RuntimeError("source commit changed after Transformer T2 launch") + for relative, expected_hash in source["tracked_files"].items(): + if sha256(os.path.join(ROOT, relative)) != expected_hash: + raise RuntimeError(f"Transformer T2 source drift: {relative}") + freeze = subprocess.run( + [sys.executable, "-m", "pip", "freeze"], + check=True, + capture_output=True, + text=True, + ).stdout + if text_sha256(freeze) != source["environment_freeze_sha256"]: + raise RuntimeError("Transformer T2 environment drift") + + +def physical_gpu_report(dry_run=False): + visible = os.environ.get("CUDA_VISIBLE_DEVICES") + if dry_run: + return { + "cuda_visible_devices": visible, + "physical_gpu_index": None, + "physical_gpu_uuid": None, + } + if visible not in {"5", "7"}: + raise RuntimeError("Transformer T2 requires physical GPU 5 or 7") + query = subprocess.run( + [ + "nvidia-smi", + "--query-gpu=index,uuid,name", + "--format=csv,noheader,nounits", + ], + check=True, + capture_output=True, + text=True, + ).stdout.splitlines() + rows = {} + for line in query: + index, uuid, name = [value.strip() for value in line.split(",", 2)] + rows[index] = {"uuid": uuid, "name": name} + if visible not in rows: + raise RuntimeError(f"physical GPU {visible} not found") + return { + "cuda_visible_devices": visible, + "physical_gpu_index": int(visible), + "physical_gpu_uuid": rows[visible]["uuid"], + "physical_gpu_name": rows[visible]["name"], + } + + +def run_job(job, source, selector, gpu, dry_run): + if not dry_run: + assert_source_unchanged(source) + if sha256(os.path.join(ROOT, selector["path"])) != selector["sha256"]: + raise RuntimeError("Transformer T2 selector changed after launch") + manifest_path = job["output"] + ".manifest.json" + if os.path.exists(manifest_path): + print(f"preserving {job['experiment_name']}", flush=True) + return + if os.path.exists(job["output"]): + raise RuntimeError(f"orphaned T2 output requires audit: {job['output']}") + print("RUN", " ".join(job["command"]), flush=True) + if dry_run: + return + os.makedirs(os.path.dirname(job["output"]), exist_ok=True) + started = time.time() + try: + result = subprocess.run( + job["command"], cwd=ROOT, timeout=job["timeout_seconds"] + ) + return_code = result.returncode + status = "completed" if return_code == 0 else "nonzero_exit" + except subprocess.TimeoutExpired: + return_code = None + status = "timeout" + output_exists = os.path.isfile(job["output"]) + if status == "completed" and not output_exists: + status = "missing_output" + manifest = { + **job, + "source": source, + "selector": selector, + "hardware_lock": gpu, + "status": status, + "return_code": return_code, + "output_exists": output_exists, + "output_sha256": sha256(job["output"]) if output_exists else None, + "driver_wall_seconds": time.time() - started, + "completed_unix_time": time.time(), + } + with open(manifest_path, "w", encoding="utf-8") as handle: + json.dump(manifest, handle, indent=2, sort_keys=True) + handle.write("\n") + print(f"DONE status={status} {job['experiment_name']}", flush=True) + + +def main(): + parser = argparse.ArgumentParser() + parser.add_argument("--selector", default=DEFAULT_SELECTOR) + parser.add_argument("--shard-index", type=int, default=0) + parser.add_argument("--num-shards", type=int, default=1) + parser.add_argument("--method", choices=METHODS) + parser.add_argument("--depth", type=int, choices=DEPTHS) + parser.add_argument("--dry-run", action="store_true") + args = parser.parse_args() + if not 0 <= args.shard_index < args.num_shards: + raise ValueError("invalid Transformer T2 shard") + selector = selector_report(args.selector) + source = source_report(args.selector) + gpu = physical_gpu_report(args.dry_run) + complete_jobs = t2_jobs(selector["selected_rates"]) + if not args.dry_run: + launch = ensure_launch(source, selector, complete_jobs) + print(f"T2 launch lock: {launch}", flush=True) + jobs = complete_jobs + if args.method: + jobs = [job for job in jobs if job["method"] == args.method] + if args.depth is not None: + jobs = [job for job in jobs if job["depth"] == args.depth] + jobs = [ + job + for index, job in enumerate(jobs) + if index % args.num_shards == args.shard_index + ] + if not jobs: + raise ValueError("no Transformer T2 jobs selected") + for job in jobs: + run_job(job, source, selector, gpu, args.dry_run) + + +if __name__ == "__main__": + main() |