#!/usr/bin/env python3 """Audit and mechanically select the frozen Transformer P1 rates.""" import argparse import hashlib import json import math import os import sys ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) sys.path.insert(0, ROOT) from experiments.transformer_crossover_grid import p1_jobs, registry_sha256 TRAIN_HASH = ( "6ec305602a99ac2802745a134e1f5e33e2231b4855525b00b9aebb730ac2626f") VALIDATION_HASH = ( "d37d30cc0c8327c270d493299c3dca54135f6d5f1c9ef60cda78076e311204b1") EXPECTED_TRAIN_TOKENS = 1000 * 32 * 64 EXPECTED_VALIDATION_TOKENS = 1742 * 64 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 read_json(path): with open(path, encoding="utf-8") as handle: return json.load(handle) def audit_completed(job, manifest, expected_source): record = read_json(job["output"]) assert manifest["output_sha256"] == sha256(job["output"]) assert record["provenance"]["git_commit"] == expected_source["git_commit"] assert record["provenance"]["git_tracked_dirty"] is False args = record["args"] expected_args = { "method": job["method"], "depth": 4, "width": 128, "heads": 4, "mlp_ratio": 4, "context_length": 64, "batch_size": 32, "eval_batch_size": 32, "train_steps": 1000, "lr": job["rate"], "schedule": "constant", "weight_decay": 0.1, "run_seed": 0, "model_seed": 2027, "loader_seed": 0, "negative_seed": 5001, "ep_sign_seed": 5002, "traffic_ratio": 4.0, "eval_every": 0, "max_val_batches": 0, } for key, expected in expected_args.items(): assert args[key] == expected, ( f"{job['experiment_name']}: drift in {key}") assert record["protocol_family"] == ( "transformer_local_learning_crossover") assert record["dataset"]["train_sha256"] == TRAIN_HASH assert record["dataset"]["validation_sha256"] == VALIDATION_HASH assert record["architecture"]["forward_parameter_count"] == 813568 assert record["evaluation_protocol"]["split"] == "validation" assert record["evaluation_protocol"]["test_evaluations"] == 0 assert record["evaluation_protocol"]["test_used_for_selection"] is False assert record["first_nonfinite_step"] is None assert record["work"]["ordinary_training_tokens"] == ( EXPECTED_TRAIN_TOKENS) assert record["work"]["ordinary_validation_tokens"] == ( EXPECTED_VALIDATION_TOKENS) assert record["work"]["completed_optimizer_steps"] == 1000 assert len(record["validation"]) == 1 assert record["final"]["tokens"] == EXPECTED_VALIDATION_TOKENS assert record["final"]["step"] == 1000 assert record["hardware"]["peak_memory_allocated_bytes"] is not None assert record["hardware"]["peak_memory_reserved_bytes"] is not None nll = float(record["final"]["nll"]) finite = math.isfinite(nll) return { "finite": finite, "final_validation_nll": nll, "final_validation_perplexity": float(record["final"]["perplexity"]), "final_validation_accuracy": float(record["final"]["accuracy"]), "completed_optimizer_steps": 1000, "peak_memory_allocated_bytes": record["hardware"]["peak_memory_allocated_bytes"], "total_wall_seconds": float(record["total_wall_seconds"]), "work": record["work"], } def audit_job(job, expected_source): manifest_path = job["output"] + ".manifest.json" assert os.path.isfile(manifest_path), ( f"missing manifest for {job['experiment_name']}") manifest = read_json(manifest_path) for key in ( "stage", "method", "architecture", "rate", "experiment_name", "output", "timeout_seconds", "command"): assert manifest[key] == job[key], ( f"{job['experiment_name']}: drift in {key}") assert manifest["source"] == expected_source hardware = manifest["hardware_lock"] assert hardware["physical_gpu_index"] in (5, 7) assert hardware["physical_gpu_uuid"] status = manifest["status"] common = { "method": job["method"], "rate": job["rate"], "experiment_name": job["experiment_name"], "manifest": os.path.relpath(manifest_path, ROOT), "status": status, "driver_wall_seconds": manifest["driver_wall_seconds"], "physical_gpu_index": hardware["physical_gpu_index"], "physical_gpu_uuid": hardware["physical_gpu_uuid"], "output_sha256": manifest["output_sha256"], } if status == "completed": assert manifest["output_exists"] is True return { **common, **audit_completed(job, manifest, expected_source), } assert status in { "timeout", "nonzero_exit", "missing_output"} return { **common, "finite": False, "final_validation_nll": None, "final_validation_perplexity": None, "final_validation_accuracy": None, "completed_optimizer_steps": None, "peak_memory_allocated_bytes": None, "total_wall_seconds": None, "work": None, } def choose(candidates): return min(candidates, key=lambda row: ( not row["finite"], ( row["final_validation_nll"] if row["final_validation_nll"] is not None else math.inf), row["rate"], )) def main(): parser = argparse.ArgumentParser() parser.add_argument( "--out", default="results/transformer_crossover/p1_selector.json") parser.add_argument("--allow-partial", action="store_true") args = parser.parse_args() jobs = p1_jobs() manifest_paths = [ job["output"] + ".manifest.json" for job in jobs if os.path.isfile(job["output"] + ".manifest.json")] missing = [ job["experiment_name"] for job in jobs if not os.path.isfile(job["output"] + ".manifest.json")] if missing and not args.allow_partial: raise AssertionError(f"missing P1 jobs: {missing}") sources = [read_json(path)["source"] for path in manifest_paths] if sources: expected_source = sources[0] assert all(source == expected_source for source in sources) else: expected_source = None records = [ audit_job(job, expected_source) for job in jobs if os.path.isfile(job["output"] + ".manifest.json")] selected = {} launch = None if not missing: launch_path = os.path.join( ROOT, "results", "transformer_crossover", "p1_launch.json") assert os.path.isfile(launch_path), "missing Transformer P1 launch lock" launch = read_json(launch_path) assert launch["stage"] == "p1" assert launch["source"] == expected_source assert launch["registry_sha256"] == registry_sha256(jobs) assert launch["num_jobs"] == len(jobs) assert launch["allowed_physical_gpus"] == [5, 7] for method in dict((job["method"], None) for job in jobs): winner = choose([ record for record in records if record["method"] == method]) selected[method] = { "rate": winner["rate"], "finite": winner["finite"], "final_validation_nll": winner["final_validation_nll"], "experiment_name": winner["experiment_name"], } report = { "gate": "pass" if not missing else "partial", "stage": "transformer_crossover_p1", "selection_rule": "minimum finite final validation NLL, then lower learning rate; " "failed/timeout records rank after finite records", "source": expected_source, "registry_sha256": hashlib.sha256(json.dumps( jobs, sort_keys=True).encode("utf-8")).hexdigest(), "num_expected_records": len(jobs), "num_audited_records": len(records), "missing_experiments": missing, "test_policy": "none", "launch_lock": launch, "records": records, "selected": selected, } os.makedirs(os.path.dirname(os.path.abspath(args.out)), exist_ok=True) with open(args.out, "w", encoding="utf-8") as handle: json.dump(report, handle, indent=2, sort_keys=True) handle.write("\n") print(json.dumps({ "gate": report["gate"], "num_audited_records": len(records), "missing_experiments": missing, "selected": selected, }, indent=2, sort_keys=True)) if __name__ == "__main__": main()