path: root/experiments/contrastive_bias_c1.py

#!/usr/bin/env python3
"""Five-seed same-path full-schedule contrastive-bias confirmation."""
import argparse
import hashlib
import json
import math
import os
from pathlib import Path
import subprocess
import time

import numpy as np


ROOT = Path(__file__).resolve().parents[1]
PROTOCOL = ROOT / "CONTRASTIVE_BIAS_CONFIRMATION.md"
RESULT_ROOT = ROOT / "results" / "contrastive_bias" / "c1"
BIAS_PATCH = (
    ROOT / "external" / "dualprop_patches" /
    "0020-experiment-add-neuron-specific-bias-to-Dual-Prop.patch"
)
UPSTREAM = "7b2595b34421e1483a721dbfdeff8cdabda3a1ff"
SEEDS = (1989, 1990, 1991, 1992, 1993)
CONDITIONS = (
    ("same_path_clean", "common", "raw"),
    ("raw", "activity", "raw"),
    ("innovation", "activity", "innovation"),
    ("oracle", "activity", "oracle"),
)


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 git_output(repo, *args):
    return subprocess.run(
        ["git", *args], cwd=repo, check=True, capture_output=True, text=True
    ).stdout.strip()


def command_for(seed, condition, kind, rule, author_python):
    name = f"dp-bias-c1-s{seed}-{condition.replace('_', '-')}"
    command = [
        author_python, "train.py", "--model", "miniCNN", "--dataset", "cifar10",
        "--num-epochs", "130", "--batch-size", "100",
        "--learning-rate", "0.025", "--learning-rate-final", "2e-6",
        "--warmup-learning-rate", "0.001", "--warmup-epochs", "10",
        "--decay-epochs", "120", "--momentum", "0.9",
        "--weight-decay", "5e-4", "--dtype", "float32",
        "--param-dtype", "float32", "--percent-train", "90",
        "--percent-val", "10", "--seeds", str(seed),
        "--feedback-seed", "1729", "--gradient-diagnostics", "none",
        "--spectral-diagnostics", "none", "--test-policy", "none",
        "--early-stop-policy", "none", "--learning-algorithm",
        "dualprop-lagr-ff", "--experiment-name", name,
        "--optimizer-schedule", "author", "--loss", "sce",
        "--alpha", "0.0", "--beta", "0.1", "--inference-sequence", "fwK",
        "--inference-passes-nudged", "16", "--dp-bias-kind", kind,
        "--dp-bias-rule", rule, "--dp-bias-ratio", "4.0",
        "--dp-bias-seed", "6100", "--dp-bias-calibration-examples", "64",
    ]
    return name, command


def jobs(author_python):
    rows = []
    for seed in SEEDS:
        for condition, kind, rule in CONDITIONS:
            name, command = command_for(
                seed, condition, kind, rule, author_python)
            rows.append({
                "seed": seed, "condition": condition, "kind": kind,
                "rule": rule, "ratio": 4.0, "experiment_name": name,
                "command": command,
                "output": str(RESULT_ROOT / (name + ".json")),
                "timeout_seconds": 60 * 60,
            })
    if len(rows) != 20 or len({row["experiment_name"] for row in rows}) != 20:
        raise AssertionError("C1 registry must contain 20 unique cells")
    return rows


def registry_sha256(rows):
    payload = [
        {key: value for key, value in row.items() if key != "output"}
        for row in rows
    ]
    return hashlib.sha256(json.dumps(
        payload, sort_keys=True, separators=(",", ":")
    ).encode()).hexdigest()


def source_report(author_root):
    if git_output(ROOT, "status", "--porcelain", "--untracked-files=no"):
        raise RuntimeError("C1 requires clean tracked SDIL source")
    if git_output(author_root, "status", "--porcelain", "--untracked-files=no"):
        raise RuntimeError("C1 requires clean tracked author source")
    tracked = [
        PROTOCOL, Path(__file__).resolve(),
        ROOT / "experiments" / "analyze_contrastive_bias_c1.py", BIAS_PATCH,
    ]
    for path in tracked:
        subprocess.run([
            "git", "ls-files", "--error-unmatch", str(path.relative_to(ROOT))
        ], cwd=ROOT, check=True, capture_output=True)
    return {
        "sdil_commit": git_output(ROOT, "rev-parse", "HEAD"),
        "author_commit": git_output(author_root, "rev-parse", "HEAD"),
        "author_upstream": UPSTREAM,
        "tracked_files": {
            str(path.relative_to(ROOT)): sha256(path) for path in tracked
        },
    }


def gpu_report(physical_index):
    output = subprocess.run([
        "nvidia-smi", f"--id={physical_index}",
        "--query-gpu=index,uuid,name,memory.total", "--format=csv,noheader,nounits",
    ], check=True, capture_output=True, text=True).stdout.strip()
    fields = [part.strip() for part in output.split(",")]
    visible = os.environ.get("CUDA_VISIBLE_DEVICES")
    if len(fields) != 4 or fields[0] != str(physical_index):
        raise RuntimeError(f"could not resolve physical GPU {physical_index}: {output}")
    if visible != str(physical_index):
        raise RuntimeError(
            f"CUDA_VISIBLE_DEVICES must equal physical GPU {physical_index}, got {visible}")
    return {
        "physical_index": int(fields[0]), "uuid": fields[1], "name": fields[2],
        "memory_total_mib": int(fields[3]), "cuda_visible_devices": visible,
    }


