verify.py

#!/usr/bin/env python3
"""Read-only verification of compact observer-dynamics evidence.

Run from any working directory. The optional --write-receipt writes only the
new canonical receipt; historical source/specification/result files stay fixed.
The full numerical graph spectra and the 48 large terminal arrays are archived
measurements, not regenerated by this bounded command.
"""
from __future__ import annotations
import argparse
import json
import math
import os
from pathlib import Path
import platform
import sys
import subprocess
import re
import time

# Set before importing numerical libraries or archived modules.
for key in ("OMP_NUM_THREADS", "OPENBLAS_NUM_THREADS", "MKL_NUM_THREADS", "VECLIB_MAXIMUM_THREADS", "NUMEXPR_NUM_THREADS"):
    os.environ[key] = "1"
try:
    from .archive_adapter import ARCHIVE, SIM, RER, check_manifest, frozen_sources, load, sha
except ImportError:  # Direct script invocation.
    from archive_adapter import ARCHIVE, SIM, RER, check_manifest, frozen_sources, load, sha


# A hash a producer takes over the raw bytes of a floating array it computed on the fly (the
# centres of a geodesic construction, ``directions_sha256``) pins the last bits of that array.
# numpy's vector kernels are chosen by the CPU's features at run time and differ in their last
# bits between processor generations: the same runner image, Python and wheels reproduced the
# archived level-2 hash on one machine and not on another, while every quantity derived from
# the array agreed to tolerance. Such a key is still compared; a difference is recorded as a
# platform note instead of a failure, and the derived quantities keep their exact or tolerance
# comparison, which is the scientific content of the receipt.
PLATFORM_HASHES = ("directions_sha256",)
PLATFORM_NOTES: list[str] = []


def compare(old, new, path="", *, rtol=1e-7, atol=1e-8):
    """Exact structural/rational comparison; tolerance only for floating values; a differing
    hash of a floating array named in PLATFORM_HASHES is a platform note, not a failure."""
    if isinstance(old, dict):
        if set(old) != set(new):
            raise AssertionError(f"Keys differ at {path}: {set(old) ^ set(new)}")
        for key in old:
            compare(old[key], new[key], f"{path}/{key}", rtol=rtol, atol=atol)
    elif isinstance(old, list):
        if len(old) != len(new):
            raise AssertionError(f"Length differs at {path}")
        for i, (a, b) in enumerate(zip(old, new)):
            compare(a, b, f"{path}/{i}", rtol=rtol, atol=atol)
    elif isinstance(old, float):
        if not math.isclose(old, new, rel_tol=rtol, abs_tol=atol):
            raise AssertionError(f"Numerical receipt mismatch at {path}: {old} != {new}")
    elif old != new:
        if path.rsplit("/", 1)[-1] in PLATFORM_HASHES:
            PLATFORM_NOTES.append(f"{path}: archived {old!r}, recomputed {new!r}")
            return
        raise AssertionError(f"Receipt mismatch at {path}: {old!r} != {new!r}")


