#!/usr/bin/env python3 """Independent Fraction oracle for the public KMeans numerical probe. The native probe emits fitted K=1 center bits and recomputed C++ feature-order hypot distance bits. This tool verifies correctly rounded exact centroid means and correctly rounded exact SUM(computed-distance**2). The latter deliberately does not assert exact-coordinate Euclidean SSE. It also checks all public readouts/metadata against serial in every admitted explicit worker mode. No runtime files are modified and no performance measurements are made. """ from __future__ import annotations import argparse from collections import Counter from fractions import Fraction import hashlib import json from pathlib import Path import struct import subprocess import sys def rational(encoded: str) -> Fraction: bits = int(encoded, 16) exponent = (bits >> 52) & 0x7FF if exponent == 0x7FF: raise ValueError("nonfinite value in numerical probe") mantissa = bits & ((1 << 52) - 1) power = -1074 if exponent: mantissa |= 1 << 52 power = exponent - 1075 if bits >> 63: mantissa = -mantissa return Fraction(mantissa << power, 1) if power >= 0 else Fraction(mantissa, 1 << -power) def rounded(value: Fraction) -> str: try: return struct.pack(">d", float(value)).hex() except OverflowError: return "OVERFLOW" def sha(path: Path) -> str: return hashlib.sha256(path.read_bytes()).hexdigest() def main() -> int: parser = argparse.ArgumentParser(description=__doc__) source = parser.add_mutually_exclusive_group(required=True) source.add_argument("--binary", type=Path) source.add_argument("--observations", type=Path) parser.add_argument("--raw-output", type=Path) parser.add_argument("--output", type=Path) parser.add_argument("--build-root", type=Path) args = parser.parse_args() if args.binary: run = subprocess.run([str(args.binary.resolve())], capture_output=True, text=True, check=True, timeout=120) raw_text = run.stdout else: raw_text = args.observations.read_text() raw = json.loads(raw_text) if raw.get("schema") != "aner-kmeans-public-numeric-probe-v1": raise ValueError("unexpected numerical probe schema") if raw.get("rounding") != "FE_TONEAREST": raise ValueError("probe did not declare the required FP environment") if args.raw_output: args.raw_output.parent.mkdir(parents=True, exist_ok=True) args.raw_output.write_text(raw_text) failures = [] observations = raw["observations"] serial = {entry["name"]: entry for entry in observations if entry["mode"] == "serial"} mode_counts = Counter() observed_pairs = set() contract_example = None for entry in observations: name, mode = entry["name"], entry["mode"] if (name, mode) in observed_pairs: raise ValueError(f"duplicate observation: {name}/{mode}") observed_pairs.add((name, mode)) mode_counts[mode] += 1 n, d = entry["rows"], entry["columns"] if n <= 0 or d <= 0 or len(entry["input_bits"]) != n * d or entry["k"] != 1: raise ValueError(f"invalid input shape or K for {name}/{mode}") inputs = [rational(value) for value in entry["input_bits"]] expected_centers = [rounded(sum(inputs[column::d], Fraction()) / n) for column in range(d)] if entry["center_bits"] != expected_centers: failures.append({"name": name, "mode": mode, "readout": "centers", "expected": expected_centers, "actual": entry["center_bits"]}) distances = [rational(value) for value in entry["computed_distance_bits"]] if len(distances) != n: raise ValueError(f"wrong distance count for {name}/{mode}") expected_inertia = rounded(sum((value * value for value in distances), Fraction())) if entry["inertia_bits"] != [expected_inertia]: failures.append({"name": name, "mode": mode, "readout": "inertia", "expected": [expected_inertia], "actual": entry["inertia_bits"]}) for readout in ("label_bits", "prediction_bits"): if entry[readout] != ["0000000000000000"] * n: failures.append({"name": name, "mode": mode, "readout": readout, "reason": "K=1 must label every row canonical cluster zero"}) if not entry["converged"] or entry["iterations"] not in (1, 2): failures.append({"name": name, "mode": mode, "readout": "convergence metadata", "actual": [entry["iterations"], entry["converged"]]}) baseline = serial.get(name) if baseline is None: raise ValueError(f"missing serial reference for {name}") for readout in ("input_bits", "center_bits", "label_bits", "prediction_bits", "computed_distance_bits", "inertia_bits", "iterations", "converged"): if entry[readout] != baseline[readout]: failures.append({"name": name, "mode": mode, "readout": readout, "reason": "differs from the same-binary serial result"}) if name == "kernel-distance-SSE-contract" and mode == "serial": centers = [rational(value) for value in entry["center_bits"]] coordinate_sse = sum(((value - centers[index % d]) ** 2 for index, value in enumerate(inputs)), Fraction()) contract_example = { "computed_distance_bits": entry["computed_distance_bits"], "kernel_distance_squares_SSE_bits": expected_inertia, "exact_coordinate_SSE_bits": rounded(coordinate_sse), "different_contracts": expected_inertia != rounded(coordinate_sse), } if not observations or not serial: raise ValueError("empty numerical probe") required_modes = ["serial"] + [f"pool-{count}" for count in (1, 2, 4) if count <= raw["hard_limit"]] for name in serial: for mode in required_modes: if (name, mode) not in observed_pairs: failures.append({"name": name, "mode": mode, "reason": "missing admitted worker mode"}) if contract_example is None or not contract_example["different_contracts"]: failures.append({"name": "kernel-distance-SSE-contract", "reason": "missing arithmetic-contract distinction"}) root = Path(__file__).resolve().parent.parent paths = [root / "src/ml.cpp", root / "src/ml_exact.hpp", root / "include/aner/ml.hpp", root / "tests/kmeans_oracle_probe.cpp", Path(__file__).resolve()] report = { "schema": "aner-kmeans-public-numeric-oracle-v1", "passed": not failures, "scope": "K=1 full fit; exact rational centroid mean and exact sum of squares of emitted C++ kernel hypot distances; same-binary serial/worker readouts", "oracle": "Python fractions.Fraction with independently rounded Fraction.__float__", "fixture_count": len(serial), "observation_count": len(observations), "mode_counts": dict(mode_counts), "hard_limit": raw["hard_limit"], "skipped_worker_counts": [count for count in (1, 2, 4) if count > raw["hard_limit"]], "source_sha256": {str(path.relative_to(root)): sha(path) for path in paths}, "raw_observation_sha256": hashlib.sha256(raw_text.encode()).hexdigest(), "binary": str(args.binary.resolve()) if args.binary else None, "binary_sha256": sha(args.binary) if args.binary else None, "python": sys.version, "kernel_vs_coordinate_SSE_example": contract_example, "failure_count": len(failures), "failures": failures[:20], } if args.build_root: report["build_root"] = str(args.build_root.resolve()) build_files = [args.build_root / "CMakeCache.txt", args.build_root / "CMakeFiles/aner_ml.dir/flags.make"] report["build_metadata"] = {} for path in build_files: if not path.exists(): continue selected_lines = path.read_text().splitlines() if path.name == "CMakeCache.txt": selected_lines = [line for line in selected_lines if line.startswith(( "CMAKE_BUILD_TYPE:", "CMAKE_CXX_COMPILER:", "CMAKE_CXX_FLAGS:", "CMAKE_CXX_FLAGS_RELEASE:", "CMAKE_CXX_FLAGS_RELWITHDEBINFO:"))] report["build_metadata"][str(path.relative_to(args.build_root))] = { "sha256": sha(path), "selected_flags": selected_lines} ninja = args.build_root / "build.ninja" if ninja.exists(): # Retain only compilation blocks relevant to these observations, # rather than dumping unrelated build targets or configuration. selected_lines = [] retain = False for line in ninja.read_text().splitlines(): if line.startswith("build "): retain = line.startswith(("build CMakeFiles/aner_ml.dir/src/ml.cpp.o:", "build CMakeFiles/aner_kmeans_oracle_probe.dir/")) if retain: selected_lines.append(line) if not line: retain = False report["build_metadata"]["build.ninja"] = {"sha256": sha(ninja), "selected_flags": selected_lines} report["library_sha256"] = {name: sha(args.build_root / name) for name in ("libaner_ml.a", "libaner_nn.a", "libaner_tensor.a", "libaner_parallel.a") if (args.build_root / name).exists()} if args.output: args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(report, indent=2) + "\n") print(json.dumps({key: report[key] for key in ("passed", "fixture_count", "observation_count", "mode_counts", "failure_count")})) for failure in failures[:10]: print(json.dumps(failure), file=sys.stderr) return 1 if failures else 0 if __name__ == "__main__": raise SystemExit(main())