fix: integrate reviewed LES normalization with explicit dev flow pins

This commit is contained in:
Codex committed 2026-10-18 07:58:41 -07:00
1 parent 8403956a9a
commit 95cb6d2b9f
6 files changed
+326 -29

No files matched your search

+21 -4
View File
@@ -59,8 +59,15 @@ def decode_macros(data, w, cx, cy):
return profile, *metrics
def source_guard(root, variant):
manifest = json.loads((HERE / "data/les_flow/source_pins.json").read_text())
EXPERIMENTS = {
"historical": "source_pins.json",
"dev8403956": "source_pins_dev8403956.json",
}
def source_guard(root, variant, experiment="historical"):
manifest_path = HERE / "data/les_flow" / EXPERIMENTS[experiment]
manifest = json.loads(manifest_path.read_text())
for name, sha in manifest["sources"].items():
expected = (
manifest["candidate_header_sha256"]
@@ -69,7 +76,7 @@ def source_guard(root, variant):
)
if hashlib.sha256((root / name).read_bytes()).hexdigest() != expected:
raise RuntimeError("Source pin mismatch: " + name)
return dict(
identity = dict(
base_commit=manifest["base_commit"],
production_origin_commit=manifest["production_origin_commit"],
source_experiment=manifest["source_experiment"],
@@ -80,12 +87,22 @@ def source_guard(root, variant):
header_sha256=hashlib.sha256((root / HEADER).read_bytes()).hexdigest(),
variant=variant,
)
if experiment != "historical":
identity.update(
experiment=experiment,
source_manifest_sha256=hashlib.sha256(manifest_path.read_bytes()).hexdigest(),
)
return identity
def main():
ap = argparse.ArgumentParser(description=__doc__)
ap.add_argument("--root", type=Path, required=True)
ap.add_argument("--variant", choices=["baseline", "corrected"], required=True)
ap.add_argument(
"--experiment", choices=tuple(EXPERIMENTS), default="historical",
help="Explicit source contract; historical pins remain the default",
)
ap.add_argument("--output", type=Path, required=True)
ap.add_argument(
"--device", type=int, default=0, help="Logical device in CUDA_VISIBLE_DEVICES"
@@ -100,7 +117,7 @@ def main():
ap.error("Output exists; choose a new evidence path")
args.output.parent.mkdir(parents=True, exist_ok=True)
root = args.root.resolve()
pins = source_guard(root, args.variant)
pins = source_guard(root, args.variant, args.experiment)
# Refuse installed-package shadowing; import only after explicit GPU opt-in.
sys.path.insert(0, str(root / "src"))
import CelerisLab