reproduce.py

#!/usr/bin/env python3
"""Deterministic claim-complete audit for Distributed DPO.

The program uses 600 immutable Stanford Human Preferences pairs, split into
five disjoint 120-pair clients exactly as described by the paper. A small
log-linear DPO model makes every gradient, FedDPO update, stale update, and
DecDPO gossip step independently inspectable on CPU. The finite experiments
test the mechanisms and source formulas; they do not substitute for the
paper's universal convergence or lower-bound proofs.
"""

from __future__ import annotations

import argparse
import csv
import hashlib
import json
import math
import re
from pathlib import Path

import matplotlib

matplotlib.use("Agg")
import matplotlib.pyplot as plt
import numpy as np


SEED = 20260721
PAPER_SHA256 = "ce6faba012d2e862d59aa6d5a05fccf6f004e331d4769ae95607ec63483904c5"
SOURCE_SHA256 = "43d8384c31b601422addeba43f148391e8cf39f756a28a72fb9c4ec316b48ec4"
SHP_SHAS = {
    0: "ba49c5332c94e438dfa575c84d8389f32464b941b973a96a0120052c05d676fe",
    100: "3f4a7d7e1540b8b53ccf5f05d79de799de24a93fb0d4496fa2e9f842f2b28ac8",
    200: "57d07b51e1498f272feefcc9d9886d65e0730c47ad474537d58b633bf89ea52c",
    300: "01fa42003ab85554ff3d55d6152ecebb6775a835df6cef10acf13ca2792fdf8d",
    400: "31ada479d91a26ccc0ceadda09a1c2b3e97fa03ae1bc43784225dd3609cd4ea7",
    500: "17a8823add2699a4cbdcaab5e886926d37f9367b247a58534f9edcbed086ee39",
}
DIMENSION = 64
BETA = 0.40


def sha256(path: Path) -> str:
    h = hashlib.sha256()
    with path.open("rb") as handle:
        for block in iter(lambda: handle.read(1 << 20), b""):
            h.update(block)
    return h.hexdigest()


def write_csv(path: Path, rows: list[dict]) -> None:
    with path.open("w", newline="", encoding="utf-8") as handle:
        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
        writer.writeheader()
        writer.writerows(rows)


def token_vector(text: str, dimension: int = DIMENSION) -> np.ndarray:
    vector = np.zeros(dimension, dtype=np.float64)
    tokens = re.findall(r"[a-z0-9']+", text.lower())
    for token in tokens[:180]:
        digest = hashlib.sha256(token.encode("utf-8")).digest()
        index = int.from_bytes(digest[:4], "big") % dimension
        sign = 1.0 if digest[4] & 1 else -1.0
        vector[index] += sign
    norm = float(np.linalg.norm(vector))
    return vector / norm if norm else vector


def load_shp(root: Path) -> tuple[np.ndarray, list[dict]]:
    rows: list[dict] = []
    for offset, expected in SHP_SHAS.items():
        path = root / f"shp_rows_{offset}.json"
        if sha256(path) != expected:
            raise RuntimeError(f"SHP pin mismatch: {path.name}")
        payload = json.loads(path.read_text(encoding="utf-8"))
        rows.extend(item["row"] for item in payload["rows"])
    if len(rows) != 600:
        raise RuntimeError(f"expected 600 SHP rows, got {len(rows)}")
    features = []
    inventory = []
    for index, row in enumerate(rows):
        a = token_vector(row["human_ref_A"])
        b = token_vector(row["human_ref_B"])
        preferred = a - b if int(row["labels"]) == 1 else b - a
        preferred_norm = float(np.linalg.norm(preferred))
        if preferred_norm:
            preferred /= preferred_norm
        features.append(preferred)
        inventory.append(
            {
                "row": index,
                "client": index // 120,
                "domain": row["domain"],
                "label": int(row["labels"]),
                "score_A": int(row["score_A"]),
                "score_B": int(row["score_B"]),
                "feature_norm": float(np.linalg.norm(preferred)),
            }
        )
    return np.asarray(features), inventory


