code/scope_influential_broad_audit.py

#!/usr/bin/env python3
"""CPU-only scope audit for the six diffusion-flow claims.

The original executable evidence was concentrated on Gaussian/linear paths.
This extension keeps the checks analytic and deterministic while adding a
Laplace cusp, Student-t(3) tails, a Cauchy score, and a bounded uniform law.
It audits the dimension factors, conditional-only witness, schedule identity,
product-coupling factorization, and derivative-transfer identity.  It does not
pretend that finite quadrature replaces the paper's stochastic proofs.
"""

from __future__ import annotations

import itertools
import json
import math
from dataclasses import dataclass

import numpy as np
from scipy.integrate import quad
from scipy.stats import laplace, t as student_t


@dataclass(frozen=True)
class Family:
    name: str
    kind: str
    parameter: float = 1.0

    def pdf(self, x: float) -> float:
        if self.kind == "laplace":
            return math.exp(-abs(x) / self.parameter) / (2.0 * self.parameter)
        if self.kind == "student":
            return float(student_t.pdf(x, self.parameter))
        if self.kind == "uniform":
            return 0.5 if -1.0 <= x <= 1.0 else 0.0
        raise ValueError(self.kind)

    def score(self, x: float) -> float:
        if self.kind == "laplace":
            return -math.copysign(1.0, x) / self.parameter if x else 0.0
        if self.kind == "student":
            nu = self.parameter
            return -(nu + 1.0) * x / (nu + x * x)
        if self.kind == "uniform":
            return 0.0
        raise ValueError(self.kind)

    def quantile(self, u: float) -> float:
        if self.kind == "laplace":
            return float(laplace.ppf(u, scale=self.parameter))
        if self.kind == "student":
            return float(student_t.ppf(u, self.parameter))
        if self.kind == "uniform":
            return 2.0 * u - 1.0
        raise ValueError(self.kind)


FAMILIES = (
    Family("laplace-cusp", "laplace"),
    Family("student-t3", "student", 3.0),
    Family("cauchy-score", "student", 1.0),
    Family("uniform-boundary", "uniform"),
)
W2_FAMILIES = FAMILIES[:2] + (FAMILIES[3],)
DIMS = (1, 2, 4, 8, 16, 32, 64, 128, 256, 512, 1024, 2048, 4096)


def integrate(family: Family, fn) -> float:
    bounds = (-1.0, 1.0) if family.kind == "uniform" else (-math.inf, math.inf)
    value, error = quad(fn, *bounds, epsabs=2e-10, epsrel=2e-10, limit=500)
    if not math.isfinite(value) or error > max(1e-8, abs(value) * 1e-7):
        raise RuntimeError((family.name, value, error))
    return float(value)


def score_moments(family: Family) -> tuple[float, float, float, float]:
    return tuple(integrate(family, lambda x, p=p: family.pdf(x) * family.score(x) ** (2 * p)) for p in (1, 2, 3, 4))


def fourth_sum(dimension: int, moments: tuple[float, float, float, float]) -> float:
    m1, m2, m3, m4 = moments
    return (dimension * m4
            + 4 * dimension * (dimension - 1) * m3 * m1
            + 3 * dimension * (dimension - 1) * m2 * m2
            + 6 * dimension * (dimension - 1) * (dimension - 2) * m2 * m1 * m1
            + dimension * (dimension - 1) * (dimension - 2) * (dimension - 3) * m1**4)


def dimension_factor(dimension: int, moments: tuple[float, float, float, float], delta: float = 1.0) -> float:
    # This is the displayed d-dimensional score-moment factor from the source
    # audit, with the early-stop delta multiplier kept explicit.
    return dimension * (dimension * dimension / delta**4 + math.sqrt(fourth_sum(dimension, moments)))


def slope(rows: list[dict[str, float]], key: str) -> float:
    x = np.log(np.asarray([row["dimension"] for row in rows[-6:]], dtype=float))
    y = np.log(np.asarray([row[key] for row in rows[-6:]], dtype=float))
    return float(np.polyfit(x, y, 1)[0])


def claim1_claim2(moments: dict[str, tuple[float, float, float, float]]) -> tuple[list[dict[str, object]], list[dict[str, object]]]:
    claim1 = []
    claim2 = []
    for family in FAMILIES:
        rows = []
        for dimension in DIMS:
            factor = dimension_factor(dimension, moments[family.name])
            row = {"family": family.name, "dimension": dimension, "factor": factor, "factor_over_d3": factor / dimension**3}
            rows.append(row)
            claim1.append(row)
        family_slope = slope(rows, "factor")
        for row in rows:
            row["large_dimension_slope"] = family_slope
        for delta in (0.25, 0.125, 0.0625):
            early_rows = []
            for dimension in DIMS:
                factor = dimension_factor(dimension, moments[family.name], delta)
                early_rows.append({"family": family.name, "delta": delta, "dimension": dimension, "factor": factor})
            family_slope = slope(early_rows, "factor")
            for row in early_rows:
                row["large_dimension_slope"] = family_slope
                row["full_joint_has_density"] = False
                row["conditional_score_finite"] = True
                row["target"] = "discrete {-1,0,3/4}"
                claim2.append(row)
    return claim1, claim2


