From df0f586331205fcbea15134cd49ddd1792366465 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 20 Oct 2026 18:44:23 +0000 Subject: [PATCH] test: final source recheck for V1 jet CV and V3 Taylor-Green runs Both runners now re-derive the recorded source identity after the run (source_provenance.final_source_check) and only then mark the receipt complete. A changed source, or a recheck that fails, rewrites the receipt as invalid and raises: - V1: in-run receipts carry run_status=INCOMPLETE; finish_receipt sets COMPLETE or INVALID_SOURCE_CHANGED (+ source_recheck/source_after). - V3: completed stays False, invalid=SOURCE_CHANGED_DURING_RUN, no screening_pass; the assessor requires source_recheck UNCHANGED (frozen mode still accepts receipts from before the field existed). Interrupted runs keep INCOMPLETE / completed=False. Host tests cover unchanged, changed and failed rechecks and incomplete runs. No threshold or solver change. Co-Authored-By: Claude Opus 5.5 --- tests/unit/test_jet_circle_cv.py | 35 ++++++++++++++ .../unit/test_validation_source_provenance.py | 17 +++++++ tests/validation/run_jet_circle_cv.py | 47 ++++++++++++++----- tests/validation/source_provenance.py | 17 +++++++ .../taylor_green/assess_projected_pairs.py | 4 ++ tests/validation/taylor_green/run_case.py | 11 +++++ .../taylor_green/test_projected_pairs.py | 43 +++++++++++++++-- 7 files changed, 159 insertions(+), 15 deletions(-) diff --git a/tests/unit/test_jet_circle_cv.py b/tests/unit/test_jet_circle_cv.py index 2991360..57b78aa 100644 --- a/tests/unit/test_jet_circle_cv.py +++ b/tests/unit/test_jet_circle_cv.py @@ -196,5 +196,40 @@ class Host(unittest.TestCase): 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, after in ((lambda: dict(recorded, checkout_dirty=True), True), + (lambda: v.verify(Path(d)), None)): + 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", "CHANGED")) + self.assertEqual(bad["source_after"] is None, after is None) + 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)) + + if __name__ == "__main__": unittest.main() diff --git a/tests/unit/test_validation_source_provenance.py b/tests/unit/test_validation_source_provenance.py index df26033..54c3f4a 100644 --- a/tests/unit/test_validation_source_provenance.py +++ b/tests/unit/test_validation_source_provenance.py @@ -111,3 +111,20 @@ def test_historical_files_reports_unreadable_blob_as_history_unavailable(tmp_pat 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"]) == ("CHANGED", None) + assert "Source hash mismatch" in failed["source_recheck_error"] diff --git a/tests/validation/run_jet_circle_cv.py b/tests/validation/run_jet_circle_cv.py index 448e3c1..6cc8bf3 100644 --- a/tests/validation/run_jet_circle_cv.py +++ b/tests/validation/run_jet_circle_cv.py @@ -244,6 +244,37 @@ def current_checkout(repo): 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 @@ -556,6 +587,7 @@ def run_case(c, ctx, output, provenance, continuation=False): case=c, **{"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", @@ -638,16 +670,7 @@ 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) - ) + provenance = source_identity(repo, a.current_checkout) sys.path.insert(0, str(repo / "src")) import CelerisLab @@ -663,7 +686,9 @@ 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") - run_case(c, ctx, a.output, provenance, continuation=a.continuation) + 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)) print( json.dumps( dict(status="COMPLETE_DIAGNOSTIC_ONLY", physics_validation_pass=False) diff --git a/tests/validation/source_provenance.py b/tests/validation/source_provenance.py index d3dd07f..807e8ae 100644 --- a/tests/validation/source_provenance.py +++ b/tests/validation/source_provenance.py @@ -15,6 +15,23 @@ def checkout_provenance(repo): 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", or "CHANGED" with source_after + (None if the recheck itself failed) and the recheck error. Callers mark a receipt + complete only for "UNCHANGED"; a changed source invalidates the run. + """ + try: + after = recheck() + except Exception as error: # a failed recheck cannot confirm the source + return dict(source_recheck="CHANGED", 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) + + class HistoryUnavailable(RuntimeError): """The clone cannot show the recorded commit (no git, shallow clone, installed copy).""" diff --git a/tests/validation/taylor_green/assess_projected_pairs.py b/tests/validation/taylor_green/assess_projected_pairs.py index 42bd69b..dc3db00 100644 --- a/tests/validation/taylor_green/assess_projected_pairs.py +++ b/tests/validation/taylor_green/assess_projected_pairs.py @@ -144,6 +144,10 @@ def assess(records, expected_revision=None, current=False): checks[label + " reference"] = (r.get("numerical_revision") == NUMERICAL_REVISION and "source_mode" not in r and r.get("reference_harness") == REFERENCE_HARNESS) + # Final source recheck (run_case); receipts from before it existed carry no field + # and are accepted only in the frozen mode. + recheck = r.get("source_recheck") if current else r.get("source_recheck", "UNCHANGED") + checks[label + " source_recheck"] = recheck == "UNCHANGED" and "invalid" not in r checks[label + " numerics"] = r.get("numerics") == NUMERICS and r.get("initialization") == INITIALIZATION runtime = r.get("runtime_source", {}) package = Path(runtime.get("checkout", "")) / "src/CelerisLab" diff --git a/tests/validation/taylor_green/run_case.py b/tests/validation/taylor_green/run_case.py index f32b3aa..ea903e1 100644 --- a/tests/validation/taylor_green/run_case.py +++ b/tests/validation/taylor_green/run_case.py @@ -319,6 +319,17 @@ 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, diff --git a/tests/validation/taylor_green/test_projected_pairs.py b/tests/validation/taylor_green/test_projected_pairs.py index d944133..60dd4ff 100644 --- a/tests/validation/taylor_green/test_projected_pairs.py +++ b/tests/validation/taylor_green/test_projected_pairs.py @@ -12,7 +12,7 @@ import sys import tempfile from types import SimpleNamespace import unittest -from unittest.mock import patch +from unittest.mock import call, patch import numpy as np @@ -322,6 +322,7 @@ class ProjectedPairsTests(unittest.TestCase): source_mode="current_checkout", source_capture=run_case.CURRENT_CAPTURE) r["numerical_revision"], r["source_mode"] = None, "current_checkout" + r["source_recheck"] = "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"))) @@ -333,6 +334,23 @@ class ProjectedPairsTests(unittest.TestCase): 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() + 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)) + for mutate in (dict(source_recheck="CHANGED"), dict(invalid="SOURCE_CHANGED_DURING_RUN")): + 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" + # Current-checkout receipts are new: a missing recheck is not accepted. + self.assertFalse(assess(current, MOCK_REVISION, current=True)["checks"][label(current[0])]) + def test_explicit_opt_in_selector_and_dry_run_without_cuda(self): with tempfile.TemporaryDirectory() as tmp: output = Path(tmp) / "case.json" @@ -393,7 +411,8 @@ 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")): + (True, 2, "probe_nonfinite"), + (False, 2, "changed"), (False, 1, "recheck_error")): calls, selectors = [], [] class FakeRun: @@ -448,13 +467,29 @@ 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: + 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: 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()) - verify.assert_called_once_with(REPO, MOCK_REVISION, current=False) + 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") + 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(selectors, ["projected_linear" if early else "constant"] if depth == 2 else []) self.assertEqual(receipt["source"], source()) self.assertEqual(receipt["runtime_identity"], RUNTIME)