def ensure_launch(source, rows):
    path = RESULT_ROOT / "launch.json"
    expected = {
        "stage": "contrastive_bias_c1", "source": source,
        "registry_sha256": registry_sha256(rows), "num_jobs": len(rows),
        "seeds": list(SEEDS), "conditions": [row[0] for row in CONDITIONS],
        "allowed_physical_gpus": [5, 7], "test_policy": "none",
    }
    if path.is_file():
        with open(path, encoding="utf-8") as handle:
            if json.load(handle) != expected:
                raise RuntimeError("C1 launch lock drift")
    else:
        path.parent.mkdir(parents=True, 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 summarize_hist(path):
    hist = np.load(path, allow_pickle=True).item()
    completed = int(hist["epochs_completed"])
    keys = (
        "val_loss", "val_accuracy", "train_loss", "train_accuracy",
        "train_time", "val_time", "raw_bias_clean_difference_rms_ratio",
        "post_bias_raw_bias_rms_ratio", "used_clean_difference_rms_ratio",
        "maximum_used_clean_difference_relative_error", "neutral_observations",
        "instruction_observations_for_predictor",
    )
    curves = {
        key: [float(value) for value in np.asarray(hist[key])[:completed]]
        for key in keys
    }
    finite = completed == 130 and all(
        math.isfinite(value)
        for key in ("val_loss", "val_accuracy", "train_loss")
        for value in curves[key]
    )
    return {
        "epochs_completed": completed, "finite": finite,
        "final_validation_accuracy": curves["val_accuracy"][-1],
        "best_validation_accuracy": float(hist["best_validation_accuracy"]),
        "best_epoch": int(hist["best_epoch"]),
        "test_accuracy": float(hist["test_accuracy"]),
        "dp_bias_initialization": hist.get("dp_bias_initialization"),
        "curves": curves,
    }


def find_hist(author_root, experiment_name):
    paths = list((author_root / "runs" / experiment_name).glob("*/hist.npy"))
    if len(paths) != 1:
        raise RuntimeError(
            f"expected one history for {experiment_name}, found {len(paths)}")
    return paths[0]


def run_job(job, author_root, source, registry_hash, gpu):
    output = Path(job["output"])
    if output.exists():
        print(f"preserving s{job['seed']} {job['condition']}", flush=True)
        return
    if git_output(ROOT, "rev-parse", "HEAD") != source["sdil_commit"]:
        raise RuntimeError("SDIL commit changed after C1 launch")
    if git_output(author_root, "rev-parse", "HEAD") != source["author_commit"]:
        raise RuntimeError("author commit changed after C1 launch")
    print("RUN", " ".join(job["command"]), flush=True)
    started = time.time()
    try:
        result = subprocess.run(
            job["command"], cwd=author_root, timeout=job["timeout_seconds"])
        return_code = result.returncode
        status = "completed" if return_code == 0 else "nonzero_exit"
    except subprocess.TimeoutExpired:
        return_code, status = None, "timeout"
    history = None
    history_path = None
    if status == "completed":
        try:
            resolved = find_hist(author_root, job["experiment_name"])
            history_path = str(resolved)
            history = summarize_hist(resolved)
        except Exception as error:
            status = "missing_or_invalid_history"
            history = {"error": repr(error)}
    record = {
        **job, "stage": "contrastive_bias_c1", "source": source,
        "registry_sha256": registry_hash, "hardware": gpu, "status": status,
        "return_code": return_code, "driver_wall_seconds": time.time() - started,
        "completed_unix_time": time.time(), "author_history": history_path,
        "history": history,
    }
    output.parent.mkdir(parents=True, exist_ok=True)
    with open(output, "w", encoding="utf-8") as handle:
        json.dump(record, handle, indent=2, sort_keys=True)
        handle.write("\n")
    print(f"DONE status={status} s{job['seed']} {job['condition']}", flush=True)


def main():
    parser = argparse.ArgumentParser()
    parser.add_argument("--author-root", type=Path, required=True)
    parser.add_argument("--author-python", required=True)
    parser.add_argument("--physical-gpu", type=int, choices=(5, 7), required=True)
    parser.add_argument("--seed-shard-index", type=int, choices=(0, 1), required=True)
    parser.add_argument("--dry-run", action="store_true")
    args = parser.parse_args()
    args.author_root = args.author_root.resolve()
    rows = jobs(args.author_python)
    selected_seeds = SEEDS[args.seed_shard_index::2]
    selected = [row for row in rows if row["seed"] in selected_seeds]
    if args.dry_run:
        for row in selected:
            print(row["seed"], row["condition"], " ".join(row["command"]))
        return
    source = source_report(args.author_root)
    gpu = gpu_report(args.physical_gpu)
    launch = ensure_launch(source, rows)
    print(f"C1 launch lock: {launch}", flush=True)
    registry_hash = registry_sha256(rows)
    for row in selected:
        run_job(row, args.author_root, source, registry_hash, gpu)


if __name__ == "__main__":
    main()