Blame for code/offload/dspark/d2/train/agreement.py

"""Fixed fast/reference gates. Private captures receive no optimizer updates.

Reference and fast arms run in separate processes to fit a single 3090. Both
consume identical materialized inputs, initialization, and synthetic targets.
"""
from __future__ import annotations

import argparse
import json
import os
from pathlib import Path
from typing import Any, Callable, cast

import torch
from torch import Tensor

from .loss import distillation_loss
from .model import load
from .perf import synthetic_targets
from .data import sha256
from .replay import capture_inputs, load_capture_set, v1_import
from .safety import assert_gpu_lease
from .fast_policy import BACKEND, tensor_groups


def prepare(a: argparse.Namespace) -> None:
    a.out.mkdir(parents=True, exist_ok=False, mode=0o700)
    rows: list[dict[str, Any]] = []
    for p in sorted(a.fixtures.glob("n1-*.pt")):
        fixture = torch.load(p, map_location="cpu", weights_only=False)
        rows.append({"key": p.stem, "population": "n1", "inputs": fixture["inputs"]})
    if len(rows) != 4:
        raise ValueError("require four N1 windows")
    v1_import(a.v1_root)
    cs = load_capture_set(a.captures)
    by_key = {s.key: s for s in cs.samples}
    parity = json.loads((a.reference_dumps / "parity.json").read_text())
    if parity["pass"] is not True or parity["samples"] != 50:
        raise ValueError("require the qualified 50-round reference population")
    for row in parity["rows"]:
        key = row["key"]
        inp, _ = capture_inputs(cs, by_key[key], "cpu")
        rows.append({"key": key, "population": "captures", "inputs": inp})
    for row in rows:
        torch.save(row, a.out / (row["key"] + ".pt"))
    manifest = {"schema": "dsv41.d2.fast-inputs.v1", "samples": len(rows),
                "files": {p.name: sha256(p) for p in sorted(a.out.glob("*.pt"))},
                "reference_population_sha256": sha256(a.reference_dumps / "parity.json")}
    (a.out / "inputs.json").write_text(json.dumps(manifest, indent=2) + "\n")


def arm(a: argparse.Namespace) -> None:
    assert_gpu_lease()
    torch.set_num_threads(8)
    torch.manual_seed(23)
    model = load(a.ckpt, "cuda", rank=a.rank, reference_math=a.command == "reference")
    if a.command == "fast":
        model.enable_fast_training()
    model.train()
    manifest = json.loads((a.inputs / "inputs.json").read_text())
    a.out.mkdir(parents=True, exist_ok=False, mode=0o700)
    for population in ("n1", "captures"):
        model.zero_grad(set_to_none=True)
        members = []
        for name, digest in manifest["files"].items():
            p = a.inputs / name
            if sha256(p) != digest:
                raise ValueError(f"input identity changed: {name}")
            data = torch.load(p, map_location="cpu", weights_only=False)
            if data["population"] == population:
                members.append(data)
        for row in members:
            inp = {k: v.to(model.activation_device) for k, v in row["inputs"].items()}
            h, hx = model.hidden_from(inp)
            target = synthetic_targets(model, inp)
            loss, _ = distillation_loss(model, h, chunk=a.chunk, **target)
            # Also exercise the separately trained confidence tensor. This
            # explicit auxiliary objective is used only for gradient comparison;
            # the loss-value gate below uses unmodified distillation_loss.
            conf = (model.confidence(hx.detach(), target["previous"]).sigmoid() - .5).square().mean()
            cast(Callable[[], None], ((loss + .1 * conf) / len(members)).backward)()
            with torch.no_grad():
                picks, _ = model.greedy(h, hx, inp["anchor"])
                previous = torch.cat((inp["anchor"][:, None], picks[:, :-1]), -1)
                if a.command == "fast":
                    ref_row = torch.load(a.reference / (row["key"] + ".pt"), map_location="cpu", weights_only=False)
                    if ref_row["inputs_sha256"] != sha256(a.inputs / (row["key"] + ".pt")):
                        raise ValueError("reference sample differs from fast input")
                    previous = ref_row["previous"].to(model.activation_device)
                logits = torch.cat([model.logit_chunk(h, previous, lo, min(lo + 4096, model.c.vocab)).cpu()
                                    for lo in range(0, model.c.vocab, 4096)], -1)
            torch.save({"drafts": picks.cpu(), "hidden": h.detach().cpu(), "previous": previous.cpu(),
                        "logits": logits, "loss": float(loss.detach()), "population": population,
                        "inputs_sha256": sha256(a.inputs / (row["key"] + ".pt"))},
                       a.out / (row["key"] + ".pt"))
            print(json.dumps({"arm": a.command, "population": population, "sample": row["key"],
                              "loss": float(loss.detach())}), flush=True)
        grads = {n: None if p.grad is None else p.grad.detach().cpu() for n, p in model.named_parameters()}
        torch.save(grads, a.out / (population + ".gradients.pt"))
    summary = {"schema": "dsv41.d2.fast-arm.v1", "arm": a.command, "vocab_chunk": a.chunk, "lora_rank": a.rank,
               "backend": BACKEND if model.fast_training else "qualified-reference",
               "frozen_identity": model.frozen_identity, "inputs_sha256": sha256(a.inputs / "inputs.json"),
               "source_sha256": {p.name: sha256(p) for p in sorted(Path(__file__).parent.glob("*.py"))},
               "gradient_objective": "mean synthetic top20+tail distillation + 0.1 confidence MSE(target=0.5); no updates",
               "zero_lora_a_gradients": "Initial LoRA B is zero; zero/zero gradients are reported explicitly.",
               "peak_reserved_bytes": torch.cuda.max_memory_reserved()}
    (a.out / "arm.json").write_text(json.dumps(summary, indent=2) + "\n")


