#!/usr/bin/env python3 """house_render_metrics.py — H3 / H3b / H5, the cross-centroid control, and the PQ floor. `PREREG-HOUSE-ARMS.md` §8.2 (the registered render hypotheses) and §7.5 (the objective side-readings). The instrument is the one Amendment A1 pins and is imported from the sealed `run_bench.py`, never rebuilt. * **H3-HOUSE** — for each base/adapter pair at the same caption and seed, is the adapter's render closer to its own training corpus's centroid? Exact one-sided sign test, **n = 24 per arm** (12 prompts x 2 seeds), clears at **k >= 17** (α = 0.0320). ⚠ Those 24 are **12 prompts x 2 seeds and are therefore prompt-clustered**, so §7.3's clustered type-I table rides this test too and the 0.0320 label is optimistic at any ρ > 0. Reported, not hidden. **Predicted: uncertain** — the mnml arc's best-powered arm returned 5/12 and 6/12. * **H3b-HOUSE** — the same test on the **dose-0.5** arm, also n = 24. * **H5-HOUSE (descriptive)** — is `d(·, C_house-wide)` non-increasing across `base -> 0.5 -> 1.0`? A non-monotone majority publishes in the coherence prereg's own registered words: *"the dial moves the audio but not along the corpus axis"*. * **BLOCK T, the separation (§7.3(i))** — the trigger-only render's distance against the plain base render at the same seed, so *"the token moved it"* and *"the weights moved it"* are separable rather than confounded. * **The cross-centroid control** (`PREREG-COHERENCE.md` §6.4) — every distance is also reported against corpora the adapter was NOT trained on (`house-artist`, and the mnml arc's `mnml` and `fm-control`), so "moved toward its corpus" can be told apart from "moved toward everything". * **THE MEMORISATION SCREEN (A2-HOUSE) — DESCRIPTIVE, and the reason is on the record.** §12 owes A2-HOUSE *"before the first training step of arm 1"*, and it was **not filed**: the only amendment on the sealed prereg is A-H1, and the first training step was 2026-09-06T09:52:02Z. §0.5's amendment rule then governs — *"If an owed amendment is not filed before the step it governs, the criterion it carries degrades to DESCRIPTIVE for that run — it does not silently become a pass"* — so §7.6(d) is descriptive for this run and the screen below **cannot be counted as a clause of "arm 1 clears"**. It is computed anyway, to A2-HOUSE's own written spec (the training-set pairwise similarity distribution published whole, its 95th percentile as the threshold, the holdout as the matched negative control, k = 20 over block A's adapter-arm clips), because a distribution is worth having even when the gate that would have used it was never armed. * **§7.5.1 THE PQ FLOOR — the one hard automatic fail, and it is BINDING** (A1 is filed and verified). Audiobox-Aesthetics **PQ >= (base-arm PQ − 0.3)**, mean over the arm, both arms scored by one A1-pinned instrument in one session. **Never rank on the PC axis.** MEASURED, NEVER GATING everywhere except the PQ floor, which §7.6(c) makes a clause of "arm 1 clears". """ from __future__ import annotations import argparse import csv import hashlib import importlib.util import json import math import sys from pathlib import Path PQ_FLOOR_DELTA = 0.3 PROMPT_IDS = [f"P{i:02d}" for i in range(1, 13)] SEED_SETS = {"primary": 4000, "second": 5000} def load_module(path: Path, name: str): spec = importlib.util.spec_from_file_location(name, path) mod = importlib.util.module_from_spec(spec) sys.modules[name] = mod spec.loader.exec_module(mod) return mod def track_vectors(V, idx, pred): import numpy as np buckets: dict[str, list[int]] = {} for r in idx: if pred(r): buckets.setdefault(r["source_sha256"], []).append(int(r["row"])) out = {} for sha, rows in buckets.items(): v = V[sorted(rows)].mean(axis=0) out[sha] = v / np.linalg.norm(v) return out def centroid(E): import numpy as np c = E.mean(axis=0) return c / np.linalg.norm(c) def sign_test(diffs): """Exact one-sided sign test, zeros dropped (the coherence prereg's §6.3).""" nz = [d for d in diffs if d != 0] n = len(nz) k = sum(1 for d in nz if d > 0) p = sum(math.comb(n, i) for i in range(k, n + 1)) / (2 ** n) if n else float("nan") return n, k, p def clears_at(n: int, alpha: float = 0.05) -> int: """The smallest k whose exact one-sided tail is <= alpha — recomputed from the rule, never held at its original value when n falls (§7.4).""" for k in range(n // 2, n + 1): if sum(math.comb(n, i) for i in range(k, n + 1)) / (2 ** n) <= alpha: return k return n + 1 def measure_pq(paths: list[Path], ckpt: Path) -> dict[str, dict]: """§7.5.1: Audiobox-Aesthetics, ONE instrument, ONE session, both arms.""" from audiobox_aesthetics.infer import initialize_predictor pred = initialize_predictor(ckpt=str(ckpt)) out: dict[str, dict] = {} B = 8 for i in range(0, len(paths), B): chunk = paths[i:i + B] res = pred.forward([{"path": str(p)} for p in chunk]) for p, r in zip(chunk, res): out[p.stem] = {k: float(v) for k, v in dict(r).items()} return out def main() -> int: import numpy as np ap = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) ap.add_argument("--bench", type=Path, default=Path.home() / "music/out/coherence-bench/run_bench.py") ap.add_argument("--bench-sha", required=True) ap.add_argument("--archive", type=Path, default=Path.home() / "music/out/coherence-bench") ap.add_argument("--house", type=Path, default=Path.home() / "music/out/house-coherence") ap.add_argument("--renders", type=Path, required=True, help="the render crop/embed workdir") ap.add_argument("--records", type=Path, required=True, help="round-4 records.json") ap.add_argument("--split", type=Path, required=True) ap.add_argument("--manifest", type=Path, required=True) ap.add_argument("--split-rule", type=Path, required=True) ap.add_argument("--listen-dir", type=Path, required=True, help="the -16 LUFS listening copies — the PQ floor scores these") ap.add_argument("--audiobox-ckpt", type=Path, default=Path.home() / "music/venvs/eval/weights/audiobox-aesthetics-checkpoint.pt") ap.add_argument("--out", type=Path, required=True) args = ap.parse_args() got = hashlib.sha256(args.bench.read_bytes()).hexdigest() if got != args.bench_sha: raise SystemExit(f"ABORT: sealed bench hashes to {got}") rb = load_module(args.bench, "run_bench_pinned") sr = load_module(args.split_rule, "split_rule_pinned") print(f"OK sealed instrument = {got}") # ---- the centroids: a property of each CORPUS, not of a draw (§6.1) ---- Vh = np.load(args.house / "embeddings.npz")["crops"] ih = list(csv.DictReader(open(args.house / "embeddings-index.csv"))) house_tv = track_vectors(Vh, ih, lambda r: r["mode"] == "norm") manifest = [json.loads(l) for l in args.manifest.read_text("utf-8").splitlines() if l.strip()] sha_of = {r["track_id"]: r["local_sha256"] for r in manifest} split = json.loads(args.split.read_text("utf-8")) train_ids = sorted(t for a in split["train"]["artists"] for t in a["track_ids"]) arm3 = sr.by_track_split([r for r in manifest if r["artist_id"] == "artist_487628"]) cents = { "house-wide": centroid(np.stack([house_tv[sha_of[t]] for t in train_ids])), "house-artist": centroid(np.stack([house_tv[sha_of[t]] for t in arm3.training_track_ids])), } Va = np.load(args.archive / "embeddings.npz")["crops"] ia = list(csv.DictReader(open(args.archive / "embeddings-index.csv"))) for c in ("mnml", "fm-control"): tv = track_vectors(Va, ia, lambda r, c=c: (r["set"] == "corpus" and r["mode"] == "norm" and r["in_train"] == "1" and r["corpus"] == c)) cents[c] = centroid(np.stack([tv[s] for s in sorted(tv)])) print("OK centroids: " + ", ".join(f"{k}(n)" for k in cents)) # ---- the render vectors, keyed by clip id ------------------------------ # ⚠ The crop plan dedupes on the SOURCE SHA, so two clips with identical bytes # share ONE embedding — and this round has exactly such a pair by design: the # determinism control `P01__base-repeat__seed4001` is byte-identical to # `P01__base__seed4001`. Keying off the crop index alone therefore loses one of # the two names. The map is built from records.json, which lists every clip and # its output_sha256, so both names resolve to the one embedding they share. Vr = np.load(args.renders / "embeddings.npz")["crops"] ir = list(csv.DictReader(open(args.renders / "embeddings-index.csv"))) rvec = track_vectors(Vr, ir, lambda r: True) doc = json.loads(args.records.read_text("utf-8")) clip_vec, unembedded = {}, [] for c in doc["clips"]: v = rvec.get(c["output_sha256"]) if v is None: unembedded.append(c["clip_id"]) else: clip_vec[c["clip_id"]] = v if unembedded: raise SystemExit(f"ABORT: {len(unembedded)} clips have no embedding: {unembedded[:5]}") dupes = len(doc["clips"]) - len(rvec) print(f"OK render clips embedded = {len(clip_vec)} " f"({len(rvec)} distinct embeddings; {dupes} byte-identical clip(s) share one)") def dists(clip): v = clip_vec[clip] return {c: float(1.0 - v @ cents[c]) for c in cents} rows = [] for clip in sorted(clip_vec): d = dists(clip) rows.append({"clip_id": clip, **{f"d_to_C_{k}": round(v, 6) for k, v in d.items()}}) (args.out.parent).mkdir(parents=True, exist_ok=True) with open(args.out.parent / "render-centroid.csv", "w", newline="") as fh: w = csv.DictWriter(fh, fieldnames=list(rows[0].keys())) w.writeheader(); w.writerows(rows) # ---- H3 / H3b: the sign tests ----------------------------------------- OWN = "house-wide" results = {"instrument": {"run_bench_sha256": got, "coherence_seal": rb.SEAL, "clap_ckpt_sha256": rb.CKPT_SHA}, "centroid_note": ("centroids from the FULL training set of each corpus, " "primary arm — a centroid is a property of the corpus, " "not of a draw (§6.1)"), "standing_constraint": "MEASURED, NEVER GATING (D-20260831-17), except §7.5.1's PQ floor"} pair_rows = [] for arm, key in (("house-1.0", "H3_HOUSE"), ("house-0.5", "H3b_HOUSE")): diffs, per = [], [] for pid in PROMPT_IDS: for setname, base in SEED_SETS.items(): seed = base + int(pid[1:]) b, a = f"{pid}__base__seed{seed}", f"{pid}__{arm}__seed{seed}" db, da = dists(b), dists(a) diff = db[OWN] - da[OWN] # > 0 == the adapter moved TOWARD diffs.append(diff) row = {"arm": arm, "prompt_id": pid, "seed": seed, "seed_set": setname, "d_base_own": round(db[OWN], 6), "d_adapter_own": round(da[OWN], 6), "diff_own": round(diff, 6), "direction": "toward" if diff > 0 else ("away" if diff < 0 else "flat")} for c in cents: if c != OWN: row[f"diff_vs_C_{c}"] = round(db[c] - da[c], 6) per.append(row); pair_rows.append(row) n, k, p = sign_test(diffs) results[key] = { "arm": arm, "n": n, "toward": k, "p_one_sided": round(p, 6), "clears_at_registered": 17, "clears_at_recomputed": clears_at(n), "verdict": "SUPPORTED" if k >= clears_at(n) else "NOT SUPPORTED", "prediction": "uncertain (the mnml arc's best-powered arm returned 5/12 and 6/12)", "clustering_caveat": ("these trials are 12 prompts x 2 seeds and are therefore " "PROMPT-CLUSTERED; §7.3's clustered type-I table rides this " "test, so the 0.0320 label is optimistic at any rho > 0"), "cross_centroid_control": { c: {"toward": sum(1 for r in per if r.get(f"diff_vs_C_{c}", 0) > 0), "n": len(per)} for c in cents if c != OWN}, "pairs": per, } print(f"{key:12} {arm:11} {k}/{n} toward, p={p:.4f}, clears at >= " f"{clears_at(n)} -> {results[key]['verdict']}") # ---- H5: monotone across base -> 0.5 -> 1.0 --------------------------- def h5(seed_sets): mono, detail = 0, [] for pid in PROMPT_IDS: for setname, base in seed_sets.items(): seed = base + int(pid[1:]) d0 = dists(f"{pid}__base__seed{seed}")[OWN] d5 = dists(f"{pid}__house-0.5__seed{seed}")[OWN] d1 = dists(f"{pid}__house-1.0__seed{seed}")[OWN] ok = d0 >= d5 >= d1 mono += ok detail.append({"prompt_id": pid, "seed": seed, "d_base": round(d0, 6), "d_0_5": round(d5, 6), "d_1_0": round(d1, 6), "monotone_non_increasing": ok}) return mono, detail m24, det24 = h5(SEED_SETS) m12, det12 = h5({"primary": SEED_SETS["primary"]}) results["H5_HOUSE"] = { "hypothesis": "d(., C_house-wide) is non-increasing across base -> 0.5 -> 1.0", "monotone_of_24": m24, "monotone_of_12_primary_seed": m12, "verdict": "SUPPORTED" if m24 > 12 else "NOT SUPPORTED", "registered_words_if_not": ("the dial moves the audio but not along the corpus axis"), "detail_24": det24, } print(f"H5_HOUSE monotone {m24}/24 (primary seed only: {m12}/12) -> " f"{results['H5_HOUSE']['verdict']}") # ---- BLOCK T: the token vs the weights (§7.3(i)) ----------------------- t_rows = [] for pid in PROMPT_IDS: seed = SEED_SETS["primary"] + int(pid[1:]) db = dists(f"{pid}__base__seed{seed}") dt = dists(f"{pid}__trigger-only__seed{seed}") da = dists(f"{pid}__house-1.0__seed{seed}") t_rows.append({ "prompt_id": pid, "seed": seed, "d_base_own": round(db[OWN], 6), "d_trigger_only_own": round(dt[OWN], 6), "d_adapter_1_0_own": round(da[OWN], 6), "token_moved_it": round(db[OWN] - dt[OWN], 6), "weights_moved_it": round(dt[OWN] - da[OWN], 6), "total": round(db[OWN] - da[OWN], 6), }) nt, kt, pt = sign_test([r["token_moved_it"] for r in t_rows]) nw, kw, pw = sign_test([r["weights_moved_it"] for r in t_rows]) results["BLOCK_T"] = { "what": ("the trigger STRING with no adapter, against the plain base render at the " "same seed — the trigger confound's control and this harness's first null"), "why_it_is_needed": ("'house-wide' tokenises to include the genre word HOUSE, so " "prepending it adds a genre term to every prompt: new " "information on P01-P06 and redundant on P07-P10 (§4.6)"), "token_moved_toward": {"n": nt, "toward": kt, "p_one_sided": round(pt, 6)}, "weights_moved_toward": {"n": nw, "toward": kw, "p_one_sided": round(pw, 6)}, "rows": t_rows, "null_note": ("scored by the EAR this is the cleanest null this harness will ever " "have — twelve pairs in which no adapter exists, so a systematic " "preference is an instrument artefact by construction (A6-HOUSE)"), } print(f"BLOCK_T token moved toward {kt}/{nt} · weights moved toward {kw}/{nw}") # ---- A2-HOUSE: the memorisation screen, DESCRIPTIVE -------------------- Etr = np.stack([house_tv[sha_of[t]] for t in train_ids]) hold_ids = sorted(t for a in split["held_out"]["artists"] for t in a["track_ids"]) Eho = np.stack([house_tv[sha_of[t]] for t in hold_ids]) G = Etr @ Etr.T iu = np.triu_indices(len(Etr), k=1) train_sims = G[iu] thresh95 = float(np.percentile(train_sims, 95)) # the matched negative control: holdout-vs-training similarity, the distribution # a render's similarity is read AGAINST (the model never saw the holdout) neg = (Eho @ Etr.T).ravel() gate_clips = [f"{pid}__house-1.0__seed{b + int(pid[1:])}" for pid in PROMPT_IDS for b in SEED_SETS.values()] pairs = [] for c in gate_clips: sims = clip_vec[c] @ Etr.T j = int(np.argmax(sims)) pairs.append({"render": c, "nearest_training_track": train_ids[j], "cosine": round(float(sims[j]), 6), "above_95th_percentile": bool(sims[j] > thresh95)}) top20 = sorted(pairs, key=lambda r: -r["cosine"])[:20] results["A2_HOUSE_MEMORISATION"] = { "status": "DESCRIPTIVE — A2-HOUSE was NOT FILED before the first training step", "why_descriptive": ( "PREREG §12 owes A2-HOUSE before the first training step of arm 1; the only " "amendment on the sealed prereg is A-H1 and the first training step was " "2026-09-06T09:52:02Z. §0.5: an owed amendment not filed before the step it " "governs degrades its criterion to DESCRIPTIVE — it does not silently become " "a pass. §7.6(d) is therefore descriptive for this run."), "training_set_pairwise_similarity": { "n_pairs": int(len(train_sims)), "min": round(float(train_sims.min()), 6), "p25": round(float(np.percentile(train_sims, 25)), 6), "median": round(float(np.median(train_sims)), 6), "p75": round(float(np.percentile(train_sims, 75)), 6), "p95_THRESHOLD": round(thresh95, 6), "p99": round(float(np.percentile(train_sims, 99)), 6), "max": round(float(train_sims.max()), 6), "mean": round(float(train_sims.mean()), 6), }, "matched_negative_control_holdout_vs_training": { "n_pairs": int(len(neg)), "median": round(float(np.median(neg)), 6), "p95": round(float(np.percentile(neg, 95)), 6), "max": round(float(neg.max()), 6), "note": ("the holdout is what the model NEVER saw; this is the distribution a " "render's similarity is read against. ⚠ §3.1(3) fault (a): the holdout " "is small-catalogue artists BY CONSTRUCTION, so this control is " "computed on small-catalogue artists only"), }, "k": 20, "render_set": "block A's 24 adapter-arm clips (house-1.0, both seeds)", "renders_above_threshold": sum(1 for p in pairs if p["above_95th_percentile"]), "top20_render_to_training_pairs": top20, "still_owed": ("A2-HOUSE's own spec has the top-20 pairs LISTENED TO in the same " "harness, question 'is this the same tune?'. That is the operator's " "ear and has not happened."), } print(f"A2-HOUSE DESCRIPTIVE (unfiled). training-set similarity p95 = {thresh95:.4f}; " f"{results['A2_HOUSE_MEMORISATION']['renders_above_threshold']}/24 gate renders above it") # ---- §7.5.1 THE PQ FLOOR — binding ------------------------------------ clip_ids = sorted(r["clip_id"] for r in doc["clips"]) paths = [args.listen_dir / f"{c}.wav" for c in clip_ids] print(f"PQ scoring {len(paths)} listening copies with the A1-pinned Audiobox instrument...") axes = measure_pq(paths, args.audiobox_ckpt) arm_of = {r["clip_id"]: r["arm"] for r in doc["clips"]} by_arm: dict[str, list[float]] = {} for cid, ax in axes.items(): by_arm.setdefault(arm_of[cid], []).append(ax["PQ"]) base_pq = sum(by_arm["base"]) / len(by_arm["base"]) floor = base_pq - PQ_FLOOR_DELTA pq = {"axis": "PQ", "base_arm_mean": round(base_pq, 4), "floor": round(floor, 4), "delta": PQ_FLOOR_DELTA, "instrument": "audiobox-aesthetics 0.0.4, A1-pinned, one session, all arms", "never_rank_on": "PC (§7.5.1)", "arms": {a: {"n": len(v), "mean_PQ": round(sum(v) / len(v), 4), "min_PQ": round(min(v), 4), "max_PQ": round(max(v), 4), "passes_floor": (sum(v) / len(v)) >= floor} for a, v in sorted(by_arm.items())}, "per_clip": {c: axes[c] for c in clip_ids}} pq["gate_arm"] = "house-1.0" pq["verdict_7_6_c"] = ("PASS" if pq["arms"]["house-1.0"]["passes_floor"] else "FAIL") results["PQ_FLOOR"] = pq for a, v in pq["arms"].items(): print(f"PQ {a:14} n={v['n']:>2} mean PQ {v['mean_PQ']:.4f} " f"{'PASS' if v['passes_floor'] else 'FAIL'} (floor {floor:.4f})") print(f"PQ §7.6(c) on the gate arm house-1.0: {pq['verdict_7_6_c']}") with open(args.out.parent / "render-pairs.csv", "w", newline="") as fh: cols = sorted({k for r in pair_rows for k in r}) w = csv.DictWriter(fh, fieldnames=cols); w.writeheader() for r in pair_rows: w.writerow({k: r.get(k, "") for k in cols}) args.out.write_text(json.dumps(results, indent=2, ensure_ascii=False) + "\n", "utf-8") print(f"\nWROTE {args.out}") return 0 if __name__ == "__main__": raise SystemExit(main())