QuantumCoder-9B / v2_executor_preflight /train_quantum_v2_preflight_pilot.py
itsalissonsilva's picture
Freeze isolated v2 executor against every fresh reference before reading model programs
df525ad verified
Raw History Blame Contribute Delete
3.42 kB
"""Freeze v2 executor with all curriculum references before opening model outputs."""
import hashlib
import json
from pathlib import Path
import numpy as np
from huggingface_hub import HfApi, hf_hub_download
import quantum_sandbox_v2 as sandbox
REPO = "itsalissonsilva/QuantumCoder-9B"
DATA_REV = "97bf11cde2ef2aeb43345a0da7eab3d8ea27993a"
def main():
api = HfApi()
assert api.model_info(REPO).private
def get(name):
return Path(hf_hub_download(REPO, "curriculum_v2/"+name, revision=DATA_REV))
source = get("rows.json")
manifest = json.loads(get("manifest.json").read_text())
assert hashlib.sha256(source.read_bytes()).hexdigest() == manifest["rows_sha256"]
rows = json.loads(source.read_text())
versions = sandbox.initialize()
isolation = sandbox.run("", 1, mode="isolation_check")
assert isolation == dict(file_opens_blocked=True, network_blocked=True, uid=65534, environment_empty=True)
for code in ["open('/etc/passwd')", "import os", "import numpy as np\nstate=np.load('/etc/passwd')", "x=().__class__", "eval('1+1')"]:
try:
sandbox.screen(code)
except (ValueError, SyntaxError):
continue
raise AssertionError("Forbidden fixture passed")
records = []
for row in rows:
result = sandbox.run(row["program"], row["qubits"])
fidelity = None
passed = False
if result.get("status") == "executed":
state = np.array(result["real"])+1j*np.array(result["imag"])
target = np.array(row["state_real"])+1j*np.array(row["state_imag"])
fidelity = float(abs(np.vdot(target, state))**2)
passed = abs(result["norm"]-1)<1e-6 and fidelity>1-1e-6
records.append(dict(id=row["id"], split=row["split"], framework=row["framework"], passed=passed, fidelity=fidelity,
execution={k:v for k,v in result.items() if k not in ["real","imag"]}))
if len(records)%100==0:
print(json.dumps(dict(stage="reference_checks", n=len(records), passed=sum(r["passed"] for r in records))), flush=True)
report = dict(data_revision=DATA_REV, manifest=manifest, versions=versions, isolation=isolation,
reference_checks=records, all_passed=all(r["passed"] for r in records),
executor_sha256=hashlib.sha256(Path(sandbox.__file__).read_bytes()).hexdigest(),
scope="V2 independently frozen supported-Python executor. V1 remains unchanged. Adds RZ warmup and documented Cirq CX/Rx/Ry/Rz aliases; same nobody/seccomp isolation. No model outputs read by this preflight.")
folder=Path("v2_executor_preflight")
folder.mkdir(exist_ok=True)
(folder/"preflight.json").write_text(json.dumps(report,indent=2))
(folder/"quantum_sandbox_v2.py").write_bytes(Path(sandbox.__file__).read_bytes())
(folder/"train_quantum_v2_preflight_pilot.py").write_bytes(Path(__file__).read_bytes())
commit=api.upload_folder(repo_id=REPO,folder_path=str(folder),path_in_repo="v2_executor_preflight",
commit_message="Freeze isolated v2 executor against every fresh reference before reading model programs")
print(json.dumps(dict(revision=commit.oid, all_passed=report["all_passed"], n=len(records))),flush=True)
assert report["all_passed"], "Reference gate failed; do not score model programs"
if __name__ == "__main__":
main()