"""Independently validate the recorded 100-ID selected GMS result. This is a post-hoc integrity and aggregation check. It deliberately does not import the numerical runner or re-execute JAX: its purpose is to make the recorded result's input identity, raw values, and reported aggregates independently traceable from a small standard-Python programme. """ from __future__ import annotations import argparse import datetime as dt import hashlib import json import math import os from pathlib import Path import platform import statistics import subprocess import sys import tempfile from typing import Any, Iterable, NoReturn import scipy from scipy.stats import t as student_t REPOSITORY_ROOT = Path(__file__).resolve().parents[1] STUDY_ROOT = REPOSITORY_ROOT / "study" DEFAULT_INPUT = STUDY_ROOT / "gms-selected-ids-0-99.json" DEFAULT_OUTPUT = STUDY_ROOT / "gms-selected-ids-0-99-validation.json" SOURCE_COMMIT = "767effe3e755067eb8a04422597fbf37eb8ab754" EXPECTED_SOURCE_ORIGIN = "https://github.com/zgbkdlm/diffres.git" EXPECTED_CONFIGURATION = { "diffusion_a": -1.0, "terminal_time": 3.0, "diffusion_steps": 128, "integrator": "jentzen_and_kloeden", "ode": True, "particles": 10_000, "components": 5, "dimension": 8, "observation_dimension": 1, "swd_projections": 1_000, } METHODS = ("diffusion_resampling", "multinomial_baseline") METRICS = ("sliced_wasserstein_l1", "posterior_mean_residual_squared_l2") class ValidationError(ValueError): """Raised when the recorded result is not internally traceable.""" def fail(message: str) -> NoReturn: raise ValidationError(message) def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as handle: for block in iter(lambda: handle.read(1024 * 1024), b""): digest.update(block) return digest.hexdigest() def utc_now() -> str: return dt.datetime.now(tz=dt.timezone.utc).isoformat().replace("+00:00", "Z") def strict_json_load(path: Path) -> dict[str, Any]: def reject_nonfinite(token: str) -> NoReturn: fail(f"{path.name} contains non-standard JSON constant {token!r}") def reject_duplicate_keys(pairs: list[tuple[str, Any]]) -> dict[str, Any]: object_value: dict[str, Any] = {} for key, value in pairs: if key in object_value: fail(f"{path.name} contains duplicate JSON object key {key!r}") object_value[key] = value return object_value try: value = json.loads( path.read_text(encoding="utf-8"), parse_constant=reject_nonfinite, object_pairs_hook=reject_duplicate_keys, ) except (OSError, json.JSONDecodeError) as error: fail(f"Cannot parse strict JSON from {path}: {error}") if not isinstance(value, dict): fail(f"Top-level value in {path.name} must be an object") return value def git_text(*arguments: str) -> str: completed = subprocess.run( ["git", *arguments], cwd=REPOSITORY_ROOT, check=False, capture_output=True, text=True, ) if completed.returncode != 0: fail(f"Git command failed: git {' '.join(arguments)}: {completed.stderr.strip()}") return completed.stdout.strip() def resolve_within_repository(relative_path: str) -> Path: candidate = (REPOSITORY_ROOT / relative_path).resolve() try: candidate.relative_to(REPOSITORY_ROOT.resolve()) except ValueError: fail(f"Recorded source path escapes repository root: {relative_path}") return candidate def require_finite_number(value: object, path: str) -> float: if isinstance(value, bool) or not isinstance(value, (int, float)): fail(f"{path} must be a numeric JSON value") converted = float(value) if not math.isfinite(converted): fail(f"{path} must be finite") return converted def require_equal(actual: object, expected: object, path: str) -> None: if actual != expected: fail(f"{path} is {actual!r}; expected {expected!r}") def require_close(actual: object, expected: float, path: str) -> None: value = require_finite_number(actual, path) if not math.isclose(value, expected, rel_tol=0.0, abs_tol=1e-12): fail(f"{path} is {value:.17g}; recomputation is {expected:.17g}") def summary(values: Iterable[float]) -> dict[str, float]: values_list = list(values) if not values_list: fail("Cannot aggregate an empty metric series") return { "mean": statistics.fmean(values_list), "population_standard_deviation": statistics.pstdev(values_list), "minimum": min(values_list), "maximum": max(values_list), } def paired_t_interval(values: Iterable[float]) -> dict[str, float | int]: values_list = list(values) if len(values_list) < 2: fail("A paired t interval needs at least two observations") mean = statistics.fmean(values_list) sample_standard_deviation = statistics.stdev(values_list) standard_error = sample_standard_deviation / math.sqrt(len(values_list)) critical_value = float(student_t.ppf(0.975, len(values_list) - 1)) half_width = critical_value * standard_error return { "sample_count": len(values_list), "mean": mean, "sample_standard_deviation": sample_standard_deviation, "standard_error": standard_error, "critical_value": critical_value, "lower": mean - half_width, "upper": mean + half_width, } def validate_source_identity(result: dict[str, Any]) -> dict[str, Any]: provenance = result.get("source_provenance") if not isinstance(provenance, dict): fail("source_provenance must be an object") require_equal(result.get("source_commit"), SOURCE_COMMIT, "source_commit") require_equal( provenance.get("expected_source_commit"), SOURCE_COMMIT, "source_provenance.expected_source_commit" ) require_equal( provenance.get("observed_source_commit"), SOURCE_COMMIT, "source_provenance.observed_source_commit" ) require_equal( provenance.get("expected_source_origin"), EXPECTED_SOURCE_ORIGIN, "source_provenance.expected_source_origin", ) require_equal( provenance.get("observed_remote_matches_expected"), True, "source_provenance.observed_remote_matches_expected", ) require_equal(provenance.get("git_metadata_available"), True, "source_provenance.git_metadata_available") require_equal( provenance.get("tracked_worktree_diff_paths"), [], "source_provenance.tracked_worktree_diff_paths" ) require_equal(provenance.get("staged_diff_paths"), [], "source_provenance.staged_diff_paths") current_commit = git_text("rev-parse", "HEAD") current_origin = git_text("config", "--get", "remote.origin.url") current_diff = git_text("diff", "--name-only") current_staged_diff = git_text("diff", "--cached", "--name-only") require_equal(current_commit, SOURCE_COMMIT, "current Git commit") require_equal(current_origin, EXPECTED_SOURCE_ORIGIN, "current Git origin") require_equal(current_diff, "", "current tracked worktree diff") require_equal(current_staged_diff, "", "current staged worktree diff") recorded_hashes = provenance.get("author_source_sha256") if not isinstance(recorded_hashes, dict) or not recorded_hashes: fail("source_provenance.author_source_sha256 must be a non-empty object") checked_hashes: dict[str, str] = {} for relative_path, recorded_hash in sorted(recorded_hashes.items()): if not isinstance(relative_path, str) or not isinstance(recorded_hash, str): fail("source_provenance.author_source_sha256 must map strings to strings") path = resolve_within_repository(relative_path) if not path.is_file(): fail(f"Recorded source file is unavailable: {relative_path}") actual_hash = sha256_file(path) require_equal(actual_hash, recorded_hash, f"SHA-256 for {relative_path}") checked_hashes[relative_path] = actual_hash return { "current_commit": current_commit, "current_origin_matches_expected": current_origin == EXPECTED_SOURCE_ORIGIN, "current_tracked_worktree_clean": current_diff == "" and current_staged_diff == "", "checked_author_source_sha256": checked_hashes, } def validate_protocol(result: dict[str, Any]) -> None: require_equal(result.get("reproduction_compatible"), True, "reproduction_compatible") configuration = result.get("configuration") if not isinstance(configuration, dict): fail("configuration must be an object") for key, expected in EXPECTED_CONFIGURATION.items(): require_equal(configuration.get(key), expected, f"configuration.{key}") require_equal(configuration.get("mc_ids"), list(range(100)), "configuration.mc_ids") require_equal(configuration.get("author_keys_path"), "experiments/rnd_keys.npy", "configuration.author_keys_path") source_paths = result.get("source_paths") required_paths = { "experiments/run_gms.sh", "experiments/gms/diffusion.py", "experiments/gms/baselines.py", "experiments/summary/print_gms_errs.py", "experiments/rnd_keys.npy", } if not isinstance(source_paths, list) or not required_paths.issubset(source_paths): fail("source_paths does not record the required released GMS files") def validate_records_and_aggregates(result: dict[str, Any]) -> dict[str, Any]: records = result.get("per_id_results") if not isinstance(records, list): fail("per_id_results must be a list") if len(records) != 100: fail(f"per_id_results has {len(records)} records; expected 100") series: dict[str, dict[str, list[float]]] = { method: {metric: [] for metric in METRICS} for method in METHODS } for expected_id, record in enumerate(records): if not isinstance(record, dict): fail(f"per_id_results[{expected_id}] must be an object") require_equal(record.get("mc_id"), expected_id, f"per_id_results[{expected_id}].mc_id") for method in METHODS: method_values = record.get(method) if not isinstance(method_values, dict): fail(f"per_id_results[{expected_id}].{method} must be an object") for metric in METRICS: series[method][metric].append( require_finite_number( method_values.get(metric), f"per_id_results[{expected_id}].{method}.{metric}" ) ) aggregate = result.get("aggregate") if not isinstance(aggregate, dict): fail("aggregate must be an object") recomputed: dict[str, Any] = {} for method in METHODS: reported_method = aggregate.get(method) if not isinstance(reported_method, dict): fail(f"aggregate.{method} must be an object") recomputed[method] = {} for metric in METRICS: reported_metric = reported_method.get(metric) if not isinstance(reported_metric, dict): fail(f"aggregate.{method}.{metric} must be an object") computed = summary(series[method][metric]) for name, value in computed.items(): require_close(reported_metric.get(name), value, f"aggregate.{method}.{metric}.{name}") recomputed[method][metric] = computed reported_paired = aggregate.get("paired_diffusion_minus_multinomial") if not isinstance(reported_paired, dict): fail("aggregate.paired_diffusion_minus_multinomial must be an object") recomputed_paired: dict[str, dict[str, float]] = {} intervals: dict[str, dict[str, float | int]] = {} wins: dict[str, int] = {} for metric in METRICS: values = [ diffusion - multinomial for diffusion, multinomial in zip( series["diffusion_resampling"][metric], series["multinomial_baseline"][metric], strict=True, ) ] reported_metric = reported_paired.get(metric) if not isinstance(reported_metric, dict): fail(f"aggregate.paired_diffusion_minus_multinomial.{metric} must be an object") computed = summary(values) for name, value in computed.items(): require_close( reported_metric.get(name), value, f"aggregate.paired_diffusion_minus_multinomial.{metric}.{name}" ) recomputed_paired[metric] = computed intervals[metric] = paired_t_interval(values) wins[metric] = sum(value < 0.0 for value in values) reported_wins = aggregate.get("diffusion_lower_is_better_wins") if not isinstance(reported_wins, dict): fail("aggregate.diffusion_lower_is_better_wins must be an object") for metric, count in wins.items(): require_equal(reported_wins.get(metric), count, f"aggregate.diffusion_lower_is_better_wins.{metric}") return { "record_count": len(records), "recomputed_aggregate": recomputed, "recomputed_paired_aggregate": recomputed_paired, "paired_diffusion_minus_multinomial_95_percent_t_intervals": intervals, "diffusion_lower_is_better_wins": wins, } def validate_result(input_path: Path) -> dict[str, Any]: result = strict_json_load(input_path) validate_protocol(result) source_identity = validate_source_identity(result) aggregation = validate_records_and_aggregates(result) return { "schema_version": 1, "validation_status": "passed", "validation_type": "post-hoc integrity and aggregate recomputation; no numerical rerun", "checked_at_utc": utc_now(), "input": { "path": input_path.relative_to(REPOSITORY_ROOT).as_posix(), "sha256": sha256_file(input_path), }, "checks": { "strict_json": "passed", "selected_protocol": "passed", "current_source_identity_and_hashes": source_identity, "raw_record_integrity_and_aggregate_recomputation": aggregation, }, "validator": { "path": Path(__file__).relative_to(REPOSITORY_ROOT).as_posix(), "sha256": sha256_file(Path(__file__)), "python_version": sys.version, "scipy_version": scipy.__version__, "platform": { "system": platform.system(), "release": platform.release(), "machine": platform.machine(), }, }, "claim_boundary": ( "This check independently recomputes the recorded aggregates and paired intervals " "from the 100 raw JSON records while checking the current pinned source identity. " "It does not rerun the numerical experiment, independently reproduce the paper, " "or establish an outperformance or significance claim." ), } def resolve_study_path(requested: Path) -> Path: path = requested if requested.is_absolute() else REPOSITORY_ROOT / requested path = path.resolve() try: path.relative_to(STUDY_ROOT.resolve()) except ValueError: fail(f"Path must be contained in {STUDY_ROOT}: {path}") return path def write_json(path: Path, payload: dict[str, Any], *, overwrite: bool) -> None: serialised = json.dumps(payload, indent=2, sort_keys=True, allow_nan=False) + "\n" with tempfile.NamedTemporaryFile( mode="w", encoding="utf-8", dir=path.parent, prefix=f".{path.name}.", delete=False ) as temporary_file: temporary_file.write(serialised) temporary_file.flush() os.fsync(temporary_file.fileno()) temporary_path = Path(temporary_file.name) try: if overwrite: os.replace(temporary_path, path) else: os.link(temporary_path, path) except FileExistsError as error: raise FileExistsError( f"Refusing to overwrite {path}; pass --overwrite for a deliberate rerun." ) from error finally: temporary_path.unlink(missing_ok=True) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--input", type=Path, default=DEFAULT_INPUT) parser.add_argument("--output", type=Path, default=DEFAULT_OUTPUT) parser.add_argument("--overwrite", action="store_true") return parser.parse_args() def main() -> None: args = parse_args() input_path = resolve_study_path(args.input) output_path = resolve_study_path(args.output) if output_path.exists() and not args.overwrite: raise FileExistsError( f"Refusing to overwrite {output_path}; pass --overwrite for a deliberate rerun." ) payload = validate_result(input_path) write_json(output_path, payload, overwrite=args.overwrite) print(f"Wrote validation result to {output_path}") if __name__ == "__main__": main()