class canonical_eigenvector_signs:
    """Fix the sign convention of symmetric eigensolver output during replay.

    A normalized eigenvector of a simple eigenvalue is defined only up to an
    overall sign, and LAPACK builds return either sign. The archived producers
    use the solver's output directly, so a receipt quantity that is odd in a
    mode vector would change sign between platforms. Inside this context each
    returned eigenvector is oriented so that its largest-magnitude component
    is positive, the first such index deciding ties within a relative 1e-9.
    The archived receipts replay exactly under this convention; the
    recomputation is then platform independent while the comparison itself
    stays exact.
    """

    @staticmethod
    def _orient(vectors):
        import numpy as np
        v = np.array(vectors, copy=True)
        if v.ndim != 2:
            return v
        for j in range(v.shape[1]):
            column = v[:, j]
            scale = np.max(np.abs(column)) if column.size else 0.0
            if scale == 0.0:
                continue
            largest = np.flatnonzero(np.abs(column) >= scale * (1 - 1e-9))
            if column[largest[0]] < 0:
                v[:, j] = -column
        return v

    def __enter__(self):
        import numpy
        import scipy.linalg
        self._saved = (scipy.linalg.eigh, numpy.linalg.eigh)
        scipy_eigh, numpy_eigh = self._saved
        orient = self._orient

        def scipy_wrapped(*args, **kwargs):
            result = scipy_eigh(*args, **kwargs)
            if isinstance(result, tuple) and len(result) == 2:
                return result[0], orient(result[1])
            return result

        def numpy_wrapped(*args, **kwargs):
            result = numpy_eigh(*args, **kwargs)
            return type(result)(result[0], orient(result[1]))

        scipy.linalg.eigh = scipy_wrapped
        numpy.linalg.eigh = numpy_wrapped
        return self

    def __exit__(self, *exc):
        import numpy
        import scipy.linalg
        scipy.linalg.eigh, numpy.linalg.eigh = self._saved
        return False


def replay(relative, function, receipt="receipt.json", excluded=()):
    module = load(relative)
    old = json.loads((SIM / relative).with_name(receipt).read_text())
    if "boundary_bulk" in relative:
        # Context-paper/old proof-prose hashes are historical provenance, not
        # scientific inputs. They are explicitly outside this replay's scope.
        original = module.sha256
        module.sha256 = lambda p: original(p) if Path(p).is_file() else None
    new = getattr(module, function)()
    for key in excluded:
        old.pop(key, None); new.pop(key, None)
    # JSON round-trip normalizes tuples and numpy scalar subclasses only.
    compare(old, json.loads(json.dumps(new)))
    print(f"Recomputed {relative}", flush=True)
    return {"producer": "sim-analysis/" + relative, "receipt": receipt,
            "comparison": "exact structures and rationals; floats rtol=1e-7 atol=1e-8",
            "excluded_context_fields": list(excluded)}


class Counts:
    def __init__(self): self.passed = self.failed = self.skipped = 0
    def pytest_runtest_logreport(self, report):
        if report.skipped:
            self.skipped += 1
        elif report.when == "call":
            self.passed += report.passed
            self.failed += report.failed
    def pytest_collectreport(self, report):
        self.skipped += report.skipped


def require_complete_controls(skipped):
    if skipped:
        raise RuntimeError("Mandatory archived controls were skipped; install the pinned root requirements")


def subprocess_pass_count(output):
    skipped = re.search(r"\b(\d+) skipped\b", output)
    require_complete_controls(int(skipped.group(1)) if skipped else 0)
    passed = re.search(r"\b(\d+) passed\b", output)
    if passed is None:
        raise RuntimeError("No passing isolated controls reported")
    return int(passed.group(1))


