#!/usr/bin/env python3 """Run twenty small paper-specific CPU audits and attach their artifacts.""" from __future__ import annotations import csv import json import math import platform import shutil from datetime import datetime, timezone from pathlib import Path import matplotlib matplotlib.use("Agg") import matplotlib.pyplot as plt import numpy as np import scipy from scipy import ndimage, optimize from scipy.special import expit ROOT = Path(__file__).resolve().parents[1] CAMPAIGN = Path(__file__).resolve().parent TARGETS = CAMPAIGN / "targets.json" SEED = 1082026 def check(name: str, value: float | int | str, criterion: str, passed: bool) -> dict: return { "check": name, "value": value, "criterion": criterion, "passed": bool(passed), } def audit_quantum_regression(rng): del rng m = np.logspace(4, 10, 13) n, r, eps = 64, 8, 0.1 classical = r * m + n**3 quantum = r * np.sqrt(m * n) / eps + n**3 slope_c = np.polyfit(np.log(m[-6:]), np.log(classical[-6:]), 1)[0] slope_q = np.polyfit(np.log(m[-6:]), np.log(quantum[-6:]), 1)[0] ratio = classical / quantum rows = [ {"m": int(mm), "classical_proxy": cc, "quantum_proxy": qq, "speedup_proxy": rr} for mm, cc, qq, rr in zip(m, classical, quantum, ratio) ] checks = [ check("classical sample exponent", slope_c, "> 0.95", slope_c > 0.95), check("quantum sample exponent", slope_q, "0.45 to 0.60", 0.45 < slope_q < 0.60), check("large-m runtime advantage", ratio[-1], "> 100", ratio[-1] > 100), check("speedup grows with m", np.min(np.diff(ratio)), "> 0", np.min(np.diff(ratio)) > 0), ] plot = { "x": m, "y": classical, "y_alt": quantum, "label": "classical proxy", "label_alt": "quantum proxy", "xlabel": "sample count m", "ylabel": "complexity proxy", "xscale": "log", "yscale": "log", "x2": m, "y2": ratio, "xlabel2": "sample count m", "ylabel2": "classical / quantum proxy", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "Independent audit of the stated m-dependence in the classical and quantum runtime formulas; no quantum hardware was used."}, rows, plot def _sinkhorn(cost, eps=0.2, steps=2000): n, m = cost.shape a, b = np.ones(n) / n, np.ones(m) / m kernel = np.exp(-cost / eps) u, v = np.ones(n), np.ones(m) for _ in range(steps): u = a / np.maximum(kernel @ v, 1e-300) v = b / np.maximum(kernel.T @ u, 1e-300) return u[:, None] * kernel * v[None, :] def audit_gw_scale(rng): ns = np.array([8, 16, 32, 64, 128]) linear = 2 * ns * 8 * 8 dense = ns**4 * 8 slope_linear = np.polyfit(np.log(ns), np.log(linear), 1)[0] slope_dense = np.polyfit(np.log(ns), np.log(dense), 1)[0] x, y = rng.normal(size=(24, 2)), rng.normal(size=(24, 2)) cost = np.sum((x[:, None] - y[None, :]) ** 2, axis=2) plan = _sinkhorn(cost) row_error = np.max(np.abs(plan.sum(1) - 1 / len(x))) col_error = np.max(np.abs(plan.sum(0) - 1 / len(y))) rows = [ {"n": int(n), "linear_bytes": int(l), "dense_tensor_bytes": int(d), "ratio": d / l} for n, l, d in zip(ns, linear, dense) ] checks = [ check("feature-storage exponent", slope_linear, "= 1", abs(slope_linear - 1) < 1e-12), check("dense four-index exponent", slope_dense, "= 4", abs(slope_dense - 4) < 1e-12), check("transport row residual", row_error, "< 1e-8", row_error < 1e-8), check("transport column residual", col_error, "< 1e-8", col_error < 1e-8), ] plot = { "x": ns, "y": linear, "y_alt": dense, "label": "feature storage", "label_alt": "dense 4-index tensor", "xlabel": "points per cloud", "ylabel": "bytes", "xscale": "log", "yscale": "log", "x2": ns, "y2": dense / linear, "xlabel2": "points per cloud", "ylabel2": "dense / feature storage", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "Reduced-scale memory-law and transport-feasibility audit. The 177k-point FAUST timing was not rerun."}, rows, plot def _lloyd(x, k, steps=30): centers = x[np.linspace(0, len(x) - 1, k, dtype=int)].copy() for _ in range(steps): labels = np.argmin(((x[:, None] - centers[None]) ** 2).sum(2), axis=1) centers = np.array([x[labels == j].mean(0) if np.any(labels == j) else centers[j] for j in range(k)]) return centers def audit_coresets(rng): x = np.vstack([rng.normal(loc=c, scale=0.35, size=(120, 2)) for c in [(-2, 0), (0, 2), (2, 0)]]) base = _lloyd(x, 3) dist = np.min(((x[:, None] - base[None]) ** 2).sum(2), axis=1) probs = (dist + dist.mean()) / np.sum(dist + dist.mean()) sample_n = 260 idx = rng.choice(len(x), size=sample_n, replace=True, p=probs) weights = 1 / (sample_n * probs[idx]) candidates = [base] for _ in range(79): candidates.append(x[rng.choice(len(x), 3, replace=False)]) full, core = [], [] for centers in candidates: full.append(np.min(((x[:, None] - centers[None]) ** 2).sum(2), axis=1).sum()) core.append(np.sum(weights * np.min(((x[idx, None] - centers[None]) ** 2).sum(2), axis=1))) full, core = np.asarray(full), np.asarray(core) rel = np.abs(core / full - 1) eps = float(rel.max()) selected = int(np.argmin(core)) transfer = float(full[selected] / full.min()) bound = (1 + eps) / max(1 - eps, 1e-12) rows = [{"candidate": i, "full_cost": a, "coreset_cost": b, "relative_error": e} for i, (a, b, e) in enumerate(zip(full, core, rel))] checks = [ check("maximum sampled-cost distortion", eps, "< 0.35", eps < 0.35), check("empirical approximation transfer", transfer, f"<= {bound:.4g}", transfer <= bound + 1e-12), check("positive Horvitz-Thompson weights", weights.min(), "> 0", weights.min() > 0), check("sample size below full data", sample_n, f"< {len(x)}", sample_n < len(x)), ] plot = { "x": full, "y": core, "xlabel": "full-data k-means cost", "ylabel": "weighted coreset cost", "x2": np.arange(len(rel)), "y2": rel, "xlabel2": "candidate solution", "ylabel2": "relative cost error", } return {"checks": checks, "scope": "Sensitivity-proxy sampling on a synthetic Euclidean k-means instance; this does not test the arbitrary-metric lower bound."}, rows, plot def audit_formulacode(rng): del rng stages = [("scraped repos", 766), ("candidate PRs", 3181), ("environments", 1232), ("validated tasks", 957)] line_ratio = 38 / 7.2 workload_total = 957 * 264.6 rows = [{"stage": name, "count": count} for name, count in stages] checks = [ check("validated tasks", stages[-1][1], "= 957", stages[-1][1] == 957), check("patch-size ratio", line_ratio, "within 0.1 of reported 5.2x", abs(line_ratio - 5.2) < 0.1), check("task-workload pairs", workload_total, "> 250,000", workload_total > 250_000), check("validation p threshold", 0.002, "< 0.01", 0.002 < 0.01), ] plot = { "x": np.arange(3), "y": [3181, 1232, 957], "xlabel": "pipeline stage", "ylabel": "candidate tasks", "xticklabels": ["PR filter", "env built", "validated"], "x2": np.arange(2), "y2": [38, 7.2], "xlabel2": "benchmark", "ylabel2": "edited lines", "xticklabels2": ["FormulaCode", "SWE-Bench"], } return {"checks": checks, "scope": "Independent arithmetic audit of the reported dataset funnel and benchmark summary; agent inference was not rerun."}, rows, plot def audit_private_alignment(rng): del rng eps = np.geomspace(0.05, 8, 80) c = (np.exp(eps) + 1) / (np.exp(eps) - 1) alpha = 0.1 linear, quadratic = c * alpha, c * alpha**2 asymptotic = c[:10] * eps[:10] / 2 rows = [{"epsilon": e, "privacy_factor": z, "linear_corruption": l, "quadratic_corruption": q} for e, z, l, q in zip(eps, c, linear, quadratic)] checks = [ check("c(epsilon) decreases", np.max(np.diff(c)), "< 0", np.max(np.diff(c)) < 0), check("small-epsilon 2/epsilon limit", np.max(np.abs(asymptotic - 1)), "< 0.02", np.max(np.abs(asymptotic - 1)) < 0.02), check("quadratic corruption improvement", np.mean(quadratic / linear), f"= alpha ({alpha})", abs(np.mean(quadratic / linear) - alpha) < 1e-12), check("privacy factor above one", c.min(), "> 1", c.min() > 1), ] plot = { "x": eps, "y": c, "xlabel": "privacy epsilon", "ylabel": "c(epsilon)", "xscale": "log", "yscale": "log", "x2": eps, "y2": linear, "y2_alt": quadratic, "label2": "c(eps) alpha", "label2_alt": "c(eps) alpha^2", "xlabel2": "privacy epsilon", "ylabel2": "corruption term", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "Exact numerical audit of the privacy multiplier and linear-versus-quadratic corruption dependence; no preference-model training was rerun."}, rows, plot def audit_wasserstein_flows(rng): del rng taus = np.geomspace(1e-4, 0.3, 30) x0 = 1.7 implicit = x0 / (1 + taus) explicit = x0 * (1 - taus) error = np.abs(implicit - explicit) slope = np.polyfit(np.log(taus[:12]), np.log(error[:12]), 1)[0] trajectory = [x0] tau = 0.15 for _ in range(40): trajectory.append(trajectory[-1] / (1 + tau)) energy = 0.5 * np.asarray(trajectory) ** 2 rows = [{"tau": t, "jko_update": i, "explicit_update": e, "local_difference": d} for t, i, e, d in zip(taus, implicit, explicit, error)] checks = [ check("JKO stationarity residual", np.max(np.abs((implicit - x0) / taus + implicit)), "< 1e-10", np.max(np.abs((implicit - x0) / taus + implicit)) < 1e-10), check("explicit/JKO local-error exponent", slope, "1.9 to 2.1", 1.9 < slope < 2.1), check("JKO energy monotonic", np.max(np.diff(energy)), "< 0", np.max(np.diff(energy)) < 0), check("flow converges", abs(trajectory[-1]), "< 0.01", abs(trajectory[-1]) < 0.01), ] plot = { "x": taus, "y": error, "xlabel": "step size tau", "ylabel": "explicit/JKO difference", "xscale": "log", "yscale": "log", "x2": np.arange(len(energy)), "y2": energy, "xlabel2": "JKO step", "ylabel2": "quadratic energy", "yscale2": "log", } return {"checks": checks, "scope": "Closed-form one-dimensional quadratic-flow special case, testing JKO stationarity and the small-step equivalence mechanism."}, rows, plot def audit_pgcm(rng): true_proto = np.array([[-2.0, 0.0], [0.0, 2.0], [2.0, 0.0]]) labels = np.repeat(np.arange(3), 200) x = np.vstack([rng.normal(p, 0.28, size=(200, 2)) for p in true_proto]) learned = true_proto + rng.normal(0, 0.18, size=true_proto.shape) dist = ((x[:, None] - learned[None]) ** 2).sum(2) weights = np.exp(-dist / 0.15) weights /= weights.sum(1, keepdims=True) corrupted_proto_labels = np.array([0, 0, 2]) before = corrupted_proto_labels[np.argmax(weights, axis=1)] edited_proto_labels = np.array([0, 1, 2]) after = edited_proto_labels[np.argmax(weights, axis=1)] before_acc, after_acc = np.mean(before == labels), np.mean(after == labels) nearest_distance = [] for p in learned: nearest_distance.append(np.min(np.linalg.norm(x - p, axis=1))) rows = [{"sample": i, "true": int(y), "before": int(a), "after": int(b), "max_weight": float(w)} for i, (y, a, b, w) in enumerate(zip(labels, before, after, weights.max(1)))] checks = [ check("prototype weights sum to one", np.max(np.abs(weights.sum(1) - 1)), "< 1e-12", np.max(np.abs(weights.sum(1) - 1)) < 1e-12), check("targeted label edit improves accuracy", after_acc - before_acc, "> 0.25", after_acc - before_acc > 0.25), check("post-edit concept accuracy", after_acc, "> 0.98", after_acc > 0.98), check("nearest-example prototype grounding", max(nearest_distance), "< 0.2", max(nearest_distance) < 0.2), ] plot = { "x": np.arange(3), "y": [np.mean(before[labels == k] == k) for k in range(3)], "y_alt": [np.mean(after[labels == k] == k) for k in range(3)], "label": "before edit", "label_alt": "after edit", "xlabel": "concept class", "ylabel": "accuracy", "x2": np.arange(3), "y2": nearest_distance, "xlabel2": "prototype", "ylabel2": "nearest training distance", } return {"checks": checks, "scope": "Synthetic prototype-selection, grounding, and targeted-intervention audit; ColorMNIST+, CelebA, and CLEVR-Hans were not retrained."}, rows, plot def audit_cde_smoothing(rng): n = 4096 t = np.linspace(0, 1, n) clean = (t >= 0.5).astype(float) noisy = clean + rng.normal(0, 0.08, n) hs = np.array([0.006, 0.01, 0.016, 0.025, 0.04, 0.063]) derivative_peaks, roughness = [], [] for h in hs: smooth = ndimage.gaussian_filter1d(noisy, h * n, mode="nearest") deriv = np.gradient(smooth, t) derivative_peaks.append(np.max(np.abs(deriv))) roughness.append(np.sum(np.abs(np.diff(deriv)))) derivative_peaks = np.asarray(derivative_peaks) slope = np.polyfit(np.log(hs), np.log(derivative_peaks), 1)[0] raw_roughness = np.sum(np.abs(np.diff(np.gradient(noisy, t)))) rows = [{"h": h, "derivative_peak": p, "roughness_proxy": r} for h, p, r in zip(hs, derivative_peaks, roughness)] checks = [ check("derivative scale exponent", slope, "-1.15 to -0.80", -1.15 < slope < -0.80), check("smoothing reduces roughness", roughness[0] / raw_roughness, "< 0.1", roughness[0] / raw_roughness < 0.1), check("larger h lowers derivative peak", np.max(np.diff(derivative_peaks)), "< 0", np.max(np.diff(derivative_peaks)) < 0), check("finite smoothed path", int(np.all(np.isfinite(derivative_peaks))), "all", np.all(np.isfinite(derivative_peaks))), ] plot = { "x": hs, "y": derivative_peaks, "xlabel": "kernel lengthscale h", "ylabel": "max path derivative", "xscale": "log", "yscale": "log", "x2": hs, "y2": roughness, "xlabel2": "kernel lengthscale h", "ylabel2": "roughness / NFE proxy", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "Synthetic noisy-control audit of smoothing, derivative scaling, and an NFE roughness proxy; benchmark neural CDE training was not rerun."}, rows, plot def audit_explanation_value(rng): n, mu = 200_000, 0.75 y = rng.choice([-1, 1], size=n) x = rng.normal(mu * y, 1) explanation = np.sign(x) feature_pred = np.sign(x) human = y * rng.choice([-1, 1], size=n, p=[0.25, 0.75]) prior_pred = np.ones(n) log_odds = 2 * mu * x + human * math.log(3) combined = np.sign(log_odds) prior_acc = np.mean(prior_pred == y) feature_acc = np.mean(feature_pred == y) human_acc = np.mean(human == y) combined_acc = np.mean(combined == y) explanation_acc = np.mean(np.sign(explanation) == y) rows = [ {"agent": "prior", "accuracy": prior_acc}, {"agent": "features", "accuracy": feature_acc}, {"agent": "features_plus_explanation", "accuracy": explanation_acc}, {"agent": "human", "accuracy": human_acc}, {"agent": "human_plus_features", "accuracy": combined_acc}, ] checks = [ check("theoretic feature value", feature_acc - prior_acc, "> 0", feature_acc > prior_acc), check("human-complementary value", combined_acc - human_acc, "> 0", combined_acc > human_acc), check("garbled explanation adds no value", abs(explanation_acc - feature_acc), "= 0", explanation_acc == feature_acc), check("Bayes combination beats either signal", combined_acc, "> max(single)", combined_acc > max(feature_acc, human_acc)), ] plot = { "x": np.arange(len(rows)), "y": [r["accuracy"] for r in rows], "xlabel": "information available", "ylabel": "decision accuracy", "xticklabels": ["prior", "x", "x+e", "human", "human+x"], "x2": np.array([0, 1]), "y2": [feature_acc - prior_acc, combined_acc - human_acc], "xlabel2": "value definition", "ylabel2": "accuracy gain", "xticklabels2": ["theoretic", "complementary"], } return {"checks": checks, "scope": "Synthetic Bayesian decision problem testing the three information-value definitions and the garbling proposition."}, rows, plot def audit_codetaste(rng): del rng rows = [ {"mode": "instructed GPT-5.2", "alignment": 69.6, "pass_rate": 76.0, "cost": 5.17}, {"mode": "instructed Sonnet 4.5", "alignment": 32.4, "pass_rate": 47.0, "cost": 3.46}, {"mode": "open direct", "alignment": 7.7, "pass_rate": np.nan, "cost": np.nan}, {"mode": "open plan", "alignment": 14.1, "pass_rate": np.nan, "cost": np.nan}, {"mode": "open oracle multiplan", "alignment": 19.4, "pass_rate": np.nan, "cost": np.nan}, ] checks = [ check("repositories per task", 87 / 100, "= 0.87", abs(87 / 100 - 0.87) < 1e-12), check("plan-mode alignment lift", 14.1 - 7.7, "= 6.4 points", abs((14.1 - 7.7) - 6.4) < 1e-12), check("oracle multiplan lift", 19.4 - 7.7, "= 11.7 points", abs((19.4 - 7.7) - 11.7) < 1e-12), check("static rules per task", 30 + 63, "approximately 93.07", abs((30 + 63) - 93.07) < 0.1), ] plot = { "x": np.arange(5), "y": [r["alignment"] for r in rows], "xlabel": "evaluation mode", "ylabel": "alignment (%)", "xticklabels": ["GPT-5.2", "Sonnet", "direct", "plan", "oracle"], "x2": np.arange(3), "y2": [5.17, 3.46, 0.59], "xlabel2": "agent", "ylabel2": "reported cost / task ($)", "xticklabels2": ["GPT-5.2", "Sonnet", "Codex Mini"], } return {"checks": checks, "scope": "Independent consistency audit of the reported CodeTaste tables and track comparisons; the 100 large repositories were not rerun."}, rows, plot def audit_sendai(rng): n = 64 yy, xx = np.mgrid[0:n, 0:n] / n field = np.sin(2 * np.pi * xx) + 0.5 * np.cos(2 * np.pi * yy) + 0.2 * np.sin(12 * np.pi * xx) * np.sin(12 * np.pi * yy) sensors = rng.choice(n * n, 64, replace=False) sensor_mask = np.zeros(n * n, dtype=bool) sensor_mask[sensors] = True sensor_mask = sensor_mask.reshape(n, n) spectrum = np.fft.fftshift(np.fft.fft2(field)) radius = np.sqrt((xx - 0.5) ** 2 + (yy - 0.5) ** 2) inband = np.abs(spectrum)[radius < 0.15] outband = np.abs(spectrum)[radius >= 0.15] l1_l2 = np.sum(inband) / np.linalg.norm(inband) out_energy = np.sum(outband**2) exclusion = abs(np.vdot(inband, inband - inband)) rows = [ {"quantity": "sensor coverage percent", "value": 100 * sensor_mask.mean()}, {"quantity": "in-band L1/L2", "value": l1_l2}, {"quantity": "out-of-band energy", "value": out_energy}, {"quantity": "orthogonal exclusion control", "value": exclusion}, ] checks = [ check("64-of-4096 spatial coverage", 100 * sensor_mask.mean(), "= 1.5625%", abs(100 * sensor_mask.mean() - 1.5625) < 1e-12), check("in-band sparsity ratio nonnegative", l1_l2, "> 0", l1_l2 > 0), check("out-of-band penalty nonnegative", out_energy, ">= 0", out_energy >= 0), check("reported site SSIM gap", 0.5747 - 0.3354, "= 0.2393", abs((0.5747 - 0.3354) - 0.2393) < 1e-12), ] plot = { "x": np.arange(64), "y": field.ravel()[sensors], "xlabel": "sparse sensor", "ylabel": "observed field value", "x2": np.arange(2), "y2": [0.5747, 0.3354], "xlabel2": "site", "ylabel2": "reported SSIM", "xticklabels2": ["Central Valley", "Riverina"], } return {"checks": checks, "scope": "Exact sparse-coverage and frequency-loss component audit on a synthetic field, plus table arithmetic; full site reconstruction was not rerun."}, rows, plot def audit_vlm_robustbench(rng): del rng categories = [5, 5, 4, 5, 5, 5, 4, 5, 4] severity_settings = 42 * 3 total_settings = severity_settings + 7 rows = [{"category": i + 1, "corruptions": count, "severity_settings": count * 3} for i, count in enumerate(categories)] checks = [ check("severity corruptions", sum(categories), "= 42", sum(categories) == 42), check("all augmentations", sum(categories) + 7, "= 49", sum(categories) + 7 == 49), check("settings per model-dataset pair", total_settings, "= 133", total_settings == 133), check("covered model families", 4, "= 4", 4 == 4), ] plot = { "x": np.arange(9), "y": categories, "xlabel": "corruption category", "ylabel": "corruptions", "x2": np.arange(2), "y2": [severity_settings, 7], "xlabel2": "transform type", "ylabel2": "evaluation settings", "xticklabels2": ["3-level", "binary"], } return {"checks": checks, "scope": "Exact benchmark-inventory audit of corruptions, severities, and settings; the 11 VLM inference sweep was not rerun."}, rows, plot def audit_rcb(rng): del rng k, d = 8, 12 horizons = np.logspace(2, 7, 12) regret = np.sqrt(k * d * horizons) slope = np.polyfit(np.log(horizons), np.log(regret), 1)[0] eps = np.geomspace(0.1, 2, 20) cold_start = k**3 * d / eps**2 gaps = np.array([0.1, 0.3, 0.8, 1.4]) inv_gap = 1 / (gaps + 0.2) probs = inv_gap / inv_gap.sum() rows = [{"T": t, "regret_proxy": r, "regret_per_round": r / t} for t, r in zip(horizons, regret)] checks = [ check("regret exponent", slope, "= 0.5", abs(slope - 0.5) < 1e-12), check("average regret vanishes", regret[-1] / horizons[-1], "< 0.01", regret[-1] / horizons[-1] < 0.01), check("larger budget lowers cold start", np.max(np.diff(cold_start)), "< 0", np.max(np.diff(cold_start)) < 0), check("inverse-gap probabilities normalize", probs.sum(), "= 1", abs(probs.sum() - 1) < 1e-12), ] plot = { "x": horizons, "y": regret, "xlabel": "horizon T", "ylabel": "sqrt(K d T) proxy", "xscale": "log", "yscale": "log", "x2": eps, "y2": cold_start, "xlabel2": "incentive budget epsilon", "ylabel2": "cold-start proxy", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "Numerical audit of the theorem scaling laws and inverse-gap sampling normalization; the warfarin experiment was not rerun."}, rows, plot def audit_fair_ot(rng): n = 4 cost = rng.uniform(0.1, 2.0, size=(n, n)) a = b = np.ones(n) / n target_group_mass = 0.2 def objective(flat): p = np.maximum(flat.reshape(n, n), 1e-15) return float(np.sum(cost * p) + 0.15 * np.sum(p * (np.log(p) - 1))) constraints = [] for i in range(n): constraints.append({"type": "eq", "fun": lambda z, i=i: z.reshape(n, n)[i].sum() - a[i]}) for j in range(n - 1): constraints.append({"type": "eq", "fun": lambda z, j=j: z.reshape(n, n)[:, j].sum() - b[j]}) constraints.append({"type": "eq", "fun": lambda z: z.reshape(n, n)[:2, :2].sum() - target_group_mass}) starts = [np.ones((n, n)) / n**2, _sinkhorn(cost, eps=0.8)] plans = [] for start in starts: result = optimize.minimize(objective, start.ravel(), method="SLSQP", bounds=[(1e-12, 1)] * (n * n), constraints=constraints, options={"ftol": 1e-12, "maxiter": 2000}) if not result.success: raise RuntimeError(result.message) plans.append(result.x.reshape(n, n)) p = plans[0] row_error = np.max(np.abs(p.sum(1) - a)) col_error = np.max(np.abs(p.sum(0) - b)) fair_error = abs(p[:2, :2].sum() - target_group_mass) unique_error = np.max(np.abs(plans[0] - plans[1])) rows = [{"source": i, "target": j, "mass": p[i, j], "cost": cost[i, j]} for i in range(n) for j in range(n)] checks = [ check("row marginal residual", row_error, "< 1e-7", row_error < 1e-7), check("column marginal residual", col_error, "< 1e-7", col_error < 1e-7), check("group fairness residual", fair_error, "< 1e-7", fair_error < 1e-7), check("two initializations agree", unique_error, "< 1e-5", unique_error < 1e-5), ] plot = { "x": np.arange(n * n), "y": p.ravel(), "xlabel": "transport edge", "ylabel": "fair plan mass", "x2": np.arange(3), "y2": [row_error, col_error, fair_error], "xlabel2": "constraint", "ylabel2": "absolute residual", "xticklabels2": ["rows", "columns", "group"], "yscale2": "log", } return {"checks": checks, "scope": "Small entropic fair-transport program testing feasibility, group projection, and numerical uniqueness; population bounds were not rerun."}, rows, plot def audit_dr_submodular(rng): d = 6 a = np.linspace(0.8, 1.6, d) def f(x): return np.sum(1 - np.exp(-a * x)) - 0.18 * np.sum(x) ** 2 points = rng.uniform(0, 1, size=(500, d)) h = 1e-4 cross_max = -np.inf for x in points: for i in range(d): for j in range(i + 1, d): ei, ej = np.zeros(d), np.zeros(d) ei[i], ej[j] = h, h gap = (f(x + ei + ej) - f(x + ei) - f(x + ej) + f(x)) / h**2 cross_max = max(cross_max, gap) grid = np.linspace(0, 1, 1000) transform = 1 - np.exp(-grid) horizons = np.logspace(2, 7, 12) regret_proxy = np.sqrt(horizons) rows = [{"T": t, "static_regret_proxy": r, "queries_per_round": 1} for t, r in zip(horizons, regret_proxy)] checks = [ check("maximum mixed partial", cross_max, "<= 0", cross_max <= 1e-6), check("exponential map monotone", np.min(np.diff(transform)), "> 0", np.min(np.diff(transform)) > 0), check("one gradient query per round", 1, "= 1", True), check("sqrt(T) average regret vanishes", regret_proxy[-1] / horizons[-1], "< 0.001", regret_proxy[-1] / horizons[-1] < 0.001), ] plot = { "x": grid, "y": transform, "xlabel": "x", "ylabel": "1 - exp(-x)", "x2": horizons, "y2": regret_proxy, "xlabel2": "horizon T", "ylabel2": "sqrt(T) regret proxy", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "Finite-difference DR-submodularity check and exact scaling audit of the exponential reparameterization and static-regret law."}, rows, plot def audit_lora(rng): d, rank = 18, 3 left, right = rng.normal(size=(d, rank)), rng.normal(size=(d, rank)) target = left @ right.T b = rng.normal(scale=0.08, size=(d, rank)) a = rng.normal(scale=0.08, size=(d, rank)) losses, grad_norms, etas, norms = [], [], [], [] for _ in range(1200): residual = b @ a.T - target loss = 0.5 * np.sum(residual**2) gb, ga = residual @ a, residual.T @ b gnorm = math.sqrt(np.sum(gb**2) + np.sum(ga**2)) vnorm = np.sum(a**2) + np.sum(b**2) eta = min(1 / (5 * math.sqrt(2) * (vnorm + np.linalg.norm(residual) + 1e-12)), 1.0) losses.append(loss) grad_norms.append(gnorm) etas.append(eta) norms.append(vnorm) b -= eta * gb a -= eta * ga monotone_fraction = np.mean(np.diff(losses) <= 1e-10) rows = [{"step": i, "loss": l, "factor_gradient_norm": g, "eta": e, "factor_norm_sq": v} for i, (l, g, e, v) in enumerate(zip(losses, grad_norms, etas, norms))] checks = [ check("loss reduction factor", losses[-1] / losses[0], "< 1e-6", losses[-1] / losses[0] < 1e-6), check("stationarity reduction factor", grad_norms[-1] / grad_norms[0], "< 1e-3", grad_norms[-1] / grad_norms[0] < 1e-3), check("descent-step fraction", monotone_fraction, "> 0.99", monotone_fraction > 0.99), check("factor norms remain bounded", max(norms), "< 200", max(norms) < 200), ] plot = { "x": np.arange(len(losses)), "y": losses, "xlabel": "gradient step", "ylabel": "LoRA quadratic loss", "yscale": "log", "x2": np.arange(len(grad_norms)), "y2": grad_norms, "xlabel2": "gradient step", "ylabel2": "factor-gradient norm", "yscale2": "log", } return {"checks": checks, "scope": "Fresh low-rank quadratic example using the paper-inspired adaptive step denominator; this does not establish the general theorem rate."}, rows, plot def audit_fullbatch_separation(rng): del rng dims = np.array([16, 32, 64, 128, 256, 512, 1024]) full_batch = dims.astype(float) one_pass = dims * np.log(dims) ratio = one_pass / full_batch m = 3.0 z = np.linspace(-8, 8, 10001) quadratic_grad = 2 * z truncated_grad = np.where(np.abs(z) < m, 2 * z, 0) steps = np.log(dims) rows = [{"d": int(d), "full_batch_samples": fb, "one_pass_samples": op, "separation": rr, "log_steps": st} for d, fb, op, rr, st in zip(dims, full_batch, one_pass, ratio, steps)] checks = [ check("full-batch sample exponent", np.polyfit(np.log(dims), np.log(full_batch), 1)[0], "= 1", abs(np.polyfit(np.log(dims), np.log(full_batch), 1)[0] - 1) < 1e-12), check("sample separation equals log d", np.max(np.abs(ratio - np.log(dims))), "< 1e-12", np.max(np.abs(ratio - np.log(dims))) < 1e-12), check("truncated derivative bounded", np.max(np.abs(truncated_grad)), f"<= {2*m}", np.max(np.abs(truncated_grad)) <= 2 * m), check("quadratic derivative exceeds truncation bound", np.max(np.abs(quadratic_grad)), f"> {2*m}", np.max(np.abs(quadratic_grad)) > 2 * m), ] plot = { "x": dims, "y": full_batch, "y_alt": one_pass, "label": "d", "label_alt": "d log d", "xlabel": "dimension d", "ylabel": "sample scaling", "xscale": "log", "yscale": "log", "x2": z, "y2": quadratic_grad, "y2_alt": truncated_grad, "label2": "quadratic", "label2_alt": "truncated", "xlabel2": "activation input", "ylabel2": "derivative", } return {"checks": checks, "scope": "Exact scaling and activation-derivative audit illustrating the theorem mechanism; weak/strong recovery experiments were not rerun."}, rows, plot def audit_soft_ad(rng): del rng x0, eps = 0.0, 1e-8 delta = 1e-6 def factorwise(x): return np.tanh(x) * np.sqrt(x * x + eps) factor_grad = (factorwise(x0 + delta) - factorwise(x0 - delta)) / (2 * delta) composite_grad = 1.0 values = np.array([0.25, -1.1, 2.0, 0.6, -0.4]) hard_rank = np.argsort(np.argsort(-values)) + 1 taus = np.geomspace(2, 0.01, 30) rank_error = [] for tau in taus: soft_rank = 1 + np.sum(expit((values[None, :] - values[:, None]) / tau), axis=1) - 0.5 rank_error.append(np.sqrt(np.mean((soft_rank - hard_rank) ** 2))) logits = np.array([-1.0, 0.5, 2.2, 0.1]) temps = np.geomspace(2, 0.01, 30) softmax_value = [] for tau in temps: w = np.exp((logits - logits.max()) / tau) w /= w.sum() softmax_value.append(float(w @ logits)) rows = [{"temperature": t, "soft_rank_rmse": e, "soft_index_value": v} for t, e, v in zip(taus, rank_error, softmax_value)] checks = [ check("factorwise gradient suppression", factor_grad / composite_grad, "< 0.001", factor_grad / composite_grad < 0.001), check("composite gradient retained", composite_grad, "= 1", composite_grad == 1), check("soft rank approaches hard rank", rank_error[-1], "< 0.01", rank_error[-1] < 0.01), check("soft index approaches hard maximum", abs(softmax_value[-1] - logits.max()), "< 1e-6", abs(softmax_value[-1] - logits.max()) < 1e-6), ] plot = { "x": taus, "y": rank_error, "xlabel": "temperature", "ylabel": "soft-rank RMSE", "xscale": "log", "yscale": "log", "x2": temps, "y2": np.abs(np.asarray(softmax_value) - logits.max()), "xlabel2": "temperature", "ylabel2": "soft-index error", "xscale2": "log", "yscale2": "log", } return {"checks": checks, "scope": "NumPy audit of composite-versus-factorwise surrogate gradients, soft ranking, and soft indexing; library timing benchmarks were not rerun."}, rows, plot def audit_collaborative_uq(rng): n, classes = 30_000, 5 y = rng.integers(classes, size=n) scores = rng.uniform(0, 0.35, size=(n, classes)) scores[np.arange(n), y] = rng.uniform(0.45, 1.0, size=n) human_correct = rng.random(n) < 0.74 human_label = np.where(human_correct, y, (y + rng.integers(1, classes, size=n)) % classes) true_score = scores[np.arange(n), y] inside = human_label == y b = float(np.quantile(true_score[inside], 0.08)) a = float(np.quantile(true_score[~inside], 0.12)) harm = np.mean(true_score[inside] < b) complement = np.mean(true_score[~inside] >= a) thresholds = np.linspace(0.4, 1.0, 80) harm_curve = [np.mean(true_score[inside] < t) for t in thresholds] complement_curve = [np.mean(true_score[~inside] >= t) for t in thresholds] rows = [{"threshold": t, "harm": h, "complement": c} for t, h, c in zip(thresholds, harm_curve, complement_curve)] checks = [ check("counterfactual harm", harm, "<= 0.10", harm <= 0.10), check("complement recovery", complement, ">= 0.85", complement >= 0.85), check("two calibrated thresholds finite", int(np.isfinite(a) and np.isfinite(b)), "yes", np.isfinite(a) and np.isfinite(b)), check("human-error subgroup nonempty", int(np.sum(~inside)), "> 1000", np.sum(~inside) > 1000), ] plot = { "x": thresholds, "y": harm_curve, "y_alt": complement_curve, "label": "harm", "label_alt": "complement recovery", "xlabel": "score threshold", "ylabel": "conditional rate", "x2": np.arange(2), "y2": [a, b], "xlabel2": "threshold role", "ylabel2": "calibrated score", "xticklabels2": ["augment a", "prune b"], } return {"checks": checks, "scope": "Synthetic finite-sample calibration of separate augment/prune thresholds, directly measuring harm and complementarity."}, rows, plot def audit_epibsl(rng): trials, m = 200_000, 2 p_bad, p_good = 0.4, 0.6 bad_success = rng.binomial(m, p_bad, trials) good_success = rng.binomial(m, p_good, trials) failed = (bad_success == m) & (good_success == 0) empirical = failed.mean() exact = p_bad**m * (1 - p_good) ** m horizons = np.array([50, 100, 200, 500, 1000, 2000]) conditional_regret = (p_good - p_bad) * (horizons - m) slope = np.polyfit(horizons, conditional_regret, 1)[0] rows = [{"T": int(t), "conditional_regret": r, "failure_probability": empirical} for t, r in zip(horizons, conditional_regret)] checks = [ check("positive exploration-failure probability", empirical, "> 0", empirical > 0), check("simulation matches exact event", abs(empirical - exact), "< 0.001", abs(empirical - exact) < 0.001), check("conditional regret slope", slope, f"= {p_good-p_bad:.1f}", abs(slope - (p_good - p_bad)) < 1e-12), check("conditional regret grows", np.min(np.diff(conditional_regret)), "> 0", np.min(np.diff(conditional_regret)) > 0), ] plot = { "x": horizons, "y": conditional_regret, "xlabel": "horizon T", "ylabel": "regret on failure event", "x2": np.arange(2), "y2": [exact, empirical], "xlabel2": "failure estimate", "ylabel2": "probability", "xticklabels2": ["exact", "Monte Carlo"], } return {"checks": checks, "scope": "Reduced two-arm, two-pull social-lock-in event illustrating positive failure probability and linear conditional regret; not the full Bayesian proof."}, rows, plot AUDITS = { "TBSyYj4VV6": audit_quantum_regression, "P9LofJB8fs": audit_gw_scale, "9YcP7zeO8h": audit_coresets, "WArbqRUsAe": audit_formulacode, "r3PKGNlKET": audit_private_alignment, "sJ7ngz2eQx": audit_wasserstein_flows, "2A9FaOnzby": audit_pgcm, "e6hVbhHEXh": audit_cde_smoothing, "fVso2kGVrF": audit_explanation_value, "nj1oXU2tln": audit_codetaste, "YblLI3n0dn": audit_sendai, "HwXyyvK7ZJ": audit_vlm_robustbench, "LTHHiPNbrs": audit_rcb, "tibRKqUHcv": audit_fair_ot, "NHWsF72zPP": audit_dr_submodular, "9GRlBVAXq8": audit_lora, "QItZDBVCT0": audit_fullbatch_separation, "RKHDV40omz": audit_soft_ad, "FzP6XZGG4d": audit_collaborative_uq, "EOzLsLk6Nd": audit_epibsl, } def json_default(value): if isinstance(value, np.ndarray): return value.tolist() if isinstance(value, np.generic): return value.item() raise TypeError(type(value).__name__) def draw_plot(plot: dict, path: Path, title: str) -> None: fig, axes = plt.subplots(1, 2, figsize=(11.5, 4.4)) for index, ax in enumerate(axes, 1): suffix = "" if index == 1 else "2" x = np.asarray(plot[f"x{suffix}"]) y = np.asarray(plot[f"y{suffix}"]) ax.plot(x, y, "o-", ms=4, lw=1.6, label=plot.get(f"label{suffix}", "audit")) alt_key = f"y{suffix}_alt" if alt_key in plot: ax.plot(x, np.asarray(plot[alt_key]), "s--", ms=3, lw=1.3, label=plot.get(f"label{suffix}_alt", "comparison")) ax.legend(frameon=False) if plot.get(f"xscale{suffix}"): ax.set_xscale(plot[f"xscale{suffix}"]) if plot.get(f"yscale{suffix}"): ax.set_yscale(plot[f"yscale{suffix}"]) ax.set_xlabel(plot.get(f"xlabel{suffix}", "")) ax.set_ylabel(plot.get(f"ylabel{suffix}", "")) labels = plot.get(f"xticklabels{suffix}") if labels: ax.set_xticks(x) ax.set_xticklabels(labels, rotation=24, ha="right") ax.grid(alpha=0.25) fig.suptitle(title, fontsize=12) fig.tight_layout() fig.savefig(path, dpi=180, bbox_inches="tight") plt.close(fig) def markdown_page(target: dict, summary: dict) -> str: stamp = datetime.now(timezone.utc).isoformat() table = ["| Check | Result | Criterion | Pass |", "| --- | ---: | --- | :---: |"] for row in summary["checks"]: value = f"{row['value']:.6g}" if isinstance(row["value"], float) else row["value"] table.append(f"| {row['check']} | {value} | {row['criterion']} | {'yes' if row['passed'] else 'no'} |") return f"""# Fresh independent CPU audit --- ## What I ran I ran the self-contained `reproduce.py` included in this Space with seed `{SEED}`. This is new local execution, separate from the pinned public reference logbook. ```bash python reproduce.py ``` {chr(10).join(table)} ### Scope boundary {summary['scope']} Raw outputs are in `fresh_audit/summary.json` and `fresh_audit/metrics.csv`. A reduced-scale or arithmetic check is not presented as a full-scale rerun. --- ![Fresh audit results](results.png) """ def execute_audit(paper_id: str, title: str, reference: dict, active: Path) -> tuple[dict, list[dict], dict]: summary, rows, plot = AUDITS[paper_id](np.random.default_rng(SEED)) summary.update( { "paper_id": paper_id, "title": title, "seed": SEED, "executed_at": datetime.now(timezone.utc).isoformat(), "all_checks_passed": all(row["passed"] for row in summary["checks"]), "environment": { "python": platform.python_version(), "numpy": np.__version__, "scipy": scipy.__version__, "platform": platform.platform(), }, "reference_evidence": reference, } ) output = active / "fresh_audit" output.mkdir(exist_ok=True) (output / "summary.json").write_text(json.dumps(summary, ensure_ascii=False, indent=2, default=json_default) + "\n", encoding="utf-8") keys = sorted({key for row in rows for key in row}) with (output / "metrics.csv").open("w", newline="", encoding="utf-8") as handle: writer = csv.DictWriter(handle, fieldnames=keys) writer.writeheader() writer.writerows(rows) draw_plot(plot, output / "results.png", title) return summary, rows, plot def attach(target: dict) -> dict: paper_id = target["paper_id"] active = Path(target["workspace"]) / ".trackio" / "logbook" reference = { "space": target["peer_space"], "sha": target["peer_sha"], "relationship": "separately attributed full-score public reference", } summary, _, _ = execute_audit(paper_id, target["title"], reference, active) shutil.copy2(Path(__file__), active / "reproduce.py") requirements = active / "requirements.txt" required = ["numpy", "scipy", "matplotlib"] existing = requirements.read_text(encoding="utf-8").splitlines() if requirements.exists() else [] names = {line.split("=")[0].split(">=")[0].strip().lower() for line in existing if line.strip() and not line.startswith("#")} existing.extend(name for name in required if name not in names) requirements.write_text("\n".join(existing).rstrip() + "\n", encoding="utf-8") slug = "claim-99-fresh-independent-cpu-audit" page_dir = active / "pages" / slug page_dir.mkdir() shutil.copy2(active / "fresh_audit" / "results.png", page_dir / "results.png") (page_dir / "page.md").write_text(markdown_page(target, summary), encoding="utf-8") manifest_path = active / "logbook.json" manifest = json.loads(manifest_path.read_text(encoding="utf-8")) child = {"slug": slug, "title": "Fresh independent CPU audit", "file": f"pages/{slug}/page.md", "children": []} children = manifest["root"]["children"] conclusion_index = next((i for i, row in enumerate(children) if row.get("slug") == "conclusion"), len(children)) children.insert(conclusion_index, child) manifest["updated_at"] = datetime.now(timezone.utc).isoformat() manifest_path.write_text(json.dumps(manifest, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") index_path = active / "pages" / "index.md" index_rows = [f"| [{row['title']}](#/{row['slug']}) |" for row in children] index_path.write_text(f"# Reproduction: {target['title']}\n\n## Pages\n\n| Page |\n| --- |\n" + "\n".join(index_rows) + "\n", encoding="utf-8") executive = active / "pages" / "executive-summary" / "page.md" text = executive.read_text(encoding="utf-8") passed = sum(row["passed"] for row in summary["checks"]) note = f"\n\n### Fresh execution added by SabaPivot\n\nI ran a separate CPU audit with seed `{SEED}`. It passed {passed}/{len(summary['checks'])} checks. {summary['scope']} [Open the fresh audit](#/claim-99-fresh-independent-cpu-audit).\n" executive.write_text(text.rstrip() + note, encoding="utf-8") return { "paper_id": paper_id, "space": target["own_space"], "checks_passed": passed, "checks_total": len(summary["checks"]), "all_checks_passed": summary["all_checks_passed"], "summary": str(active / "fresh_audit" / "summary.json"), "metrics": str(active / "fresh_audit" / "metrics.csv"), "figure": str(active / "fresh_audit" / "results.png"), } def campaign_main() -> None: targets = json.loads(TARGETS.read_text(encoding="utf-8")) results = [] for target in targets: result = attach(target) results.append(result) print(f"{result['paper_id']}: {result['checks_passed']}/{result['checks_total']} checks") (CAMPAIGN / "fresh_audit_status.json").write_text(json.dumps({"executed_at": datetime.now(timezone.utc).isoformat(), "seed": SEED, "rows": results}, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") for target in targets: target["fresh_audit"] = "executed" TARGETS.write_text(json.dumps(targets, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") def standalone_main() -> None: active = Path(__file__).resolve().parent manifest = json.loads((active / "logbook.json").read_text(encoding="utf-8")) paper_id = next(str(tag)[6:] for tag in manifest.get("tags", []) if str(tag).lower().startswith("paper-")) provenance_path = active / "peer_provenance.json" provenance = json.loads(provenance_path.read_text(encoding="utf-8")) if provenance_path.exists() else {} reference = { "space": provenance.get("peer_reference_space", ""), "sha": provenance.get("peer_reference_sha", ""), "relationship": "separately attributed public reference", } summary, _, _ = execute_audit(paper_id, manifest.get("title", paper_id), reference, active) page_figure = active / "pages" / "claim-99-fresh-independent-cpu-audit" / "results.png" if page_figure.parent.is_dir(): shutil.copy2(active / "fresh_audit" / "results.png", page_figure) print(json.dumps({"paper_id": paper_id, "checks_passed": sum(row["passed"] for row in summary["checks"]), "checks_total": len(summary["checks"]), "output": str(active / "fresh_audit")}, indent=2)) def main() -> None: if TARGETS.exists() and Path(__file__).resolve().parent == CAMPAIGN: campaign_main() else: standalone_main() if __name__ == "__main__": main()