"""Bounded numerical transport. Canonical assessment belongs to Rust composition.""" import contextlib import hashlib import json import math import os from pathlib import Path import resource import sys import unittest MEMORY = 6 * 1024**3 PHASES = {"forward": -math.pi / 2, "reverse": math.pi / 2, "null": 0.0} def emit(record): # The strict protocol rejects negative zero. Normalise this representation # before producing the native JSON stream; the runner retains emitted bytes. def clean(value): if isinstance(value, dict): return {key: clean(item) for key, item in value.items()} if isinstance(value, list): return [clean(item) for item in value] if type(value) is float and value == 0: return 0.0 return value print(json.dumps(clean(record), allow_nan=False, separators=(",", ":")), flush=True) def restrict(memory=MEMORY): if sys.platform != "linux" or os.geteuid() == 0: raise ValueError("Linux with an unprivileged account is required") resource.setrlimit(resource.RLIMIT_AS, (memory, memory)) resource.setrlimit(resource.RLIMIT_CORE, (0, 0)) resource.setrlimit(resource.RLIMIT_FSIZE, (0, 0)) resource.setrlimit(resource.RLIMIT_CPU, (7200, 7200)) if resource.getrlimit(resource.RLIMIT_AS) != (memory, memory): raise ValueError("address-space ceiling not established") def no_children(event, _args): if event in {"os.fork", "os.forkpty", "os.posix_spawn", "subprocess.Popen", "os.system"}: raise ValueError("primary numerical process cannot create children") def checked_radiation(values): quantities = {"energy_flux", "momentum_flux", "source_reaction"} if (not isinstance(values, dict) or set(values) != quantities or any(type(v) not in (int, float) or not math.isfinite(v) for v in values.values()) or values["energy_flux"] <= 0 or abs(values["momentum_flux"] + values["source_reaction"]) > 1e-12): raise ValueError("invalid radiation observations") return values def check_field_identity(data, name, width, phase, config, measurements): h, half, cells, steps, dt = config expected = {"status": "finite-cell field qualification only; not a canonical result", "backend": "numpy-2.5.3", "configuration": name, "width": width, "phase": phase, "source_rule": "clipped-cell-mass-at-node-v1", "spacing": h, "time_step": dt, "steps": steps, "control_half_width": cells*h, "outer_half_width": half*h, "earliest_boundary_return": 2*half*h-cells*h-math.pi/4-3*width-h/2} if not isinstance(data, dict) or set(data) != set(expected) | set(measurements): raise ValueError("malformed field observations") for key, value in expected.items(): if isinstance(value, float): valid = type(data[key]) in (int, float) and math.isfinite(data[key]) and abs(data[key]-value) <= 1e-12 else: valid = type(data[key]) is type(value) and data[key] == value if not valid: raise ValueError("field observation identity drifted") if (any(type(data[key]) not in (int, float) or not math.isfinite(data[key]) for key in measurements) or type(data["source_nodes"]) is not int or data["source_nodes"] != (59582 if width == 0.25 else 31250)): raise ValueError("invalid field measurements") def prepare(): restrict() if sys.version_info[:2] != (3, 14): raise ValueError("Python series is not qualified") root = Path.cwd() contract = json.loads((root / "contract.json").read_bytes()) for item in contract["implementation"]: payload = (root / item["path"]).read_bytes() if len(payload) != item["bytes"] or "sha256:" + hashlib.sha256(payload).hexdigest() != item["identity"]: raise ValueError("staged input identity drifted") sys.path[:0] = [str(root / "deposition"), str(root)] import numpy if numpy.__version__ != "2.5.3": raise ValueError("NumPy version is not qualified") if os.uname().machine not in {"aarch64", "x86_64"}: raise ValueError("unsupported architecture") from numpy._core import _multiarray_umath with open(_multiarray_umath.__file__, "rb") as binary: payload = binary.read(64 * 1024**2 + 1) if len(payload) > 64 * 1024**2: raise ValueError("NumPy binary exceeds identification budget") emit({"type": "runtime", "python": ".".join(map(str, sys.version_info[:3])), "numpy": numpy.__version__, "platform": sys.platform, "architecture": os.uname().machine, "numpy_binary_identity": "sha256:" + hashlib.sha256(payload).hexdigest(), "memory_bytes": MEMORY, "memory_enforcement": "linux-rlimit-as"}) return contract def qualify(contract): # Qualification has its own deadline, before the primary clock starts. # The frozen test harness runs one bounded audit subprocess, not a primary # case. Its raw diagnostics remain on stderr and cannot become a verdict. modules = ["test_00_backend", "test_audit", "test_validation", "test_qualification"] suite = unittest.defaultTestLoader.loadTestsFromNames(modules) with contextlib.redirect_stdout(sys.stderr): result = unittest.TextTestRunner(stream=sys.stderr, verbosity=1).run(suite) emit({"type": "qualification", "passed": result.wasSuccessful(), "tests": result.testsRun, "failures": len(result.failures), "errors": len(result.errors), "skipped": len(result.skipped), "required_checks": contract["qualification"]["required_checks"]}) return 0 if result.wasSuccessful() and not result.skipped else 3 def primary(contract): import audit import field import radiation # The frozen modules are trusted numerical code, not an arbitrary-code # sandbox. The OS prevents additional processes, and the audit hook gives # an explicit diagnostic for Python-level attempts. resource.setrlimit(resource.RLIMIT_NPROC, (0, 0)) sys.addaudithook(no_children) for config in contract["configurations"]: expected = (float(config["spacing"]), config["outer_half_cells"], config["control_half_cells"], config["steps"], float(config["time_step"])) if field.CONFIGS[config["id"]] != expected: raise ValueError("field configuration drifted") for width in ("0.25", "0.2"): for label, phase in PHASES.items(): for nodes in (33, 65): emit({"type": "reference", "width": width, "phase": label, "nodes": nodes, "values": checked_radiation(audit.reference(float(width), phase, nodes))}) for scenario in contract["scenarios"]: name, width, phase = scenario["configuration"], float(scenario["width"]), PHASES[scenario["phase"]] print("primary case " + scenario["id"], file=sys.stderr, flush=True) data = field.simulate(name, width, phase) check_field_identity(data, name, width, phase, field.CONFIGS[name], audit.MEASUREMENTS) # Malformed output stops subsequent work; valid numerical violations do # not. No threshold assessment or canonical state occurs in this worker. eb = data["energy_change"] + data["integrated_energy_flux"] - data["integrated_source_work"] mb = data["momentum_change"] + data["integrated_momentum_flux"] + data["integrated_source_force"] if (any(not math.isfinite(data[key]) for key in audit.MEASUREMENTS) or abs(eb - data["energy_residual"]) > 1e-12 or abs(mb - data["momentum_residual"]) > 1e-12 or data["energy_change"] < 0 or data["integrated_energy_flux"] <= 0 or data["integrated_source_work"] <= 0 or data["peak_abs_stored_momentum"] < 0): raise ValueError("invalid field observations") emit({"type": "case", "id": scenario["id"], "field": data, "radiation": checked_radiation(radiation.integrated(radiation.axes(field.CONFIGS[name][0], width), phase))}) emit({"type": "complete", "cases": len(contract["scenarios"])}) return 0 def main(): try: if len(sys.argv) != 2 or sys.argv[1] not in {"qualify", "primary"}: raise ValueError("unsupported worker operation") contract = prepare() return qualify(contract) if sys.argv[1] == "qualify" else primary(contract) except (ValueError, ImportError) as error: emit({"type": "refusal", "detail": type(error).__name__}) return 2 except MemoryError: emit({"type": "resource_failure", "detail": "memory_limit"}) return 3 except Exception as error: # Do not leak private paths from exception messages into structured # observations. Native stderr is retained separately by the runner. emit({"type": "resource_failure", "detail": type(error).__name__}) return 3 if __name__ == "__main__": raise SystemExit(main())