#!/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, 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 from mqt import qcec remaining = max(5.0, min(eq_timeout, 560 - (time.time() - T0))) try: res = qcec.verify(ref_n, sub_n, timeout=remaining, run_simulation_checker=True, run_alternating_checker=True, run_construction_checker=True, run_zx_checker=True) except TypeError: res = qcec.verify(ref_n, sub_n, timeout=remaining) except Exception as e: write("inconclusive", f"equivalence checker failed: {str(e)[:300]}", 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": "error", "qcec_seconds": 0, "reference": cr, "submission": cs}) crit = str(res.equivalence).split(".")[-1] ev = {"qubits": cs["qubits"], "t_count": cs["t_count"], "cx_count": cs["two_qubit_count"], "total_gates": cs["total_gates"], "depth": cs["depth"], "qcec_result": crit, "qcec_seconds": round(res.check_time, 3), "reference": cr, "submission": cs} 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)})", score=float(score), evidence=ev) if crit == "not_equivalent": write("refuted", "submitted circuit is not equivalent to the reference circuit", evidence=ev) write("inconclusive", f"equivalence could not be decided within the time budget ({crit})", evidence=ev) 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]}")