Stage bounded offline public re-scoring script for a clean execution check before release

release_candidate/reproduce_execution.py

@@ -0,0 +1,69 @@
+"""Offline re-scoring of frozen programs; Linux/root/libseccomp are required.
+
+Never replace the sandbox with direct exec. All generated programs run only in
+its credential-cleared nobody workers with filesystem/network denial.
+"""
+import ast
+import hashlib
+import json
+import re
+from collections import defaultdict
+from pathlib import Path
+import numpy as np
+import quantum_sandbox_v2 as sandbox
+
+ROOT=Path(__file__).resolve().parent
+
+
+def score(program,row):
+    wrapped=re.fullmatch(r"\s*```(?:python)?\s*\n(.*?)\n```\s*",program,flags=re.S)
+    clean=wrapped.group(1) if wrapped else program
+    imports=set()
+    try:
+        for node in ast.walk(ast.parse(clean)):
+            if isinstance(node,ast.Import):imports.update(a.name.split(".")[0] for a in node.names)
+            elif isinstance(node,ast.ImportFrom):imports.add((node.module or "").split(".")[0])
+    except SyntaxError:
+        pass
+    result=sandbox.run(clean,row["qubits"])
+    passed=False
+    if result.get("status")=="executed":
+        actual=np.array(result["real"])+1j*np.array(result["imag"])
+        target=np.array(row["state_real"])+1j*np.array(row["state_imag"])
+        passed=abs(result["norm"]-1)<1e-6 and abs(np.vdot(target,actual))**2>1-1e-6 and row["framework"] in imports
+    return bool(passed)
+
+
+def main():
+    frozen=json.loads((ROOT/"v2_execution/execution.json").read_text())
+    preflight=json.loads((ROOT/"v2_executor_preflight/preflight.json").read_text())
+    assert preflight["all_passed"] and len(preflight["reference_checks"])==1368
+    assert hashlib.sha256(Path(sandbox.__file__).read_bytes()).hexdigest()==frozen["executor_sha256"]==preflight["executor_sha256"]
+    file=ROOT/"curriculum_v2/rows.json"
+    assert hashlib.sha256(file.read_bytes()).hexdigest()==preflight["manifest"]["rows_sha256"]
+    rows={r["id"]:r for r in json.loads(file.read_text()) if r["split"]!="train"}
+    assert len(rows)==720
+    versions=sandbox.initialize()
+    assert versions==frozen["versions"]
+    assert sandbox.run("",1,mode="isolation_check")==preflight["isolation"]
+    for row in rows.values():
+        assert score(row["program"],row),"Reference failed; stop before scoring generated programs"
+    outputs=json.loads((ROOT/"curriculum_v2_run/generation_report.json").read_text())["conditions"]
+    totals=defaultdict(lambda:dict(n=0,passed=0))
+    for condition in ["base","adapted"]:
+        assert len(outputs[condition])==720 and {r["id"] for r in outputs[condition]}==set(rows)
+        for output in outputs[condition]:
+            row=rows[output["id"]];passed=score(output["program"],row)
+            for framework in ["all",row["framework"]]:
+                key=condition+"/"+row["split"]+"/"+framework
+                totals[key]["n"]+=1;totals[key]["passed"]+=passed
+    for key,value in totals.items():
+        condition,split,framework=key.split("/")
+        original=frozen["summary"][condition][split+"/"+framework]
+        assert value==dict(n=original["n"],passed=original["passed"]),(key,value,original)
+    result=dict(versions=versions,all_counts_reproduced=True,summary=dict(totals))
+    (ROOT/"reproduced_counts.json").write_text(json.dumps(result,indent=2)+"\n")
+    print(json.dumps(result,indent=2))
+
+
+if __name__=="__main__":main()