Compare commits

..
Author SHA1 Message Date
ClaudeandClaude Opus 5.5 a454f02355 Jets: compute jet link shapes once per boundary preparation
surface_link_velocity already runs jet_link_shapes; CurvedLinkSoA.
assign_surface and _prepare_blocked_boundary then ran it again with the
same jets and hits to fill the host jet cache. A private _surface_links
returns (velocity, aperture, shape, jet_id) from the one call; both
callers use it. surface_link_velocity keeps its two-value public
contract (now a thin wrapper). Same values, same validation order, same
cache contents; host geometry preparation only.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2026-10-21 05:57:11 +00:00
17 changed files with 93 additions and 567 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`.
+14 -3
View File
@@ -207,19 +207,30 @@ def surface_link_velocity(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
jets=(), hits=None, *, _shifted_blocked=False):
"""Per-link prescribed surface velocity (n, 2) FP32 and aperture mask.
See _surface_links, which also returns the unit-amplitude jet shapes.
"""
return _surface_links(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry, jets, hits,
_shifted_blocked=_shifted_blocked)[:2]
def _surface_links(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
jets=(), hits=None, *, _shifted_blocked=False):
"""(velocity, aperture, jet shape, jet id); the last two are None without jets.
Host-only and storage-neutral: rows follow the given physical cut-link
order. Returns (None, None) when neither an object carries surface
order. Returns all None when neither an object carries surface
actuation nor a jet is given. Jets need true FP64 ``hits`` (n, 2) and may
not share a link with each other or with a legacy aperture.
Shifted storage is privately enabled only by the all-blocked stepper.
"""
enabled = [obj for obj in objects if obj.surface_actuation is not None]
if not enabled and not jets:
return None, None
return None, None, None, None
from .preprocess.flag_overlay import validate_cut_link_ownership
validate_cut_link_ownership(flags, fluid_idx, dir, cfg.nq)
surface_velocity = np.zeros((len(dir), 2), np.float32)
surface_aperture = np.zeros(len(dir), bool)
shape = jet_id = None
for obj in enabled:
obj.validate_surface_actuation(cfg)
selected = body_id == obj.obj_id
@@ -252,4 +263,4 @@ def surface_link_velocity(objects, cfg, flags, fluid_idx, dir, body_id, rx, ry,
raise ValueError("Jet overlaps a SurfaceActuation aperture")
surface_velocity[driven] = jet_link_velocity(jets, shape, jet_id)[driven]
surface_aperture |= driven
return surface_velocity, surface_aperture
return surface_velocity, surface_aperture, shape, jet_id
+2 -4
View File
@@ -94,12 +94,10 @@ class CurvedLinkSoA:
"""
self.surface_velocity = self.aperture = None
self.jets, self.jet_shape, self.jet_id = tuple(jets), None, None
self.surface_velocity, self.aperture = surface_link_velocity(
from ..body.jets import _surface_links
self.surface_velocity, self.aperture, self.jet_shape, self.jet_id = _surface_links(
objects, cfg, flags, self.fluid_idx, self.dir, self.body_id, self.rx, self.ry,
jets=self.jets, hits=hits)
if self.jets:
from ..body.jets import jet_link_shapes
self.jet_shape, self.jet_id = jet_link_shapes(self.jets, hits)
def set_jet_amplitude(self, index, amplitude, stream=None):
"""Rewrite jet rows from cached shapes; re-upload only the velocity column.
+6 -4
View File
@@ -8,7 +8,7 @@ from dataclasses import dataclass, replace
import numpy as np
from .blocks import blocked_link_tables
from ..body.jets import jet_link_shapes, surface_link_velocity
from ..body.jets import _surface_links
from .descriptors import D2Q9_EX, D2Q9_EY
from .refinement import ACTIVE, OUTER_GHOST
@@ -142,11 +142,13 @@ def _prepare_blocked_boundary(boundary, layout, cfg, jets, *, _shifted_blocked=F
tables.update(body_id=np.asarray(body_id, np.int32), rx=np.asarray(rx, np.float32),
ry=np.asarray(ry, np.float32), aperture=np.zeros(len(dirs), bool))
actuated = bool(jets) or any(o.surface_actuation is not None for o in boundary.objects)
jet_cache = None
if actuated:
cfg = replace(cfg)
cfg.validate()
tables["surface_velocity"], tables["aperture"] = surface_link_velocity(
tables["surface_velocity"], tables["aperture"], shape, jet_id = _surface_links(
boundary.objects, cfg, boundary.flags, fluid_idx, dirs, body_id, rx, ry,
jets=jets, hits=boundary.curved_hits, _shifted_blocked=_shifted_blocked)
return _BlockedBoundary(boundary, tables, jets, actuated,
jet_link_shapes(jets, boundary.curved_hits) if jets else None)
if jets:
jet_cache = (shape, jet_id)
return _BlockedBoundary(boundary, tables, jets, actuated, jet_cache)
@@ -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 -28
View File
@@ -166,34 +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"])
verify(REPO)
if __name__ == "__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")
+10 -49
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,36 +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()
+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__":
+7 -40
View File
@@ -3,7 +3,6 @@
import argparse
import hashlib
import subprocess
import json
import os
from pathlib import Path
@@ -221,29 +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 run_case(c, ctx, output, provenance, continuation=False):
from CelerisLab.body.geometry.circle import CircleGeometry
from CelerisLab.body.objects import SimObject
@@ -554,7 +530,8 @@ 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",
windows=rows,
physics_validation_pass=False,
@@ -594,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")
@@ -621,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,
@@ -638,16 +611,10 @@ def main():
):
p.error("Root must select one verified idle GPU via CUDA_VISIBLE_DEVICES")
repo = a.repo.resolve()
if a.current_checkout:
provenance = dict(**checkout_provenance(repo), **current_checkout(repo))
if provenance["checkout_dirty"]:
raise RuntimeError("Current-checkout runs require a clean checkout")
else:
source = verify(repo)
provenance = dict(
source_experiment=source["experiment"], source_mode="pinned",
**checkout_provenance(repo)
)
source = verify(repo)
provenance = dict(
source_experiment=source["experiment"], **checkout_provenance(repo)
)
sys.path.insert(0, str(repo / "src"))
import CelerisLab
-71
View File
@@ -13,74 +13,3 @@ def checkout_provenance(repo):
text=True,
)
return dict(checkout_revision=revision, checkout_dirty=bool(dirty))
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,10 +73,7 @@ def valid_builds(record, n, depth):
return True
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"))
@@ -90,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
@@ -136,14 +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 + " 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"
@@ -202,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",
@@ -219,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)
+4 -26
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,7 +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)]
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,8 +182,7 @@ def main():
report = dict(
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,
@@ -291,48 +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"
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_explicit_opt_in_selector_and_dry_run_without_cuda(self):
with tempfile.TemporaryDirectory() as tmp:
output = Path(tmp) / "case.json"
@@ -361,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())
@@ -454,7 +412,7 @@ class ProjectedPairsTests(unittest.TestCase):
else:
self.assertEqual(run_case.main(), 0)
receipt = json.loads(output.read_text())
verify.assert_called_once_with(REPO, MOCK_REVISION, current=False)
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)