experiment: freeze Transformer P1 selector grid

2 files changed, 485 insertions, 0 deletions

diff --git a/experiments/analyze_transformer_crossover_p1.py b/experiments/analyze_transformer_crossover_p1.py
new file mode 100644
index 0000000..ab4c846
--- /dev/null
+++ b/experiments/analyze_transformer_crossover_p1.py
@@ -0,0 +1,227 @@
+#!/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
+
+
+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 = {}
+    if not missing:
+        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",
+        "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()
diff --git a/experiments/transformer_crossover_grid.py b/experiments/transformer_crossover_grid.py
new file mode 100644
index 0000000..0fa8ba8
--- /dev/null
+++ b/experiments/transformer_crossover_grid.py
@@ -0,0 +1,258 @@
+#!/usr/bin/env python3
+"""Immutable Transformer selector registry and failure-continuing driver."""
+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")
+METHODS = (
+    "bp", "fa", "dfa", "pepita", "ff", "ep", "dualprop",
+    "clean_kp", "sdil",
+)
+RATE_GRIDS = {
+    "bp": (0.0003, 0.001, 0.003),
+    "fa": (0.0003, 0.001, 0.003),
+    "dfa": (0.0003, 0.001, 0.003),
+    "pepita": (0.0001, 0.0003, 0.001),
+    "ff": (0.0001, 0.0003, 0.001),
+    "ep": (0.0001, 0.0003, 0.001),
+    "dualprop": (0.0001, 0.0003, 0.001),
+    "clean_kp": (0.0003, 0.001, 0.003),
+    "sdil": (0.0003, 0.001, 0.003),
+}
+
+
+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 git_output(*args):
+    return subprocess.run(
+        ["git", *args], cwd=ROOT, check=True, capture_output=True,
+        text=True).stdout.strip()
+
+
+def source_report():
+    dirty = git_output(
+        "status", "--porcelain", "--untracked-files=no")
+    if dirty:
+        raise RuntimeError(
+            "Transformer crossover requires clean tracked source")
+    paths = [
+        PROTOCOL,
+        os.path.abspath(__file__),
+        os.path.join(
+            ROOT, "experiments", "analyze_transformer_crossover_p1.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"),
+    ]
+    for path in paths:
+        tracked = subprocess.run(
+            ["git", "ls-files", "--error-unmatch",
+             os.path.relpath(path, ROOT)],
+            cwd=ROOT, capture_output=True).returncode == 0
+        if not tracked:
+            raise RuntimeError(f"untracked crossover source: {path}")
+    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 rate_tag(rate):
+    return f"{rate:g}".replace(".", "p")
+
+
+def p1_command(method, rate):
+    name = f"transformer-p1-{method}-d4-lr{rate_tag(rate)}"
+    output = os.path.join(RESULT_ROOT, "p1", name + ".json")
+    command = [
+        sys.executable,
+        "experiments/transformer_crossover_native.py",
+        "--method", method,
+        "--out", output,
+        "--device", "cuda",
+        "--depth", "4",
+        "--width", "128",
+        "--heads", "4",
+        "--mlp_ratio", "4",
+        "--context_length", "64",
+        "--batch_size", "32",
+        "--eval_batch_size", "32",
+        "--train_steps", "1000",
+        "--lr", str(rate),
+        "--schedule", "constant",
+        "--min_lr", "0.0001",
+        "--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": "p1",
+        "method": method,
+        "architecture": "transformer4",
+        "rate": rate,
+        "experiment_name": name,
+        "output": output,
+        "timeout_seconds": 48 * 60 * 60,
+        "command": command,
+    }
+
+
+def p1_jobs():
+    # Interleave the expensive methods so two shards do not inherit all state
+    # jobs from contiguous method groups.
+    order = (
+        "ep", "dualprop", "ff", "pepita", "dfa", "fa",
+        "clean_kp", "sdil", "bp",
+    )
+    jobs = []
+    for grid_index in range(3):
+        for method in order:
+            jobs.append(p1_command(
+                method, RATE_GRIDS[method][grid_index]))
+    return jobs
+
+
+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(
+            "formal Transformer jobs require CUDA_VISIBLE_DEVICES=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, gpu, dry_run):
+    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 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,
+        "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("--stage", choices=("p1",), default="p1")
+    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("--dry-run", action="store_true")
+    args = parser.parse_args()
+    if not 0 <= args.shard_index < args.num_shards:
+        raise ValueError("invalid shard")
+    source = source_report()
+    gpu = physical_gpu_report(args.dry_run)
+    jobs = p1_jobs()
+    if args.method:
+        jobs = [job for job in jobs if job["method"] == args.method]
+    jobs = [
+        job for index, job in enumerate(jobs)
+        if index % args.num_shards == args.shard_index]
+    if not jobs:
+        raise ValueError("no jobs selected")
+    for job in jobs:
+        run_job(job, source, gpu, args.dry_run)
+
+
+if __name__ == "__main__":
+    main()