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>
This commit is contained in:
ClaudeandClaude Opus 5.5 committed 2026-10-20 18:15:50 +00:00
1 parent 750e8a483e
commit 941eb28b79
6 files changed
+161 -26

No files matched your search

+41 -7
View File
@@ -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