#!/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)