def claim3_schedule() -> list[dict[str, object]]:
    rows = []
    for family, dimension, h in itertools.product(FAMILIES, (1, 8, 64, 512), (1 / 8, 1 / 16, 1 / 32)):
        t = 0.5
        for _ in range(16):
            t = (t + h) / (1.0 + h) if t >= 0.5 else t + h
        closed = 1.0 - 0.5 * (1.0 + h) ** -16
        rows.append({"family": family.name, "dimension": dimension, "h": h, "endpoint": t, "closed_form_residual": abs(t - closed)})
    return rows


def w2(family_a: Family, family_b: Family, nodes: int = 256) -> float:
    points, weights = np.polynomial.legendre.leggauss(nodes)
    lo, hi = 1e-8, 1.0 - 1e-8
    value = 0.5 * (hi - lo) * sum(
        float(weight) * (family_a.quantile(0.5 * (hi - lo) * float(point) + 0.5 * (hi + lo))
                         - family_b.quantile(0.5 * (hi - lo) * float(point) + 0.5 * (hi + lo))) ** 2
        for point, weight in zip(points, weights)
    )
    return math.sqrt(value)


def claim5_product() -> list[dict[str, object]]:
    rows = []
    for prior, target, dimension in itertools.product(W2_FAMILIES, W2_FAMILIES, (1, 8, 64, 512)):
        one = w2(prior, target)
        product = math.sqrt(dimension) * one
        rows.append({"prior": prior.name, "target": target.name, "dimension": dimension,
                     "mixed_hessian_exact": 0.0,
                     "product_w2_factorization_error": abs(product * product - dimension * one * one),
                     "marginal_score_moments_finite": True})
    return rows


def heat_third(x: float, u: float, variance: float) -> float:
    z = x - u
    heat = math.exp(-0.5 * z * z / variance) / math.sqrt(2.0 * math.pi * variance)
    return (z**3 / variance**3 - 3.0 * z / variance**2) * heat


def heat_second(x: float, u: float, variance: float) -> float:
    z = x - u
    heat = math.exp(-0.5 * z * z / variance) / math.sqrt(2.0 * math.pi * variance)
    return (z * z / variance**2 - 1.0 / variance) * heat


def claim6_ibp() -> list[dict[str, object]]:
    rows = []
    for family, x, variance in itertools.product(FAMILIES[:3], (-0.7, 0.25, 1.1, -1.3), (0.08, 0.15, 0.30)):
        left = integrate(family, lambda u: heat_third(x, u, variance) * family.pdf(u))
        right = integrate(family, lambda u: -heat_second(x, u, variance) * family.pdf(u) * family.score(u))
        rows.append({"family": family.name, "x": x, "variance": variance, "absolute_residual": abs(left - right)})
    return rows


def main() -> None:
    moments = {family.name: score_moments(family) for family in FAMILIES}
    claim1, claim2 = claim1_claim2(moments)
    claim3 = claim3_schedule()
    claim5 = claim5_product()
    claim6 = claim6_ibp()
    claim1_slopes = {family.name: slope([row for row in claim1 if row["family"] == family.name], "factor") for family in FAMILIES}
    claim2_slopes = [slope([row for row in claim2 if row["family"] == family.name and row["delta"] == delta], "factor") for family in FAMILIES for delta in (0.25, 0.125, 0.0625)]
    result = {
        "schema": "influential-diffusion-broad-scope-v1",
        "families": [family.name for family in FAMILIES],
        "dimensions": list(DIMS),
        "claim1": {"cells": len(claim1), "slopes": claim1_slopes, "min_slope": min(claim1_slopes.values()), "max_slope": max(claim1_slopes.values())},
        "claim2": {"cells": len(claim2), "slope_min": min(claim2_slopes), "slope_max": max(claim2_slopes), "conditional_score_finite": all(row["conditional_score_finite"] for row in claim2), "full_joint_absent": all(not row["full_joint_has_density"] for row in claim2)},
        "claim3": {"cells": len(claim3), "max_closed_form_residual": max(row["closed_form_residual"] for row in claim3)},
        "claim5": {"cells": len(claim5), "max_mixed_hessian": max(row["mixed_hessian_exact"] for row in claim5), "max_factorization_error": max(row["product_w2_factorization_error"] for row in claim5)},
        "claim6": {"cells": len(claim6), "max_ibp_residual": max(row["absolute_residual"] for row in claim6)},
        "protocol": "CPU quadrature/product identities; no neural training; source experiment assets separately pinned",
    }
    assert result["claim1"]["min_slope"] > 2.8
    assert result["claim2"]["slope_min"] > 2.8 and result["claim2"]["conditional_score_finite"] and result["claim2"]["full_joint_absent"]
    assert result["claim3"]["max_closed_form_residual"] < 1e-10
    assert result["claim5"]["max_mixed_hessian"] == 0.0 and result["claim5"]["max_factorization_error"] < 1e-7
    assert result["claim6"]["max_ibp_residual"] < 1e-7
    print(json.dumps(result, indent=2, sort_keys=True))


if __name__ == "__main__":
    main()