Portable formal replay wrapper (requires the external Lean kit)

Quantum and Circuits

replay_formal.py

Download original file

This is the original source text. Reading it does not run or verify the program.

Read the typeset paper

Skip source text

1 #!/usr/bin/env python3
2 """Replay the two supplied Lean projects with an existing offline runtime.
3 
4 The wrapper uses the original project verifiers. It never downloads a runtime,
5 changes proof sources, or enables the optional Linux compatibility adapter
6 without --compat-shim. --dry-run does not start Lean or write any files.
7 """
8 from __future__ import annotations
9 
10 import argparse
11 from datetime import datetime, timezone
12 import hashlib
13 import json
14 import os
15 from pathlib import Path
16 import platform
17 import shlex
18 import shutil
19 import subprocess
20 import sys
21 import tempfile
22 
23 
24 PACKAGE = Path(__file__).resolve().parents[1]
25 QUANTUM = PACKAGE / "sources/quantum/Quantum_Arithmetic_Complete_Release/lean/checked_checkpoint"
26 CIRCUITS = PACKAGE / "sources/circuits/Komorebi_Circuit_Obstructions_Research_Edition_1.0/source/komorebi-circuit-obstructions-v1.2.0"
27 
28 
29 def digest(path: Path) -> str:
30     return hashlib.sha256(path.read_bytes()).hexdigest()
31 
32 
33 def make_plan(runtime: Path, output: Path) -> list[dict]:
34     toolchain = runtime / "toolchain"
35     return [
36         {
37             "name": "quantum",
38             "cwd": str(QUANTUM),
39             "argv": [sys.executable, str(QUANTUM / "scripts/verify_lean.py"),
40                      "--toolchain", str(toolchain),
41                      "--packages", str(runtime / "workspace/.lake/packages"),
42                      "--out", str(output / "quantum")],
43             "receipt": str(output / "quantum/REPORT.json"),
44         },
45         {
46             "name": "circuits",
47             "cwd": str(CIRCUITS),
48             "argv": [sys.executable, str(CIRCUITS / "scripts/verify_lean.py"),
49                      "--toolchain", str(toolchain),
50                      "--out", str(output / "circuits")],
51             "receipt": str(output / "circuits/verification.json"),
52         },
53     ]
54 
55 
56 def main() -> int:
57     parser = argparse.ArgumentParser(description=__doc__)
58     parser.add_argument("--runtime", required=True, type=Path,
59                         help="Root of the extracted Komorebi_Portable_Lean_4.34.0-rc2 kit")
60     parser.add_argument("--out", required=True, type=Path,
61                         help="New output directory; an existing directory is never overwritten")
62     parser.add_argument("--compat-shim", action="store_true",
63                         help="Explicitly preload the kit's Linux executable-path adapter")
64     parser.add_argument("--dry-run", action="store_true",
65                         help="Print commands only; no runtime, compiler or output directory is needed")
66     args = parser.parse_args()
67     runtime, output = args.runtime.expanduser().resolve(), args.out.expanduser().resolve()
68     plan = make_plan(runtime, output)
69     if args.dry_run:
70         print("DRY RUN: no commands executed and no proof claim made.")
71         print("Compatibility adapter:", "explicitly requested" if args.compat_shim else "not enabled by wrapper")
72         for step in plan:
73             print(f"\n[{step['name']}] cwd: {step['cwd']}")
74             print(shlex.join(step["argv"]))
75             print("Expected receipt:", step["receipt"])
76         return 0
77 
78     if platform.system() != "Linux" or platform.machine() not in ("x86_64", "AMD64"):
79         parser.error("The stored offline kit requires Linux x86_64, including compatible WSL2.")
80     if output.exists():
81         parser.error(f"Output already exists; choose a new directory: {output}")
82     required = [runtime / "toolchain/bin" / name for name in ("lean", "lake", "leanchecker")]
83     required += [Path(step["argv"][1]) for step in plan]
84     if args.compat_shim:
85         required.append(runtime / "compat/app_path.so")
86     missing = [str(path) for path in required if not path.is_file()]
87     packages = runtime / "workspace/.lake/packages"
88     if not packages.is_dir():
89         missing.append(str(packages))
90     if missing:
91         parser.error("Required files or cache are missing:\n" + "\n".join(missing))
92     if any(not os.access(path, os.X_OK) for path in required[:3]):
93         parser.error("The supplied Lean, Lake and leanchecker binaries must be executable.")
94 
95     # Keep external Lean settings from substituting a different library root.
96     env = os.environ.copy()
97     for key in ("LEAN_PATH", "LEAN_SRC_PATH", "LEAN_SYSROOT", "LAKE_HOME"):
98         env.pop(key, None)
99     output.mkdir(parents=True, exist_ok=False)
100     summary = {
101         "status": "RUNNING", "started_at": datetime.now(timezone.utc).isoformat(),
102         "runtime": str(runtime), "compat_shim_requested": args.compat_shim,
103         "official_declaration_counts": {"quantum": 96, "circuits": 115},
104         "scope": "The supplied formal subsets only; supplementary mode-one checks are separate.",
105         "runs": [],
106     }
107     summary_path = output / "replay_summary.json"
108     summary_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
109     adapter_directory = None
110     try:
111         if args.compat_shim:
112             # The Linux loader cannot represent spaces inside LD_PRELOAD paths.
113             adapter_directory = tempfile.TemporaryDirectory(prefix="komorebi-lean-", dir="/tmp")
114             adapter = Path(adapter_directory.name) / "app_path.so"
115             shutil.copy2(runtime / "compat/app_path.so", adapter)
116             previous = env.get("LD_PRELOAD", "")
117             env["LD_PRELOAD"] = str(adapter) + (":" + previous if previous else "")
118             summary["compat_shim_sha256"] = digest(adapter)
119         for step in plan:
120             log = output / (step["name"] + ".console.log")
121             record = {**step, "verifier_sha256": digest(Path(step["argv"][1])),
122                       "console_log": log.name, "status": "FAIL"}
123             print(f"Running {step['name']} verifier...", flush=True)
124             try:
125                 with log.open("w", encoding="utf-8") as stream:
126                     completed = subprocess.run(step["argv"], cwd=step["cwd"], env=env,
127                                                stdout=stream, stderr=subprocess.STDOUT)
128                 record["exit_status"] = completed.returncode
129                 receipt = Path(step["receipt"])
130                 if receipt.is_file():
131                     result = json.loads(receipt.read_text(encoding="utf-8"))
132                     record["receipt_status"] = result.get("status")
133                     record["receipt_sha256"] = digest(receipt)
134                 if completed.returncode == 0 and record.get("receipt_status") == "PASS":
135                     record["status"] = "PASS"
136             except (OSError, ValueError) as exc:
137                 record["failure"] = str(exc)
138             if log.is_file():
139                 record["console_log_sha256"] = digest(log)
140             summary["runs"].append(record)
141             summary_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
142             print(f"{step['name']}: {record['status']} (see {log})", flush=True)
143         summary["status"] = "PASS" if all(r["status"] == "PASS" for r in summary["runs"]) else "FAIL"
144     except OSError as exc:
145         summary["status"] = "FAIL"
146         summary["failure"] = str(exc)
147     finally:
148         if adapter_directory is not None:
149             adapter_directory.cleanup()
150         summary["finished_at"] = datetime.now(timezone.utc).isoformat()
151         summary_path.write_text(json.dumps(summary, indent=2) + "\n", encoding="utf-8")
152     print(f"{summary['status']}: {summary_path}")
153     return 0 if summary["status"] == "PASS" else 1
154 
155 
156 if __name__ == "__main__":
157     raise SystemExit(main())

Back to source controls

Other PDFs for this work

Original file and source record

This copy matches the recorded source file. Its publication here does not change the work’s review status.

SHA-256: 360eef240360b65a9c819f31952fad232047035342c6c0e82f160616078ace5c

Original source on GitHub

Back to Quantum and Circuits