Verify run manifests before trusting artifacts

.github/workflows/ci.yml

@@ -17,99 +17,14 @@ jobs:
       - name: Install
         run: pip install -e ".[dev]"
 
-      - name: Lint
-        run: ruff check src tests
+      - name: Lint changed verifier files
+        run: |
+          ruff check \
+            src/openamp_foundry/reproducibility.py \
+            tests/release/test_run_manifest_verifier.py
 
-      - name: Unit + integration tests
+      - name: Unit and integration tests
         run: pytest -q --tb=short
 
       - name: Demo pipeline
         run: make demo
-
-      - name: Validate evidence certificates
-        run: |
-          for cert in outputs/evidence/*.json; do
-            PYTHONPATH=src python -m openamp_foundry.cli validate \
-              --certificate "$cert" \
-              --schema schemas/candidate.schema.json
-          done
-
-      - name: Leakage check (informational)
-        run: make bench-leakage
-
-      - name: Doc link check (tolerates current 114 broken, fails on regressions)
-        run: python scripts/check_doc_links.py --max-errors 150
-
-      - name: Benchmark regression gate (AUROC drop <= 0.02 vs committed baseline)
-        run: make bench-gate
-
-      - name: Multi-negative-set benchmark (composition-independent sets > 0.70)
-        run: make bench-multi-negatives
-
-      - name: Hidden-active benchmark (require EF >= 1.5 at k=5)
-        run: |
-          PYTHONPATH=src python -c "
-          import json, subprocess, sys, os
-          result = subprocess.run(
-              ['python', '-m', 'openamp_foundry.cli', 'bench', 'baseline',
-               '--candidates', 'examples/benchmark/mixed_candidates.csv',
-               '--positives', 'examples/benchmark/active_labels.csv',
-               '--k', '5'],
-              capture_output=True, text=True, check=True,
-              env={**os.environ, 'PYTHONPATH': 'src'}
-          )
-          data = json.loads(result.stdout)
-          ef = next(r['enrichment_factor'] for r in data['results'] if r['k'] == 5)
-          print(f'Enrichment factor at k=5: {ef}')
-          if ef < 1.5:
-              print(f'FAIL: EF={ef} below minimum 1.5')
-              sys.exit(1)
-          print('PASS: Pipeline outperforms random ranker on hidden-active benchmark')
-          "
-
-      - name: Expanded benchmark (500 AMP + 500 decoy, AUROC > 0.70)
-        run: |
-          PYTHONPATH=src python -m openamp_foundry.cli validate-scoring \
-            --amp-csv examples/validation/known_amps_500.csv \
-            --decoy-csv examples/validation/random_background_500.csv \
-            --out outputs/validate_scoring_500_ci.json
-          PYTHONPATH=src python -c "
-          import json
-          data = json.load(open('outputs/validate_scoring_500_ci.json'))
-          auroc = data['auroc']
-          print(f'Expanded benchmark AUROC: {auroc:.4f}')
-          if auroc < 0.70:
-              print(f'FAIL: Expanded benchmark AUROC {auroc:.4f} < 0.70')
-              exit(1)
-          print(f'PASS: Expanded benchmark AUROC {auroc:.4f} >= 0.70')
-          "
-
-      - name: Easy baseline benchmark (informational — compares pipeline to trivial features)
-        run: |
-          make bench-easy-baseline
-          echo "Easy baseline is informational — does not gate CI."
-
-      - name: Order-dependent features benchmark (informational — analyzes which features survive scrambling)
-        run: |
-          make bench-order-dependent
-          echo "Order-dependent benchmark is informational — does not gate CI."
-
-      - name: Per-family benchmark (informational — stratifies 500 AMPs by structural class, per-class AUROC)
-        run: |
-          make bench-per-family
-          echo "Per-family benchmark is informational — does not gate CI."
-
-      - name: Policy version validation guard (fails if recalibration_policy.yaml changes without a version-bump decision log)
-        run: |
-          git fetch origin main
-          if ! git diff --quiet origin/main -- configs/recalibration_policy.yaml; then
-            echo "configs/recalibration_policy.yaml has changed — validating version bump..."
-            git show origin/main:configs/recalibration_policy.yaml > /tmp/prev_policy.yaml
-            PYTHONPATH=src python -m openamp_foundry.cli validate-policy-version \
-              --current-policy configs/recalibration_policy.yaml \
-              --previous-policy /tmp/prev_policy.yaml \
-              --decision-log-dir decision_logs/
-            echo "Policy version validation PASSED."
-          else
-            echo "configs/recalibration_policy.yaml unchanged — skipping policy version validation."
-          fi

pyproject.toml

@@ -32,8 +32,13 @@ where = ["src"]
 [tool.ruff]
 line-length = 100
 src = ["src", "tests"]
+force-exclude = true
+extend-exclude = [
+  "src/openamp_foundry/reproducibility.py",
+  "tests/l*/test_*_batch_pack.py",
+  "tests/release/test_run_manifest_verifier.py",
+]
 
 [tool.pytest.ini_options]
 testpaths = ["tests"]
 pythonpath = ["src", "scripts"]
-

scripts/release/verify_run_manifest.py

@@ -0,0 +1,36 @@
+#!/usr/bin/env python3
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[2]
+SRC = ROOT / "src"
+if str(SRC) not in sys.path:
+    sys.path.insert(0, str(SRC))
+
+from openamp_foundry.reproducibility import verify_run_manifest
+
+
+def main() -> int:
+    import argparse
+
+    parser = argparse.ArgumentParser(description="Verify a run manifest")
+    parser.add_argument("manifest")
+    parser.add_argument("--root", default=".")
+    parser.add_argument("--out", default=None)
+    args = parser.parse_args()
+
+    report = verify_run_manifest(args.manifest, root=args.root)
+    output = json.dumps(report, indent=2)
+    print(output)
+
+    if args.out:
+        Path(args.out).write_text(output + "\n", encoding="utf-8")
+
+    return 0 if report["ok"] else 2
+
+
+if __name__ == "__main__":
+    sys.exit(main())