def cosine(a: Tensor | None, b: Tensor | None) -> dict[str, float | bool]:
    if a is None or b is None:
        return {"pass": False, "cosine": 0., "missing_gradient": True}
    x, y = a.double().flatten(), b.double().flatten()
    nx, ny = x.norm(), y.norm()
    if nx == 0 or ny == 0:
        equal = bool(nx == 0 and ny == 0)
        return {"pass": equal, "cosine": float(equal), "both_zero": equal}
    c = float(x.dot(y) / (nx * ny))
    return {"pass": c >= .999, "cosine": c, "reference_norm": float(nx), "fast_norm": float(ny)}


def population_gates(rows: list[dict[str, Any]], gradients: dict[str, dict[str, float | bool]]) -> dict[str, Any]:
    if not rows or not gradients:
        raise ValueError("empty agreement population or gradient groups")
    kl = torch.tensor([r["kl"] for r in rows], dtype=torch.float64)
    medians = kl.quantile(.5, dim=0).tolist()
    p99 = kl.quantile(.99, dim=0).tolist()
    pos1 = sum(r["position1"] for r in rows) / len(rows)
    exact = sum(r["exact"] for r in rows) / len(rows)
    loss_error = max(r["loss_relative_error"] for r in rows)
    gates = {"drafts": pos1 >= .99 and exact >= .95,
             "kl": max(medians) <= 1e-3 and max(p99) <= 1e-2,
             "loss": loss_error <= .005,
             "gradients": all(g["pass"] and not g.get("both_zero", False) for g in gradients.values())}
    return {"pass": all(gates.values()), "gates": gates, "samples": len(rows),
            "position1_agreement": pos1, "exact_blocks": exact, "kl_median_by_position": medians,
            "kl_p99_by_position": p99, "max_loss_relative_error": loss_error,
            "gradient_groups": gradients, "rows": rows}


