test: gate V1 side files and V3 legacy receipts; early output rejection

- analyze_jet_cv_continuation: samples are analyzed only if the sibling
  re*_s*.json receipt is the same case with run_status COMPLETE and
  source_recheck UNCHANGED; receipts without run_status (pre-recheck
  runners) are refused unless --legacy-receipts, and labelled
  LEGACY_NO_FINAL_SOURCE_RECHECK. Results carry receipt_gate.
- V3: receipts are schema 2 and need source_recheck UNCHANGED; a missing
  recheck is accepted only for frozen-mode schema 1 receipts whose
  run_case.py hash is one of the pre-recheck runners (9805383, 2cf9853,
  db3d098; hashes checked against git history in a test).
- V1/V3 refuse --output inside the checkout before source checks or CUDA
  (shared source_provenance.reject_output_inside).
- A failing recheck is recorded as RECHECK_FAILED (still invalid); the
  in-run V1 receipt write is tmp+replace.
Negative host tests for each gate. No threshold, solver or evidence edits.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

tests/unit/test_jet_circle_cv.py

@@ -196,5 +196,51 @@ 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, 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())
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_jet_cv_measurements.py

@@ -151,6 +151,8 @@ 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,
@@ -163,6 +165,7 @@ 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
@@ -178,5 +181,36 @@ 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()

tests/unit/test_validation_source_provenance.py

@@ -111,3 +111,29 @@ 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"]) == ("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)

tests/validation/analyze_jet_cv_continuation.py

@@ -14,11 +14,35 @@ def integrate(t, y):
     )
 
 
-def analyze(directory):
+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):
     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:
@@ -97,6 +121,7 @@ def analyze(directory):
         )
     return dict(
         scale=s,
+        receipt_gate=gate,
         complete_schedule=bool(complete_schedule),
         sample_checksum_verified=checksum_verified,
         windows=records,
@@ -115,8 +140,10 @@ 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) for d in a.directories]
+    results = [analyze(d, a.legacy_receipts) for d in a.directories]
     payload = dict(
         status="TRANSIENT_DIAGNOSTIC_COMPARISON",
         results=results,

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,13 +587,14 @@ 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",
                 )
-                (output / f"re{c['re_d']}_s{c['scale']}.json").write_text(
-                    json.dumps(payload, indent=2) + "\n"
-                )
+                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)
                 print(
                     json.dumps(
                         dict(
@@ -638,16 +670,9 @@ 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)
-        )
+    from source_provenance import reject_output_inside
+    reject_output_inside(a.output, repo)
+    provenance = source_identity(repo, a.current_checkout)
     sys.path.insert(0, str(repo / "src"))
     import CelerisLab
 
@@ -663,7 +688,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)

tests/validation/source_provenance.py

@@ -15,6 +15,31 @@ 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"; "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)."""
 

tests/validation/taylor_green/assess_projected_pairs.py

@@ -73,6 +73,27 @@ 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
@@ -144,6 +165,7 @@ 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)
+        checks[label + " source_recheck"] = rechecked(r, current)
         checks[label + " numerics"] = r.get("numerics") == NUMERICS and r.get("initialization") == INITIALIZATION
         runtime = r.get("runtime_source", {})
         package = Path(runtime.get("checkout", "")) / "src/CelerisLab"

tests/validation/taylor_green/run_case.py

@@ -182,6 +182,8 @@ 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)
     runtime = runtime_source(repo)
     if args.dry_run:
@@ -201,7 +203,7 @@ def main():
     masks = [k == ACTIVE for k in h.kinds]
     main_steps, early_steps = sample_steps(args.n, args.early_pressure)
     report = dict(
-        schema=1,
+        schema=2,  # 2: final source recheck (source_recheck) is recorded
         source=source,
         numerical_revision=None if args.current_checkout else NUMERICAL_REVISION,
         **({"source_mode": "current_checkout"} if args.current_checkout else {}),
@@ -319,6 +321,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,

tests/validation/taylor_green/test_projected_pairs.py

@@ -12,13 +12,13 @@ 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
 
 import run_case
 from assess_projected_pairs import (BUILD_SOURCES, CASES, NUMERICS, NUMERICAL_REVISION,
-                                    SOURCE_SHA256, assess)
+                                    LEGACY_RUN_CASE_SHA256, 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": "a" * 64},
+                harness_sha256={"run_case.py": LEGACY_RUN_CASE_SHA256["2cf9853"]},
                 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(
-            re_L=re, n=n, depth=depth, explosion=scheme, U=U, mach=float(U*np.sqrt(3)),
+            schema=1, 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,
@@ -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["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")))
@@ -333,6 +334,62 @@ 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()  # 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"
@@ -393,7 +450,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 +506,31 @@ 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" 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)
                 self.assertEqual(selectors, ["projected_linear" if early else "constant"] if depth == 2 else [])
                 self.assertEqual(receipt["source"], source())
                 self.assertEqual(receipt["runtime_identity"], RUNTIME)