#!/usr/bin/env python3 """Deterministic parity proof between canonical and wrapper verifier entrypoints.""" from __future__ import annotations import os import re import subprocess import sys from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Tuple SCRIPT_DIR = Path(__file__).resolve().parent AUDIT_FILE = SCRIPT_DIR / "ARTIFACT_PATH_AUDIT.md" LOG_ROOT = SCRIPT_DIR / "parity_logs" CANONICAL_ENV = "BENCHMARK_VERIFIER_CANONICAL" WRAPPER_ENV = "BENCHMARK_VERIFIER_WRAPPER" SUCCESS_RUN_ID = os.environ.get("VERIFIER_PARITY_SUCCESS_RUN_ID", "2026-03-04") FAILURE_RUN_ID = os.environ.get("VERIFIER_PARITY_FAILURE_RUN_ID", "not-a-run-id") SCENARIOS: List[Tuple[str, str]] = [ ("success", SUCCESS_RUN_ID), ("intentional_failure", FAILURE_RUN_ID), ] MARKER_PATTERNS = [ ("PASS_MARKER", "Artifact contract check passed."), ("FAIL_MARKER", "Artifact contract check failed."), ("INVALID_RUN_ID", "INVALID_RUN_ID"), ("RUN_ID_FORMAT", "expected YYYY-MM-DD format"), ("RUN_MISMATCH", "MISMATCH:"), ("HASH_MISMATCH", "HASH_MISMATCH"), ("MISSING", "MISSING:"), ] def resolve_path(env_key: str, fallback_name: str, require_wrapper: bool = False) -> Path: value = os.environ.get(env_key, "").strip() if value: path = Path(value) if path.exists(): return path.resolve() raise FileNotFoundError(f"{env_key} set but not found: {path}") direct = (SCRIPT_DIR / fallback_name).resolve() if direct.exists(): return direct if not require_wrapper: raise FileNotFoundError(f"canonical verifier missing: {direct}") # Heuristic fallback: any script containing both verify/artifact and wrapper in filename. candidates = [ p.resolve() for p in sorted(SCRIPT_DIR.glob("*.sh")) if "verify" in p.name.lower() and "artifact" in p.name.lower() and "wrapper" in p.name.lower() and p.is_file() ] if not candidates: raise FileNotFoundError( "wrapper verifier not found; set " "BENCHMARK_VERIFIER_WRAPPER to the wrapper entrypoint path" ) return candidates[0] def detect_marker(text: str) -> str: for label, pattern in MARKER_PATTERNS: if re.search(pattern, text): return label return "NO_KEY_MARKER" def run_entrypoint(verifier: Path, run_id: str, scenario: str, tag: str, run_dir: Path) -> Dict[str, object]: log_path = run_dir / f"{scenario}_{tag}.log" env = os.environ.copy() env["RUN_ID"] = run_id proc = subprocess.run( [str(verifier)], env=env, capture_output=True, text=True, check=False, ) output = (proc.stdout or "") + (proc.stderr or "") log_path.write_text(output, encoding="utf-8") marker = detect_marker(output) marker_line = "" if output: for line in output.splitlines(): if any(pat in line for _, pat in MARKER_PATTERNS) or "Artifact contract" in line: marker_line = line.strip() break return { "verifier": str(verifier), "run_id": run_id, "scenario": scenario, "exit_code": proc.returncode, "marker": marker, "marker_line": marker_line, "log": str(log_path), "raw_output_len": len(output), } def write_matrix(artifact: Path, matrix: List[Dict[str, object]], generated_at: str, canonical: Path, wrapper: Path) -> None: lines = [] lines.append("\n## Deterministic verifier parity evidence") lines.append(f"Generated: {generated_at}") lines.append(f"Canonical verifier: `{canonical}`") lines.append(f"Wrapper verifier: `{wrapper}`") lines.append("Scenarios: success + intentional_failure") lines.append("") lines.append("| Scenario | RUN_ID | Canonical exit | Wrapper exit | Canonical marker | Wrapper marker | Parity |") lines.append("|---|---|---:|---:|---|---|---|") all_pass = True for row in matrix: canonical_entry = row["canonical"] wrapper_entry = row["wrapper"] parity = bool(canonical_entry["exit_code"] == wrapper_entry["exit_code"] and canonical_entry["marker"] == wrapper_entry["marker"]) row["parity"] = parity if not parity: all_pass = False lines.append( f"| {row['scenario']} | {row['run_id']} | " f"{canonical_entry['exit_code']} | {wrapper_entry['exit_code']} | " f"{canonical_entry['marker']} | {wrapper_entry['marker']} | {'PASS' if parity else 'FAIL'} |" ) lines.append("") lines.append(f"Overall parity: {'PASS' if all_pass else 'FAIL'}") lines.append("") lines.append("### Scenario evidence logs") lines.append("") for row in matrix: for tag in ("canonical", "wrapper"): entry = row[tag] lines.append(f"- **{row['scenario']}::{tag}** -> `{entry['log']}` (exit={entry['exit_code']}, marker={entry['marker']})") lines.append("") artifact.write_text("\n".join(lines) + "\n", encoding="utf-8") def append_to_audit(matrix: List[Dict[str, object]], generated_at: str, canonical: Path, wrapper: Path) -> None: lines = [] lines.append("\n## Deterministic verifier parity evidence") lines.append(f"Generated: {generated_at}") lines.append(f"Canonical verifier: `{canonical}`") lines.append(f"Wrapper verifier: `{wrapper}`") lines.append("Scenarios: success + intentional_failure") lines.append("") lines.append("| Scenario | RUN_ID | Canonical exit | Wrapper exit | Canonical marker | Wrapper marker | Parity |") lines.append("|---|---|---:|---:|---|---|---|") all_pass = True for row in matrix: canonical_entry = row["canonical"] wrapper_entry = row["wrapper"] parity = bool(canonical_entry["exit_code"] == wrapper_entry["exit_code"] and canonical_entry["marker"] == wrapper_entry["marker"]) row["parity"] = parity all_pass = all_pass and parity lines.append( f"| {row['scenario']} | {row['run_id']} | " f"{canonical_entry['exit_code']} | {wrapper_entry['exit_code']} | " f"{canonical_entry['marker']} | {wrapper_entry['marker']} | {'PASS' if parity else 'FAIL'} |" ) lines.append("") lines.append(f"Overall parity: {'PASS' if all_pass else 'FAIL'}") existing = AUDIT_FILE.read_text(encoding="utf-8") if AUDIT_FILE.exists() else "" header = "# Artifact path literal audit\n" if not existing.startswith(header): existing = header + "\n" + existing footer = "\n".join(lines) + "\n" AUDIT_FILE.write_text(existing + footer, encoding="utf-8") def main() -> int: canonical = resolve_path(CANONICAL_ENV, "verify_benchmark_artifacts.sh") wrapper = resolve_path(WRAPPER_ENV, "verify_benchmark_artifacts_wrapper.sh", require_wrapper=True) generated_at = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") timestamp = generated_at.replace(":", "-") run_dir = (LOG_ROOT / timestamp) run_dir.mkdir(parents=True, exist_ok=True) matrix: List[Dict[str, object]] = [] for scenario, run_id in SCENARIOS: canonical_result = run_entrypoint(canonical, run_id, scenario, "canonical", run_dir) wrapper_result = run_entrypoint(wrapper, run_id, scenario, "wrapper", run_dir) matrix.append( { "scenario": scenario, "run_id": run_id, "canonical": canonical_result, "wrapper": wrapper_result, } ) matrix_path = run_dir / "verifier_parity_matrix.json" matrix_payload = { "generated_at": generated_at, "canonical_verifier": str(canonical), "wrapper_verifier": str(wrapper), "scenarios": SCENARIOS, "matrix": matrix, } matrix_path.write_text(__import__("json").dumps(matrix_payload, indent=2), encoding="utf-8") append_to_audit(matrix, generated_at, canonical, wrapper) write_matrix(matrix_path, matrix, generated_at, canonical, wrapper) return 0 if __name__ == "__main__": raise SystemExit(main())