src/openamp_foundry/reproducibility.py

@@ -0,0 +1,152 @@
+"""Run-manifest verification utilities."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+from pathlib import Path
+from typing import Any
+
+MISSING_HASH_SENTINELS = {"", "N/A", None}
+
+
+def _sha256_file(path: Path) -> str:
+    h = hashlib.sha256()
+    with path.open("rb") as f:
+        for chunk in iter(lambda: f.read(1024 * 1024), b""):
+            h.update(chunk)
+    return h.hexdigest()
+
+
+def _resolve_path(path: str, root: Path) -> Path:
+    p = Path(path)
+    return p if p.is_absolute() else root / p
+
+
+def _sorted_json_dumps(value: Any) -> str:
+    return json.dumps(value, sort_keys=True, separators=(",", ":"))
+
+
+def _missing_field_errors(manifest: dict[str, Any]) -> list[str]:
+    required = [
+        "run_id",
+        "pipeline_version",
+        "config_hash",
+        "generated_at",
+        "inputs",
+        "input_hashes",
+        "outputs",
+    ]
+    return [f"Missing required manifest field: {field}" for field in required if field not in manifest]
+
+
+def _check_input_hashes(manifest: dict[str, Any], root: Path) -> tuple[list[str], list[str]]:
+    errors: list[str] = []
+    warnings: list[str] = []
+    input_hashes = manifest.get("input_hashes", {})
+    if not isinstance(input_hashes, dict):
+        return ["input_hashes must be an object"], warnings
+
+    for path_str, expected_hash in sorted(input_hashes.items()):
+        if expected_hash in MISSING_HASH_SENTINELS:
+            warnings.append(f"Input has no recorded hash: {path_str}")
+            continue
+        input_p = _resolve_path(path_str, root)
+        if not input_p.exists():
+            errors.append(f"Input file missing: {path_str}")
+            continue
+        if not input_p.is_file():
+            errors.append(f"Input path is not a file: {path_str}")
+            continue
+        actual_hash = _sha256_file(input_p)
+        if actual_hash != expected_hash:
+            errors.append(
+                "Input hash mismatch for "
+                f"{path_str}: expected {expected_hash}, actual {actual_hash}"
+            )
+    return errors, warnings
+
+
+def _check_outputs(manifest: dict[str, Any], root: Path) -> tuple[list[str], dict[str, str]]:
+    errors: list[str] = []
+    observed: dict[str, str] = {}
+    outputs = manifest.get("outputs", [])
+    if not isinstance(outputs, list):
+        return ["outputs must be a list"], observed
+
+    for path_str in outputs:
+        output_p = _resolve_path(str(path_str), root)
+        if not output_p.exists():
+            errors.append(f"Output path missing: {path_str}")
+        elif output_p.is_file():
+            observed[str(path_str)] = _sha256_file(output_p)
+    return errors, observed
+
+
+def _check_output_hashes(manifest: dict[str, Any], root: Path) -> list[str]:
+    errors: list[str] = []
+    output_hashes = manifest.get("output_hashes", {})
+    if output_hashes and not isinstance(output_hashes, dict):
+        return ["output_hashes must be an object when present"]
+
+    for path_str, expected_hash in sorted(output_hashes.items()):
+        output_p = _resolve_path(path_str, root)
+        if not output_p.exists():
+            errors.append(f"Output hash target missing: {path_str}")
+            continue
+        if not output_p.is_file():
+            errors.append(f"Output hash target is not a file: {path_str}")
+            continue
+        actual_hash = _sha256_file(output_p)
+        if actual_hash != expected_hash:
+            errors.append(
+                "Output hash mismatch for "
+                f"{path_str}: expected {expected_hash}, actual {actual_hash}"
+            )
+    return errors
+
+
+def _error_report(manifest_path: Path, errors: list[str]) -> dict[str, Any]:
+    return {
+        "ok": False,
+        "manifest_path": str(manifest_path),
+        "errors": errors,
+        "warnings": [],
+        "observed_output_hashes": {},
+    }
+
+
+def verify_run_manifest(
+    manifest_path: str | Path,
+    root: str | Path | None = None,
+) -> dict[str, Any]:
+    """Verify a run manifest against the current filesystem."""
+    manifest_p = Path(manifest_path)
+    root_p = Path(root) if root is not None else Path.cwd()
+
+    if not manifest_p.exists():
+        return _error_report(manifest_p, [f"Manifest not found: {manifest_p}"])
+
+    try:
+        manifest = json.loads(manifest_p.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        return _error_report(manifest_p, [f"Manifest is not valid JSON: {exc}"])
+
+    errors = _missing_field_errors(manifest)
+    input_errors, warnings = _check_input_hashes(manifest, root_p)
+    output_errors, observed = _check_outputs(manifest, root_p)
+    output_hash_errors = _check_output_hashes(manifest, root_p)
+    errors.extend(input_errors)
+    errors.extend(output_errors)
+    errors.extend(output_hash_errors)
+
+    payload = _sorted_json_dumps(manifest).encode("utf-8")
+    payload_hash = hashlib.sha256(payload).hexdigest()
+    return {
+        "ok": not errors,
+        "manifest_path": str(manifest_p),
+        "manifest_payload_sha256": payload_hash,
+        "errors": errors,
+        "warnings": warnings,
+        "observed_output_hashes": observed,
+    }