analysis/thread_statistics/validate_artifacts.py

"""Bounded CPU validation against already completed synthetic output files.

Run from the repository root using .venv/bin/python -m
analysis.thread_statistics.validate_artifacts. Writes only this thread's
output directory. No model loading, training, generation, or GPU imports.
"""

import hashlib
import json
import os
from pathlib import Path
import time

import numpy as np

from .composition import composition_point_difference, paired_composition_bootstrap


def _hash(path):
    return hashlib.sha256(path.read_bytes()).hexdigest()


def main():
    expected_env = {"CUDA_VISIBLE_DEVICES": "", "OMP_NUM_THREADS": "2", "OPENBLAS_NUM_THREADS": "2"}
    if any(os.environ.get(key) != value for key, value in expected_env.items()):
        raise RuntimeError(f"run with required CPU environment: {expected_env}")
    out = Path("outputs/thread_statistics")
    out.mkdir(parents=True, exist_ok=True)
    values = {0: [], 25: []}
    provenance, metrics, reference = [], {0: [], 25: []}, None
    for budget in (0, 25):
        for seed in range(5):
            path = Path(f"outputs/synthetic/synthetic_standard_n{budget}_noise0_s{seed}")
            config = json.loads((path / "config.json").read_text())
            metric = json.loads((path / "metrics.json").read_text())
            if config["seed"] != seed or config["n_interventions"] != budget or metric["seed"] != seed:
                raise ValueError(f"run metadata does not match expected pairing: {path}")
            with np.load(path / "predictions.npz", allow_pickle=False) as archive:
                keys = ("composition_pair_correct", "composition_pair_valid", "composition_pairs", "composition_weights", "composition_accuracy", "z_true", "labels")
                record = {key: archive[key] for key in keys}
            if reference is None:
                reference = record
            for key in ("composition_pairs", "composition_weights", "composition_pair_valid", "z_true", "labels"):
                np.testing.assert_array_equal(reference[key], record[key], err_msg=f"{path}: {key} alignment")
            valid = record["composition_pair_valid"]
            counts = valid.sum(axis=0)
            accuracy = np.where(valid, record["composition_pair_correct"], 0).sum(axis=0, dtype=np.float64) / counts
            np.testing.assert_allclose(accuracy, record["composition_accuracy"], rtol=0, atol=5e-16)
            np.testing.assert_allclose(accuracy.mean(), metric["composition_accuracy"], rtol=0, atol=5e-16)
            values[budget].append(record)
            metrics[budget].append(metric["composition_accuracy"])
            provenance.append({
                "run": str(path), "seed": seed, "n_interventions": budget,
                "metric_timestamp": metric["timestamp"],
                "sha256": {name: _hash(path / name) for name in ("predictions.npz", "metrics.json", "config.json")},
            })
    stack = lambda budget, key: np.stack([r[key] for r in values[budget]])
    point = composition_point_difference(
        stack(0, "composition_pair_correct"), stack(25, "composition_pair_correct"),
        stack(0, "composition_pair_valid"), stack(25, "composition_pair_valid"),
        seed_ids_a=list(range(5)), seed_ids_b=list(range(5)),
        pair_ids_a=np.arange(len(reference["composition_pairs"])), pair_ids_b=np.arange(len(reference["composition_pairs"])),
        rubric_ids_a=np.arange(64), rubric_ids_b=np.arange(64),
    )
    np.testing.assert_allclose(point["difference"], np.mean(metrics[25]) - np.mean(metrics[0]), rtol=0, atol=5e-16)
    pairs = reference["composition_pairs"]
    degree = np.bincount(pairs.ravel(), minlength=len(reference["z_true"]))
    # Shared endpoints induce components, not one independent cluster per pair.
    parents = np.arange(len(degree))

    def find(x):
        while parents[x] != x:
            parents[x] = parents[parents[x]]
            x = parents[x]
        return x

    for i, j in pairs:
        parents[find(i)] = find(j)
    component_sizes = np.bincount([find(i) for i in range(len(degree)) if degree[i] > 0])
    synthetic = {
        "kind": "point-estimate verification of already completed synthetic runs; no new experiment",
        "comparison": "standard N=25 minus standard N=0, noise=0, paired seeds 0..4",
        "point_result": point,
        "ci95": None,
        "ci_status": "not attempted: saved composition pairs join reused independent probe observations, not disjoint prompt clusters",
        "pair_dependence": {
            "pairs": len(pairs), "probe_observations": len(degree),
            "self_pairs": int((pairs[:, 0] == pairs[:, 1]).sum()),
            "probe_observations_in_multiple_pair_endpoints": int((degree > 1).sum()),
            "largest_endpoint_component_observations": int(component_sizes.max()),
            "distinct_unordered_pairs": len(np.unique(np.sort(pairs, axis=1), axis=0)),
        },
        "provenance": provenance,
    }
    (out / "completed_synthetic_point_check.json").write_text(json.dumps(synthetic, indent=2, allow_nan=False) + "\n")

    # A generated correctness fixture is only an algorithm/time benchmark.
    # It is explicitly not a model or research performance outcome.
    rng = np.random.default_rng(105)
    a = rng.integers(0, 2, size=(5, 1500, 64), dtype=np.int8)
    b = rng.integers(0, 2, size=a.shape, dtype=np.int8)
    valid = rng.random((1500, 64)) > 0.2
    valid[::3] = True  # Each rubric has support in every cluster.
    valid = np.broadcast_to(valid, a.shape)
    started = time.perf_counter()
    result = paired_composition_bootstrap(
        a, b, valid, valid, np.repeat(np.arange(500), 3),
        seed_ids_a=list(range(5)), seed_ids_b=list(range(5)),
    )
    elapsed = time.perf_counter() - started
    assert result["bootstrap_samples"] == 10_000 and result["interval_status"] == "ok"
    expected = ((np.where(valid, b, 0).sum(axis=1) - np.where(valid, a, 0).sum(axis=1)) / valid.sum(axis=1)).mean()
    np.testing.assert_allclose(result["difference"], expected, atol=5e-16, rtol=0)
    benchmark = {
        "kind": "algorithm timing and numerical validation on generated binary test fixtures; NOT an experiment",
        "environment": expected_env, "numpy_version": np.__version__,
        "shape_seed_pair_rubric": list(a.shape), "prompt_clusters": 500,
        "bootstrap_samples": result["bootstrap_samples"], "elapsed_seconds": elapsed,
        "point_matches_direct_formula": True, "interval_status": result["interval_status"],
    }
    (out / "algorithm_benchmark.json").write_text(json.dumps(benchmark, indent=2, allow_nan=False) + "\n")
    print(json.dumps({"synthetic_saved_point_difference": point["difference"], "benchmark": benchmark}, indent=2))


if __name__ == "__main__":
    main()