test: check pinned source manifests against their own commits

Historical-evidence pins (dfg_re100 / jet_circle_cv / refined_dfg manifests)
now compare each recorded digest with the file at the recorded commit, read
with git, instead of with today's working tree. Missing history (no git,
shallow clone, installed copy) raises HistoryUnavailable and the test is
reported as skipped with that reason, never passed. Receipt/plan logic tests
take the selected manifest directly, so their assertions run again; the
pinned modes' mismatch rejections stay as negative controls.

V1 (run_jet_circle_cv) keeps its pinned mode and gains --current-checkout:
it records HEAD, the src tree and per-file hashes labelled as current
checkout, never as the frozen qualification, and requires a clean tree.

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

tests/unit/test_jet_circle_cv.py

@@ -166,7 +166,34 @@ class Host(unittest.TestCase):
         self.assertEqual(a["steps"] * 4, b["steps"])
 
     def test_source_pin(self):
-        verify(REPO)
+        """Historical integrity: the manifest matches its pinned commit, read with git."""
+        from run_jet_circle_cv import verify_historical
+        from source_provenance import HistoryUnavailable
+        try:
+            verify_historical(REPO)
+        except HistoryUnavailable as error:
+            self.skipTest(f"historical integrity NOT checked: {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"])
 
 
 if __name__ == "__main__":

tests/unit/test_re100.py

@@ -67,8 +67,12 @@ def test_production_unmodified():
         (ROOT / "tests/validation/data/dfg_re100/source_manifest.json").read_text()
     )
     assert m["commit"] == h.PIN
-    for name, digest in m["production_sha256"].items():
-        assert hashlib.sha256((ROOT / name).read_bytes()).hexdigest() == digest, name
+    from source_provenance import HistoryUnavailable, historical_sha256
+    try:  # the manifest certifies its own commit, not today's working tree
+        actual = historical_sha256(ROOT, m["commit"], m["production_sha256"])
+    except HistoryUnavailable as error:
+        pytest.skip(f"historical integrity NOT checked: {error}")
+    assert actual == m["production_sha256"]
 
 
 def test_reference():

tests/unit/test_refined_dfg.py

@@ -12,12 +12,30 @@ 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 = json.loads((r.HERE / "data/refined_dfg/source_manifest.json").read_text())
+    historical = historical_manifest("SRT")
     assert historical["commit"].startswith("cd3a657")
-    mrt_history = json.loads((r.HERE / "data/refined_dfg/source_manifest_mrt.json").read_text())
+    mrt_history = historical_manifest("MRT")
     assert mrt_history["commit"].startswith("3ea131e")
-    assert r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["commit"].startswith("a233c90")
+    assert historical_manifest("MRT", "window")["commit"].startswith("a233c90")
+    from source_provenance import HistoryUnavailable, historical_sha256
+    for manifest in (historical, mrt_history, historical_manifest("MRT", "window")):
+        try:  # historical integrity: each manifest matches the tree of its own commit
+            actual = historical_sha256(r.DEFAULT_SOURCE, manifest["commit"], manifest["sha256"])
+        except HistoryUnavailable as error:
+            unittest.TestCase().skipTest(f"historical integrity NOT checked: {error}")
+        assert actual == manifest["sha256"], manifest["commit"]
+    # Negative controls: the pinned modes refuse a working tree that differs from their pin.
     with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
         r.verify_source(r.DEFAULT_SOURCE, "MRT")
     with unittest.TestCase().assertRaisesRegex(RuntimeError, "Source mismatch"):
@@ -143,12 +161,13 @@ def test_axis_specific_sampling_phase():
 
 def test_full_requires_completed_pilot():
     with unittest.TestCase().assertRaisesRegex(ValueError, "pilot-results"):