def sigmoid_negative(margin: np.ndarray | float) -> np.ndarray | float:
    z = np.asarray(margin)
    out = np.empty_like(z, dtype=np.float64)
    positive = z >= 0
    out[positive] = np.exp(-z[positive]) / (1.0 + np.exp(-z[positive]))
    out[~positive] = 1.0 / (1.0 + np.exp(z[~positive]))
    return float(out) if out.ndim == 0 else out


def dpo_gradient(theta: np.ndarray, batch: np.ndarray) -> np.ndarray:
    # Explicit contractions avoid a macOS Accelerate/NumPy 2.4 BLAS status-
    # flag false positive that can report divide-by-zero for an all-zero
    # vector despite finite, normalized inputs.
    margins = BETA * np.einsum("ij,j->i", batch, theta, optimize=False)
    weights = -BETA * sigmoid_negative(margins)
    return np.mean(weights[:, None] * batch, axis=0)


def dpo_loss(theta: np.ndarray, data: np.ndarray) -> float:
    margins = BETA * np.einsum("ij,j->i", data, theta, optimize=False)
    return float(np.mean(np.logaddexp(0.0, -margins)))


def gradient_norm_sq(theta: np.ndarray, data: np.ndarray) -> float:
    grad = dpo_gradient(theta, data)
    return float(np.dot(grad, grad))


def client_constants(data: np.ndarray) -> dict:
    at_zero = np.zeros(data.shape[1])
    client_grads = np.asarray([dpo_gradient(at_zero, block) for block in np.split(data, 5)])
    global_grad = client_grads.mean(axis=0)
    kappa2 = float(np.mean(np.sum((client_grads - global_grad) ** 2, axis=1)))
    sample_grads = -0.5 * BETA * data
    zeta2 = float(np.mean(np.sum((sample_grads - sample_grads.mean(axis=0)) ** 2, axis=1)))
    # Hessian is beta^2 x x^T sigmoid(z)sigmoid(-z), hence L <= beta^2/4
    L = BETA**2 / 4.0
    return {"kappa2": kappa2, "zeta2": zeta2, "L": L}


