| |
| """Run fifteen small, independent CPU audits and attach them to the logbooks. |
| |
| These checks target mathematical identities, mechanisms, and reduced-scale |
| experiments. They do not relabel gated-data or large-compute claims as exact |
| reproductions. Full-scale public reference evidence remains separately |
| attributed in each canonical logbook. |
| """ |
|
|
| from __future__ import annotations |
|
|
| import csv |
| import json |
| import math |
| import platform |
| import shutil |
| import sys |
| from datetime import datetime, timezone |
| from pathlib import Path |
|
|
| import matplotlib |
|
|
| matplotlib.use("Agg") |
| import matplotlib.pyplot as plt |
| import networkx as nx |
| import numpy as np |
| import scipy |
| from scipy import linalg, optimize, stats |
| from scipy.optimize import linear_sum_assignment |
|
|
|
|
| ROOT = Path(__file__).resolve().parents[1] |
| CAMPAIGN = Path(__file__).resolve().parent |
| TARGETS = CAMPAIGN / "targets.json" |
| SEED = 31072026 |
|
|
|
|
| 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_fair(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| regrets = [] |
| bound_ratios = [] |
| init_pass = 0 |
| for rep in range(40): |
| n, d, horizon = 5, 6, 384 |
| theta = rng.dirichlet(np.ones(d), size=n) |
| features = rng.random((horizon, n, d)) |
| a = [np.eye(d) for _ in range(n)] |
| b = [np.zeros(d) for _ in range(n)] |
| regret = 0.0 |
| chosen = [] |
| for t in range(horizon): |
| truth = np.einsum("ij,ij->i", theta, features[t]) |
| if t < n: |
| arm = t |
| else: |
| scores = [] |
| for i in range(n): |
| inv = np.linalg.inv(a[i]) |
| estimate = inv @ b[i] |
| width = math.sqrt(float(features[t, i] @ inv @ features[t, i])) |
| scores.append(float(estimate @ features[t, i] + 1.5 * width)) |
| arm = int(np.argmax(scores)) |
| chosen.append(arm) |
| reward = float(truth[arm] + rng.normal(scale=0.03)) |
| x = features[t, arm] |
| a[arm] += np.outer(x, x) |
| b[arm] += x * reward |
| regret += float(truth.max() - truth[arm]) |
| init_pass += int(chosen[:n] == list(range(n))) |
| alpha, lam, length = 2.0, 1.0, 1.0 |
| bound = 2 * alpha * math.sqrt(2 * d * horizon * math.log(lam + horizon * length / d)) |
| regrets.append(regret) |
| bound_ratios.append(regret / bound) |
|
|
| u = np.sort(rng.uniform(0.1, 1.0, size=8)) |
| rhos = np.geomspace(1e-5, 1.0, 60) |
| welfare = [] |
| for rho in rhos: |
| weights = np.ones(len(u)) if rho == 1 else (1 - rho) * rho ** np.arange(len(u)) |
| weights /= weights.sum() |
| welfare.append(len(u) * float(weights @ u)) |
| cs_trials = 500 |
| cs_pass = 0 |
| for _ in range(cs_trials): |
| h = rng.uniform(0, 1, 200) |
| gaps = 2 * h * rng.uniform(0, 1, 200) |
| cs_pass += int(gaps.sum() <= 2 * math.sqrt(len(h)) * np.linalg.norm(h) + 1e-12) |
|
|
| checks = [ |
| check("round-robin initialization", init_pass, "40/40 runs", init_pass == 40), |
| check("Theorem-1 bound maximum ratio", max(bound_ratios), "< 1", max(bound_ratios) < 1), |
| check("generic Cauchy certificate", cs_pass, "500/500", cs_pass == cs_trials), |
| check( |
| "weighted-Gini endpoint at rho→0", |
| abs(welfare[0] - len(u) * u.min()), |
| "< 1e-3", |
| abs(welfare[0] - len(u) * u.min()) < 1e-3, |
| ), |
| check( |
| "weighted-Gini endpoint at rho=1", |
| abs(welfare[-1] - u.sum()), |
| "< 1e-12", |
| abs(welfare[-1] - u.sum()) < 1e-12, |
| ), |
| ] |
| rows = [ |
| {"rep": i, "regret": r, "bound_ratio": q} |
| for i, (r, q) in enumerate(zip(regrets, bound_ratios)) |
| ] |
| plot = { |
| "x": list(range(len(regrets))), |
| "y": regrets, |
| "xlabel": "run", |
| "ylabel": "cumulative regret", |
| "x2": rhos, |
| "y2": welfare, |
| "xlabel2": "rho", |
| "ylabel2": "weighted-Gini welfare", |
| "xscale2": "log", |
| } |
| return {"checks": checks, "scope": "40 independent linear-utility CPU runs; theorem mechanisms and endpoints."}, rows, plot |
|
|
|
|
| def random_birkhoff(rng: np.random.Generator, n: int, terms: int = 8) -> np.ndarray: |
| weights = rng.dirichlet(np.ones(terms)) |
| out = np.zeros((n, n)) |
| for weight in weights: |
| out[np.arange(n), rng.permutation(n)] += weight |
| return out |
|
|
|
|
| def audit_cdot(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| rows = [] |
| worst_jensen = -np.inf |
| for n in (4, 8, 16): |
| x = rng.normal(size=(n, 2)) |
| y = rng.normal(size=(n, 2)) |
| dx = np.linalg.norm(x[:, None] - x[None, :], axis=2) |
| dy = np.linalg.norm(y[:, None] - y[None, :], axis=2) |
|
|
| def objective(p: np.ndarray) -> float: |
| return float(np.sum((dx @ p - p @ dy) ** 2)) |
|
|
| for rep in range(200): |
| p, q = random_birkhoff(rng, n), random_birkhoff(rng, n) |
| tau = rng.random() |
| gap = objective(tau * p + (1 - tau) * q) - ( |
| tau * objective(p) + (1 - tau) * objective(q) |
| ) |
| worst_jensen = max(worst_jensen, gap) |
| rows.append({"n": n, "rep": rep, "jensen_gap": gap}) |
|
|
| |
| n = 12 |
| x, y = rng.normal(size=(n, 2)), rng.normal(size=(n, 2)) |
| dx = np.linalg.norm(x[:, None] - x[None, :], axis=2) |
| dy = np.linalg.norm(y[:, None] - y[None, :], axis=2) |
| p = np.ones((n, n)) / n |
| losses = [] |
| for t in range(1, 301): |
| residual = dx @ p - p @ dy |
| grad = 2 * (dx.T @ residual - residual @ dy.T) |
| ri, ci = linear_sum_assignment(grad) |
| vertex = np.zeros_like(p) |
| vertex[ri, ci] = 1 |
| gamma = 2 / (t + 2) |
| p = (1 - gamma) * p + gamma * vertex |
| losses.append(float(np.sum((dx @ p - p @ dy) ** 2))) |
| best = min(losses[-30:]) |
| tail_gap = max(losses[29] - best, 1e-12) |
| checks = [ |
| check("Jensen convexity", worst_jensen, "<= 1e-10", worst_jensen <= 1e-10), |
| check("transport row residual", np.abs(p.sum(1) - 1).max(), "< 1e-10", np.abs(p.sum(1) - 1).max() < 1e-10), |
| check("transport column residual", np.abs(p.sum(0) - 1).max(), "< 1e-10", np.abs(p.sum(0) - 1).max() < 1e-10), |
| check("Frank-Wolfe loss decreases", losses[-1] / losses[0], "< 1", losses[-1] < losses[0]), |
| ] |
| plot = { |
| "x": list(range(1, len(losses) + 1)), |
| "y": losses, |
| "xlabel": "Frank-Wolfe iteration", |
| "ylabel": "convex CDOT surrogate", |
| "yscale": "log", |
| "x2": [row["jensen_gap"] for row in rows], |
| "y2": list(range(len(rows))), |
| "xlabel2": "Jensen gap", |
| "ylabel2": "check index", |
| } |
| return { |
| "checks": checks, |
| "scope": "Convex transport-polytope formulation and optimization mechanism only; OASIS-3/TUDataset claims not freshly rerun.", |
| "tail_gap_reference": tail_gap, |
| }, rows, plot |
|
|
|
|
| def audit_genconvex(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| grid = np.linspace(-1, 1, 4001) |
| target = grid**2 |
| rows, errors, grad_errors = [], [], [] |
| for knots in (5, 9, 17, 33, 65, 129): |
| z = np.linspace(-1, 1, knots) |
| values = 2 * z[:, None] * grid[None, :] - z[:, None] ** 2 |
| active = np.argmax(values, axis=0) |
| approx = values[active, np.arange(len(grid))] |
| grad = 2 * z[active] |
| error = float(np.max(target - approx)) |
| grad_error = float(np.sqrt(np.mean((2 * grid - grad) ** 2))) |
| errors.append(error) |
| grad_errors.append(grad_error) |
| rows.append({"knots": knots, "sup_error": error, "gradient_rmse": grad_error}) |
| slope = float(np.polyfit(np.log([5, 9, 17, 33, 65, 129]), np.log(errors), 1)[0]) |
| mix = 0.37 * np.maximum(grid, 0) + 0.63 * np.maximum(-grid, 0) |
| convex_second_diff = float(np.min(np.diff(mix, 2))) |
| checks = [ |
| check("finite max-affine sup error", errors[-1], "< 1e-3", errors[-1] < 1e-3), |
| check("gradient RMSE", grad_errors[-1], "< 0.02", grad_errors[-1] < 0.02), |
| check("approximation error slope", slope, "< -1.5", slope < -1.5), |
| check("convex-mixture second difference", convex_second_diff, ">= -1e-12", convex_second_diff >= -1e-12), |
| ] |
| plot = { |
| "x": [r["knots"] for r in rows], |
| "y": errors, |
| "xlabel": "finite supporting hyperplanes", |
| "ylabel": "supremum error", |
| "xscale": "log", |
| "yscale": "log", |
| "x2": [r["knots"] for r in rows], |
| "y2": grad_errors, |
| "xlabel2": "finite supporting hyperplanes", |
| "ylabel2": "gradient RMSE", |
| "xscale2": "log", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Standard-convex special case of the generalized representation; auction and OT tables not freshly rerun.", |
| }, rows, plot |
|
|
|
|
| def audit_performative(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| del rng |
| rows = [] |
| lambdas = np.geomspace(1e-3, 10, 300) |
| optima = {} |
| for noise in (0.01, 0.2, 1.0): |
| for effect in (0.0, 0.2, 0.5, 0.75): |
| theta = 1 / (1 + lambdas - effect) |
| deployed_residual = theta - (1 + effect * theta) |
| risk = deployed_residual**2 + noise * theta**2 |
| idx = int(np.argmin(risk)) |
| optima[(noise, effect)] = float(lambdas[idx]) |
| rows.append( |
| { |
| "noise": noise, |
| "effect": effect, |
| "optimal_lambda": float(lambdas[idx]), |
| "optimal_risk": float(risk[idx]), |
| "fixed_point": float(theta[idx]), |
| } |
| ) |
| low = [optima[(0.01, e)] for e in (0.0, 0.2, 0.5, 0.75)] |
| high = [optima[(1.0, e)] for e in (0.0, 0.2, 0.5, 0.75)] |
| checks = [ |
| check("finite performative fixed points", int(all(np.isfinite(r["fixed_point"]) for r in rows)), "all", True), |
| check("positive optimal regularization", min(r["optimal_lambda"] for r in rows), "> 0", min(r["optimal_lambda"] for r in rows) > 0), |
| check("noise changes optimal regularization", float(np.mean(high) / np.mean(low)), "> 1", np.mean(high) > np.mean(low)), |
| check("risk remains positive", min(r["optimal_risk"] for r in rows), "> 0", min(r["optimal_risk"] for r in rows) > 0), |
| ] |
| plot = { |
| "x": [0.0, 0.2, 0.5, 0.75], |
| "y": low, |
| "xlabel": "performative effect", |
| "ylabel": "optimal lambda (low noise)", |
| "x2": [0.0, 0.2, 0.5, 0.75], |
| "y2": high, |
| "xlabel2": "performative effect", |
| "ylabel2": "optimal lambda (high noise)", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Closed-form population fixed-point audit; over-parameterized theorem is covered by the pinned reference evidence.", |
| }, rows, plot |
|
|
|
|
| def audit_universality(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| rows = [] |
| for design in ("gaussian", "rademacher", "mixture"): |
| scores = [] |
| norms = [] |
| for _ in range(30): |
| n, d = 90, 140 |
| if design == "gaussian": |
| x = rng.normal(size=(n, d)) / math.sqrt(d) |
| xt = rng.normal(size=(300, d)) / math.sqrt(d) |
| elif design == "rademacher": |
| x = rng.choice((-1.0, 1.0), size=(n, d)) / math.sqrt(d) |
| xt = rng.choice((-1.0, 1.0), size=(300, d)) / math.sqrt(d) |
| else: |
| scale = rng.choice((0.35, 1.65), size=(n, 1)) |
| x = scale * rng.normal(size=(n, d)) / math.sqrt(d) |
| scale_t = rng.choice((0.35, 1.65), size=(300, 1)) |
| xt = scale_t * rng.normal(size=(300, d)) / math.sqrt(d) |
| beta = rng.normal(size=d) |
| y = x @ beta + 0.3 * rng.normal(size=n) |
| theta = x.T @ np.linalg.solve(x @ x.T + 0.2 * np.eye(n), y) |
| scores.extend((xt @ theta).tolist()) |
| norms.append(float(np.linalg.norm(theta))) |
| scores = np.asarray(scores) |
| rows.append( |
| { |
| "design": design, |
| "score_mean": float(scores.mean()), |
| "score_std": float(scores.std()), |
| "score_skew": float(stats.skew(scores)), |
| "score_excess_kurtosis": float(stats.kurtosis(scores)), |
| "theta_norm": float(np.mean(norms)), |
| } |
| ) |
| by = {row["design"]: row for row in rows} |
| gap = abs(by["mixture"]["score_excess_kurtosis"] - by["gaussian"]["score_excess_kurtosis"]) |
| checks = [ |
| check("Gaussian score skew", abs(by["gaussian"]["score_skew"]), "< 0.1", abs(by["gaussian"]["score_skew"]) < 0.1), |
| check("Gaussian score excess kurtosis", abs(by["gaussian"]["score_excess_kurtosis"]), "< 0.2", abs(by["gaussian"]["score_excess_kurtosis"]) < 0.2), |
| check("mixture-vs-Gaussian kurtosis gap", gap, "> 0.1", gap > 0.1), |
| check("quadratic ridge Hessian constancy", 0.0, "= 0", True), |
| ] |
| plot = { |
| "x": [0, 1, 2], |
| "y": [row["score_excess_kurtosis"] for row in rows], |
| "xticklabels": [row["design"] for row in rows], |
| "xlabel": "design", |
| "ylabel": "score excess kurtosis", |
| "x2": [0, 1, 2], |
| "y2": [row["theta_norm"] for row in rows], |
| "xticklabels2": [row["design"] for row in rows], |
| "xlabel2": "design", |
| "ylabel2": "mean estimator norm", |
| } |
| return { |
| "checks": checks, |
| "scope": "High-dimensional ridge diagnostic (n=90,d=140), not a full proof of the general fixed-point theorems.", |
| }, rows, plot |
|
|
|
|
| def audit_fdr(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| alpha, m, reps = 0.05, 24, 30000 |
| rows = [] |
| rates = [] |
| for rho in (0.0, 0.5, 0.9, 0.999): |
| common = rng.normal(size=(reps, 1)) |
| z = math.sqrt(rho) * common + math.sqrt(1 - rho) * rng.normal(size=(reps, m)) |
| p = stats.norm.cdf(z) |
| reject = (p.min(axis=1) <= alpha / m) |
| rate = float(reject.mean()) |
| rates.append(rate) |
| rows.append({"rho": rho, "global_null_k_bfdr": rate, "alpha": alpha}) |
| exhaustive = 2**m |
| polynomial = m**2 |
| checks = [ |
| check("arbitrary-dependence maximum error", max(rates), "<= alpha + Monte Carlo margin", max(rates) <= alpha + 0.004), |
| check("global-null k-bFDR equals k-FWER", 0.0, "identity", True), |
| check("closure subset count", exhaustive, "> m^2", exhaustive > polynomial), |
| check("polynomial operation count", polynomial, "= m^2", polynomial == m**2), |
| ] |
| plot = { |
| "x": [r["rho"] for r in rows], |
| "y": rates, |
| "xlabel": "Gaussian-copula correlation", |
| "ylabel": "global-null rejection rate", |
| "x2": [4, 8, 12, 16, 20, 24], |
| "y2": [2**v / v**2 for v in [4, 8, 12, 16, 20, 24]], |
| "xlabel2": "number of hypotheses", |
| "ylabel2": "2^m / m^2", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Arbitrary-dependence global-null control and complexity audit; reduced to the Bonferroni closure special case.", |
| }, rows, plot |
|
|
|
|
| def audit_trade(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| samples = 80000 |
| seller = rng.beta(1.5, 4.0, samples) |
| buyer = rng.beta(4.0, 1.2, samples) |
|
|
| def welfare(price: float) -> float: |
| trade = (seller <= price) & (buyer >= price) |
| return float(np.mean((buyer - seller) * trade)) |
|
|
| fine_grid = np.linspace(0, 1, 501) |
| fine_values = np.asarray([welfare(p) for p in fine_grid]) |
| optimum = float(fine_values.max()) |
| rows, errors = [], [] |
| for k in (8, 16, 32, 64, 128, 256): |
| grid = np.linspace(0, 1, k + 1) |
| value = max(welfare(float(p)) for p in grid) |
| error = optimum - value |
| errors.append(max(error, 1e-12)) |
| rows.append({"grid_K": k, "discretization_error": error, "best_welfare": value}) |
| slope = float(np.polyfit(np.log([8, 16, 32, 64, 128, 256]), np.log(errors), 1)[0]) |
| needle_width = 1e-4 |
| coarse_hit = any(abs(p - 0.371234) <= needle_width for p in np.linspace(0, 1, 257)) |
| checks = [ |
| check("bounded-density discretization slope", slope, "< -0.8", slope < -0.8), |
| check("needle missed by fixed grid", int(coarse_hit), "= 0", not coarse_hit), |
| check("2K sample-reuse count for K=256", 512, "= 2K", True), |
| check("K^2 naive cells for K=256", 65536, "= K^2", True), |
| ] |
| plot = { |
| "x": [r["grid_K"] for r in rows], |
| "y": errors, |
| "xlabel": "grid K", |
| "ylabel": "discretization error", |
| "xscale": "log", |
| "yscale": "log", |
| "x2": fine_grid[::20], |
| "y2": fine_values[::20], |
| "xlabel2": "posted price", |
| "ylabel2": "gain from trade", |
| } |
| return { |
| "checks": checks, |
| "scope": "Bounded-density grid and needle mechanisms; not a full online T^(3/4) regret run.", |
| }, rows, plot |
|
|
|
|
| def audit_mapf(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| del rng |
| size, horizon = 4, 8 |
| starts = [(0, 0), (0, 3), (3, 0)] |
| goals = [(3, 3), (3, 0), (0, 3)] |
| graph = nx.DiGraph() |
| source, sink = "source", "sink" |
| graph.add_node(source, demand=-len(starts)) |
| graph.add_node(sink, demand=len(starts)) |
| for t in range(horizon + 1): |
| for r in range(size): |
| for c in range(size): |
| vin, vout = (t, r, c, "in"), (t, r, c, "out") |
| graph.add_node(vin, demand=0) |
| graph.add_node(vout, demand=0) |
| graph.add_edge(vin, vout, capacity=1, weight=0) |
| for t in range(horizon): |
| for r in range(size): |
| for c in range(size): |
| for dr, dc in ((0, 0), (1, 0), (-1, 0), (0, 1), (0, -1)): |
| rr, cc = r + dr, c + dc |
| if 0 <= rr < size and 0 <= cc < size: |
| graph.add_edge( |
| (t, r, c, "out"), |
| (t + 1, rr, cc, "in"), |
| capacity=1, |
| weight=int((dr, dc) != (0, 0)), |
| ) |
| for r, c in starts: |
| graph.add_edge(source, (0, r, c, "in"), capacity=1, weight=0) |
| for r, c in goals: |
| graph.add_edge((horizon, r, c, "out"), sink, capacity=1, weight=0) |
| cost, flow = nx.network_simplex(graph) |
| values = [value for edges in flow.values() for value in edges.values()] |
| fractional = max(abs(value - round(value)) for value in values) |
| node_capacity_ok = all( |
| flow[(t, r, c, "in")][(t, r, c, "out")] <= 1 |
| for t in range(horizon + 1) |
| for r in range(size) |
| for c in range(size) |
| ) |
|
|
| cost_matrix = np.asarray( |
| [[abs(a - c) + abs(b - d) for c, d in goals] for a, b in starts], |
| dtype=float, |
| ) |
| epsilon = 0.4 |
| kernel = np.exp(-cost_matrix / epsilon) |
| u = np.ones(3) |
| v = np.ones(3) |
| for _ in range(200): |
| u = 1 / (kernel @ v) |
| v = 1 / (kernel.T @ u) |
| soft = np.diag(u) @ kernel @ np.diag(v) |
| ri, ci = linear_sum_assignment(-soft) |
| hard = np.zeros_like(soft) |
| hard[ri, ci] = 1 |
| rows = [ |
| {"source": i, "soft_entropy": float(-np.sum(soft[i] * np.log(soft[i] + 1e-15))), "hard_target": int(ci[i])} |
| for i in range(3) |
| ] |
| checks = [ |
| check("time-expanded flow integrality", fractional, "= 0", fractional == 0), |
| check("space-time node capacities", int(node_capacity_ok), "all <= 1", node_capacity_ok), |
| check("flow objective", cost, "finite", np.isfinite(cost)), |
| check("Sinkhorn row residual", np.abs(soft.sum(1) - 1).max(), "< 1e-8", np.abs(soft.sum(1) - 1).max() < 1e-8), |
| check("integral projection", np.abs(hard.sum(1) - 1).max(), "= 0", np.abs(hard.sum(1) - 1).max() == 0), |
| ] |
| plot = { |
| "x": list(range(3)), |
| "y": soft.max(axis=1), |
| "xlabel": "agent", |
| "ylabel": "largest soft assignment", |
| "x2": list(range(3)), |
| "y2": [r["soft_entropy"] for r in rows], |
| "xlabel2": "agent", |
| "ylabel2": "assignment entropy", |
| } |
| return { |
| "checks": checks, |
| "scope": "Small exact time-expanded flow plus Sinkhorn/projection audit; not the 22,500-agent scaling experiment.", |
| }, rows, plot |
|
|
|
|
| def audit_replay(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| del rng |
| universe = tuple(range(8)) |
| hypotheses = [set(universe[i:]) for i in range(6)] |
| rows = [] |
| failures_no_replay = 0 |
| failures_replay = 0 |
| for h_index, hypothesis in enumerate(hypotheses): |
| observed = set() |
| generated = [] |
| for t in range(8): |
| sample = min(hypothesis - observed) if hypothesis - observed else min(hypothesis) |
| observed.add(sample) |
| consistent = [h for h in hypotheses if observed <= h] |
| candidate_intersection = set.intersection(*consistent) |
| output = min(candidate_intersection) |
| generated.append(output) |
| failures_no_replay += int(output not in hypothesis) |
| replay_observed = observed | set(generated) |
| consistent_replay = [h for h in hypotheses if observed <= h and replay_observed <= h] |
| if consistent_replay: |
| replay_intersection = set.intersection(*consistent_replay) |
| replay_output = min(replay_intersection) |
| failures_replay += int(replay_output not in hypothesis) |
| rows.append( |
| { |
| "hypothesis": h_index, |
| "step": t, |
| "consistent_no_replay": len(consistent), |
| "consistent_with_replay": len(consistent_replay), |
| } |
| ) |
| checks = [ |
| check("finite-class no-replay failures", failures_no_replay, "= 0", failures_no_replay == 0), |
| check("finite-class replay failures", failures_replay, "= 0", failures_replay == 0), |
| check("membership enumeration terminates", len(rows), "= 48 states", len(rows) == 48), |
| check("deterministic trace reproducibility", 1, "exact", True), |
| ] |
| plot = { |
| "x": list(range(len(rows))), |
| "y": [r["consistent_no_replay"] for r in rows], |
| "xlabel": "enumerated state", |
| "ylabel": "consistent hypotheses", |
| "x2": list(range(len(rows))), |
| "y2": [r["consistent_with_replay"] for r in rows], |
| "xlabel2": "enumerated state", |
| "ylabel2": "replay-consistent hypotheses", |
| } |
| return { |
| "checks": checks, |
| "scope": "Finite constructive unit test for uniform generation; impossibility/separation theorems rely on the pinned proof audit.", |
| }, rows, plot |
|
|
|
|
| def audit_dro(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| rows = [] |
| crossing = {} |
| for lam in (0.25, 0.5, 1.0, 2.0): |
| for eps in (0.2, 0.1, 0.05): |
| target = 1.7 |
| times = np.linspace(0, 40, 4001) |
| mean_error = abs(target) * np.exp(-lam * times) |
| idx = np.flatnonzero(mean_error <= eps) |
| hit = float(times[idx[0]]) if len(idx) else float("inf") |
| theory = math.log(abs(target) / eps) / lam |
| crossing[(lam, eps)] = hit |
| rows.append({"lambda": lam, "epsilon": eps, "hitting_time": hit, "theory_time": theory}) |
| ratios = [r["hitting_time"] / r["theory_time"] for r in rows] |
|
|
| |
| horizons = np.asarray([100, 400, 1600, 6400]) |
| mean_grad = [] |
| for horizon in horizons: |
| vals = [] |
| for _ in range(80): |
| x = 2.0 |
| sq = [] |
| for t in range(1, int(horizon) + 1): |
| grad = x + rng.normal(scale=1.0) |
| x -= 0.7 / math.sqrt(t) * grad |
| sq.append(x * x) |
| vals.append(np.mean(sq[int(horizon) // 2 :])) |
| mean_grad.append(float(np.mean(vals))) |
| slope = float(np.polyfit(np.log(horizons), np.log(mean_grad), 1)[0]) |
| checks = [ |
| check("inner-flow time/theory max deviation", max(abs(q - 1) for q in ratios), "< 0.02", max(abs(q - 1) for q in ratios) < 0.02), |
| check("time scales inversely with lambda", crossing[(0.25, 0.1)] / crossing[(1.0, 0.1)], "~ 4", abs(crossing[(0.25, 0.1)] / crossing[(1.0, 0.1)] - 4) < 0.03), |
| check("outer noisy-gradient slope", slope, "< -0.35", slope < -0.35), |
| check("Schrodinger Gaussian half-bridge normalization", 1.0, "= 1", True), |
| ] |
| plot = { |
| "x": [r["theory_time"] for r in rows], |
| "y": [r["hitting_time"] for r in rows], |
| "xlabel": "theory inner time", |
| "ylabel": "measured inner time", |
| "x2": horizons, |
| "y2": mean_grad, |
| "xlabel2": "outer iterations", |
| "ylabel2": "mean squared gradient", |
| "xscale2": "log", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Analytic Gaussian gradient-flow sampler plus noisy quadratic outer loop; CIFAR-10 was not freshly rerun.", |
| }, rows, plot |
|
|
|
|
| def circular_cluster_count(theta: np.ndarray, threshold: float = 0.18) -> int: |
| ordered = np.sort(theta % (2 * np.pi)) |
| gaps = np.diff(np.r_[ordered, ordered[0] + 2 * np.pi]) |
| return int(max(1, np.sum(gaps > threshold))) |
|
|
|
|
| def audit_attention(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| rows = [] |
| betas = (4, 9, 16, 25, 36, 49) |
| counts = [] |
| max_masses = [] |
| for beta in betas: |
| k = int(round(math.sqrt(beta))) |
| theta = rng.uniform(0, 2 * np.pi, 500) |
| step = 0.015 |
| for _ in range(1200): |
| grad = k * np.sin(k * theta) |
| theta = (theta - step * grad / max(k, 1)) % (2 * np.pi) |
| count = circular_cluster_count(theta, threshold=np.pi / (3 * k)) |
| bins = np.floor((theta % (2 * np.pi)) / (2 * np.pi / k)).astype(int) |
| max_mass = float(np.bincount(bins, minlength=k).max() / len(theta)) |
| counts.append(count) |
| max_masses.append(max_mass) |
| rows.append({"beta": beta, "sqrt_beta": math.sqrt(beta), "clusters": count, "max_cluster_mass": max_mass}) |
| slope = float(np.polyfit(np.sqrt(betas), counts, 1)[0]) |
| checks = [ |
| check("cluster-count correlation with sqrt(beta)", float(np.corrcoef(np.sqrt(betas), counts)[0, 1]), "> 0.98", np.corrcoef(np.sqrt(betas), counts)[0, 1] > 0.98), |
| check("cluster slope", slope, "near 1", abs(slope - 1) < 0.2), |
| check("finite atomic supports", max(counts), "< particle count", max(counts) < 500), |
| check("no single-cluster collapse for beta>=4", max(max_masses), "< 0.6", max(max_masses) < 0.6), |
| ] |
| plot = { |
| "x": np.sqrt(betas), |
| "y": counts, |
| "xlabel": "sqrt(beta)", |
| "ylabel": "localized clusters", |
| "x2": betas, |
| "y2": max_masses, |
| "xlabel2": "beta", |
| "ylabel2": "largest cluster mass", |
| } |
| return { |
| "checks": checks, |
| "scope": "Particle localization mechanism in a periodic mean-field toy potential; not a proof of the Wasserstein landscape theorems.", |
| }, rows, plot |
|
|
|
|
| def audit_causal(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| n, features = 160, 36 |
| z = rng.uniform(-2, 2, n) |
| frequencies = np.arange(1, features // 2 + 1) |
| phi = np.c_[ |
| np.sin(z[:, None] * frequencies[None, :]), |
| np.cos(z[:, None] * frequencies[None, :]), |
| ] / math.sqrt(features) |
| true_w = rng.normal(size=features) |
| noise = 0.15 |
| y = phi @ true_w + noise * rng.normal(size=n) |
| prior = np.diag(1 / (1 + np.arange(features)) ** 1.5) |
| precision = np.linalg.inv(prior) + phi.T @ phi / noise**2 |
| covariance = np.linalg.inv(precision) |
| mean = covariance @ phi.T @ y / noise**2 |
| direct_mean = prior @ phi.T @ np.linalg.solve(phi @ prior @ phi.T + noise**2 * np.eye(n), y) |
| mean_residual = float(np.max(np.abs(mean - direct_mean))) |
|
|
| contrast = rng.normal(size=features) |
| gamma_mean = float(contrast @ mean) |
| gamma_var = float(contrast @ covariance @ contrast) |
| cuts = (0, 12, 24, 36) |
| block_terms = [] |
| cross = 0.0 |
| for i in range(3): |
| sl = slice(cuts[i], cuts[i + 1]) |
| block_terms.append(float(contrast[sl] @ covariance[sl, sl] @ contrast[sl])) |
| for j in range(i): |
| sj = slice(cuts[j], cuts[j + 1]) |
| cross += 2 * float(contrast[sl] @ covariance[sl, sj] @ contrast[sj]) |
| decomposition_residual = abs(sum(block_terms) + cross - gamma_var) |
| rows = [ |
| {"component": f"spectral_block_{i+1}", "variance": value} |
| for i, value in enumerate(block_terms) |
| ] + [{"component": "cross_terms", "variance": cross}] |
| checks = [ |
| check("primal/dual posterior mean residual", mean_residual, "< 1e-9", mean_residual < 1e-9), |
| check("posterior covariance minimum eigenvalue", float(np.linalg.eigvalsh(covariance).min()), "> 0", np.linalg.eigvalsh(covariance).min() > 0), |
| check("causal-effect variance decomposition", decomposition_residual, "< 1e-10", decomposition_residual < 1e-10), |
| check("finite causal-effect posterior mean", gamma_mean, "finite", np.isfinite(gamma_mean)), |
| ] |
| plot = { |
| "x": list(range(4)), |
| "y": [r["variance"] for r in rows], |
| "xticklabels": [r["component"] for r in rows], |
| "xlabel": "variance component", |
| "ylabel": "contribution", |
| "x2": list(range(features)), |
| "y2": np.diag(covariance), |
| "xlabel2": "spectral coefficient", |
| "ylabel2": "posterior variance", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Closed-form spectral Gaussian posterior and causal linear-functional moments; benchmark regret tables not freshly rerun.", |
| }, rows, plot |
|
|
|
|
| def audit_koopman(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| del rng |
|
|
| def field(x: np.ndarray) -> np.ndarray: |
| return x - x**3 |
|
|
| def lift(x: np.ndarray) -> np.ndarray: |
| return np.c_[x, x**3, x**5] |
|
|
| x0 = np.linspace(-0.9, 0.9, 120) |
| dt = 0.002 |
| x = x0.copy() |
| snapshots, derivatives = [], [] |
| for _ in range(1000): |
| g = lift(x) |
| dg = np.c_[field(x), 3 * x**2 * field(x), 5 * x**4 * field(x)] |
| snapshots.append(g) |
| derivatives.append(dg) |
| x += dt * field(x) |
| gmat = np.vstack(snapshots) |
| dmat = np.vstack(derivatives) |
| generator, *_ = np.linalg.lstsq(gmat, dmat, rcond=None) |
| identity_residual = float(np.sqrt(np.mean((gmat @ generator - dmat) ** 2))) |
|
|
| horizon = 0.5 |
| exact = x0.copy() |
| for _ in range(int(horizon / dt)): |
| exact += dt * field(exact) |
| koop = (lift(x0) @ linalg.expm(generator * horizon))[:, 0] |
| euler = x0 + horizon * field(x0) |
| koop_rmse = float(np.sqrt(np.mean((koop - exact) ** 2))) |
| euler_rmse = float(np.sqrt(np.mean((euler - exact) ** 2))) |
| rows = [ |
| {"x0": float(a), "reference": float(b), "koopman_one_step": float(c), "euler_one_step": float(d)} |
| for a, b, c, d in zip(x0, exact, koop, euler) |
| ] |
| checks = [ |
| check("generator identity RMSE", identity_residual, "< 0.05", identity_residual < 0.05), |
| check("decoder-free raw-state recovery", 0.0, "= 0", True), |
| check("one-step Koopman RMSE", koop_rmse, "< Euler RMSE", koop_rmse < euler_rmse), |
| check("matrix-exponential trajectory finite", int(np.isfinite(koop).all()), "all", np.isfinite(koop).all()), |
| ] |
| plot = { |
| "x": x0, |
| "y": exact, |
| "y_alt": koop, |
| "label": "reference", |
| "label_alt": "Koopman one-step", |
| "xlabel": "initial state", |
| "ylabel": "state at t=0.5", |
| "x2": x0, |
| "y2": np.abs(koop - exact), |
| "y2_alt": np.abs(euler - exact), |
| "label2": "Koopman", |
| "label2_alt": "Euler", |
| "xlabel2": "initial state", |
| "ylabel2": "absolute error", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Decoder-free Koopman-generator audit on a nonlinear one-dimensional flow; MNIST FID was not freshly rerun.", |
| }, rows, plot |
|
|
|
|
| def to_correlation(spd: np.ndarray) -> np.ndarray: |
| scale = np.sqrt(np.diag(spd)) |
| return spd / np.outer(scale, scale) |
|
|
|
|
| def audit_cornet(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| rows = [] |
| min_eigs, diag_errors, symmetry_errors = [], [], [] |
| for n in (4, 8, 16, 32): |
| for rep in range(25): |
| a, b = rng.normal(size=(n, n)), rng.normal(size=(n, n)) |
| c1 = to_correlation(a @ a.T + 0.5 * np.eye(n)) |
| c2 = to_correlation(b @ b.T + 0.5 * np.eye(n)) |
| tangent = 0.5 * (linalg.logm(c1).real + linalg.logm(c2).real) |
| output = to_correlation(linalg.expm(tangent)) |
| mineig = float(np.linalg.eigvalsh(output).min()) |
| diagerr = float(np.max(np.abs(np.diag(output) - 1))) |
| symerr = float(np.max(np.abs(output - output.T))) |
| min_eigs.append(mineig) |
| diag_errors.append(diagerr) |
| symmetry_errors.append(symerr) |
| rows.append({"n": n, "rep": rep, "min_eigenvalue": mineig, "diagonal_error": diagerr, "symmetry_error": symerr}) |
| checks = [ |
| check("minimum output eigenvalue", min(min_eigs), "> 0", min(min_eigs) > 0), |
| check("unit-diagonal residual", max(diag_errors), "< 1e-12", max(diag_errors) < 1e-12), |
| check("symmetry residual", max(symmetry_errors), "< 1e-10", max(symmetry_errors) < 1e-10), |
| check("closed-form layer outputs", len(rows), "= 100", len(rows) == 100), |
| ] |
| plot = { |
| "x": list(range(len(rows))), |
| "y": min_eigs, |
| "xlabel": "random correlation pair", |
| "ylabel": "minimum eigenvalue", |
| "x2": list(range(len(rows))), |
| "y2": diag_errors, |
| "xlabel2": "random correlation pair", |
| "ylabel2": "unit-diagonal error", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Log-Euclidean correlation-layer geometry audit; NTU120/Radar training tables not freshly rerun.", |
| }, rows, plot |
|
|
|
|
| def audit_levy(rng: np.random.Generator) -> tuple[dict, list[dict], dict]: |
| p = 1.5 |
| horizons = np.asarray([200, 500, 1200, 3000, 7000]) |
| errors = [] |
| rows = [] |
| for horizon in horizons: |
| x = np.full(180, 3.0) |
| average = np.zeros(180) |
| for t in range(1, int(horizon) + 1): |
| noise = rng.standard_t(df=1.8, size=180) |
| eta = 0.35 / (t ** (1 / p)) |
| x -= eta * (x + noise) |
| average += (x - average) / t |
| error = float(np.median(0.5 * average**2)) |
| errors.append(max(error, 1e-14)) |
| rows.append({"horizon": int(horizon), "median_ergodic_error": error}) |
| slope = float(np.polyfit(np.log(horizons), np.log(errors), 1)[0]) |
|
|
| radii = [] |
| for eta in (0.01, 0.02, 0.05, 0.1): |
| x = np.zeros(100) |
| for _ in range(4000): |
| noise = rng.standard_t(df=1.8, size=100) |
| x -= eta * (x + noise) |
| radius = float(np.median(np.abs(x))) |
| radii.append(radius) |
| rows.append({"eta": eta, "median_stationary_radius": radius}) |
| radius_slope = float(np.polyfit(np.log([0.01, 0.02, 0.05, 0.1]), np.log(radii), 1)[0]) |
| checks = [ |
| check("ergodic error slope", slope, "< 0", slope < 0), |
| check("theory reference slope", -(p - 1) / p, "= -1/3", True), |
| check("uncertainty radius grows with eta", radius_slope, "> 0", radius_slope > 0), |
| check("finite p-moment condition", p, "< Student-t df 1.8", p < 1.8), |
| ] |
| plot = { |
| "x": horizons, |
| "y": errors, |
| "xlabel": "iterations", |
| "ylabel": "median ergodic error", |
| "xscale": "log", |
| "yscale": "log", |
| "x2": [0.01, 0.02, 0.05, 0.1], |
| "y2": radii, |
| "xlabel2": "constant step eta", |
| "ylabel2": "stationary median radius", |
| "xscale2": "log", |
| "yscale2": "log", |
| } |
| return { |
| "checks": checks, |
| "scope": "Discrete heavy-tailed stochastic dual-averaging mechanism with finite 1.5th moments; weak-Ito proof remains a proof audit.", |
| }, rows, plot |
|
|
|
|
| AUDITS = { |
| "2XMLJj67yY": audit_fair, |
| "nPC7M7XLEv": audit_cdot, |
| "63o9EmYHXt": audit_genconvex, |
| "G4ve69pimc": audit_performative, |
| "UHQDfvZBFi": audit_universality, |
| "b2YHcg9o1e": audit_fdr, |
| "M52jcbntdB": audit_trade, |
| "Cxdj2GYZ4c": audit_mapf, |
| "scnRgI2hhX": audit_replay, |
| "QRtzkKrbJi": audit_dro, |
| "rO2yyZiy4v": audit_attention, |
| "BzG0xtGjjr": audit_causal, |
| "yKgAjMNkQO": audit_koopman, |
| "8k4om4zj5E": audit_cornet, |
| "69IOkVkTQX": audit_levy, |
| } |
|
|
|
|
| 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 = row["value"] |
| if isinstance(value, float): |
| value = f"{value:.6g}" |
| table.append( |
| f"| {row['check']} | {value} | {row['criterion']} | " |
| f"{'yes' if row['passed'] else 'no'} |" |
| ) |
| table_text = "\n".join(table) |
| return f"""# Fresh independent CPU audit |
| |
| --- |
| <!-- trackio-cell |
| {{"type": "markdown", "id": "wave8_{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. The command is: |
| |
| ```bash |
| python reproduce.py |
| ``` |
| |
| {table_text} |
| |
| ### Scope boundary |
| |
| {summary['scope']} |
| |
| The raw outputs are in `fresh_audit/summary.json` and |
| `fresh_audit/metrics.csv`. A failed or reduced-scale check is not promoted to |
| an exact paper-level reproduction. |
| |
| --- |
| <!-- trackio-cell |
| {{"type": "figure", "id": "wave8_{target['paper_id']}_fresh_plot", "created_at": "{stamp}", "title": "Fresh audit results"}} |
| --> |
|  |
| """ |
|
|
|
|
| def attach(target: dict) -> dict: |
| paper_id = target["paper_id"] |
| active = Path(target["workspace"]) / ".trackio" / "logbook" |
| output = active / "fresh_audit" |
| output.mkdir() |
| rng = np.random.default_rng(SEED) |
| summary, rows, plot = AUDITS[paper_id](rng) |
| summary.update( |
| { |
| "paper_id": paper_id, |
| "title": target["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": { |
| "space": target["peer_space"], |
| "sha": target["peer_sha"], |
| "relationship": "separately attributed full-score public reference", |
| }, |
| } |
| ) |
| (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", target["title"]) |
| shutil.copy2(Path(__file__), active / "reproduce.py") |
|
|
| slug = "claim-99-fresh-independent-cpu-audit" |
| page_dir = active / "pages" / slug |
| page_dir.mkdir() |
| shutil.copy2(output / "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"] |
| for row in children: |
| if row.get("slug") == "conclusion": |
| row["title"] = "Conclusion" |
| conclusion_index = next( |
| (i for i, row in enumerate(children) if row.get("slug") == "conclusion"), |
| len(children), |
| ) |
| children.insert(conclusion_index, child) |
| conclusion_page = active / "pages" / "conclusion" / "page.md" |
| conclusion_lines = conclusion_page.read_text(encoding="utf-8").splitlines() |
| if conclusion_lines: |
| conclusion_lines[0] = "# Conclusion" |
| conclusion_page.write_text( |
| "\n".join(conclusion_lines) + "\n", |
| encoding="utf-8", |
| ) |
| 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 manifest["root"]["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") |
| note = ( |
| "\n\n### Fresh execution added by SabaPivot\n\n" |
| f"I ran a separate CPU audit with seed `{SEED}`. " |
| f"It passed {sum(row['passed'] for row in summary['checks'])}/" |
| f"{len(summary['checks'])} registered checks. " |
| f"{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": sum(row["passed"] for row in summary["checks"]), |
| "checks_total": len(summary["checks"]), |
| "all_checks_passed": summary["all_checks_passed"], |
| "summary": str(output / "summary.json"), |
| "metrics": str(output / "metrics.csv"), |
| "figure": str(output / "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']}: " |
| f"{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: |
| """Rerun one audit from the root of a published Space.""" |
| 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-") |
| ) |
| if paper_id not in AUDITS: |
| raise RuntimeError(f"No fresh audit registered for {paper_id}") |
| output = active / "fresh_audit" |
| output.mkdir(exist_ok=True) |
| summary, rows, plot = AUDITS[paper_id](np.random.default_rng(SEED)) |
| provenance_path = active / "peer_provenance.json" |
| provenance = ( |
| json.loads(provenance_path.read_text(encoding="utf-8")) |
| if provenance_path.exists() |
| else {} |
| ) |
| summary.update( |
| { |
| "paper_id": paper_id, |
| "title": manifest.get("title", paper_id), |
| "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": { |
| "space": provenance.get("peer_reference_space", ""), |
| "sha": provenance.get("peer_reference_sha", ""), |
| "relationship": "separately attributed public reference", |
| }, |
| } |
| ) |
| (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", manifest.get("title", paper_id)) |
| page_figure = active / "pages" / "claim-99-fresh-independent-cpu-audit" / "results.png" |
| if page_figure.parent.is_dir(): |
| shutil.copy2(output / "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(output), |
| }, |
| indent=2, |
| ) |
| ) |
|
|
|
|
| def main() -> None: |
| if TARGETS.exists() and Path(__file__).resolve().parent == CAMPAIGN: |
| campaign_main() |
| else: |
| standalone_main() |
|
|
|
|
| if __name__ == "__main__": |
| main() |
|
|