def compare(a: argparse.Namespace) -> None:
    torch.set_num_threads(8)
    reference = json.loads((a.reference / "arm.json").read_text())
    fast = json.loads((a.fast / "arm.json").read_text())
    for field in ("frozen_identity", "inputs_sha256", "gradient_objective", "vocab_chunk"):
        if reference[field] != fast[field]:
            raise ValueError(f"agreement arm identity differs: {field}")
    by_pop: dict[str, list[dict[str, Any]]] = {"n1": [], "captures": []}
    for p in sorted(a.reference.glob("*.pt")):
        if p.name.endswith(".gradients.pt"):
            continue
        ref = torch.load(p, map_location="cpu", weights_only=False)
        arm_data = torch.load(a.fast / p.name, map_location="cpu", weights_only=False)
        if ref["inputs_sha256"] != arm_data["inputs_sha256"]:
            raise ValueError("sample identity differs")
        if not torch.equal(ref["previous"], arm_data["previous"]):
            raise ValueError("logit KL conditionals differ")
        logp = ref["logits"].double().log_softmax(-1)
        logq = arm_data["logits"].double().log_softmax(-1)
        kl = (logp.exp() * (logp - logq)).sum(-1).squeeze(0)
        row = {"key": p.stem, "position1": bool(ref["drafts"][0, 0] == arm_data["drafts"][0, 0]),
               "exact": bool(torch.equal(ref["drafts"], arm_data["drafts"])), "kl": kl.tolist(),
               "loss_relative_error": abs(arm_data["loss"] - ref["loss"]) / max(abs(ref["loss"]), 1e-30)}
        by_pop[ref["population"]].append(row)
    populations = {}
    inferred_rank: int | None = None
    for population, rows in by_pop.items():
        rg = torch.load(a.reference / (population + ".gradients.pt"), map_location="cpu", weights_only=False)
        fg = torch.load(a.fast / (population + ".gradients.pt"), map_location="cpu", weights_only=False)
        if rg.keys() != fg.keys():
            raise ValueError("gradient tensor names differ")
        ranks = {rg[n].shape[0] for n in rg if n.endswith(".a")}
        if len(ranks) != 1 or any(rg[n].shape != fg[n].shape for n in rg if rg[n] is not None and fg[n] is not None):
            raise ValueError("gradient tensor shapes/ranks differ")
        rank = next(iter(ranks))
        if inferred_rank is not None and inferred_rank != rank:
            raise ValueError("gradient populations have different LoRA ranks")
        inferred_rank = rank
        if any(arm_meta.get("lora_rank", rank) != rank for arm_meta in (reference, fast)):
            raise ValueError("declared LoRA rank differs from measured gradients")
        groups = tensor_groups(set(rg))
        gradients = {}
        for name, members in groups.items():
            if any(rg[n] is None or fg[n] is None for n in members):
                gradients[name] = cosine(None, None)
            else:
                gradients[name] = cosine(torch.cat([rg[n].flatten() for n in members]),
                                         torch.cat([fg[n].flatten() for n in members]))
        populations[population] = population_gates(rows, gradients)
        populations[population]["gradient_tensors"] = {n: cosine(rg[n], fg[n]) for n in rg}
        populations[population]["gradient_group_members"] = groups
        individual_pass = all(g["pass"] for g in populations[population]["gradient_tensors"].values())
        populations[population]["gates"]["gradients"] &= individual_pass
        populations[population]["pass"] = all(populations[population]["gates"].values())
        del rg, fg
    report = {"schema": "dsv41.d2.fast-agreement.v1", "backend": BACKEND,
              "frozen_identity": reference["frozen_identity"], "inputs_sha256": reference["inputs_sha256"],
              "source_sha256": fast["source_sha256"], "populations": populations,
              "pass": all(p["pass"] for p in populations.values()),
              "reference_arm_sha256": sha256(a.reference / "arm.json"),
              "fast_arm_sha256": sha256(a.fast / "arm.json"), "batch_size": 1,
              "vocab_chunk": reference["vocab_chunk"],
              "lora_rank": inferred_rank,
              "no_optimizer_updates": True, "gradient_objective": reference["gradient_objective"]}
    a.out.write_text(json.dumps(report, indent=2) + "\n")
    print(json.dumps({"pass": report["pass"], "populations": {n: {k: v for k, v in p.items()
          if k not in ("rows", "gradient_groups", "gradient_tensors", "gradient_group_members")}
          for n, p in populations.items()}}), flush=True)


def main() -> None:
    os.umask(0o077)
    ap = argparse.ArgumentParser(__doc__)
    sub = ap.add_subparsers(dest="command", required=True)
    prep = sub.add_parser("prepare")
    prep.add_argument("--fixtures", type=Path, required=True)
    prep.add_argument("--reference-dumps", type=Path, required=True)
    prep.add_argument("--v1-root", required=True)
    prep.add_argument("--captures", required=True)
    prep.add_argument("--out", type=Path, required=True)
    for command in ("reference", "fast"):
        run = sub.add_parser(command)
        run.add_argument("--ckpt", required=True)
        run.add_argument("--inputs", type=Path, required=True)
        run.add_argument("--out", type=Path, required=True)
        run.add_argument("--chunk", type=int, default=4096)
        run.add_argument("--rank", type=int, choices=(32, 64), default=32)
        if command == "fast":
            run.add_argument("--reference", type=Path, required=True)
    cmp = sub.add_parser("compare")
    cmp.add_argument("--reference", type=Path, required=True)
    cmp.add_argument("--fast", type=Path, required=True)
    cmp.add_argument("--out", type=Path, required=True)
    a = ap.parse_args()
    if a.command == "prepare":
        prepare(a)
    elif a.command == "compare":
        compare(a)
    else:
        arm(a)


if __name__ == "__main__":
    main()