Compare commits

..
Author SHA1 Message Date
ClaudeandClaude Opus 5.5 1fa498c7d7 kernels: shared block-window to global coordinate helper
BlockStep (slot-reservoir wall rows) and BlockChannelStep computed the
global cell of a block-window cell with the same four lines.
block_global_xy (core/layout.cuh, 2D, __forceinline__) now holds them;
both kernels call it with their own domain width. Same integer
operations: no numerical, ABI or scheduling change. The slot-reservoir
kernel layout is unchanged (no split).

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:24:27 +00:00
ClaudeandClaude Opus 5.5 781333407e kernels: one shared unforced MRT collision for blocked, channel and patch paths
BlockStep, BlockChannelStep and refinement.cu patch_collide each carried
the same compute_rho_u -> omega clamp -> collide_mrt_no_force sequence.
collide_mrt_unforced (collision_mrt.cuh, __forceinline__) now holds it
once; the three sites call it. Same operations in the same order: no
numerical, ABI or SRT change. The host source contract now requires one
helper definition and exactly one call per file, and forbids the old
inline sequence.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 01:15:47 +00:00
22 changed files with 119 additions and 888 deletions

No files matched your search

-5
View File
@@ -46,9 +46,4 @@ CUDA_VISIBLE_DEVICES=<n> python tests/validation/run_dfg_re20.py --case both \
The uniform D80 control separates method error from refinement error. `--plan` writes `plan.json` only. A GPU run writes `plan.json` and `result.json`
(per case: lattice parameters, every sample, window ranges, status, verdict).
Receipts: `plan.json` and `result.json` carry `schema` 2 and `provenance` (HEAD, src tree,
runner hash; clean checkout required, refused if it changes during the run). Receipts
written before this (schema 1, e.g. evidence 70f6617) are a bare list of cases. Read both
with `run_dfg_re20.read_result(path)` (`["cases"][0]` replaces the old `[0]`); existing
evidence scripts stay valid for schema-1 receipts and need this reader for new ones.
Host contract tests: `tests/unit/test_dfg_re20.py`.
+3 -9
View File
@@ -168,21 +168,15 @@ extern "C" __global__ void BlockStep(const fpxx* in, fpxx* out, unsigned int hal
stream_pull_load(k, f, in + base, j);
if (cell_kind == KIND_ACTIVE) {
if (d_block_wall.nx != 0) {
const unsigned int bx = NX - 2 * halo, by = NY - 2 * halo;
const unsigned int nbx = d_block_wall.nx / bx;
const unsigned int gx = (blockIdx.z % nbx) * bx + x - halo;
const unsigned int gy = (blockIdx.z / nbx) * by + y - halo;
unsigned int gx, gy;
block_global_xy(x, y, halo, d_block_wall.nx, gx, gy);
if (gy == 1)
reconstruct_bottom_velocity(f, 0.0f, bottom_wall_normal_velocity(
gx, d_block_wall.nx, d_block_wall.slot_width, d_block_wall.amplitude));
else if (gy == d_block_wall.ny - 2) reconstruct_top_reservoir(f, RHO);
}
#if COLLISION_MODEL == 2
// Uniform MRT sequence (collide_dispatch, zero forcing); SRT path below unchanged.
float rho, ux, uy;
compute_rho_u(f, rho, ux, uy);
collide_mrt_no_force(f, rho, ux, uy,
fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
collide_mrt_unforced(f); // SRT path below unchanged.
#else
float rho, ux, uy, feq[NQ];
compute_rho_u(f, rho, ux, uy);
@@ -53,6 +53,18 @@ __device__ __forceinline__ void index_from_thread(unsigned int& x,
k = (unsigned long)y * NX + x;
}
// Global cell of window cell (x, y) in block blockIdx.z of a row-major,
// full-domain block layout domain_nx cells wide. NX/NY are the local strides.
__device__ __forceinline__ void block_global_xy(unsigned int x, unsigned int y,
unsigned int halo, unsigned int domain_nx,
unsigned int& gx, unsigned int& gy)
{
const unsigned int bx = NX - 2 * halo, by = NY - 2 * halo;
const unsigned int nbx = domain_nx / bx;
gx = (blockIdx.z % nbx) * bx + x - halo;
gy = (blockIdx.z / nbx) * by + y - halo;
}
// ---------------------------------------------------------------------------
// Neighbor list (periodic BCs via modular arithmetic)
// ---------------------------------------------------------------------------
@@ -114,6 +114,16 @@ __device__ __forceinline__ void collide_mrt_no_force(float* __restrict__ g,
collide_mrt(g, rho, ux, uy, Fin, omega);
}
// Unforced MRT with the clamped device omega, as in collide_dispatch with zero
// forcing. Shared by the blocked, channel-root and dense-patch collisions.
__device__ __forceinline__ void collide_mrt_unforced(float* __restrict__ g)
{
float rho, ux, uy;
compute_rho_u(g, rho, ux, uy);
collide_mrt_no_force(g, rho, ux, uy,
fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
}
#elif LATTICE_MODEL == LATTICE_D3Q19
__device__ __forceinline__ void collide_mrt(float* __restrict__ g,
+1 -4
View File
@@ -37,10 +37,7 @@ __device__ unsigned int incoming_counts[NQ * TOTAL_CELLS];
__device__ __forceinline__ void patch_collide(float* f, unsigned long k) {
#if COLLISION_MODEL == 2
float rho, ux, uy;
compute_rho_u(f, rho, ux, uy);
collide_mrt_no_force(f, rho, ux, uy,
fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
collide_mrt_unforced(f);
#else
float rho, ux, uy, feq[NQ];
compute_rho_u(f, rho, ux, uy);
@@ -15,10 +15,8 @@ extern "C" __global__ void BlockChannelStep(
unsigned int x, y; unsigned long k;
index_from_thread(x, y, k);
if (x < halo || y < halo || x >= NX - halo || y >= NY - halo) return;
const unsigned int bx = NX - 2 * halo, by = NY - 2 * halo;
const unsigned int nbx = d_block_channel.nx / bx;
const unsigned int gx = (blockIdx.z % nbx) * bx + x - halo;
const unsigned int gy = (blockIdx.z / nbx) * by + y - halo;
unsigned int gx, gy;
block_global_xy(x, y, halo, d_block_channel.nx, gx, gy);
const unsigned long base = (unsigned long)blockIdx.z * NQ * TOTAL_CELLS;
const unsigned long cell = (unsigned long)blockIdx.z * TOTAL_CELLS + k;
float f[NQ];
@@ -55,11 +53,7 @@ extern "C" __global__ void BlockChannelStep(
}
if (!wall && !outlet) {
#if COLLISION_MODEL == 2
// Match uniform MRT arithmetic (collide_dispatch, zero forcing).
float rho, ux, uy;
compute_rho_u(f, rho, ux, uy);
collide_mrt_no_force(f, rho, ux, uy,
fminf(OMEGA_COLLISION_MAX, fmaxf(OMEGA_COLLISION_MIN, d_params.omega)));
collide_mrt_unforced(f);
#else
// Match uniform SRT arithmetic, including its shifted equilibrium path.
float rho, ux, uy, feq[NQ], forcing[NQ] = {};
@@ -1,66 +0,0 @@
"""Checkpoint save -> load round trip for a scene with passive bodies. Requires GPU.
A fresh Simulation with the same scene, restored from the checkpoint must hold the same populations,
flags, body count and step count, and must stay bitwise identical to the original
after both continue for the same number of steps. Checkpoints with Jets are
refused (documented limitation) and are not covered here.
"""
import os
import tempfile
import unittest
import numpy as np
from ._cases import make_sim
# Three separated cylinders (pinball-like); cut-link receivers stay in bulk fluid.
CIRCLES = (((40, 32), 6), ((56, 22), 6), ((56, 42), 6))
def scene():
sim = make_sim()
sim.initialize()
for center, radius in CIRCLES:
sim.add_body("circle", center=center, radius=radius)
sim.sync_bodies()
return sim
class CheckpointBodiesGPU(unittest.TestCase):
def test_round_trip_is_bitwise_and_continues_identically(self):
sim = scene()
sim.run(60)
with tempfile.NamedTemporaryFile(suffix=".h5", delete=False) as f:
path = f.name
restored = None
try:
saved = sim.save_checkpoint(path)
# The loader requires the same body count and geometry: rebuild the scene.
restored = scene()
restored.load_checkpoint(saved)
self.assertEqual(restored.bodies.count, len(CIRCLES))
self.assertEqual(restored.stepper.step_count, sim.stepper.step_count)
np.testing.assert_array_equal(restored.get_flags(), sim.get_flags())
a, b = sim.get_ddf(), restored.get_ddf()
self.assertEqual(a.dtype, b.dtype)
self.assertEqual(a.tobytes(), b.tobytes())
sim.run(40)
restored.run(40)
self.assertEqual(restored.stepper.step_count, sim.stepper.step_count)
a, b = sim.get_ddf(), restored.get_ddf()
self.assertTrue(np.isfinite(a).all())
self.assertEqual(a.tobytes(), b.tobytes())
for i in range(len(CIRCLES)):
self.assertTrue(np.all(np.isfinite(restored.read_force(i))))
finally:
if restored is not None:
restored.close()
os.unlink(path)
sim.close()
if __name__ == "__main__":
unittest.main()
-54
View File
@@ -75,59 +75,5 @@ class Re20Contract(unittest.TestCase):
self.assertEqual((out["status"], out["verdict"]), ("INCOMPLETE", None))
class Re20Provenance(unittest.TestCase):
def test_records_current_checkout(self):
record = v.provenance(rr.DEFAULT_SOURCE)
self.assertEqual(record["source_mode"], "current_checkout")
self.assertEqual(len(record["checkout_revision"]), 40)
self.assertEqual(len(record["src_tree"]), 40)
self.assertEqual(len(record["runner_sha256"]), 64)
self.assertFalse(record["checkout_dirty"])
self.assertEqual(v.RESULT_SCHEMA, 2)
def test_reader_accepts_both_schemas_and_rejects_unknown(self):
import json
import tempfile
from pathlib import Path
case = dict(case="refined", status="NOT_STEADY", verdict=None, rows=[])
with tempfile.TemporaryDirectory() as d:
old, new, bad = Path(d, "old.json"), Path(d, "new.json"), Path(d, "bad.json")
old.write_text(json.dumps([case]))
new.write_text(json.dumps(dict(schema=2, provenance=dict(src_tree="a" * 40),
cases=[case])))
bad.write_text(json.dumps(dict(schema=3, cases=[case])))
self.assertEqual(v.read_result(old), dict(schema=1, provenance=None, cases=[case]))
self.assertEqual(v.read_result(new)["cases"][0], case)
self.assertEqual(v.read_result(new)["provenance"]["src_tree"], "a" * 40)
with self.assertRaisesRegex(ValueError, "unknown DFG Re20 result schema"):
v.read_result(bad)
def test_dirty_checkout_is_refused(self):
from unittest.mock import patch
# Three git calls in order: HEAD:src, HEAD, status --porcelain.
with patch("subprocess.check_output",
side_effect=["e" * 40 + "\n", "f" * 40 + "\n", " M src/x.py\n"]) as git:
with self.assertRaisesRegex(RuntimeError, "clean checkout"):
v.provenance(rr.DEFAULT_SOURCE)
self.assertEqual(git.call_count, 3)
self.assertEqual([c.args[0][-1] for c in git.call_args_list[:2]], ["HEAD:src", "HEAD"])
self.assertIn("--porcelain", git.call_args_list[2].args[0])
def test_output_inside_checkout_is_refused_before_any_work(self):
import tempfile
from pathlib import Path
from unittest.mock import patch
source = rr.DEFAULT_SOURCE.resolve()
for output in (source, source / "re20-out", source / "src" / "x"):
argv = ["run_dfg_re20.py", "--plan", "--source", str(source), "--output", str(output)]
with patch("sys.argv", argv), \
patch.object(v, "provenance", side_effect=AssertionError("work started")):
with self.assertRaisesRegex(SystemExit, "outside the source checkout"):
v.main()
self.assertFalse((source / "re20-out").exists())
with tempfile.TemporaryDirectory() as d:
v.reject_output_inside(Path(d) / "out", source) # outside: accepted
if __name__ == "__main__":
unittest.main()
+1 -74
View File
@@ -166,80 +166,7 @@ class Host(unittest.TestCase):
self.assertEqual(a["steps"] * 4, b["steps"])
def test_source_pin(self):
"""Historical integrity: the manifest matches its pinned commit, read with git."""
from run_jet_circle_cv import verify_historical
from source_provenance import HistoryUnavailable, history_unavailable
try:
verify_historical(REPO)
except HistoryUnavailable as error:
history_unavailable(self, error)
def test_pinned_mode_still_rejects_changed_source(self):
"""Negative control: the frozen-source guard refuses a tree that differs from PIN."""
import json
import run_jet_circle_cv as v
manifest = json.loads((v.HERE / "data/jet_circle_cv/source_manifest.json").read_text())
import hashlib
current = {f: hashlib.sha256((REPO / f).read_bytes()).hexdigest()
for f in manifest["files"] if (REPO / f).exists()}
if current == manifest["files"]:
self.skipTest("working tree equals PIN; no changed source to reject")
with self.assertRaisesRegex(RuntimeError, "Source hash mismatch"):
verify(REPO)
def test_current_checkout_provenance(self):
from run_jet_circle_cv import current_checkout
record = current_checkout(REPO)
self.assertEqual(record["source_mode"], "current_checkout")
self.assertIn("not the frozen", record["source_capture"])
self.assertEqual(len(record["src_tree"]), 40)
self.assertTrue(record["src_sha256"])
def test_final_source_recheck_completes_or_invalidates_receipt(self):
import json
import tempfile
import run_jet_circle_cv as v
recorded = dict(source_mode="current_checkout", checkout_revision="a" * 40,
checkout_dirty=False, src_tree="b" * 40)
payload = dict(status="PILOT_DIAGNOSTIC_ONLY", run_status="INCOMPLETE", windows=[1, 2])
with tempfile.TemporaryDirectory() as d:
path = Path(d) / "re100_s1.json"
path.write_text(json.dumps(payload))
done = v.finish_receipt(path, payload, recorded, lambda: dict(recorded))
self.assertEqual(json.loads(path.read_text()), done)
self.assertEqual((done["run_status"], done["source_recheck"]), ("COMPLETE", "UNCHANGED"))
for recheck, state in ((lambda: dict(recorded, checkout_dirty=True), "CHANGED"),
(lambda: v.verify(Path(d)), "RECHECK_FAILED")):
path.write_text(json.dumps(payload))
with self.assertRaisesRegex(RuntimeError, "Source changed during the run"):
v.finish_receipt(path, payload, recorded, recheck)
bad = json.loads(path.read_text())
self.assertEqual((bad["run_status"], bad["source_recheck"]),
("INVALID_SOURCE_CHANGED", state))
self.assertEqual(bad["source_after"] is None, state == "RECHECK_FAILED")
self.assertEqual(bad["windows"], [1, 2])
self.assertFalse(Path(str(path) + ".tmp").exists())
def test_receipts_written_during_the_run_are_incomplete(self):
"""Static check (run_case needs CUDA): the in-run payload is INCOMPLETE and only
main() -> finish_receipt can mark it COMPLETE."""
import inspect
import run_jet_circle_cv as v
self.assertIn('run_status="INCOMPLETE"', inspect.getsource(v.run_case))
self.assertNotIn("COMPLETE\"", inspect.getsource(v.run_case).replace("INCOMPLETE\"", ""))
self.assertIn("finish_receipt(", inspect.getsource(v.main))
def test_output_inside_checkout_is_refused_before_source_or_cuda(self):
from unittest import mock
import run_jet_circle_cv as v
argv = ["run_jet_circle_cv.py", "--repo", str(REPO), "--output", str(REPO / "v1-out")]
with mock.patch("sys.argv", argv), mock.patch.dict("os.environ", CUDA_VISIBLE_DEVICES="0"), \
mock.patch.object(v, "source_identity", side_effect=AssertionError("work started")):
with self.assertRaisesRegex(SystemExit, "outside the source checkout"):
v.main()
self.assertFalse((REPO / "v1-out").exists())
verify(REPO)
if __name__ == "__main__":
-34
View File
@@ -151,8 +151,6 @@ class Measurements(unittest.TestCase):
(path / "samples_s2.json").write_text(
json.dumps(dict(case=c, sample_file="samples_s2.npz"))
)
(path / f"re{c['re_d']}_s2.json").write_text(json.dumps(
dict(case=c, run_status="COMPLETE", source_recheck="UNCHANGED")))
np.savez(
path / "samples_s2.npz",
t_reference=t,
@@ -165,7 +163,6 @@ class Measurements(unittest.TestCase):
velocity_error=np.zeros(n),
)
result = analyze(path)
self.assertEqual(result["receipt_gate"], "COMPLETE_SOURCE_UNCHANGED")
self.assertEqual(len(result["windows"]), 32)
np.testing.assert_allclose(
result["windows"][0]["normalized_cv_residual"], 0, atol=1e-14
@@ -181,36 +178,5 @@ class Measurements(unittest.TestCase):
np.testing.assert_allclose(integrate(t, y), [112, 8], atol=1e-15)
def test_receipt_gate_refuses_incomplete_invalid_and_unmarked_legacy(self):
import json
from analyze_jet_cv_continuation import receipt_gate
c = case(2, steps=3200)
with tempfile.TemporaryDirectory() as tmp:
path = Path(tmp)
receipt = path / f"re{c['re_d']}_s2.json"
with self.assertRaisesRegex(ValueError, "No sibling run receipt"):
receipt_gate(path, c)
for fields, message in (
(dict(run_status="INCOMPLETE"), "INCOMPLETE"),
(dict(run_status="INVALID_SOURCE_CHANGED", source_recheck="CHANGED"), "INVALID"),
(dict(run_status="COMPLETE", source_recheck="RECHECK_FAILED"), "RECHECK_FAILED"),
(dict(run_status="COMPLETE"), "source_recheck=None"),
(dict(), "Legacy run receipt"),
):
receipt.write_text(json.dumps(dict(case=c, **fields)))
with self.assertRaisesRegex(ValueError, message):
receipt_gate(path, c)
receipt.write_text(json.dumps(dict(case=dict(c, scale=1), run_status="COMPLETE",
source_recheck="UNCHANGED")))
with self.assertRaisesRegex(ValueError, "different case"):
receipt_gate(path, c)
receipt.write_text(json.dumps(dict(case=c)))
self.assertEqual(receipt_gate(path, c, legacy=True), "LEGACY_NO_FINAL_SOURCE_RECHECK")
receipt.write_text(json.dumps(dict(case=c, run_status="INCOMPLETE")))
with self.assertRaisesRegex(ValueError, "INCOMPLETE"):
receipt_gate(path, c, legacy=True) # legacy opt-in never admits new statuses
if __name__ == "__main__":
unittest.main()
+21 -43
View File
@@ -13,15 +13,8 @@ from unittest.mock import patch
sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "validation"))
from run_uniform_les_flow import source_guard, decode_macros, HERE, HEADER
from source_provenance import (HistoryUnavailable, historical_files, historical_sha256,
historical_tree, history_unavailable)
ROOT = HERE.parents[1]
A135252 = "a135252ae78b47b57d7e77d19001700b3bedac8b"
def a135252_pins():
return json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
class GuardTests(unittest.TestCase):
@@ -48,35 +41,20 @@ class GuardTests(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, "velocity/profile"):
decode_macros(invalid, w, cx, cy)
def materialize(self, root):
"""The pinned a135252 tree, read with git (the guard's own qualification)."""
try:
historical_files(ROOT, A135252, a135252_pins()["sources"], root)
except HistoryUnavailable as error:
history_unavailable(self, error)
def test_a135252_pins_qualify_their_own_commit(self):
with TemporaryDirectory() as directory:
root = Path(directory)
self.materialize(root)
with patch("run_uniform_les_flow.subprocess.check_output",
return_value="isolated-fixture\n"):
self.assertEqual(source_guard(root, "corrected", "a135252")["base_commit"], A135252)
def test_current_tree_is_not_the_a135252_qualification(self):
"""Explicit record: the guard refuses today's src wherever it differs from the pin."""
pins = a135252_pins()["sources"]
changed = sorted(n for n, sha in pins.items() if (ROOT / n).exists()
and hashlib.sha256((ROOT / n).read_bytes()).hexdigest() != sha)
if not changed:
self.skipTest("current src equals the a135252 pin")
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
source_guard(ROOT, "corrected", "a135252")
def test_current_corrected_pins(self):
self.assertEqual(
source_guard(ROOT, "corrected", "a135252")["base_commit"],
"a135252ae78b47b57d7e77d19001700b3bedac8b",
)
def test_corrected_and_tamper_guard(self):
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
with TemporaryDirectory() as directory:
root = Path(directory)
self.materialize(root)
for name in pins["sources"]:
dest = root / name
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(ROOT / name, dest)
with patch(
"run_uniform_les_flow.subprocess.check_output",
return_value="isolated-fixture\n",
@@ -107,25 +85,25 @@ class GuardTests(unittest.TestCase):
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
source_guard(ROOT, "corrected") # Omission still uses the historical profile.
def test_manifest_pins_the_a135252_corrected_tree(self):
pins = a135252_pins()
self.assertEqual(pins["base_commit"], A135252)
def test_current_manifest_pins_actual_corrected_tree(self):
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
tracked = subprocess.check_output(["git", "-C", str(ROOT), "ls-files", "src"], text=True)
self.assertEqual(set(pins["sources"]), set(tracked.splitlines()))
for name, expected in pins["sources"].items():
self.assertEqual(hashlib.sha256((ROOT / name).read_bytes()).hexdigest(), expected)
self.assertEqual(pins["sources"][HEADER], pins["candidate_header_sha256"])
self.assertNotEqual(pins["sources"][HEADER], pins["baseline_header_sha256"])
with self.assertRaisesRegex(RuntimeError, "Source pin mismatch"):
source_guard(ROOT, "baseline", "a135252")
try:
names = historical_tree(ROOT, A135252)
actual = historical_sha256(ROOT, A135252, pins["sources"])
except HistoryUnavailable as error:
history_unavailable(self, error)
self.assertEqual(set(pins["sources"]), set(names))
self.assertEqual(actual, pins["sources"])
def test_named_baseline_accepts_only_reversed_header(self):
pins = json.loads((HERE / "data/les_flow/source_pins_a135252.json").read_text())
with TemporaryDirectory() as directory:
root = Path(directory)
self.materialize(root)
for name in pins["sources"]:
dest = root / name
dest.parent.mkdir(parents=True, exist_ok=True)
shutil.copyfile(ROOT / name, dest)
subprocess.run(["git", "apply", "--reverse", str(HERE / "les/data/smagorinsky_normalization.patch")], cwd=root, check=True)
with patch("run_uniform_les_flow.subprocess.check_output", return_value="baseline-fixture\n"):
source_guard(root, "baseline", "a135252")
+3 -21
View File
@@ -62,31 +62,13 @@ def test_periodic_synthetic():
assert not p.assess(rows()[:100])["periodic"]
def test_manifest_matches_its_pinned_commit():
"""Historical integrity only: the manifest describes the h.PIN tree (read with git)."""
def test_production_unmodified():
m = json.loads(
(ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
)
assert m["commit"] == h.PIN
import unittest
from source_provenance import HistoryUnavailable, history_unavailable, verify_manifest
try:
verify_manifest(ROOT, m["commit"], m["production_sha256"])
except HistoryUnavailable as error:
history_unavailable(unittest.TestCase(), error)
def test_current_src_is_not_the_pinned_qualification():
"""Explicit record: today's src differs from h.PIN, so its qualification does not carry over."""
m = json.loads(
(ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
)
changed = sorted(n for n, d in m["production_sha256"].items()
if not (ROOT / n).exists()
or hashlib.sha256((ROOT / n).read_bytes()).hexdigest() != d)
if not changed:
pytest.skip("current src equals the pinned tree")
assert changed # e.g. src/CelerisLab/lbm/channel.py since 47b05bc
for name, digest in m["production_sha256"].items():
assert hashlib.sha256((ROOT / name).read_bytes()).hexdigest() == digest, name
def test_reference():
+11 -29
View File
@@ -12,33 +12,16 @@ import run_refined_dfg as r
h, periodic = r.helpers(r.DEFAULT_SOURCE)
def historical_manifest(collision="SRT", observation_mode="last_base_step"):
"""The manifest verify_source would select, without hashing the working tree.
Receipt/plan logic tests use it; source integrity is checked once in test_source_pin.
"""
name = ("source_manifest_window.json" if observation_mode == "window" else
"source_manifest.json" if collision == "SRT" else "source_manifest_mrt.json")
return json.loads((r.HERE / "data/refined_dfg" / name).read_text())
def test_source_pin():
historical = historical_manifest("SRT")
historical = json.loads((r.HERE / "data/refined_dfg/source_manifest.json").read_text())
assert historical["commit"].startswith("cd3a657")
mrt_history = historical_manifest("MRT")
mrt_history = json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())
assert mrt_history["commit"].startswith("3ea131e")
assert historical_manifest("MRT", "window")["commit"].startswith("a233c90")
# Negative controls first, so missing history can never hide them: every pinned
# mode refuses a working tree that differs from its pin.
for collision, mode in (("MRT", "last_base_step"), ("SRT", "last_base_step"), ("MRT", "window")):
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, collision, mode)
from source_provenance import HistoryUnavailable, history_unavailable, verify_manifest
for manifest in (historical, mrt_history, historical_manifest("MRT", "window")):
try: # historical integrity: each manifest matches the tree of its own commit
verify_manifest(r.DEFAULT_SOURCE, manifest["commit"], manifest["sha256"])
except HistoryUnavailable as error:
history_unavailable(unittest.TestCase(), error)
assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["commit"].startswith("a233c90")
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, "MRT")
with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
r.verify_source(r.DEFAULT_SOURCE, "SRT")
def test_matched_units_and_planes():
@@ -160,13 +143,12 @@ def test_axis_specific_sampling_phase():
def test_full_requires_completed_pilot():
with unittest.TestCase().assertRaisesRegex(ValueError, "pilot-results"):
r.verify_pilot(None, historical_manifest("MRT", "window"))
r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window"))
def pilot_fixture(path, observation_mode="window", jet_scenario="none", case_selection="both", uniform_diameter=80, placement="wide"):
"""Synthetic contract fixture only; never a physical result or run receipt."""
manifest = ((historical_manifest("MRT", "window") if jet_scenario == "none" else
r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario))
manifest = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario)
if observation_mode == "window" else
json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text()))
selection = dict(case_selection=case_selection, uniform_diameter=uniform_diameter, placement=placement)
@@ -366,7 +348,7 @@ def test_record_actual_checkout_separately():
actual = checkout_provenance(r.DEFAULT_SOURCE)
assert len(actual["checkout_revision"]) == 40
assert isinstance(actual["checkout_dirty"], bool)
assert historical_manifest("MRT", "window")["production_origin_commit"].startswith("d930")
assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["production_origin_commit"].startswith("d930")
def test_window_plan_preserves_physics():
@@ -376,7 +358,7 @@ def test_window_plan_preserves_physics():
assert baseline.pop("observation_mode") == "last_base_step"
assert window.pop("observation_mode") == "window"
assert baseline == window
assert historical_manifest(collision, "window")["observation_mode"] == "window"
assert r.verify_source(r.DEFAULT_SOURCE, collision, "window")["observation_mode"] == "window"
with unittest.TestCase().assertRaises(ValueError):
r.verify_source(r.DEFAULT_SOURCE, "MRT", "unknown")
+6 -1
View File
@@ -37,8 +37,13 @@ class RefinementMRTHost(unittest.TestCase):
self.assertIn("STORE_PRECISION != 0", source)
self.assertIn('#if COLLISION_MODEL == 2\n#include "operators/collision_mrt.cuh"\n#endif', source)
self.assertIn("USE_DDF_SHIFTING ||", refinement) # dense/mixed shifted stays rejected
# One shared unforced MRT body; each blocked/patch path calls it exactly once.
mrt = (KERNELS / "operators/collision_mrt.cuh").read_text()
self.assertEqual(mrt.count("void collide_mrt_unforced("), 1)
for source in (blocks, channel, refinement):
self.assertEqual(source.count("collide_mrt_no_force("), 1)
self.assertEqual(source.count("collide_mrt_unforced(f);"), 1)
self.assertNotIn("collide_mrt_no_force(", source)
self.assertNotIn("compute_rho_u(f, rho, ux, uy);\n collide_mrt", source)
if __name__ == "__main__": unittest.main()
+10 -75
View File
@@ -8,19 +8,17 @@ import sys
ROOT = Path(__file__).resolve().parents[2]
sys.path.insert(0, str(ROOT / "tests/validation"))
import pytest
from source_provenance import (HistoryUnavailable, checkout_provenance, historical_sha256,
historical_tree, history_unavailable, verify_manifest)
import unittest
from source_provenance import checkout_provenance
import re100_helpers
import run_jet_circle_cv
ORIGIN = "47b05bc8abfc957fe876b0c63bdb386e3c6cbaa5"
def test_manifests_match_their_recorded_source_and_preserve_origin():
"""Historical evidence integrity: each manifest describes the tree of its own commit."""
manifests = []
def test_manifests_cover_current_source_and_preserve_origin():
names = subprocess.check_output(
["git", "-C", str(ROOT), "ls-files", "src"], text=True
).splitlines()
for group, hashes_key, revision_key in (
("dfg_re100", "production_sha256", "commit"),
("jet_circle_cv", "files", "revision"),
@@ -30,22 +28,18 @@ def test_manifests_match_their_recorded_source_and_preserve_origin():
ROOT / "tests/validation/data" / group / "source_manifest.json"
).read_text()
)
# Structural checks need no history and always run.
assert manifest[revision_key] == ORIGIN
assert set(manifest[hashes_key]) == set(names)
assert manifest["source_origin_commits"][-1] == ORIGIN
assert (
manifest["historical_checkpoints"]["production"]
== "d79e3f6f0a424f5bcc3a1a6d610313e49daf97a2"
)
assert "not inherited from d79" in manifest["experiment"]
manifests.append(manifest[hashes_key])
try:
names = historical_tree(ROOT, ORIGIN)
for digests in manifests:
assert set(digests) == set(names)
verify_manifest(ROOT, ORIGIN, digests)
except HistoryUnavailable as error:
history_unavailable(unittest.TestCase(), error)
for name, expected in manifest[hashes_key].items():
assert (
hashlib.sha256((ROOT / name).read_bytes()).hexdigest() == expected
), name
assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN
@@ -78,62 +72,3 @@ def test_runtime_checkout_identity_is_separate_from_source_origin(tmp_path):
assert checkout_provenance(tmp_path) == dict(
checkout_revision=expected, checkout_dirty=True
)
def test_historical_check_rejects_tampering_and_reports_missing_history(tmp_path):
with pytest.raises(HistoryUnavailable):
historical_sha256(tmp_path, ORIGIN, ["src/x.py"]) # not a repository
with pytest.raises(HistoryUnavailable):
historical_sha256(ROOT, "0" * 40, ["src/x.py"]) # commit absent
manifest = json.loads(
(ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text())
digests = dict(manifest["production_sha256"])
name = sorted(digests)[0]
tampered = dict(digests, **{name: ("0" if digests[name][0] != "0" else "1") + digests[name][1:]})
try:
historical_tree(ROOT, ORIGIN)
except HistoryUnavailable as error:
history_unavailable(unittest.TestCase(), error)
with pytest.raises(RuntimeError, match="Historical manifest mismatch"):
verify_manifest(ROOT, ORIGIN, tampered) # one flipped digest is caught
with pytest.raises(ValueError, match="not part of"):
historical_sha256(ROOT, ORIGIN, ["src/CelerisLab/does_not_exist.py"])
def test_historical_files_reports_unreadable_blob_as_history_unavailable(tmp_path):
"""A blob missing from the clone is HistoryUnavailable, not a raw CalledProcessError."""
import source_provenance
from unittest.mock import patch
missing = subprocess.CompletedProcess(["git", "show"], 128, b"", b"fatal: missing blob")
with patch.object(source_provenance, "historical_sha256", return_value={"src/a.py": "0" * 64}), \
patch.object(source_provenance.subprocess, "run", return_value=missing):
with pytest.raises(HistoryUnavailable, match="blob not in this clone"):
source_provenance.historical_files(ROOT, ORIGIN, ["src/a.py"], tmp_path)
assert not (tmp_path / "src/a.py").exists()
def test_final_source_check_unchanged_changed_and_failed_recheck():
from source_provenance import final_source_check
recorded = dict(revision="a" * 40, dirty=False)
assert final_source_check(recorded, lambda: dict(recorded)) == dict(source_recheck="UNCHANGED")
changed = final_source_check(recorded, lambda: dict(recorded, dirty=True))
assert changed == dict(source_recheck="CHANGED", source_after=dict(recorded, dirty=True),
source_recheck_error=None)
def fails():
raise RuntimeError("Source hash mismatch: src/x.py")
failed = final_source_check(recorded, fails)
assert (failed["source_recheck"], failed["source_after"]) == ("RECHECK_FAILED", None)
assert "Source hash mismatch" in failed["source_recheck_error"]
def test_reject_output_inside_checkout(tmp_path):
from source_provenance import reject_output_inside
for output in (ROOT, ROOT / "out", ROOT / "src" / "x.json"):
with pytest.raises(SystemExit, match="outside the source checkout"):
reject_output_inside(output, ROOT)
reject_output_inside(tmp_path / "out", ROOT)
@@ -14,35 +14,11 @@ def integrate(t, y):
)
def receipt_gate(directory, c, legacy=False):
"""Side files are used only if the sibling run receipt finished with an unchanged source.
Receipts from runners before the final source recheck have no run_status; they are
refused unless legacy=True (--legacy-receipts), and the result is labelled as such.
"""
path = Path(directory) / f"re{c['re_d']}_s{c['scale']}.json"
if not path.exists():
raise ValueError(f"No sibling run receipt {path.name}")
receipt = json.loads(path.read_text())
if receipt.get("case") != c:
raise ValueError("Sibling run receipt describes a different case")
if "run_status" not in receipt:
if not legacy:
raise ValueError("Legacy run receipt without run_status/final source recheck; "
"pass --legacy-receipts to analyze it as such")
return "LEGACY_NO_FINAL_SOURCE_RECHECK"
if receipt["run_status"] != "COMPLETE" or receipt.get("source_recheck") != "UNCHANGED":
raise ValueError(f"Run receipt is {receipt['run_status']} "
f"(source_recheck={receipt.get('source_recheck')})")
return "COMPLETE_SOURCE_UNCHANGED"
def analyze(directory, legacy=False):
def analyze(directory):
meta_path = next(Path(directory).glob("samples_s*.json"))
meta = json.loads(meta_path.read_text())
c = meta["case"]
s = c["scale"]
gate = receipt_gate(directory, c, legacy)
sample_path = meta_path.parent / meta["sample_file"]
checksum_verified = False
if "sample_sha256" in meta:
@@ -121,7 +97,6 @@ def analyze(directory, legacy=False):
)
return dict(
scale=s,
receipt_gate=gate,
complete_schedule=bool(complete_schedule),
sample_checksum_verified=checksum_verified,
windows=records,
@@ -140,10 +115,8 @@ def main():
p = argparse.ArgumentParser(description=__doc__)
p.add_argument("directories", nargs="+", type=Path)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--legacy-receipts", action="store_true",
help="Accept receipts written before the final source recheck (labelled)")
a = p.parse_args()
results = [analyze(d, a.legacy_receipts) for d in a.directories]
results = [analyze(d) for d in a.directories]
payload = dict(
status="TRANSIENT_DIAGNOSTIC_COMPARISON",
results=results,
+2 -47
View File
@@ -57,8 +57,6 @@ WINDOW_TSTAR = 10.0 # trailing 5 s, 21 samples
BANDS = dict(cd=(5.57, 5.59), cl=(0.0104, 0.0110), p_diff=(0.1172, 0.1176))
STEADY_RANGE = {k: (hi - lo) / 10 for k, (lo, hi) in BANDS.items()}
STEADY_SLOPE = {k: (hi - lo) / 50 for k, (lo, hi) in BANDS.items()}
# 2: {schema, provenance, cases}; the earlier receipts (schema 1) are a bare list of cases.
RESULT_SCHEMA = 2
DEVIATIONS = ("pressure-NEQ outlet (neq_extrap), not do-nothing", "local Zou-He inlet")
_re100_config = rr.config
@@ -108,21 +106,6 @@ def summarize(rows):
verdict=inside)
def read_result(path):
"""Cases and provenance from a result.json of either schema.
Schema 1 (receipts before RESULT_SCHEMA=2, e.g. evidence 70f6617) is a bare list of
cases without provenance; schema 2 is {schema, provenance, cases}. Readers that
indexed [0] on schema 1 must use read_result(path)["cases"][0].
"""
data = json.loads(Path(path).read_text())
if isinstance(data, list):
return dict(schema=1, provenance=None, cases=data)
if not isinstance(data, dict) or data.get("schema") != RESULT_SCHEMA:
raise ValueError(f"unknown DFG Re20 result schema: {data.get('schema') if isinstance(data, dict) else type(data).__name__}")
return dict(schema=RESULT_SCHEMA, provenance=data["provenance"], cases=data["cases"])
def healthy(row):
return row["max_mach"] <= 0.2 and row["rho_min"] >= 0.5 and row["rho_max"] <= 1.5
@@ -211,27 +194,6 @@ def plan(collision="SRT"):
label="compared against the official reference bands (non-canonical neq_extrap outlet, zou_he_local inlet); not a full benchmark certification")
def provenance(source):
"""Current checkout identity: HEAD, dirty flag, src tree and this runner's hash."""
import hashlib
import subprocess
from source_provenance import checkout_provenance
tree = subprocess.check_output(["git", "-C", str(source), "rev-parse", "HEAD:src"],
text=True).strip()
record = dict(source_mode="current_checkout", **checkout_provenance(source), src_tree=tree,
runner_sha256=hashlib.sha256(Path(__file__).read_bytes()).hexdigest())
if record["checkout_dirty"]: # HEAD would not describe the code that runs
raise RuntimeError("DFG Re20 runs require a clean checkout")
return record
def reject_output_inside(output, source):
"""Receipts inside the checkout would make it dirty and change the recorded source."""
output = Path(output).resolve()
if output == source or source in output.parents:
raise SystemExit(f"--output must be outside the source checkout {source}")
def main():
parser = argparse.ArgumentParser(description=__doc__.splitlines()[0])
parser.add_argument("--case", choices=["refined", "uniform", "both"], default="refined")
@@ -242,10 +204,8 @@ def main():
parser.add_argument("--max-wall-seconds", type=float)
a = parser.parse_args()
a.source = a.source.resolve()
reject_output_inside(a.output, a.source)
record = dict(plan(a.collision), schema=RESULT_SCHEMA, provenance=provenance(a.source))
a.output.mkdir(parents=True, exist_ok=False)
(a.output / "plan.json").write_text(json.dumps(record, indent=2, allow_nan=False))
(a.output / "plan.json").write_text(json.dumps(plan(a.collision), indent=2, allow_nan=False))
if a.plan:
return
if not a.max_wall_seconds or a.max_wall_seconds <= 0 or not os.environ.get("CUDA_VISIBLE_DEVICES"):
@@ -255,12 +215,7 @@ def main():
a.deadline = time.monotonic() + a.max_wall_seconds
cases = ["refined", "uniform"] if a.case == "both" else [a.case]
results = [run_case(a, h, case) for case in cases]
after = provenance(a.source)
if after != record["provenance"]:
raise RuntimeError("Source checkout changed during the run")
(a.output / "result.json").write_text(json.dumps(dict(schema=RESULT_SCHEMA, provenance=after,
cases=results),
indent=2, allow_nan=False))
(a.output / "result.json").write_text(json.dumps(results, indent=2, allow_nan=False))
if __name__ == "__main__":
+11 -71
View File
@@ -3,7 +3,6 @@
import argparse
import hashlib
import subprocess
import json
import os
from pathlib import Path
@@ -221,60 +220,6 @@ def verify(repo):
return manifest
def verify_historical(repo):
"""The pinned manifest describes the PIN tree (read with git, not the working tree)."""
from source_provenance import verify_manifest
manifest = json.loads((HERE / "data/jet_circle_cv/source_manifest.json").read_text())
if manifest["revision"] != PIN:
raise RuntimeError("Wrong pinned revision")
verify_manifest(repo, PIN, manifest["files"])
return manifest
def current_checkout(repo):
"""Provenance for a run on the current checkout; never labelled as the pinned source."""
files = subprocess.check_output(["git", "-C", str(repo), "ls-files", "src"],
text=True).splitlines()
return dict(source_mode="current_checkout",
source_capture="current checkout, not the frozen pinned qualification",
source_revision=checkout_provenance(repo)["checkout_revision"],
src_tree=subprocess.check_output(["git", "-C", str(repo), "rev-parse",
"HEAD:src"], text=True).strip(),
src_sha256={f: hashlib.sha256((repo / f).read_bytes()).hexdigest()
for f in files})
def source_identity(repo, current):
"""Recorded source of a run; recomputed after the run by finish_receipt."""
if current:
provenance = dict(**checkout_provenance(repo), **current_checkout(repo))
if provenance["checkout_dirty"]:
raise RuntimeError("Current-checkout runs require a clean checkout")
return provenance
source = verify(repo)
return dict(source_experiment=source["experiment"], source_mode="pinned",
**checkout_provenance(repo))
def finish_receipt(path, payload, provenance, recheck):
"""Mark the receipt COMPLETE only if the source is unchanged; otherwise INVALID.
Receipts written during the run carry run_status="INCOMPLETE", so an interrupted
run never leaves a complete-looking receipt.
"""
from source_provenance import final_source_check
fields = final_source_check(provenance, recheck)
unchanged = fields["source_recheck"] == "UNCHANGED"
payload = dict(payload, **fields,
run_status="COMPLETE" if unchanged else "INVALID_SOURCE_CHANGED")
temporary = Path(str(path) + ".tmp")
temporary.write_text(json.dumps(payload, indent=2) + "\n")
temporary.replace(path)
if not unchanged:
raise RuntimeError("Source changed during the run; receipt marked invalid")
return payload
def run_case(c, ctx, output, provenance, continuation=False):
from CelerisLab.body.geometry.circle import CircleGeometry
from CelerisLab.body.objects import SimObject
@@ -585,16 +530,16 @@ def run_case(c, ctx, output, provenance, continuation=False):
recorder.flush()
payload = dict(
case=c,
**{"source_revision": PIN, **provenance},
source_revision=PIN,
**provenance,
status="PILOT_DIAGNOSTIC_ONLY",
run_status="INCOMPLETE", # final source recheck in main() completes it
windows=rows,
physics_validation_pass=False,
precision="FP32 evolution, FP64 host diagnostics",
)
receipt = output / f"re{c['re_d']}_s{c['scale']}.json"
Path(str(receipt) + ".tmp").write_text(json.dumps(payload, indent=2) + "\n")
Path(str(receipt) + ".tmp").replace(receipt)
(output / f"re{c['re_d']}_s{c['scale']}.json").write_text(
json.dumps(payload, indent=2) + "\n"
)
print(
json.dumps(
dict(
@@ -626,9 +571,6 @@ def main():
help="Fixed measurement campaign: Re100, scale1/2, 3200 reference steps",
)
p.add_argument("--plan", action="store_true")
p.add_argument("--current-checkout", action="store_true",
help="Run on the current clean checkout and record its provenance "
"instead of requiring the frozen pinned source")
a = p.parse_args()
if a.steps <= 0 or a.steps > 3200:
p.error("steps must be 1..3200")
@@ -653,8 +595,7 @@ def main():
)
plan = dict(
case=c,
required_revision=None if a.current_checkout else PIN,
source_mode="current_checkout" if a.current_checkout else "pinned",
required_revision=PIN,
required_free_device_mib=1024,
estimated_peak_device_mib=128 * a.scale**2,
estimated_output_mib=64 if a.continuation else 2,
@@ -670,9 +611,10 @@ def main():
):
p.error("Root must select one verified idle GPU via CUDA_VISIBLE_DEVICES")
repo = a.repo.resolve()
from source_provenance import reject_output_inside
reject_output_inside(a.output, repo)
provenance = source_identity(repo, a.current_checkout)
source = verify(repo)
provenance = dict(
source_experiment=source["experiment"], **checkout_provenance(repo)
)
sys.path.insert(0, str(repo / "src"))
import CelerisLab
@@ -688,9 +630,7 @@ def main():
free, total = cuda.mem_get_info()
if free < max(1024, 2 * plan["estimated_peak_device_mib"]) * 1024**2:
raise RuntimeError("Insufficient free device memory")
payload = run_case(c, ctx, a.output, provenance, continuation=a.continuation)
finish_receipt(a.output / f"re{c['re_d']}_s{c['scale']}.json", payload, provenance,
lambda: source_identity(repo, a.current_checkout))
run_case(c, ctx, a.output, provenance, continuation=a.continuation)
print(
json.dumps(
dict(status="COMPLETE_DIAGNOSTIC_ONLY", physics_validation_pass=False)
-96
View File
@@ -13,99 +13,3 @@ def checkout_provenance(repo):
text=True,
)
return dict(checkout_revision=revision, checkout_dirty=bool(dirty))
def final_source_check(recorded, recheck):
"""Re-derive the source identity after a run and compare it with the recorded one.
Returns receipt fields: source_recheck "UNCHANGED"; "CHANGED" with source_after; or
"RECHECK_FAILED" (source_after None) with the error. Callers mark a receipt complete
only for "UNCHANGED"; anything else invalidates the run.
"""
try:
after = recheck()
except Exception as error: # a failed recheck cannot confirm the source: invalid too
return dict(source_recheck="RECHECK_FAILED", source_after=None,
source_recheck_error=f"{type(error).__name__}: {error}")
if after == recorded:
return dict(source_recheck="UNCHANGED")
return dict(source_recheck="CHANGED", source_after=after, source_recheck_error=None)
def reject_output_inside(output, repo):
"""Refuse run outputs inside the checkout before any work: they would dirty it."""
from pathlib import Path
output, repo = Path(output).resolve(), Path(repo).resolve()
if output == repo or repo in output.parents:
raise SystemExit(f"--output must be outside the source checkout {repo}")
class HistoryUnavailable(RuntimeError):
"""The clone cannot show the recorded commit (no git, shallow clone, installed copy)."""
def historical_sha256(repo, commit, paths):
"""SHA-256 of each path as stored in `commit`, read with git, never from the working tree.
Raises HistoryUnavailable when git or the commit object is missing, so callers can
report an unchecked integrity record explicitly instead of passing or failing silently.
"""
import hashlib
git = ["git", "-C", str(repo)]
try:
subprocess.run([*git, "cat-file", "-e", f"{commit}^{{commit}}"], check=True,
capture_output=True)
except (OSError, subprocess.CalledProcessError) as error:
raise HistoryUnavailable(f"commit {commit} is not available in {repo}") from error
digests = {}
for path in paths:
listed = subprocess.run([*git, "ls-tree", "--name-only", commit, "--", path],
capture_output=True, text=True)
if listed.returncode:
raise HistoryUnavailable(f"tree of {commit} is not available in {repo}")
if listed.stdout.strip() != path:
raise ValueError(f"{path} is not part of {commit}")
shown = subprocess.run([*git, "show", f"{commit}:{path}"], capture_output=True)
if shown.returncode: # e.g. a blobless partial clone without network access
raise HistoryUnavailable(f"content of {path} at {commit} is not available in {repo}")
digests[path] = hashlib.sha256(shown.stdout).hexdigest()
return digests
def verify_manifest(repo, commit, digests):
"""Raise RuntimeError naming the first path whose recorded digest differs at `commit`."""
actual = historical_sha256(repo, commit, digests)
for path, digest in digests.items():
if actual[path] != digest:
raise RuntimeError(f"Historical manifest mismatch at {commit}: {path}")
return actual
def historical_tree(repo, commit, prefix="src"):
"""Tracked file names under `prefix` at `commit` (HistoryUnavailable if missing)."""
historical_sha256(repo, commit, [])
return subprocess.check_output(["git", "-C", str(repo), "ls-tree", "-r", "--name-only",
commit, "--", prefix], text=True).splitlines()
def historical_files(repo, commit, paths, dest):
"""Write each path as stored in `commit` under `dest` (HistoryUnavailable if missing)."""
from pathlib import Path
historical_sha256(repo, commit, paths) # HistoryUnavailable / ValueError before writing
for path in paths:
shown = subprocess.run(["git", "-C", str(repo), "show", f"{commit}:{path}"],
capture_output=True)
if shown.returncode:
raise HistoryUnavailable(f"cannot read {commit}:{path} (blob not in this clone)")
target = Path(dest) / path
target.parent.mkdir(parents=True, exist_ok=True)
target.write_bytes(shown.stdout)
def history_unavailable(testcase, error):
"""Skip with the reason, or fail when CELERIS_REQUIRE_HISTORY=1 (release gating)."""
import os
message = f"historical integrity NOT checked: {error}"
if os.environ.get("CELERIS_REQUIRE_HISTORY") == "1":
raise AssertionError(message)
testcase.skipTest(message)
@@ -73,31 +73,7 @@ def valid_builds(record, n, depth):
return True
# run_case.py (sha256) of the runners before the final source recheck (git: 9805383,
# 2cf9853 = the a135252 qualification runner, db3d098); they wrote schema 1 receipts.
LEGACY_RUN_CASE_SHA256 = {
"9805383": "fbceab8c1aaaf357195a4fb9e6a699a195326e480e274692a277ce3df54ee6d9",
"2cf9853": "0789d2e1ffa581078bd11e6251b660b2c268363171fda0c6b26ac41205f1de3a",
"db3d098": "618c14ba94584aa9c0b714e0e2064a746d2b882973a13424f09dd7e54c68f7f1",
}
def rechecked(r, current):
"""Schema 2 receipts need source_recheck UNCHANGED; a missing recheck is accepted only
for frozen-mode schema 1 receipts written by a verified legacy runner."""
if "invalid" in r:
return False
if r.get("schema") == 2:
return r.get("source_recheck") == "UNCHANGED"
harness = r.get("source", {}).get("harness_sha256") or {}
return (not current and r.get("schema") == 1 and "source_recheck" not in r
and harness.get("run_case.py") in LEGACY_RUN_CASE_SHA256.values())
def assess(records, expected_revision=None, current=False):
"""current=True assesses current-checkout receipts (run_case --current-checkout) with
the same gates; it never accepts them as the frozen a135252 qualification, and the
frozen mode never accepts current-checkout receipts."""
def assess(records, expected_revision=None):
indexed, duplicates = {}, []
for record in records:
key = tuple(record.get(k) for k in ("re_L", "n", "depth", "explosion"))
@@ -111,19 +87,11 @@ def assess(records, expected_revision=None, current=False):
sources = [r.get("source", {}) for r in records]
checks["same_source_and_harness"] = bool(sources) and len({
json.dumps(s, sort_keys=True) for s in sources}) == 1
if current:
checks["clean_current_checkout_source"] = bool(sources) and expected_revision is not None and all(
s.get("dirty") is False and s.get("source_mode") == "current_checkout"
and bool(s.get("source_tree_sha256")) and bool(s.get("harness_sha256"))
and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
and s.get("revision") == expected_revision for s in sources)
else:
checks["clean_a135252_numerical_source"] = bool(sources) and all(
s.get("dirty") is False and s.get("source_tree_sha256") == SOURCE_SHA256
and "source_mode" not in s
and bool(s.get("revision")) and bool(s.get("harness_sha256"))
and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
and (expected_revision is None or s.get("revision") == expected_revision) for s in sources)
checks["clean_a135252_numerical_source"] = bool(sources) and all(
s.get("dirty") is False and s.get("source_tree_sha256") == SOURCE_SHA256
and bool(s.get("revision")) and bool(s.get("harness_sha256"))
and s.get("build_capture_sha256") == BUILD_CAPTURE_SHA256
and (expected_revision is None or s.get("revision") == expected_revision) for s in sources)
runtimes = [r.get("runtime_identity") for r in records]
checks["same_recorded_runtime"] = bool(runtimes) and all(
isinstance(v, dict) and all(v.get(k) for k in
@@ -157,15 +125,8 @@ def assess(records, expected_revision=None, current=False):
and r.get("mach") == U * math.sqrt(3)
)
checks[label + " actual_builds"] = valid_builds(r, n, depth)
if current:
checks[label + " reference"] = (r.get("numerical_revision") is None
and r.get("source_mode") == "current_checkout"
and r.get("reference_harness") == REFERENCE_HARNESS)
else:
checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION
and "source_mode" not in r
and r.get("reference_harness") == REFERENCE_HARNESS)
checks[label + " source_recheck"] = rechecked(r, current)
checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION
and r.get("reference_harness") == REFERENCE_HARNESS)
checks[label + " numerics"] = r.get("numerics") == NUMERICS and r.get("initialization") == INITIALIZATION
runtime = r.get("runtime_source", {})
package = Path(runtime.get("checkout", "")) / "src/CelerisLab"
@@ -224,10 +185,7 @@ def assess(records, expected_revision=None, current=False):
valid_case_and_controls=valid, baseline_passed=baseline_passed)
return dict(passed=all(checks.values()), missing=missing, duplicates=duplicates, checks=checks,
comparisons=comparisons, scheme_results=scheme_results, early_pressure=early,
source_mode="current_checkout" if current else "frozen",
numerical_revision=None if current else NUMERICAL_REVISION,
source_tree_sha256=(sources[0].get("source_tree_sha256") if current and sources
else None if current else SOURCE_SHA256),
numerical_revision=NUMERICAL_REVISION, source_tree_sha256=SOURCE_SHA256,
reference_harness=REFERENCE_HARNESS, expected_revision=expected_revision,
floors=FLOORS, baseline_factor=BASELINE_FACTOR,
main_statistic="maximum of four post-initial samples at t*=.12,.24,.36,.48",
@@ -241,13 +199,10 @@ if __name__ == "__main__":
p.add_argument("files", type=Path, nargs="+")
p.add_argument("--output", type=Path, required=True)
p.add_argument("--expected-revision", required=True)
p.add_argument("--current-checkout", action="store_true",
help="assess current-checkout receipts; never the frozen qualification")
args = p.parse_args()
if args.output.exists():
p.error("Output already exists; choose a new path to retain prior evidence")
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision,
current=args.current_checkout)
result = assess([json.loads(path.read_text()) for path in args.files], args.expected_revision)
args.output.write_text(json.dumps(result, indent=2, allow_nan=False) + "\n")
print(json.dumps(result, indent=2, allow_nan=False))
raise SystemExit(0 if result["passed"] else 1)
+5 -40
View File
@@ -64,16 +64,8 @@ def provenance(repo):
)
CURRENT_CAPTURE = "current checkout, not the frozen a135252 qualification"
def verify_source(repo, revision, current=False):
"""Exact clean harness commit, fixed production tree, and reviewed capture helper.
current=True records the clean checkout itself instead of requiring the frozen
numerical tree; such receipts carry source_mode="current_checkout" and are only
assessed with assess(..., current=True), never as the frozen qualification.
"""
def verify_source(repo, revision):
"""Exact clean harness commit, fixed production tree, and reviewed capture helper."""
import oracle
from tests.validation.multigrid import check_projected_pairs as gate
@@ -85,17 +77,6 @@ def verify_source(repo, revision, current=False):
if (Path(gate.__file__).resolve() != helper
or hashlib.sha256(helper.read_bytes()).hexdigest() != BUILD_CAPTURE_SHA256):
raise RuntimeError("Wrong build-capture helper")
if current:
result = provenance(repo)
if result["revision"] != revision or result["dirty"]:
raise ValueError("requires the exact expected clean checkout")
result.update(source_mode="current_checkout", source_capture=CURRENT_CAPTURE,
build_capture_sha256=BUILD_CAPTURE_SHA256)
return result
if subprocess.run(["git", "-C", str(repo), "cat-file", "-e", f"{NUMERICAL_REVISION}^{{commit}}"],
capture_output=True).returncode:
raise RuntimeError(f"frozen numerical revision {NUMERICAL_REVISION} is not available "
"in this clone; the frozen qualification cannot be checked")
if gate.preflight(repo, revision, "candidate") != NUMERICAL_REVISION:
raise RuntimeError("Wrong production revision")
result = provenance(repo)
@@ -168,8 +149,6 @@ def main():
p.add_argument("--depth", type=int, choices=[1, 2, 3], required=True)
p.add_argument("--output", type=Path, required=True)
p.add_argument("--explosion", choices=["constant", "projected_linear"], default="constant")
p.add_argument("--current-checkout", action="store_true",
help="record the clean current checkout instead of the frozen a135252 tree")
p.add_argument("--early-pressure", action="store_true",
help="also save 65 scalar pressure samples at t*=k*.001875, k=0..64")
args = p.parse_args()
@@ -182,9 +161,7 @@ def main():
p.error("--run-gpu and exactly one explicit CUDA_VISIBLE_DEVICES are required")
repo = args.repo.resolve()
sys.path[:0] = [str(repo / "src"), str(repo)]
from tests.validation.source_provenance import reject_output_inside
reject_output_inside(args.output, repo)
source = verify_source(repo, args.expected_revision, current=args.current_checkout)
source = verify_source(repo, args.expected_revision)
runtime = runtime_source(repo)
if args.dry_run:
print(json.dumps(dict(source=source, runtime_source=runtime, dry_run=True)))
@@ -203,10 +180,9 @@ def main():
masks = [k == ACTIVE for k in h.kinds]
main_steps, early_steps = sample_steps(args.n, args.early_pressure)
report = dict(
schema=2, # 2: final source recheck (source_recheck) is recorded
schema=1,
source=source,
numerical_revision=None if args.current_checkout else NUMERICAL_REVISION,
**({"source_mode": "current_checkout"} if args.current_checkout else {}),
numerical_revision=NUMERICAL_REVISION,
reference_harness=REFERENCE_HARNESS,
actual_build_sidecars=[],
runtime_identity=None,
@@ -321,17 +297,6 @@ def main():
),
flush=True,
)
from tests.validation.source_provenance import final_source_check
recheck = final_source_check(
dict(source=source, runtime_source=runtime),
lambda: dict(source=verify_source(repo, args.expected_revision,
current=args.current_checkout),
runtime_source=runtime_source(repo)))
report.update(recheck)
if recheck["source_recheck"] != "UNCHANGED":
report.update(invalid="SOURCE_CHANGED_DURING_RUN") # completed stays False
save()
raise RuntimeError("Source changed during the run; receipt marked invalid")
report.update(
completed=True,
seconds=time.monotonic() - started,
@@ -12,13 +12,13 @@ import sys
import tempfile
from types import SimpleNamespace
import unittest
from unittest.mock import call, patch
from unittest.mock import patch
import numpy as np
import run_case
from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
LEGACY_RUN_CASE_SHA256, SOURCE_SHA256, assess)
SOURCE_SHA256, assess)
from oracle import LIMITS, TIMES, U, rectangles
REPO = Path(__file__).resolve().parents[3]
@@ -31,7 +31,7 @@ RUNTIME = dict(device=dict(name="host mock", pci_bus_id="mock", capability=[7, 0
def source():
return dict(revision=MOCK_REVISION, dirty=False, source_tree_sha256=SOURCE_SHA256,
harness_sha256={"run_case.py": LEGACY_RUN_CASE_SHA256["2cf9853"]},
harness_sha256={"run_case.py": "a" * 64},
build_capture_sha256=run_case.BUILD_CAPTURE_SHA256)
@@ -53,7 +53,7 @@ def records(early=False):
h = build_hierarchy(cfg, refinements)
main_steps, early_steps = run_case.sample_steps(n, early)
out.append(dict(
schema=1, re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
viscosity=cfg.viscosity, tau_by_level=[0.5+3*c.viscosity for c in h.configs],
refinements=refinements, numerics=NUMERICS.copy(), initialization=run_case.INITIALIZATION,
source=source(), numerical_revision=NUMERICAL_REVISION,
@@ -291,105 +291,6 @@ class ProjectedPairsTests(unittest.TestCase):
else:
self.assertEqual(run_case.verify_source(REPO, expected), source())
def test_frozen_mode_reports_missing_history_not_a_diff(self):
from tests.validation.multigrid import check_projected_pairs as gate
missing = SimpleNamespace(returncode=128)
with patch.object(run_case.subprocess, "run", return_value=missing), \
patch.object(gate, "preflight") as preflight:
with self.assertRaisesRegex(RuntimeError, "not available in this clone"):
run_case.verify_source(REPO, MOCK_REVISION)
preflight.assert_not_called()
def test_current_checkout_source_is_recorded_not_frozen(self):
from tests.validation.multigrid import check_projected_pairs as gate
current = dict(source(), source_tree_sha256="c" * 64)
with patch.object(run_case, "provenance", return_value=dict(current)), \
patch.object(gate, "preflight") as preflight:
result = run_case.verify_source(REPO, MOCK_REVISION, current=True)
preflight.assert_not_called()
self.assertEqual(result["source_mode"], "current_checkout")
self.assertEqual(result["source_capture"], run_case.CURRENT_CAPTURE)
for bad in (dict(current, dirty=True), dict(current, revision=OLD_REVISION)):
with self.subTest(bad=bad), patch.object(run_case, "provenance", return_value=bad):
with self.assertRaisesRegex(ValueError, "exact expected clean checkout"):
run_case.verify_source(REPO, MOCK_REVISION, current=True)
def test_current_and_frozen_receipts_never_cross_assess(self):
frozen = records()
current = deepcopy(frozen)
for r in current:
r["source"] = dict(r["source"], source_tree_sha256="c" * 64,
source_mode="current_checkout",
source_capture=run_case.CURRENT_CAPTURE)
r["numerical_revision"], r["source_mode"] = None, "current_checkout"
r["schema"], r["source_recheck"] = 2, "UNCHANGED"
ok = assess(current, MOCK_REVISION, current=True)
self.assertTrue(ok["checks"]["clean_current_checkout_source"])
self.assertTrue(all(v for k, v in ok["checks"].items() if k.endswith(" reference")))
self.assertEqual((ok["source_mode"], ok["numerical_revision"]), ("current_checkout", None))
self.assertEqual(ok["source_tree_sha256"], "c" * 64)
self.assertFalse(assess(current, MOCK_REVISION)["checks"]["clean_a135252_numerical_source"])
self.assertFalse(assess(current, MOCK_REVISION)["passed"])
self.assertFalse(assess(frozen, MOCK_REVISION, current=True)["checks"]["clean_current_checkout_source"])
self.assertFalse(assess(current, None, current=True)["checks"]["clean_current_checkout_source"])
self.assertTrue(assess(frozen, MOCK_REVISION)["checks"]["clean_a135252_numerical_source"])
def test_assessor_requires_unchanged_final_source(self):
frozen = records() # schema 1 receipts from the verified legacy runner 2cf9853
label = lambda r: f"{r['n']}/{r['depth']}/{r['explosion']} source_recheck"
self.assertTrue(all(assess(frozen, MOCK_REVISION)["checks"][label(r)] for r in frozen))
new = deepcopy(frozen)
for r in new:
r.update(schema=2, source_recheck="UNCHANGED")
self.assertTrue(all(assess(new, MOCK_REVISION)["checks"][label(r)] for r in new))
unknown_runner = dict(frozen[0]["source"], harness_sha256={"run_case.py": "a" * 64})
for case, mutate in (
("schema 1, unverified runner", dict(source=unknown_runner)),
("schema 2 without recheck", dict(schema=2)),
("schema 2 changed", dict(schema=2, source_recheck="CHANGED")),
("schema 2 recheck failed", dict(schema=2, source_recheck="RECHECK_FAILED")),
("schema 1 with a recheck field", dict(source_recheck="UNCHANGED")),
("unknown schema", dict(schema=3, source_recheck="UNCHANGED")),
("no schema", dict(schema=None)),
("invalid", dict(schema=2, source_recheck="UNCHANGED",
invalid="SOURCE_CHANGED_DURING_RUN")),
):
with self.subTest(case):
bad = deepcopy(frozen)
bad[0].update(mutate)
out = assess(bad, MOCK_REVISION)
self.assertFalse(out["checks"][label(bad[0])])
self.assertFalse(out["passed"])
current = deepcopy(frozen)
for r in current:
r["source"] = dict(r["source"], source_mode="current_checkout")
r["numerical_revision"], r["source_mode"] = None, "current_checkout"
# Legacy acceptance is frozen-mode only.
self.assertFalse(assess(current, MOCK_REVISION, current=True)["checks"][label(current[0])])
def test_legacy_runner_hashes_are_the_recorded_commits(self):
sys.path.insert(0, str(REPO))
from tests.validation.source_provenance import (HistoryUnavailable, historical_sha256,
history_unavailable)
try:
for commit, digest in LEGACY_RUN_CASE_SHA256.items():
path = "tests/validation/taylor_green/run_case.py"
self.assertEqual(historical_sha256(REPO, commit, [path])[path], digest)
except HistoryUnavailable as error:
history_unavailable(self, error)
current = hashlib.sha256((REPO / "tests/validation/taylor_green/run_case.py").read_bytes())
self.assertNotIn(current.hexdigest(), LEGACY_RUN_CASE_SHA256.values())
def test_output_inside_checkout_is_refused_before_source_check(self):
argv = ["run_case.py", "--repo", str(REPO), "--re", "100", "--n", "128", "--depth", "1",
"--output", str(REPO / "tg-out.json"), "--expected-revision", MOCK_REVISION,
"--dry-run"]
with patch.object(sys, "argv", argv), \
patch.object(run_case, "verify_source", side_effect=AssertionError("work started")):
with self.assertRaisesRegex(SystemExit, "outside the source checkout"):
run_case.main()
self.assertFalse((REPO / "tg-out.json").exists())
def test_explicit_opt_in_selector_and_dry_run_without_cuda(self):
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / "case.json"
@@ -418,7 +319,7 @@ class ProjectedPairsTests(unittest.TestCase):
patch.object(run_case, "runtime_identity", side_effect=AssertionError("CUDA reached")), \
redirect_stdout(stdout):
self.assertEqual(run_case.main(), 0)
verify.assert_called_once_with(REPO, MOCK_REVISION, current=False)
verify.assert_called_once_with(REPO, MOCK_REVISION)
self.assertTrue(json.loads(stdout.getvalue())["dry_run"])
self.assertFalse(output.exists())
@@ -450,8 +351,7 @@ class ProjectedPairsTests(unittest.TestCase):
# Replace the GPU context and steppers before import. No CUDA module or flow is used.
for early, depth, fail in ((False, 1, None), (False, 2, None), (True, 2, None),
(True, 2, "resource"), (True, 2, "nonfinite"),
(True, 2, "probe_nonfinite"),
(False, 2, "changed"), (False, 1, "recheck_error")):
(True, 2, "probe_nonfinite")):
calls, selectors = [], []
class FakeRun:
@@ -506,31 +406,13 @@ class ProjectedPairsTests(unittest.TestCase):
patch.object(run_case, "runtime_identity", return_value=deepcopy(RUNTIME)), \
patch.object(run_case, "build_capture", return_value=nullcontext()), \
patch.object(run_case, "metrics", side_effect=metric), redirect_stdout(io.StringIO()):
if fail in ("changed", "recheck_error"):
verify.side_effect = [source(), dict(source(), source_tree_sha256="d" * 64)
if fail == "changed" else RuntimeError("Wrong numerical source fingerprint")]
with self.assertRaisesRegex(RuntimeError, "Source changed during the run"):
run_case.main()
elif fail:
if fail:
with self.assertRaises(MemoryError if fail == "resource" else FloatingPointError):
run_case.main()
else:
self.assertEqual(run_case.main(), 0)
receipt = json.loads(output.read_text())
rechecked = fail in (None, "changed", "recheck_error")
self.assertEqual(verify.call_args_list,
[call(REPO, MOCK_REVISION, current=False)] * (2 if rechecked else 1))
if fail in ("changed", "recheck_error"):
# Full trajectory, but invalid: never a complete-looking receipt.
self.assertEqual(receipt["source_recheck"],
"CHANGED" if fail == "changed" else "RECHECK_FAILED")
self.assertEqual(receipt["invalid"], "SOURCE_CHANGED_DURING_RUN")
self.assertFalse(receipt["completed"])
self.assertNotIn("screening_pass", receipt)
self.assertEqual(receipt["source_after"] is None, fail == "recheck_error")
continue
self.assertEqual(receipt.get("source_recheck"), None if fail else "UNCHANGED")
self.assertEqual(receipt["schema"], 2)
verify.assert_called_once_with(REPO, MOCK_REVISION)
self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else [])
self.assertEqual(receipt["source"], source())
self.assertEqual(receipt["runtime_identity"], RUNTIME)