#!/usr/bin/env python3 """TfP sandbox checker: equivalence-certified circuit-cost record. Reads /input/manifest.json. Reference inputs: reference_circuit the canonical circuit (OpenQASM 2) the submission must equal spec JSON: {"gate_set": [...], "ccx_policy": "reject"|"decompose_7t", "max_extra_qubits": 0, "max_two_qubit_ratio": 2.0, "equivalence_timeout_seconds": 420, "metric": "t_count"} Artifact role: circuit the submitted OpenQASM 2 circuit Writes /output/result.json {result, reason, score?} and /output/evidence/summary.json. result: verified -> equivalent (up to global phase), gate set respected, caps respected; score = metric refuted -> provably not equivalent, or an explicit rule violated (gate set, qubit cap, ratio cap) inconclusive -> unreadable circuit, unsupported construct, or the equivalence checker ran out of time/information """ import json, math, os, subprocess, sys, time, traceback INPUT = os.environ.get("TFP_INPUT", "/input") OUTPUT = os.environ.get("TFP_OUTPUT", "/output") T0 = time.time() def write(result, reason, score=None, evidence=None): os.makedirs(os.path.join(OUTPUT, "evidence"), exist_ok=True) out = {"result": result, "reason": reason[:4000]} if result == "verified" and score is not None: out["score"] = score ev = {"qubits": 0, "t_count": 0, "cx_count": 0, "total_gates": 0, "depth": 0, "qcec_result": "not_run", "qcec_seconds": 0} ev.update(evidence or {}) ev.update({"result": result, "elapsed_seconds": round(time.time() - T0, 2)}) json.dump(ev, open(os.path.join(OUTPUT, "evidence", "summary.json"), "w"), indent=1, sort_keys=True) json.dump(out, open(os.path.join(OUTPUT, "result.json"), "w")) print(f"{result}: {reason}") sys.exit(0) def main(): m = json.load(open(os.path.join(INPUT, "manifest.json"))) def find(kind, key): for e in m.get(kind, []): if e.get("key" if kind == "reference" else "role") == key: return e["path"] return None ref_path, spec_path, sub_path = find("reference", "reference_circuit"), find("reference", "spec"), find("artifacts", "circuit") if not (ref_path and spec_path and sub_path): write("inconclusive", "manifest lacks reference_circuit, spec or the circuit artifact") spec = json.load(open(spec_path)) gate_set = set(spec.get("gate_set", ["h", "s", "sdg", "t", "tdg", "x", "z", "cx"])) ccx_policy = spec.get("ccx_policy", "reject") max_extra = int(spec.get("max_extra_qubits", 0)) max_ratio = float(spec.get("max_two_qubit_ratio", 2.0)) eq_timeout = float(spec.get("equivalence_timeout_seconds", 420)) metric = spec.get("metric", "t_count") from qiskit import QuantumCircuit, transpile from qiskit.qasm2 import loads as qasm2_loads, LEGACY_CUSTOM_INSTRUCTIONS def load(path, what): text = open(path, encoding="utf-8", errors="replace").read() if len(text) > 8_000_000: write("inconclusive", f"{what} exceeds 8 MB") try: return qasm2_loads(text, custom_instructions=LEGACY_CUSTOM_INSTRUCTIONS) except Exception as e: write("inconclusive", f"{what} is not parseable OpenQASM 2: {str(e)[:300]}") ref = load(ref_path, "reference circuit") sub = load(sub_path, "submitted circuit") # Rule checks on the submission (explicit rules -> refuted) forbidden = {"measure", "reset", "barrier", "delay", "if_else", "while_loop", "for_loop", "switch_case", "initialize", "store"} names = [inst.operation.name for inst in sub.data] if any(n in forbidden or n.startswith("if") for n in names): write("refuted", "submitted circuit contains measurement, reset or classical control, which this Problem forbids") if sub.num_clbits: write("refuted", "submitted circuit declares classical bits; none are allowed") if sub.num_qubits > ref.num_qubits + max_extra: write("refuted", f"submitted circuit uses {sub.num_qubits} qubits; the reference uses {ref.num_qubits} and at most {max_extra} extra ancilla qubits are allowed") if sub.num_qubits < ref.num_qubits: write("refuted", f"submitted circuit uses {sub.num_qubits} qubits but the reference register has {ref.num_qubits}") # Normalise to the canonical gate set. rz(k*pi/4) and friends map to Clifford+T names; anything else is out of set. def normalise(circ, policy): from qiskit.circuit.library import TGate, TdgGate, SGate, SdgGate, ZGate out = QuantumCircuit(circ.num_qubits) qmap = {q: i for i, q in enumerate(circ.qubits)} bad = set() for inst in circ.data: op, qs = inst.operation, [qmap[q] for q in inst.qubits] n = op.name if n in ("rz", "p", "u1"): try: th = float(op.params[0]) % (2 * math.pi) except Exception: bad.add(f"{n}(symbolic)"); continue k = round(th / (math.pi / 4)) if abs(th - k * math.pi / 4) > 1e-9: bad.add(f"{n}({float(op.params[0]):.6g})"); continue k %= 8 seq = {0: [], 1: [TGate()], 2: [SGate()], 3: [SGate(), TGate()], 4: [ZGate()], 5: [ZGate(), TGate()], 6: [SdgGate()], 7: [TdgGate()]}[k] for g in seq: out.append(g, qs) continue if n == "ccx": if policy == "decompose_7t": a, b, c = qs for g, q in [("h", [c]), ("cx", [b, c]), ("tdg", [c]), ("cx", [a, c]), ("t", [c]), ("cx", [b, c]), ("tdg", [c]), ("cx", [a, c]), ("t", [b]), ("t", [c]), ("h", [c]), ("cx", [a, b]), ("t", [a]), ("tdg", [b]), ("cx", [a, b])]: getattr(out, g)(*q) continue if "ccx" not in gate_set: bad.add("ccx"); continue if n == "id": continue if n == "cz" and "cz" not in gate_set: out.h(qs[1]); out.cx(qs[0], qs[1]); out.h(qs[1]); continue if n == "swap" and "swap" not in gate_set: out.cx(qs[0], qs[1]); out.cx(qs[1], qs[0]); out.cx(qs[0], qs[1]); continue if n == "y": out.z(qs[0]); out.x(qs[0]); continue if n in ("rx", "sx", "sxdg", "ry") and n not in gate_set: if n == "ry": bad.add("ry"); continue th = (math.pi / 2 if n == "sx" else -math.pi / 2 if n == "sxdg" else float(op.params[0])) % (2 * math.pi) k = round(th / (math.pi / 4)) if abs(th - k * math.pi / 4) > 1e-9: bad.add(f"{n}({th:.6g})"); continue k %= 8 out.h(qs[0]) for g in {0: [], 1: [TGate()], 2: [SGate()], 3: [SGate(), TGate()], 4: [ZGate()], 5: [ZGate(), TGate()], 6: [SdgGate()], 7: [TdgGate()]}[k]: out.append(g, qs) out.h(qs[0]); continue if n in gate_set: getattr(out, n)(*qs) if hasattr(out, n) else out.append(op, qs) continue bad.add(n) return out, bad ref_n, ref_bad = normalise(ref, "decompose_7t" if "ccx" not in gate_set else "keep") sub_n, sub_bad = normalise(sub, ccx_policy) if sub_bad: write("refuted", f"submitted circuit uses gates outside the allowed set {sorted(gate_set)}: {sorted(sub_bad)}") if ref_bad: write("inconclusive", f"reference circuit uses unsupported gates {sorted(ref_bad)}; the Problem's reference input is misconfigured") def counts(c): ops = c.count_ops() t = ops.get("t", 0) + ops.get("tdg", 0) two = sum(v for k, v in ops.items() if k in ("cx", "cz", "swap", "ccx")) return {"t_count": t, "toffoli_count": ops.get("ccx", 0), "two_qubit_count": two, "total_gates": sum(ops.values()), "depth": c.depth(), "qubits": c.num_qubits, "ops": dict(ops)} cr, cs = counts(ref_n), counts(sub_n) if cr["two_qubit_count"] and cs["two_qubit_count"] > max_ratio * cr["two_qubit_count"]: write("refuted", f"submitted circuit has {cs['two_qubit_count']} two-qubit gates, more than {max_ratio}x the reference's {cr['two_qubit_count']}", evidence={"qubits": cs["qubits"], "t_count": cs["t_count"], "cx_count": cs["two_qubit_count"], "total_gates": cs["total_gates"], "depth": cs["depth"], "qcec_result": "not_run", "qcec_seconds": 0, "reference": cr, "submission": cs}) # Equivalence certificate. QCEC runs in a child process so that a crash inside the native library (a signal, never a # Python exception) is reported as inconclusive with a diagnosis instead of killing the checker; if the combined run # crashes, each decision procedure is tried on its own and the verdicts are combined conservatively. from qiskit.qasm2 import dumps as qasm2_dumps scratch = os.path.join(OUTPUT, "scratch"); os.makedirs(scratch, exist_ok=True) ref_q, sub_q = os.path.join(scratch, "ref_n.qasm"), os.path.join(scratch, "sub_n.qasm") open(ref_q, "w").write(qasm2_dumps(ref_n)); open(sub_q, "w").write(qasm2_dumps(sub_n)) child = os.path.join(scratch, "qcec_child.py"); open(child, "w").write(QCEC_CHILD) base_ev = {"qubits": cs["qubits"], "t_count": cs["t_count"], "cx_count": cs["two_qubit_count"], "total_gates": cs["total_gates"], "depth": cs["depth"], "reference": cr, "submission": cs} attempts = [] def run_qcec(mode, budget): t0 = time.time() try: p = subprocess.run([sys.executable, "-X", "faulthandler", child, ref_q, sub_q, mode, str(budget)], capture_output=True, text=True, timeout=budget + 30, cwd=scratch) except subprocess.TimeoutExpired: attempts.append({"mode": mode, "outcome": "timeout", "seconds": round(time.time() - t0, 2)}); return None rec = {"mode": mode, "exit": p.returncode, "seconds": round(time.time() - t0, 2)} if p.returncode == 0: try: rec.update(json.loads(p.stdout.strip().splitlines()[-1])); rec["outcome"] = "ok" except Exception: rec["outcome"] = "unreadable"; rec["tail"] = (p.stdout + p.stderr)[-600:] else: rec["outcome"] = f"signal {-p.returncode}" if p.returncode < 0 else f"exit {p.returncode}" rec["tail"] = (p.stdout + p.stderr)[-600:] attempts.append(rec); print(json.dumps(rec)[:1500], flush=True) return rec if rec["outcome"] == "ok" else None def remaining(): return max(5.0, min(eq_timeout, 560 - (time.time() - T0))) # TFP_QCEC_SKIP_COMBINED is a local test hook only (the harness passes the checker a fixed environment): it exercises # the fallback path that TfP's runner takes, where the combined run crashes. full = None if os.environ.get("TFP_QCEC_SKIP_COMBINED") else run_qcec("all", remaining()) decided_by = "combined" if full: crit, secs = full["equivalence"], full["check_time"] else: # The combined run died. Decide with whichever single procedure survives, cheapest decisive one first: ZX proves # equivalence of Clifford+T rewrites fastest but can only confirm; simulation refutes fastest but can only # refute; alternating and construction are complete (equivalent / not_equivalent) and get the rest of the budget. crit, secs = "no_information", 0.0 caps = {"zx": 120.0, "simulation": 60.0, "alternating": None, "construction": None} for mode in ("zx", "simulation", "alternating", "construction"): if crit in ("equivalent", "equivalent_up_to_global_phase", "equivalent_up_to_phase", "not_equivalent"): break r = run_qcec(mode, remaining() if caps[mode] is None else min(caps[mode], remaining())) if not r: continue e = r["equivalence"]; secs += r["check_time"] if mode == "simulation" and e != "not_equivalent": continue if mode == "zx" and e not in ("equivalent", "equivalent_up_to_global_phase"): continue if e in ("equivalent", "equivalent_up_to_global_phase", "equivalent_up_to_phase", "not_equivalent"): crit, decided_by = e, mode ev = dict(base_ev, qcec_result=crit, qcec_seconds=round(secs, 3), qcec_attempts=attempts) diag = "; ".join(f"{a['mode']}: {a['outcome']}" for a in attempts if a.get("outcome") != "ok") if diag or decided_by != "combined": diag = (diag + "; " if diag else "") + f"decided by {decided_by}" ev["qcec_decided_by"] = decided_by if crit in ("equivalent", "equivalent_up_to_global_phase", "equivalent_up_to_phase"): score = cs[metric] if metric in cs else cs["t_count"] write("verified", f"submitted circuit is equivalent to the reference ({crit}); {metric} = {score} (reference {cr.get(metric)})" + (f" [qcec fallback: {diag}]" if diag else ""), score=float(score), evidence=ev) if crit == "not_equivalent": write("refuted", "submitted circuit is not equivalent to the reference circuit" + (f" [qcec fallback: {diag}]" if diag else ""), evidence=ev) if not full and all(a.get("outcome") != "ok" for a in attempts): write("inconclusive", f"the equivalence checker crashed in every configuration ({diag}); environment: {env_probe()}; last output: {(attempts[-1].get('tail') or '')[-700:]}", evidence=ev) write("inconclusive", f"equivalence could not be decided within the time budget ({crit})" + (f" [qcec fallback: {diag}]" if diag else ""), evidence=ev) def env_probe(): try: import platform, resource flags = next((l.split(":", 1)[1].split() for l in open("/proc/cpuinfo") if l.startswith("flags")), []) lim = {n: resource.getrlimit(getattr(resource, n))[0] for n in ("RLIMIT_STACK", "RLIMIT_AS", "RLIMIT_NPROC", "RLIMIT_DATA")} mem = open("/proc/meminfo").readline().split()[1] return f"{platform.machine()} cpus={os.cpu_count()} affinity={len(os.sched_getaffinity(0))} memkb={mem} limits={lim} avx={'avx' in flags} avx2={'avx2' in flags} avx512f={'avx512f' in flags} sse4_2={'sse4_2' in flags} fma={'fma' in flags} flags={len(flags)}" except Exception as e: return f"probe failed: {e}" QCEC_CHILD = r""" import sys, json, time, faulthandler faulthandler.enable() ref_path, sub_path, mode, budget = sys.argv[1], sys.argv[2], sys.argv[3], float(sys.argv[4]) from qiskit.qasm2 import loads, LEGACY_CUSTOM_INSTRUCTIONS ref = loads(open(ref_path).read(), custom_instructions=LEGACY_CUSTOM_INSTRUCTIONS) sub = loads(open(sub_path).read(), custom_instructions=LEGACY_CUSTOM_INSTRUCTIONS) print("qcec child: circuits loaded", flush=True) from mqt import qcec print("qcec child: mqt.qcec imported", getattr(qcec, "__version__", "?"), flush=True) on = lambda name: mode in ("all", name) kw = {"run_simulation_checker": on("simulation"), "run_alternating_checker": on("alternating"), "run_construction_checker": on("construction"), "run_zx_checker": on("zx")} t0 = time.time() try: res = qcec.verify(ref, sub, timeout=budget, **kw) except TypeError: res = qcec.verify(ref, sub, timeout=budget) crit = str(res.equivalence).split(".")[-1] print(json.dumps({"equivalence": crit, "check_time": round(float(res.check_time), 3), "mode": mode}), flush=True) """ if __name__ == "__main__": try: main() except SystemExit: raise except Exception as e: traceback.print_exc() write("inconclusive", f"checker error: {type(e).__name__}: {str(e)[:300]}")