#!/usr/bin/env python3 """Check up to three ordered frozen reduced-surface probes as an affine chain. This checks saved handoffs and the common physical shape span. Individual FEM artifact audits remain separate; --case-audit binds their completed JSON reviews without rerunning their larger matrix checks. Passing this audit establishes neither an unrestricted Newton solve nor a physically valid stellar endpoint. """ import argparse import csv import hashlib import json import math import os from pathlib import Path import sys for _name in ("OPENBLAS_NUM_THREADS", "OMP_NUM_THREADS", "MKL_NUM_THREADS", "VECLIB_MAXIMUM_THREADS"): os.environ[_name] = "1" import numpy as np PREFIX = "reduced_surface_" EPS = np.finfo(float).eps def digest(path): result = hashlib.sha256() with Path(path).open("rb") as stream: for chunk in iter(lambda: stream.read(1024 * 1024), b""): result.update(chunk) return result.hexdigest() def rows(path): with Path(path).open(newline="") as stream: reader = csv.DictReader(stream) if not reader.fieldnames or len(set(reader.fieldnames)) != len(reader.fieldnames): raise ValueError(f"Missing or duplicate CSV columns: {path}") result = list(reader) if any(None in row or any(value is None for value in row.values()) for row in result): raise ValueError(f"Malformed CSV data: {path}") return result def finite(value): result = float(value) if not math.isfinite(result): raise ValueError(f"Nonfinite value: {value}") return result def integer(value): result = finite(value) if result != int(result): raise ValueError(f"Nonintegral value: {value}") return int(result) def load(path, matrix=False): with Path(path).open() as stream: shape = tuple(int(value) for value in stream.readline().split()) values = np.loadtxt(stream, ndmin=1) if len(shape) != (2 if matrix else 1) or min(shape) <= 0 or values.ndim != 1 or values.size != math.prod(shape): raise ValueError(f"Invalid saved array dimensions: {path}") if not np.isfinite(values).all(): raise ValueError(f"Nonfinite saved array: {path}") return values.reshape(shape) def norm(values): # Stable scaling keeps an overflowing squared norm from becoming an # apparently permissive infinite threshold in the handoff checks. scale = float(np.max(np.abs(values), initial=0.)) result = 0. if scale == 0. else scale * math.sqrt(float(np.sum((values / scale) ** 2))) return finite(result) def compare(actual, expected, atol, rtol): if actual.shape != expected.shape: return {"passed": False, "reason": "shape_mismatch"} difference = actual - expected bound = atol + rtol * np.maximum(np.abs(actual), np.abs(expected)) maximum = float(np.max(np.abs(difference), initial=0.)) scale = max(norm(actual), norm(expected)) return {"passed": bool(np.all(np.abs(difference) <= bound)), "bitwise_equal": actual.tobytes() == expected.tobytes(), "maximum_absolute_difference": maximum, "difference_norm": norm(difference), "relative_difference_norm": norm(difference) / scale if scale else 0., "absolute_tolerance": atol, "relative_tolerance": rtol} def span_check(basis, increment, reference, atol, rtol): coefficients, _, rank, singular = np.linalg.lstsq(basis, increment, rcond=None) if not np.isfinite(coefficients).all() or not np.isfinite(singular).all(): raise ValueError("Nonfinite least-squares coefficients or singular values") # This small validation contraction needs no BLAS dispatch. Explicit # products also make a real overflow distinguishable from platform BLAS # floating-status warnings observed despite finite saved input vectors. with np.errstate(over="raise", invalid="raise", divide="raise"): reconstructed = np.sum(basis * coefficients[None, :], axis=1) if not np.isfinite(reconstructed).all(): raise ValueError("Nonfinite reconstructed shape increment") residual = increment - reconstructed error, size = norm(residual), norm(increment) # Subtracting two full shape bases has a small absolute rounding floor; # report it separately rather than inflating the relative span tolerance. subtraction_floor = float(64 * EPS * max(norm(reference), norm(reference + increment), 1e-300)) threshold = float(atol * math.sqrt(increment.size) + subtraction_floor + rtol * size) return {"passed": bool(int(rank) == basis.shape[1] and error <= threshold), "rank": int(rank), "requested_rank": basis.shape[1], "increment_norm": size, "projection_error_norm": error, "relative_projection_error": error / size if size else 0., "projection_error_threshold": threshold, "subtraction_roundoff_allowance": subtraction_floor, "largest_singular_value": float(singular[0]), "smallest_singular_value": float(singular[-1]), "coefficients": coefficients.tolist()} def inspect(directories, atol=1e-14, rtol=5e-10, span_rtol=5e-10, reviews=()): failures, checks, cases, links = [], 0, [], [] def check(label, condition): nonlocal checks checks += 1 if not condition: failures.append(label) individual = {} review_hashes = {} try: folders = [Path(path).resolve() for path in directories] if not 1 <= len(folders) <= 3 or len(set(folders)) != len(folders): raise ValueError("Provide one to three distinct case directories in chain order") for review_path in reviews: review_path = Path(review_path).resolve() review = json.loads(review_path.read_text()) review_hashes[str(review_path)] = digest(review_path) for case in review.get("cases", [review]): if "directory" not in case: raise ValueError("Individual review lacks its case directory") key = str(Path(case["directory"]).resolve()) if key in individual: raise ValueError(f"Duplicate individual case review: {key}") individual[key] = case for index, folder in enumerate(folders): used = {} def file(name, extension=".txt"): path = folder / (PREFIX + name + extension) used[path.name] = digest(path) return path summary_rows = rows(file("summary", ".csv")) summary = {row["metric"]: finite(row["value"]) for row in summary_rows} if len(summary) != len(summary_rows): raise ValueError("Duplicate summary metrics") layout = rows(file("layout", ".csv")) shape_rows = [row for row in layout if row["kind"] == "value" and row["block"] == "surface_deformation"] if len(shape_rows) != 1: raise ValueError("Missing or duplicate physical shape value block") offset, size = (integer(shape_rows[0][key]) for key in ("offset", "size")) data = {name: load(file(name)) for name in ("physical_base", "base", "residual", "state_factors", "residual_factors", "reduced_residual", "reduced_defect")} basis = load(file("basis_raw"), matrix=True) n = data["base"].size check(f"case{index}:shape_layout", 0 <= offset <= n - size and size > 0 and basis.shape[0] == size) check(f"case{index}:array_dimensions", all(data[key].shape == (n,) for key in ("physical_base", "residual", "state_factors", "residual_factors"))) check(f"case{index}:positive_factors", bool(np.all(data["state_factors"] > 0) and np.all(data["residual_factors"] > 0))) roundtrip = compare(data["base"], data["state_factors"] * data["physical_base"], atol, rtol) check(f"case{index}:physical_normalized_roundtrip", roundtrip["passed"]) check(f"case{index}:reduced_model", summary["identity_control"] == 0 and summary["shape_value_offset"] == offset and summary["shape_size"] == size and summary["mode_count"] == basis.shape[1]) reduced_norm, defect_norm = norm(data["reduced_residual"]), norm(data["reduced_defect"]) relative, absolute = summary["linear_relative_tolerance"], summary["linear_absolute_tolerance"] check(f"case{index}:linear_tolerance", 0 < relative < 1 and absolute >= 0 and reduced_norm > 0) threshold = max(relative * reduced_norm, absolute) reduced_gate = defect_norm <= threshold check(f"case{index}:reported_reduced_gate", summary["reduced_verified_gate"] in (0, 1) and bool(summary["reduced_verified_gate"]) == reduced_gate) trials = rows(file("trials", ".csv")) unit_rows = [row for row in trials if integer(row["sample"]) == 0] if len(unit_rows) != 1 or finite(unit_rows[0]["alpha"]) != 1.: raise ValueError("Expected exactly one alpha=1 sample zero") unit = unit_rows[0] unit_prepared = integer(unit["prepared"]) check(f"case{index}:unit_prepared_boolean", unit_prepared in (0, 1)) candidate = None merit = {"prepared": bool(unit_prepared)} if unit_prepared: candidate = {name: load(file(name + "_a0")) for name in ("physical_candidate", "candidate", "residual", "reduced_residual")} full_pass = norm(candidate["residual"]) <= (1. - 1e-4) * norm(data["residual"]) reduced_pass = norm(candidate["reduced_residual"]) <= (1. - 1e-4) * reduced_norm for label, passed in (("full", full_pass), ("reduced", reduced_pass)): flag = integer(unit[label + "_merit_pass"]) check(f"case{index}:reported_{label}_unit_merit", flag in (0, 1) and bool(flag) == passed) merit.update(full_merit_pass=full_pass, reduced_merit_pass=reduced_pass, full_norm=norm(candidate["residual"]), reduced_norm=norm(candidate["reduced_residual"])) else: check(f"case{index}:unprepared_merit_flags", integer(unit["full_merit_pass"]) == integer(unit["reduced_merit_pass"]) == 0) qualifies = bool(reduced_gate and unit_prepared and merit.get("full_merit_pass") and merit.get("reduced_merit_pass")) # Only predecessor probes must offer a successful handoff. A final # failed or rejected unit trial honestly terminates a valid prefix. if index < len(folders) - 1: check(f"case{index}:qualifies_for_next_base", qualifies) run_path, provenance_path = folder.parent / (folder.name + "_run.json"), folder.parent / "provenance.json" run = json.loads(run_path.read_text()) provenance = json.loads(provenance_path.read_text()) check(f"case{index}:completed_producer", bool(run.get("finished_utc")) and run.get("exit_code") == 0 and not run.get("timed_out", False)) for name, observed in used.items(): check(f"case{index}:producer_hash:{name}", run.get("output_hashes", {}).get(name) == observed) binding = individual.get(str(folder)) bound = False if reviews: check(f"case{index}:individual_review_present", binding is not None) if binding is not None: source = binding.get("observations", {}).get("provenance", {}) bound = bool(binding.get("artifact_pass") and source.get("bound") and source.get("run_sha256") == digest(run_path) and source.get("provenance_sha256") == digest(provenance_path)) check(f"case{index}:individual_review_bound", bound) record = {"directory": str(folder), "input_hashes": used, "run_sha256": digest(run_path), "provenance_sha256": digest(provenance_path), "individual_audit_bound": bound, "normalization_roundtrip": roundtrip, "reduced_verified_gate": reduced_gate, "reduced_defect_norm": defect_norm, "reduced_gate_threshold": threshold, "unit_trial": merit, "qualifies_for_continuation": qualifies, "physical_shape_span": None} if index == 0: first_basis, first_shape = basis, data["physical_base"][offset:offset + size].copy() shared_hashes = {name: used[PREFIX + name + extension] for name, extension in (("layout", ".csv"), ("state_factors", ".txt"), ("residual_factors", ".txt"), ("basis_raw", ".txt"))} first_provenance = provenance else: for name, expected in shared_hashes.items(): extension = ".csv" if name == "layout" else ".txt" check(f"case{index}:identical_{name}", used[PREFIX + name + extension] == expected) for key in ("binary_sha256", "mesh_sha256"): check(f"case{index}:identical_{key}", bool(provenance.get(key)) and provenance.get(key) == first_provenance.get(key)) if previous_candidate is None: raise ValueError("A chain predecessor has no prepared unit candidate") comparisons = {label: compare(data[current], previous_candidate[prior], atol, rtol) for label, current, prior in (("physical", "physical_base", "physical_candidate"), ("normalized", "base", "candidate"), ("residual", "residual", "residual"))} for label, observation in comparisons.items(): check(f"handoff{index}:{label}", observation["passed"]) links.append({"from": str(folders[index - 1]), "to": str(folder), **comparisons}) observation = span_check(first_basis, data["physical_base"][offset:offset + size] - first_shape, first_shape, atol, span_rtol) check(f"case{index}:same_affine_shape_span", observation["passed"]) record["physical_shape_span"] = observation cases.append(record) previous_candidate = candidate except (OSError, ValueError, KeyError, TypeError, IndexError, FloatingPointError, np.linalg.LinAlgError) as error: check("complete_consistent_chain_inputs:" + str(error), False) passed = not failures all_bound = bool(cases) and len(cases) == len(directories) and all(case["individual_audit_bound"] for case in cases) return {"schema_version": 1, "passed": passed, "checks_count": checks, "failures": failures, "individual_audits_all_bound": all_bound, "complete_artifact_validation": passed and all_bound, "case_audit_hashes": review_hashes, "cases": cases, "handoffs": links, "tolerances": {"handoff_absolute": atol, "handoff_relative": rtol, "shape_span_relative": span_rtol}, "scope": "Frozen saved-vector affine chain only; individual artifact audits are separate. No new FEM evaluation, unrestricted convergence, admissibility certificate, or physical validation is claimed."} def self_test(): # A good endpoint difference can be small in total norm yet outside the # allowed shape span. The orthogonal perturbation must still be caught. basis = np.array([[1., 0.], [0., 1.], [1., 1.]]) reference = np.array([.1, .2, .3]) increment = basis @ np.array([.03, -.01]) span = span_check(basis, increment, reference, 1e-14, 5e-10) assert span["passed"] assert json.loads(json.dumps(span, allow_nan=False))["passed"] is True assert not span_check(basis, increment + 1e-5 * np.array([1., 1., -1.]), reference, 1e-14, 5e-10)["passed"] assert not span_check(np.column_stack((basis[:, 0], basis[:, 0])), increment, reference, 1e-14, 5e-10)["passed"] assert compare(reference, reference.copy(), 1e-14, 5e-10)["bitwise_equal"] roundtrip = compare(reference + 2e-15, reference, 1e-14, 5e-10) assert roundtrip["passed"] and not roundtrip["bitwise_equal"] assert not compare(reference + 1e-5, reference, 1e-14, 5e-10)["passed"] # Two tiny producer folders exercise the handoff logic itself, including # the distinction between an honest stopped prefix and a falsely extended # chain. These are not substitutes for independent per-case FEM audits. import tempfile with tempfile.TemporaryDirectory() as temporary: root = Path(temporary) (root / "provenance.json").write_text(json.dumps({"binary_sha256": "synthetic", "mesh_sha256": "synthetic"})) def csv_save(folder, name, columns, values): with (folder / (PREFIX + name + ".csv")).open("w", newline="") as stream: writer = csv.writer(stream) writer.writerow(columns) writer.writerows(values) def save(folder, name, array): array = np.asarray(array) with (folder / (PREFIX + name + ".txt")).open("w") as stream: stream.write(" ".join(str(value) for value in array.shape) + "\n") np.savetxt(stream, array.ravel(), fmt="%.17g") def seal(folder): record = {"finished_utc": "synthetic_complete", "exit_code": 0, "output_hashes": {path.name: digest(path) for path in folder.iterdir()}} (root / (folder.name + "_run.json")).write_text(json.dumps(record)) first, second = root / "first", root / "second" for index, folder in enumerate((first, second)): folder.mkdir() base = np.array([1., .1 * index, .1 * index, 1.]) candidate = base + np.array([0., .1, .1, 0.]) residual = np.full(4, .5 ** index) reduced = np.full(3, .5 ** index) for name, array in {"physical_base": base, "base": base, "residual": residual, "state_factors": np.ones(4), "residual_factors": np.ones(4), "reduced_residual": reduced, "reduced_defect": .01 * reduced, "basis_raw": np.ones((2, 1)), "physical_candidate_a0": candidate, "candidate_a0": candidate, "residual_a0": .5 * residual, "reduced_residual_a0": .5 * reduced}.items(): save(folder, name, array) csv_save(folder, "summary", ("metric", "value"), {"identity_control": 0, "shape_value_offset": 1, "shape_size": 2, "mode_count": 1, "linear_relative_tolerance": .03, "linear_absolute_tolerance": 0., "reduced_verified_gate": 1}.items()) csv_save(folder, "layout", ("kind", "block", "offset", "size"), (("value", "head", 0, 1), ("value", "surface_deformation", 1, 2), ("value", "tail", 3, 1))) csv_save(folder, "trials", ("sample", "alpha", "prepared", "full_merit_pass", "reduced_merit_pass"), ((0, 1, 1, 1, 1),)) seal(folder) result = inspect((first, second)) assert result["passed"], result["failures"] assert json.loads(json.dumps(result, allow_nan=False))["passed"] is True assert not result["complete_artifact_validation"] original = (first / (PREFIX + "reduced_defect.txt")).read_bytes() save(first, "reduced_defect", np.ones(3)) seal(first) result = inspect((first, second)) assert not result["passed"] and "case0:qualifies_for_next_base" in result["failures"] (first / (PREFIX + "reduced_defect.txt")).write_bytes(original) seal(first) save(second, "physical_base", np.array([1., .10001, .1, 1.])) seal(second) result = inspect((first, second)) assert not result["passed"] and "handoff1:physical" in result["failures"] print("12 tiny affine-span, JSON, handoff, complete-prefix and corruption checks passed") def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("directories", nargs="*", type=Path) parser.add_argument("--output", type=Path) parser.add_argument("--case-audit", type=Path, action="append", default=[]) parser.add_argument("--absolute-tolerance", type=float, default=1e-14) parser.add_argument("--relative-tolerance", type=float, default=5e-10) parser.add_argument("--span-relative-tolerance", type=float, default=5e-10) parser.add_argument("--self-test", action="store_true") args = parser.parse_args() if args.self_test: self_test() return 0 if not 1 <= len(args.directories) <= 3 or args.output is None: parser.error("Provide one to three ordered case directories and --output") tolerances = (args.absolute_tolerance, args.relative_tolerance, args.span_relative_tolerance) if any(not math.isfinite(value) or value < 0 for value in tolerances): parser.error("Tolerances must be finite and nonnegative") output = args.output.resolve() if output.exists() or any(output.is_relative_to(path.resolve()) for path in args.directories): parser.error("Use a fresh output outside immutable case directories") result = inspect(args.directories, *tolerances, reviews=args.case_audit) result["checker_sha256"] = digest(__file__) output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n") print(f"{'PASS' if result['passed'] else 'FAIL'}: {result['checks_count']} chain checks; {output}") return 0 if result["passed"] else 1 if __name__ == "__main__": sys.exit(main())