-        r.verify_pilot(None, r.verify_source(r.DEFAULT_SOURCE, "MRT", "window"))
+        r.verify_pilot(None, historical_manifest("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 = (r.verify_source(r.DEFAULT_SOURCE, "MRT", "window", jet_scenario)
+    manifest = ((historical_manifest("MRT", "window") if jet_scenario == "none" else
+                 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)
@@ -348,7 +367,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 r.verify_source(r.DEFAULT_SOURCE, "MRT", "window")["production_origin_commit"].startswith("d930")
+    assert historical_manifest("MRT", "window")["production_origin_commit"].startswith("d930")
 
 
 def test_window_plan_preserves_physics():
@@ -358,7 +377,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 r.verify_source(r.DEFAULT_SOURCE, collision, "window")["observation_mode"] == "window"
+        assert historical_manifest(collision, "window")["observation_mode"] == "window"
     with unittest.TestCase().assertRaises(ValueError):
         r.verify_source(r.DEFAULT_SOURCE, "MRT", "unknown")
 

tests/unit/test_validation_source_provenance.py

@@ -8,17 +8,21 @@ import sys
 
 ROOT = Path(__file__).resolve().parents[2]
 sys.path.insert(0, str(ROOT / "tests/validation"))
-from source_provenance import checkout_provenance
+import pytest
+from source_provenance import (HistoryUnavailable, checkout_provenance, historical_sha256,
+                               historical_tree)
 import re100_helpers
 import run_jet_circle_cv
 
 ORIGIN = "47b05bc8abfc957fe876b0c63bdb386e3c6cbaa5"
 
 
-def test_manifests_cover_current_source_and_preserve_origin():
-    names = subprocess.check_output(
-        ["git", "-C", str(ROOT), "ls-files", "src"], text=True
-    ).splitlines()
+def test_manifests_match_their_recorded_source_and_preserve_origin():
+    """Historical evidence integrity: each manifest describes the tree of its own commit."""
+    try:
+        names = historical_tree(ROOT, ORIGIN)
+    except HistoryUnavailable as error:
+        pytest.skip(f"historical integrity NOT checked: {error}")
     for group, hashes_key, revision_key in (
         ("dfg_re100", "production_sha256", "commit"),
         ("jet_circle_cv", "files", "revision"),
@@ -36,10 +40,7 @@ def test_manifests_cover_current_source_and_preserve_origin():
             == "d79e3f6f0a424f5bcc3a1a6d610313e49daf97a2"
         )
         assert "not inherited from d79" in manifest["experiment"]
-        for name, expected in manifest[hashes_key].items():
-            assert (
-                hashlib.sha256((ROOT / name).read_bytes()).hexdigest() == expected
-            ), name
+        assert historical_sha256(ROOT, ORIGIN, manifest[hashes_key]) == manifest[hashes_key]
     assert re100_helpers.PIN == run_jet_circle_cv.PIN == ORIGIN
 
 
@@ -72,3 +73,18 @@ 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())
+    name = sorted(manifest["production_sha256"])[0]
+    try:
+        actual = historical_sha256(ROOT, ORIGIN, [name])
+    except HistoryUnavailable as error:
+        pytest.skip(f"historical integrity NOT checked: {error}")
+    assert actual != {name: "0" * 64}  # a tampered digest would not match

tests/validation/run_jet_circle_cv.py

@@ -3,6 +3,7 @@
 
 import argparse
 import hashlib
+import subprocess
 import json
 import os
 from pathlib import Path
@@ -220,6 +221,30 @@ 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 historical_sha256
+    manifest = json.loads((HERE / "data/jet_circle_cv/source_manifest.json").read_text())
+    if manifest["revision"] != PIN:
+        raise RuntimeError("Wrong pinned revision")
+    if historical_sha256(repo, PIN, manifest["files"]) != manifest["files"]:
+        raise RuntimeError("Historical manifest does not match its pinned revision")
+    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
@@ -530,8 +555,7 @@ 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,
@@ -571,6 +595,9 @@ 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")
@@ -595,7 +622,8 @@ def main():
         )
     plan = dict(
         case=c,
-        required_revision=PIN,
+        required_revision=None if a.current_checkout else PIN,
+        source_mode="current_checkout" if a.current_checkout else "pinned",
         required_free_device_mib=1024,
         estimated_peak_device_mib=128 * a.scale**2,
         estimated_output_mib=64 if a.continuation else 2,
@@ -611,10 +639,16 @@ def main():
     ):
         p.error("Root must select one verified idle GPU via CUDA_VISIBLE_DEVICES")
     repo = a.repo.resolve()
-    source = verify(repo)
-    provenance = dict(
-        source_experiment=source["experiment"], **checkout_provenance(repo)
-    )
+    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)
+        )
     sys.path.insert(0, str(repo / "src"))
     import CelerisLab
 

tests/validation/source_provenance.py

@@ -13,3 +13,38 @@ 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:
+        try:
+            blob = subprocess.run([*git, "show", f"{commit}:{path}"], check=True,
+                                  capture_output=True).stdout
+        except subprocess.CalledProcessError as error:
+            raise ValueError(f"{path} is not part of {commit}") from error
+        digests[path] = hashlib.sha256(blob).hexdigest()
+    return digests
+
+
+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()