File size: 3,422 Bytes
df525ad | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 | """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()
|