def main():
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--write-receipt", action="store_true")
    parser.add_argument("--replay-refinement", action="store_true", help="also reconstruct all 61 fresh-noise RNG streams (about one minute)")
    args = parser.parse_args()
    started = time.monotonic()
    inventory = check_manifest()
    results = []
    with frozen_sources(), canonical_eigenvector_signs():
        for source, function, receipt in [
            ("codex/dynamics/stationary_control.py", "build", "stationary_receipt.json"),
            ("codex/inflation/native_history/run.py", "build", "receipt.json"),
            ("codex/necessity/universality/certify.py", "build", "receipt.json"),
            ("codex/necessity/universality/nonlinear.py", "build", "nonlinear_receipt.json"),
            ("codex/necessity/temporal_filters/certify.py", "execute", "receipt.json"),
            ("codex/necessity/event_records/run.py", "execute", "receipt.json"),
            ("codex/source_derivation/native/run.py", "build", "receipt.json"),
        ]:
            results.append(replay(source, function, receipt))
        results.append(replay("codex/necessity/boundary_bulk/certify.py", "build_receipt", excluded=("read_only_source_hashes",)))
        causal = load("scripts/causal_cosmology_audit.py")
        compare(json.loads(causal.OUTPUT.read_text()), causal.build())
        golden = load("scripts/codex_golden_fourier.py")
        compare(json.loads((SIM / "data/codex_audit_20260925/golden_fourier.json").read_text()), golden.make_receipt())
        try:
            from .retained_measurements import scale_ensemble, orbit_census, conditional_measured_comparison
        except ImportError:
            from retained_measurements import scale_ensemble, orbit_census, conditional_measured_comparison
        retained_measurements = {"scale_ensemble": scale_ensemble(), "carrier_orbits": orbit_census(),
                                 "conditional_cmb": conditional_measured_comparison()}
        innovations = load("codex/source_derivation/innovations/verify.py").check()
        nested = load("codex/source_derivation/innovations/nested.py")
        nested.main(["--check"])
        if args.replay_refinement:
            extract = load("codex/source_derivation/innovations/extract.py")
            compare(json.loads((SIM / "codex/source_derivation/innovations/receipt.json").read_text()), extract.build(), rtol=5e-10, atol=1e-15)
        import pytest
        tests = sorted(str(p) for p in (SIM / "codex").rglob("test_*.py") if "causal" not in p.parts)
        tests += sorted(str(p) for p in (SIM / "tests").glob("test_*.py"))
        tests += [str(SIM / "scripts/test_codex_causal_cosmology_audit.py")]
        tests += sorted(str(p) for p in Path(__file__).parent.glob("test_*.py"))
        tests += [str(ARCHIVE / "lean/test_verify.py")]
        counts = Counts()
        code = pytest.main(["-q", "-p", "no:cacheprovider", *tests], plugins=[counts])
        if code != 0:
            raise SystemExit(code)
        require_complete_controls(counts.skipped)
        # The causal fixture vendors a distinct historical source-net revision.
        # Isolate Python's module cache so neither revision masks the other.
        causal_tests = subprocess.run([sys.executable, "-m", "pytest", "-q", "-p", "no:cacheprovider", str(SIM / "codex/causal/test_experiment.py")], capture_output=True, text=True)
        print(causal_tests.stdout, end="")
        if causal_tests.returncode:
            print(causal_tests.stderr, file=sys.stderr)
            raise SystemExit(causal_tests.returncode)
        counts.passed += subprocess_pass_count(causal_tests.stdout)
    # Recheck immutable evidence after producers/tests to catch accidental writes.
    if check_manifest() != inventory:
        raise AssertionError("Archived inputs changed during verification")
    value = {"schema": "oph.observer-dynamics.verification.v1", "verified": True,
             "inventory": inventory, "recomputed_receipts": results,
             "causal_layer_and_FLRW_receipt_recomputed": True,
             "golden_fourier_receipt_recomputed": True,
             "conditional_observer_spectrum_recomputed_from_retained_transfer": True,
             "retained_measurements": retained_measurements,
             "code_sha256": {p.name: sha(p) for p in sorted(Path(__file__).parent.glob("*.py"))},
             "refinement": innovations, "fresh_noise_rng_replayed": args.replay_refinement,
             "tests": vars(counts), "python": platform.python_version(),
             "platform_notes": list(PLATFORM_NOTES),
             "verifier_sha256": sha(__file__), "adapter_sha256": sha(Path(__file__).with_name("archive_adapter.py")),
             "elapsed_seconds": time.monotonic() - started,
             "boundaries": ["No microscopic replay of the 48 large terminal arrays or large geometry caches.",
                            "Transport spectra and 120-case boosted-clock receipt are hash checked; archived small independent controls are rerun.",
                            "Boundary/bulk context-prose hashes are preserved historical metadata, not replayed scientific inputs.",
                            "Lean compilation has its own canonical receipt in lean/."]}
    if args.write_receipt:
        (ARCHIVE / "verification.json").write_text(json.dumps(value, indent=2, sort_keys=True) + "\n")
    print(json.dumps(value, indent=2, sort_keys=True))


if __name__ == "__main__": main()