| |
| """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 |
| |
| --- |
| <!-- trackio-cell |
| {{"type": "markdown", "id": "wave9_{target['paper_id']}_fresh", "created_at": "{stamp}", "title": "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. |
| |
| --- |
| <!-- trackio-cell |
| {{"type": "figure", "id": "wave9_{target['paper_id']}_plot", "created_at": "{stamp}", "title": "Fresh audit results"}} |
| --> |
|  |
| """ |
|
|
|
|
| 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() |
|
|
