scripts/audit_optimization_bounds_statistics.py
#!/usr/bin/env python3
"""Independently rerun frozen JSON-only analysis and compare all scientific artifacts.
Reads saved JSON/CSV (including their gzip archives) through the frozen analyzer.
It never loads RGB/depth media, model weights, optimized states or checkpoints.
"""
import argparse
import hashlib
import importlib.util
import json
import math
import tempfile
from pathlib import Path
SCIENTIFIC_FILES = (
"optimization_results.json",
"optimization_sufficiency.json",
"bounds_analysis.json",
"query_oracle_diagnostic.json",
"bootstrap_results.json",
"cost_analysis.json",
)
def load_runner():
path = Path(__file__).resolve().with_name("analyze_optimization_bounds.py")
spec = importlib.util.spec_from_file_location("frozen_optimization_bounds_analyzer", path)
module = importlib.util.module_from_spec(spec)
spec.loader.exec_module(module)
return module.run
def numeric_comparison(original, reproduced):
"""Report numerical error and structural mismatches; byte equality remains decisive."""
maximum = 0.0
numeric_count = 0
mismatch_count = 0
examples = []
def mismatch(path, kind):
nonlocal mismatch_count
mismatch_count += 1
if len(examples) < 20:
examples.append({"path": path, "kind": kind})
def visit(left, right, path):
nonlocal maximum, numeric_count
if isinstance(left, bool) or isinstance(right, bool):
if type(left) is not type(right) or left != right:
mismatch(path, "boolean_or_type")
elif isinstance(left, (int, float)) and isinstance(right, (int, float)):
if not math.isfinite(left) or not math.isfinite(right):
mismatch(path, "nonfinite_number")
return
numeric_count += 1
maximum = max(maximum, abs(left - right))
if type(left) is not type(right):
mismatch(path, "numeric_type")
elif isinstance(left, dict) and isinstance(right, dict):
if set(left) != set(right):
mismatch(path, "dictionary_keys")
for key in sorted(set(left) & set(right)):
visit(left[key], right[key], f"{path}.{key}")
elif isinstance(left, list) and isinstance(right, list):
if len(left) != len(right):
mismatch(path, "list_length")
for i, (a, b) in enumerate(zip(left, right, strict=False)):
visit(a, b, f"{path}[{i}]")
elif type(left) is not type(right) or left != right:
mismatch(path, "value_or_type")
visit(original, reproduced, "$")
return {
"max_abs_numeric_error": maximum,
"numeric_values_compared": numeric_count,
"structural_or_nonnumeric_mismatches": mismatch_count,
"mismatch_examples": examples,
}
def compare_file(original, reproduced):
original_bytes, reproduced_bytes = original.read_bytes(), reproduced.read_bytes()
return {
"original": str(original.resolve()),
"reproduced_filename": reproduced.name,
"byte_exact": original_bytes == reproduced_bytes,
"original_sha256": hashlib.sha256(original_bytes).hexdigest(),
"reproduced_sha256": hashlib.sha256(reproduced_bytes).hexdigest(),
**numeric_comparison(json.loads(original_bytes), json.loads(reproduced_bytes)),
}
def run_audit(root, reference=None, *, runner=None):
root = Path(root).resolve()
source = Path(reference).resolve() if reference is not None else root
report_path = (
root
/ "audit"
/ (
"published_statistics_reproduction_audit.json"
if reference is not None
else "statistics_reproduction_audit.json"
)
)
report = {
"status": "FAIL",
"comparison_original_root": str(root),
"analysis_input_root": str(source),
"published_reference": reference is not None,
"raw_only": True,
"media_or_checkpoint_loaded": False,
"access_scope": "JSON/CSV/gzip only; no media or state deserialization",
"comparison_rule": "Every required scientific JSON must match byte-for-byte",
"files": {},
}
try:
expected = [(root, name) for name in SCIENTIFIC_FILES]
expected += [
(root / "secondary_analysis", name)
for name in (*SCIENTIFIC_FILES, "secondary_decision_audit.json")
]
for directory, name in expected:
if not (directory / name).is_file():
raise FileNotFoundError(f"Both phases must finish before audit: {directory / name}")
runner = load_runner() if runner is None else runner
with tempfile.TemporaryDirectory(
prefix="mcss-optimization-bounds-reproduction-"
) as temporary:
primary_output, secondary_output = (
Path(temporary) / "primary",
Path(temporary) / "secondary",
)
runner(source, primary_output, include_secondary=False)
runner(source, secondary_output, include_secondary=True)
for phase, directory, reproduced_directory, filenames in (
("primary", root, primary_output, SCIENTIFIC_FILES),
(
"secondary",
root / "secondary_analysis",
secondary_output,
(*SCIENTIFIC_FILES, "secondary_decision_audit.json"),
),
):
for name in filenames:
report["files"][f"{phase}/{name}"] = compare_file(
directory / name, reproduced_directory / name
)
for phase, directory in (("primary", primary_output), ("secondary", secondary_output)):
metadata = directory / "statistics_reproduction.json"
report[f"{phase}_input_provenance"] = json.loads(metadata.read_text())
comparisons = list(report["files"].values())
report["compared_files"] = len(comparisons)
report["max_abs_numeric_error"] = max(r["max_abs_numeric_error"] for r in comparisons)
report["byte_mismatch_count"] = sum(not r["byte_exact"] for r in comparisons)
report["structural_or_nonnumeric_mismatches"] = sum(
r["structural_or_nonnumeric_mismatches"] for r in comparisons
)
if (
report["byte_mismatch_count"] == 0
and report["max_abs_numeric_error"] == 0
and report["structural_or_nonnumeric_mismatches"] == 0
):
report["status"] = "PASS"
except Exception as error:
report["error"] = {"type": type(error).__name__, "message": str(error)}
report["audit_source_sha256"] = hashlib.sha256(Path(__file__).read_bytes()).hexdigest()
report_path.parent.mkdir(parents=True, exist_ok=True)
report_path.write_text(json.dumps(report, indent=2, sort_keys=True, allow_nan=False) + "\n")
return report
if __name__ == "__main__":
parser = argparse.ArgumentParser(description=__doc__)
parser.add_argument(
"--root", type=Path, required=True, help="Original complete experiment outputs"
)
parser.add_argument(
"--reference", type=Path, help="Optional published raw archive to reanalyze"
)
args = parser.parse_args()
report = run_audit(args.root, args.reference)
print(
json.dumps(
{
key: report.get(key)
for key in (
"status",
"compared_files",
"byte_mismatch_count",
"max_abs_numeric_error",
"error",
)
},
indent=2,
)
)
raise SystemExit(0 if report["status"] == "PASS" else 1)