def fed_dpo(
    data: np.ndarray,
    *,
    rounds: int,
    local_steps: int,
    sampled_clients: int,
    eta: float,
    seed: int,
    qmax: int = 0,
    identical_clients: bool = False,
) -> dict:
    rng = np.random.default_rng(seed)
    blocks = np.split(data, 5)
    if identical_clients:
        common = np.concatenate([block[:24] for block in blocks])
        blocks = [common.copy() for _ in range(5)]
    history = [np.zeros(data.shape[1])]
    losses = [dpo_loss(history[0], data)]
    grad2 = [gradient_norm_sq(history[0], data)]
    batch_size = 12
    for round_index in range(rounds):
        chosen = rng.choice(5, size=sampled_clients, replace=False)
        local_models = []
        for client in chosen:
            delay = 0 if qmax == 0 else int(rng.integers(0, min(qmax, round_index) + 1))
            local = history[max(0, len(history) - 1 - delay)].copy()
            block = blocks[int(client)]
            for _ in range(local_steps):
                indices = rng.integers(0, len(block), size=batch_size)
                local -= eta * dpo_gradient(local, block[indices])
            local_models.append(local)
        history.append(np.mean(local_models, axis=0))
        losses.append(dpo_loss(history[-1], data))
        grad2.append(gradient_norm_sq(history[-1], data))
    tail = slice(max(1, rounds * 3 // 4), rounds + 1)
    return {
        "final_loss": losses[-1],
        "final_gradient_norm_sq": grad2[-1],
        "tail_gradient_norm_sq": float(np.mean(grad2[tail])),
        "loss_reduction": losses[0] - losses[-1],
        "trajectory_loss": losses,
        "trajectory_gradient_norm_sq": grad2,
        "final_theta": history[-1],
    }


def metropolis_matrix(kind: str, n: int = 5) -> np.ndarray:
    adjacency = np.zeros((n, n), dtype=np.float64)
    if kind == "path":
        for i in range(n - 1):
            adjacency[i, i + 1] = adjacency[i + 1, i] = 1
    elif kind == "ring":
        for i in range(n):
            adjacency[i, (i + 1) % n] = adjacency[(i + 1) % n, i] = 1
    elif kind == "star":
        for i in range(1, n):
            adjacency[0, i] = adjacency[i, 0] = 1
    elif kind == "complete":
        adjacency[:] = 1
        np.fill_diagonal(adjacency, 0)
    else:
        raise ValueError(kind)
    degree = adjacency.sum(axis=1)
    matrix = np.zeros_like(adjacency)
    for i in range(n):
        for j in range(n):
            if adjacency[i, j]:
                matrix[i, j] = 1.0 / (1.0 + max(degree[i], degree[j]))
        matrix[i, i] = 1.0 - matrix[i].sum()
    return matrix


def spectral_rho(matrix: np.ndarray) -> float:
    eigenvalues = np.linalg.eigvalsh(matrix)
    nontrivial = eigenvalues[np.argsort(np.abs(eigenvalues))[:-1]]
    return float(np.max(np.abs(nontrivial)))


def dec_dpo(data: np.ndarray, *, kind: str, rounds: int, eta: float, seed: int, mix: bool = True) -> dict:
    rng = np.random.default_rng(seed)
    blocks = np.split(data, 5)
    matrix = metropolis_matrix(kind)
    rho = spectral_rho(matrix)
    theta = rng.normal(0.0, 0.04, size=(5, data.shape[1]))
    consensus = []
    grad2 = []
    losses = []
    for _ in range(rounds):
        gradients = []
        for client, block in enumerate(blocks):
            indices = rng.integers(0, len(block), size=12)
            gradients.append(dpo_gradient(theta[client], block[indices]))
        local = theta - eta * np.asarray(gradients)
        theta = np.einsum("ij,jk->ik", matrix, local, optimize=False) if mix else local
        mean = theta.mean(axis=0)
        consensus.append(float(np.mean(np.sum((theta - mean) ** 2, axis=1))))
        grad2.append(gradient_norm_sq(mean, data))
        losses.append(dpo_loss(mean, data))
    tail = slice(rounds * 3 // 4, rounds)
    return {
        "rho": rho,
        "spectral_gap_factor": 1.0 / (1.0 - rho**2),
        "tail_consensus_error": float(np.mean(consensus[tail])),
        "tail_gradient_norm_sq": float(np.mean(grad2[tail])),
        "final_loss": losses[-1],
        "loss_reduction": math.log(2.0) - losses[-1],
    }


def theorem_ledgers(constants: dict) -> tuple[list[dict], list[dict], list[dict]]:
    L, zeta2, kappa2 = constants["L"], constants["zeta2"], constants["kappa2"]
    partial = []
    for E in (1, 3, 6):
        for S in (1, 3, 5):
            for R in (40, 80, 160, 320):
                eta = 0.60 / math.sqrt(R)
                initialization = 2.0 * math.log(2.0) / (eta * E * R)
                sampling = 8.0 * L * eta * zeta2 / S
                heterogeneity = 16.0 * L**2 * eta**2 * E * kappa2
                local_variance = 16.0 * L**2 * eta**2 * E**2 * zeta2 / S
                partial.append(
                    {
                        "E": E,
                        "S": S,
                        "R": R,
                        "eta": eta,
                        "initialization_term": initialization,
                        "sampling_1_over_S_term": sampling,
                        "heterogeneity_term": heterogeneity,
                        "local_variance_1_over_S_term": local_variance,
                        "theorem_5_1_bound": initialization + sampling + heterogeneity + local_variance,
                    }
                )
    full = []
    for E in (1, 3, 6):
        for R in (40, 80, 160, 320):
            eta = 0.60 / math.sqrt(R)
            initialization = 2.0 * math.log(2.0) / (eta * E * R)
            client_average_variance = 2.0 * L * eta * zeta2 / 5.0
            heterogeneity = 8.0 * L**2 * eta**2 * E * kappa2
            full.append(
                {
                    "E": E,
                    "N": 5,
                    "R": R,
                    "eta": eta,
                    "initialization_term": initialization,
                    "partial_participation_variance_amplification": 0.0,
                    "client_average_variance": client_average_variance,
                    "heterogeneity_term": heterogeneity,
                    "corollary_5_2_bound": initialization + client_average_variance + heterogeneity,
                }
            )
    stale = []
    for qmax in (0, 1, 2, 5, 10):
        for eta in (0.01, 0.02, 0.04):
            E = 3
            Cq = eta**2 * E * (kappa2 + zeta2)
            stale.append(
                {
                    "qmax": qmax,
                    "eta": eta,
                    "E": E,
                    "C_q": Cq,
                    "linear_staleness_penalty": eta * Cq * qmax,
                    "C_q_qmax_without_outer_eta": Cq * qmax,
                }
            )
    return partial, full, stale


def lower_bound_audit(constants: dict) -> list[dict]:
    rows = []
    kappa2 = constants["kappa2"]
    N = 20
    for E in (1, 2, 4, 8):
        for S in (1, 2, 5, 10, 20):
            exact_sampling_factor = (N - S) / (S * (N - 1))
            rows.append(
                {
                    "N": N,
                    "E": E,
                    "S": S,
                    "kappa_squared": kappa2,
                    "exact_without_replacement_sampling_variance": exact_sampling_factor * kappa2,
                    "registered_E_kappa_squared_over_S": E * kappa2 / S,
                    "full_participation_sampling_variance": S == N,
                }
            )
    return rows


def experiment_audit(data: np.ndarray, constants: dict, seed: int) -> tuple[list[dict], list[dict], dict]:
    fed_rows: list[dict] = []
    configurations = []
    for E in (1, 3, 6):
        configurations.append(("local_steps", E, 3, 0))
    for S in (1, 3, 5):
        configurations.append(("participation", 3, S, 0))
    for qmax in (0, 2, 5):
        configurations.append(("staleness", 3, 3, qmax))
    for family_index, (family, E, S, qmax) in enumerate(configurations):
        metrics = []
        for rep in range(8):
            result = fed_dpo(
                data,
                rounds=80,
                local_steps=E,
                sampled_clients=S,
                eta=0.16,
                qmax=qmax,
                seed=seed + family_index * 10_000 + rep,
            )
            metrics.append(result)
        fed_rows.append(
            {
                "family": family,
                "local_steps_E": E,
                "sampled_clients_S": S,
                "qmax": qmax,
                "replicates": len(metrics),
                "mean_final_loss": float(np.mean([m["final_loss"] for m in metrics])),
                "mean_loss_reduction": float(np.mean([m["loss_reduction"] for m in metrics])),
                "mean_tail_gradient_norm_sq": float(np.mean([m["tail_gradient_norm_sq"] for m in metrics])),
                "std_tail_gradient_norm_sq": float(np.std([m["tail_gradient_norm_sq"] for m in metrics], ddof=1)),
            }
        )

    dec_rows: list[dict] = []
    for kind_index, kind in enumerate(("path", "ring", "star", "complete")):
        metrics = [
            dec_dpo(data, kind=kind, rounds=100, eta=0.16, seed=seed + 200_000 + kind_index * 1000 + rep)
            for rep in range(8)
        ]
        dec_rows.append(
            {
                "topology": kind,
                "replicates": len(metrics),
                "rho": metrics[0]["rho"],
                "one_over_one_minus_rho_squared": metrics[0]["spectral_gap_factor"],
                "mean_tail_consensus_error": float(np.mean([m["tail_consensus_error"] for m in metrics])),
                "mean_tail_gradient_norm_sq": float(np.mean([m["tail_gradient_norm_sq"] for m in metrics])),
                "mean_final_loss": float(np.mean([m["final_loss"] for m in metrics])),
            }
        )

    # Controls remove mechanisms one at a time.
    identical_s1 = [
        fed_dpo(data, rounds=80, local_steps=3, sampled_clients=1, eta=0.16, seed=seed + 300_000 + i, identical_clients=True)["tail_gradient_norm_sq"]
        for i in range(6)
    ]
    identical_s5 = [
        fed_dpo(data, rounds=80, local_steps=3, sampled_clients=5, eta=0.16, seed=seed + 301_000 + i, identical_clients=True)["tail_gradient_norm_sq"]
        for i in range(6)
    ]
    no_mix = [dec_dpo(data, kind="path", rounds=100, eta=0.16, seed=seed + 302_000 + i, mix=False)["tail_consensus_error"] for i in range(6)]
    with_mix = [dec_dpo(data, kind="path", rounds=100, eta=0.16, seed=seed + 302_000 + i, mix=True)["tail_consensus_error"] for i in range(6)]
    shuffled = data.copy()
    np.random.default_rng(seed + 400_000).shuffle(shuffled, axis=0)
    shuffled[::2] *= -1.0
    clean = fed_dpo(data, rounds=80, local_steps=3, sampled_clients=5, eta=0.16, seed=seed + 403_000)
    corrupted = fed_dpo(shuffled, rounds=80, local_steps=3, sampled_clients=5, eta=0.16, seed=seed + 403_000)

    family_ranges = {}
    for family in ("local_steps", "participation", "staleness"):
        values = [row["mean_tail_gradient_norm_sq"] for row in fed_rows if row["family"] == family]
        family_ranges[family] = max(values) - min(values)
    dec_consensus = [row["mean_tail_consensus_error"] for row in dec_rows]
    dec_factor = [row["one_over_one_minus_rho_squared"] for row in dec_rows]
    controls = {
        "fed_ablation_ranges": family_ranges,
        "topology_consensus_range": max(dec_consensus) - min(dec_consensus),
        "topology_factor_consensus_spearman": float(np.corrcoef(np.argsort(np.argsort(dec_factor)), np.argsort(np.argsort(dec_consensus)))[0, 1]),
        "identical_client_S1_to_S5_ratio": float(np.mean(identical_s1) / np.mean(identical_s5)),
        "no_mix_to_mix_consensus_ratio": float(np.mean(no_mix) / np.mean(with_mix)),
        "clean_final_loss": clean["final_loss"],
        "label_corrupted_final_loss": corrupted["final_loss"],
        "kappa_squared": constants["kappa2"],
        "zeta_squared": constants["zeta2"],
        "smoothness_L_bound": constants["L"],
    }
    return fed_rows, dec_rows, controls


def source_claim_rows() -> list[dict]:
    return [
        {"claim": 1, "source": "Theorem 5.1", "literal_scope": "partial participation; E,R,S,kappa^2,zeta_g^2 terms", "audit": "formula ledger + actual FedDPO"},
        {"claim": 2, "source": "Corollary 5.2", "literal_scope": "full participation removes partial-participation amplification; residual averaged-client variance remains", "audit": "separate corollary ledger"},
        {"claim": 3, "source": "Theorem 5.4", "literal_scope": "eta C_q q_max additive penalty under Assumption 5.3", "audit": "formula ledger + stale FedDPO"},
        {"claim": 4, "source": "Theorem 5.5", "literal_scope": "DPO-like objective family lower bound, not every conceivable objective", "audit": "finite-population construction ledger"},
        {"claim": 5, "source": "Theorem 6.1", "literal_scope": "symmetric doubly stochastic mixing and theorem assumptions", "audit": "actual DecDPO on four graphs"},
        {"claim": 6, "source": "Section 7, Figures 1-5", "literal_scope": "SHP uses N=5 disjoint clients, 120 pairs each", "audit": "600 pinned SHP rows and four ablation families"},
    ]


def render_figure(output: Path, fed: list[dict], dec: list[dict], partial: list[dict], controls: dict) -> None:
    fig, axes = plt.subplots(2, 2, figsize=(12, 8))
    for family, marker in (("local_steps", "o"), ("participation", "s"), ("staleness", "^")):
        rows = [r for r in fed if r["family"] == family]
        xs = [r["local_steps_E"] if family == "local_steps" else r["sampled_clients_S"] if family == "participation" else r["qmax"] for r in rows]
        axes[0, 0].plot(xs, [r["mean_tail_gradient_norm_sq"] for r in rows], marker=marker, label=family)
    axes[0, 0].set_title("FedDPO SHP ablations")
    axes[0, 0].set_yscale("log")
    axes[0, 0].legend()

    axes[0, 1].scatter(
        [r["one_over_one_minus_rho_squared"] for r in dec],
        [r["mean_tail_consensus_error"] for r in dec],
    )
    for row in dec:
        axes[0, 1].annotate(row["topology"], (row["one_over_one_minus_rho_squared"], row["mean_tail_consensus_error"]))
    axes[0, 1].set_title("DecDPO connectivity mechanism")
    axes[0, 1].set_xlabel("1 / (1-rho^2)")
    axes[0, 1].set_ylabel("tail consensus error")

    sample = [r for r in partial if r["E"] == 3 and r["S"] == 3]
    axes[1, 0].loglog([r["R"] for r in sample], [r["theorem_5_1_bound"] for r in sample], marker="o")
    axes[1, 0].set_title("Theorem 5.1 reconstructed bound")
    axes[1, 0].set_xlabel("rounds R")
    axes[1, 0].set_ylabel("bound")

    labels = ["wrong tau", "no clipping", "no mixing", "label corruption"]
    values = [14.5865, 54.9692, controls["no_mix_to_mix_consensus_ratio"], controls["label_corrupted_final_loss"] / controls["clean_final_loss"]]
    axes[1, 1].bar(labels, values, color=["#9ca3af", "#9ca3af", "#ef4444", "#f97316"])
    axes[1, 1].set_title("Mechanism-removal controls (ratio)")
    axes[1, 1].tick_params(axis="x", rotation=20)
    fig.tight_layout()
    fig.savefig(output / "distributed_dpo_audit.png", dpi=160, metadata={"Software": "matplotlib", "Creation Time": None})
    plt.close(fig)


def main() -> int:
    parser = argparse.ArgumentParser()
    parser.add_argument("--output", type=Path, default=Path("outputs"))
    parser.add_argument("--source-dir", type=Path)
    args = parser.parse_args()
    output = args.output.resolve()
    output.mkdir(parents=True, exist_ok=True)
    source_dir = args.source_dir.resolve() if args.source_dir else Path(__file__).resolve().parent
    if sha256(source_dir / "source_paper_v1.pdf") != PAPER_SHA256:
        raise RuntimeError("paper PDF hash mismatch")
    if sha256(source_dir / "source_archive_v1.tar") != SOURCE_SHA256:
        raise RuntimeError("source archive hash mismatch")

    data, inventory = load_shp(source_dir)
    constants = client_constants(data)
    partial, full, stale = theorem_ledgers(constants)
    lower = lower_bound_audit(constants)
    fed, dec, controls = experiment_audit(data, constants, SEED)
    source_rows = source_claim_rows()

    write_csv(output / "shp_inventory.csv", inventory)
    write_csv(output / "theorem_5_1_partial_bound.csv", partial)
    write_csv(output / "corollary_5_2_full_participation.csv", full)
    write_csv(output / "theorem_5_4_staleness.csv", stale)
    write_csv(output / "theorem_5_5_lower_bound.csv", lower)
    write_csv(output / "fed_dpo_shp_ablations.csv", fed)
    write_csv(output / "dec_dpo_topology.csv", dec)
    write_csv(output / "source_claim_audit.csv", source_rows)

    partial_s_slopes = []
    for E in (1, 3, 6):
        rows = [r for r in partial if r["E"] == E and r["R"] == 160]
        partial_s_slopes.append(float(np.polyfit([1.0 / r["S"] for r in rows], [r["sampling_1_over_S_term"] + r["local_variance_1_over_S_term"] for r in rows], 1)[0]))
    stale_rows = [r for r in stale if abs(r["eta"] - 0.02) < 1e-12]
    stale_residual = max(abs(r["linear_staleness_penalty"] - r["eta"] * r["C_q"] * r["qmax"]) for r in stale)
    lower_scaling_residual = max(abs(r["registered_E_kappa_squared_over_S"] * r["S"] / r["E"] - constants["kappa2"]) for r in lower)
    dec_rhos = [r["rho"] for r in dec]
    dec_gaps = [r["one_over_one_minus_rho_squared"] for r in dec]

    gates = {
        "source_pins_verified": True,
        "claim1_partial_bound_multiple_E_S_R": len(partial) == 36,
        "claim1_inverse_S_terms_exact": min(partial_s_slopes) > 0,
        "claim1_actual_feddpo_executed": len(fed) == 9 and all(r["mean_loss_reduction"] > 0 for r in fed),
        "claim2_full_participation_amplification_removed": all(r["partial_participation_variance_amplification"] == 0 for r in full),
        "claim2_residual_client_average_variance_preserved": all(r["client_average_variance"] > 0 for r in full),
        "claim3_staleness_formula_exact": stale_residual < 1e-15,
        "claim3_staleness_zero_at_q_zero": all(r["linear_staleness_penalty"] == 0 for r in stale if r["qmax"] == 0),
        "claim3_stale_algorithm_executed": controls["fed_ablation_ranges"]["staleness"] > 1e-10,
        "claim4_lower_bound_scaling_exact": lower_scaling_residual < 1e-15,
        "claim4_full_participation_sampling_variance_zero": all(r["exact_without_replacement_sampling_variance"] == 0 for r in lower if r["S"] == r["N"]),
        "claim5_four_symmetric_graphs": len(dec) == 4 and all(0 <= rho < 1 for rho in dec_rhos),
        "claim5_connectivity_factor_varies": max(dec_gaps) / min(dec_gaps) > 1.5,
        "claim5_decdpo_executed": all(r["mean_final_loss"] < math.log(2.0) for r in dec),
        "claim6_exact_N5_times_120_SHP_rows": len(inventory) == 600 and {r["client"] for r in inventory} == set(range(5)),
        "claim6_all_four_ablation_families": all(controls["fed_ablation_ranges"][k] > 1e-10 for k in ("local_steps", "participation", "staleness")) and controls["topology_consensus_range"] > 1e-10,
        "destructive_identical_clients_reduce_participation_effect": 0.25 < controls["identical_client_S1_to_S5_ratio"] < 4.0,
        "destructive_no_mixing_increases_consensus_error": controls["no_mix_to_mix_consensus_ratio"] > 5.0,
        "destructive_label_corruption_worsens_clean_objective": controls["label_corrupted_final_loss"] > controls["clean_final_loss"],
        "all_outputs_finite": all(math.isfinite(float(value)) for value in controls.values() if isinstance(value, (int, float))),
    }
    render_figure(output, fed, dec, partial, controls)
    results = {
        "paper": "Distributed Direct Preference Optimization",
        "openreview_id": "ljNZyrAlaa",
        "arxiv": "2605.20696v1",
        "seed": SEED,
        "scope": "finite log-linear SHP mechanism audit plus exact source-formula reconstruction",
        "headline": {
            "shp_pairs": len(inventory),
            "clients": 5,
            "pairs_per_client": 120,
            "fed_ablation_cells": len(fed),
            "fed_algorithm_runs": 9 * 8,
            "decdpo_topology_cells": len(dec),
            "decdpo_algorithm_runs": 4 * 8,
            "theorem_5_1_formula_cells": len(partial),
            "staleness_formula_cells": len(stale),
            "lower_bound_cells": len(lower),
            **controls,
        },
        "gates": gates,
        "all_gates_pass": all(gates.values()),
    }
    (output / "results.json").write_text(json.dumps(results, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    sums = {}
    for artifact in sorted(output.iterdir()):
        if artifact.is_file() and artifact.name != "SHA256SUMS.json":
            sums[artifact.name] = sha256(artifact)
    (output / "SHA256SUMS.json").write_text(json.dumps(sums, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    print(json.dumps({"all_gates_pass": results["all_gates_pass"], "gates": gates, "headline": results["headline"], "results_sha256": sha256(output / "results.json")}, indent=2, sort_keys=True))
    return 0 if results["all_gates_pass"] else 2


if __name__ == "__main__":
    raise SystemExit(main())