"""Offline experiment catalog and argv compiler. No torch, network or GPU calls. Unsupported paper cells are first-class blocked entries, never fake commands. All enabled commands use existing validated entry points and unique outputs. """ from __future__ import annotations import hashlib import json from pathlib import Path MODELS = ("ouro_1_4b", "ouro_2_6b", "loopformer_3x8", "parcae_370m") METHODS = ("bf16", "symmetric", "smoothquant", "quarot", "spinquant", "flatquant", "loopq") FIG6_SIZES = (32, 64, 128, 256, 512, 1024, 2048) def load_profile(path): profile = json.loads(Path(path).read_text()) if profile.get("schema_version") != 1 or profile.get("exact_author_protocol") is not False: raise ValueError("profile must explicitly disclose independent reproduction") seeds = profile["seeds"] if not seeds or len(seeds) != len(set(seeds)) or any(type(s) is not int or s < 0 for s in seeds): raise ValueError("seeds must be distinct nonnegative integers") tasks = profile["evaluation"] if set(tasks) != {"hellaswag", "winogrande", "lambada_openai", "arc_challenge", "mmlu", "wikitext"}: raise ValueError("all six evaluation tasks are required") if any(type(t["fewshot"]) is not int or t["fewshot"] < 0 for t in tasks.values()) or tasks["wikitext"]["fewshot"] != 0: raise ValueError("invalid few-shot protocol") import math if any(not math.isfinite(profile["parity"][k]) or profile["parity"][k] < 0 for k in ("max_logprob_error", "max_hidden_error")): raise ValueError("finite nonnegative parity thresholds required") return profile def profile_digest(profile): return hashlib.sha256(json.dumps(profile, sort_keys=True).encode()).hexdigest() def catalog(profile): jobs = {} def add(identifier, kind, *, family, dependencies=(), blocked=(), **details): if identifier in jobs: raise ValueError(f"duplicate experiment {identifier}") jobs[identifier] = dict(id=identifier, kind=kind, family=family, dependencies=list(dependencies), blockers=list(blocked), **details) return identifier bfparity = add("gate_bf16_prefill", "parity", family="implementation", bits=16, seed=0) for bits in (4, 8): add(f"gate_smoke_a{bits}", "smoke", family="implementation", bits=bits, seed=0, dependencies=(bfparity,)) def variant(bits, seed, *, ablation=None, samples=None, budget=4): tag = f"a{bits}" + (f"_{ablation}" if ablation else "") tag += f"_n{samples}" if samples is not None else "" tag += f"_b{budget}" if budget != 4 else "" tag += f"_s{seed}" cal, parity = "cal_" + tag, "parity_" + tag if cal not in jobs: details = dict(bits=bits, seed=seed, ablation=ablation, samples=samples, budget=0 if ablation == "no_slt" else budget) add(cal, "calibration", family="calibration", dependencies=("gate_smoke_a4", "gate_smoke_a8"), **details) add(parity, "parity", family="implementation", dependencies=(cal,), calibration=cal, **details) return cal, parity for model in MODELS: for method in METHODS: for bits in ((16,) if method == "bf16" else (4, 8)): for seed in profile["seeds"]: blockers = [] if model != "ouro_1_4b": blockers.append("missing validated backbone/calibration/export/runtime adapter") if method not in {"bf16", "loopq"}: blockers.append("missing paper-matched baseline integration; direct RTN is not a substitute") deps, cal = [bfparity, "gate_smoke_a4", "gate_smoke_a8"], None if not blockers and method == "loopq": cal, parity = variant(bits, seed) deps = [parity] add(f"table1_{model}_{method}_a{bits}_s{seed}", "evaluation", family="table1", dependencies=deps if not blockers else (), blocked=blockers, model=model, method=method, bits=bits, seed=seed, calibration=cal, tasks=list(profile["evaluation"])) for bits in (4, 8): for ablation in ("no_las", "no_slt", "no_cta"): for seed in profile["seeds"]: cal, parity = variant(bits, seed, ablation=ablation) add(f"table2_{ablation}_a{bits}_s{seed}", "evaluation", family="table2", dependencies=(parity,), model="ouro_1_4b", method="loopq", ablation=ablation, bits=bits, seed=seed, calibration=cal, tasks=list(profile["evaluation"])) for axis, values in (("budget", (0, 2, 4, 8)), ("samples", FIG6_SIZES)): for value in values: for seed in profile["seeds"]: kw = {axis: value} cal, parity = variant(4, seed, **kw) add(f"local_{axis}_{value}_s{seed}", "evaluation", family="local_sensitivity", dependencies=(parity,), bits=4, seed=seed, method="loopq", model="ouro_1_4b", study=(axis == "samples" or value != 4), calibration=cal, tasks=["lambada_openai", "wikitext"], author_identity_unverified=True) for bits in (4, 8): for seed in profile["seeds"]: add(f"control_direct_a{bits}_s{seed}", "evaluation", family="local_control", dependencies=(bfparity, "gate_smoke_a4", "gate_smoke_a8"), bits=bits, seed=seed, method="direct", tasks=list(profile["evaluation"]), calibration=None) fixture = add("local_lambada_fixture", "fixture", family="local_analysis", dependencies=(f"table1_ouro_1_4b_bf16_a16_s{profile['seeds'][0]}",)) for seed in profile["seeds"]: cal, parity = variant(4, seed) add(f"local_trajectory_s{seed}", "trajectory", family="local_analysis", dependencies=(parity, fixture), calibration=cal, seed=seed, bits=4) add(f"local_selection_s{seed}", "selection", family="local_analysis", dependencies=(cal,), calibration=cal, seed=seed, bits=4) # Explicit paper analysis cells; no hidden reduction to the Ouro-only subset. add("figure1", "analysis", family="figure1", blocked=("P99/transition fixture and reduction protocol not frozen",)) for model in ("ouro_1_4b", "parcae_370m"): for method in ("symmetric", "smoothquant", "quarot", "flatquant", "loopq"): add(f"figure2_{model}_{method}", "analysis", family="figure2", model=model, method=method, blocked=("requires frozen LAMBADA input fixture; Ouro LoopQ measurement script exists", *(() if model == "ouro_1_4b" and method == "loopq" else ("missing model/baseline instrumentation",)))) add("figure3", "analysis", family="figure3", blocked=("scan logs available; author Score/Cost denominator unspecified",)) add("figure4", "analysis", family="figure4", blocked=("local budget sweep configured; exact author model/protocol identity unverified",)) add("figure6", "analysis", family="figure6", blocked=("local sample sweep configured; exact author model/protocol identity unverified",)) for method in ("bf16", "smoothquant", "quarot", "flatquant", "loopq"): for loops in range(1, 17): add(f"figure5_{method}_loops{loops}", "analysis", family="figure5", method=method, loops=loops, bits=16 if method == "bf16" else 4, tasks=["lambada_openai", "wikitext"], blocked=("Parcae adapter missing; 8-loop calibration and LAS/SLT/CTA extrapolation policy unresolved",)) for model in MODELS: for method in ("bf16", "loopq"): add(f"table5_{model}_{method}", "memory", family="table5", model=model, method=method, blocked=("requires validated deployment loader and measured resident/peak memory; analytic payload is not loaded memory",)) for gate in ("decode_kv", "batch_variable_length", "long_context", "real_resume", "packed_runtime"): add(f"gate_{gate}", "validation", family="implementation", blocked=("real-model acceptance procedure still required on Pod",)) for job in jobs.values(): if any(dep not in jobs for dep in job["dependencies"]): raise ValueError("missing dependency") return jobs def compile_commands(job, profile, *, data_root, gpu, python="python"): if job["blockers"]: raise ValueError("blocked experiment: " + "; ".join(job["blockers"])) if gpu not in ("0", "1"): raise ValueError("only physical GPU 0/1 permitted") root = Path(data_root).resolve() if root == Path("/"): raise ValueError("data root cannot be filesystem root") out = root / job["id"] def script(name, *args): return [python, f"loopQ/scripts/{name}.py", *map(str, args)] def parity(artifact, destination, *, advisory=False): args = ["--model", "ouro", "--gpu", gpu, "--prompts-file", "loopQ/configs/parity_prompts.json", "--max-logprob-error", profile["parity"]["max_logprob_error"], "--max-hidden-error", profile["parity"]["max_hidden_error"], "--output", destination] if artifact is not None: args += ["--artifact", artifact] if advisory: args += ["--acceptance-advisory"] return script("verify_runtime_parity", *args) if job["kind"] in {"calibration", "smoke"}: opt, alg = profile["optimizer"], profile["algorithm"] args = ["--gpu", gpu, "--seed", job["seed"], "--activation-bits", job["bits"], "--optimizer", opt["name"], "--learning-rate", opt["learning_rate"], "--cta-learning-rate", opt["cta_learning_rate"], "--weight-decay", opt["weight_decay"], "--adam-betas", *opt["betas"], "--adam-epsilon", opt["epsilon"], "--total-steps", 2 if job["kind"] == "smoke" else opt["final_steps"], "--slt-round-steps", 1 if job["kind"] == "smoke" else opt["slt_round_steps"], "--slt-budget", job.get("budget", 4), "--output", out / "components.pt"] for key, value in alg.items(): args += ["--" + key.replace("_", "-"), value] if job["kind"] == "smoke": args += ["--smoke-small", "--smoke-samples", 2, "--smoke-max-length", 16] if job.get("samples") is not None: args += ["--study-samples", job["samples"]] if job.get("ablation"): args += ["--ablation", job["ablation"]] commands = [script("calibrate_ouro", *args), script("export_ouro_artifact", "--calibrated-components", out / "components.pt", "--activation-bits", job["bits"], "--output", out / "artifact.pt")] if job["kind"] == "smoke": commands += [script("smoke_ouro_vllm", "--artifact", out / "artifact.pt", "--gpu", gpu, "--output", out / "generation.json"), parity(out / "artifact.pt", out / "parity.json", advisory=True)] return commands if job["kind"] == "parity": artifact = root / job["calibration"] / "artifact.pt" if job.get("calibration") else None return [parity(artifact, out / "parity.json")] if job["kind"] == "fixture": return [script("freeze_lambada_inputs", "--evaluation", root / job["dependencies"][0] / "lambada_openai", "--samples", 32, "--output", out / "inputs.json")] if job["kind"] == "trajectory": return [script("measure_ouro_trajectory", "--artifact", root / job["calibration"] / "artifact.pt", "--inputs", root / "local_lambada_fixture/inputs.json", "--gpu", gpu, "--max-length", 256, "--output", out / "trajectory.json")] if job["kind"] == "selection": return [script("inspect_calibration", "--components", root / job["calibration"] / "components.pt", "--output", out / "selection.json")] if job["kind"] == "evaluation": method, bits = job["method"], job["bits"] configuration = ("bf16" if method == "bf16" else f"direct_w4a{bits}" if method == "direct" else f"loopq_{job['ablation']}" if job.get("ablation") else f"loopq_w4a{bits}") commands = [] for task in job["tasks"]: args = ["--task", task, "--configuration", configuration, "--gpu", gpu, "--fewshot", profile["evaluation"][task]["fewshot"], "--seed", job["seed"], "--protocol-note", profile["protocol_note"], "--output", out / task] if method == "loopq": args += ["--artifact", root / job["calibration"] / "artifact.pt", "--parity-report", root / job["dependencies"][0] / "parity.json"] if job.get("ablation"): args += ["--activation-bits", bits] if job.get("study"): args += ["--study-evaluation"] commands.append(script("run_paper_benchmark", *args)) return commands raise ValueError("no executor for this job kind")