"""Board measured-mcz (site key mcz): Clifford+T circuits with measurements and feedforward that implement the multi-controlled-Z gate U_N|x> = (-1)^(x_0 x_1 ... x_{N-1})|x> on N data qubits (SPEC_v4 s4, review_v4_fable s4 M1-M4). Artifact {"ancillas": a, "ops": [...]}. Wires 0..N-1 are data (unknown input), N..N+a-1 are workspace, live and |0> at the start. Instructions (exact JSON arrays): X Z H S SDG T TDG [name, q]; CX CZ [name, p, q] (distinct wires); MX MZ [name, q] (destructive, live workspace only; one record bit, 0 = +1 eigenvalue); RZ [name, q] (dead workspace only, prepares |0>); IF [bits, value, gate] (gate is one unitary of the nine above, run iff the XOR of the listed earlier records equals value; an empty list is the constant 0). Measurements and RZ are unconditional, so liveness and record numbering are static. Every unitary operand must be live whether or not an IF fires. Accept iff for every complete record r (all 2^M of them, zero-probability ones included) the branch map K_r equals c_r (U_N tensor |0..0>_live) for one scalar c_r that does not depend on the data input, and sum |c_r|^2 = 1 (checked as an invariant). Score (t, gates, ancillas): t = largest executed T/TDG count over the records with c_r != 0, gates = len(ops), ancillas = declared workspace count; all three minimized. Exact method: amplitudes are (a0 + a1 z + a2 z^2 + a3 z^3) / sqrt(2)^h with z = e^(i pi/4), z^4 = -1, four Python ints and a branch exponent h that depends only on the record (H and MX are run or not by the record alone). Every one of the 2^N basis columns is simulated depth-first over all measurement branches (sparse dict basis index -> 4-tuple; a dead wire's index bit is 0 and only RZ revives it). Column 0 fixes c_r; every other column must equal c_r * (+-1) as four integers at equal h, nothing is normalized or reduced. By linearity the columns determine each K_r, and the sign (-1) is on the all-ones input only. Budget: process time (from the start of the call) is polled every CHECK_EVERY simulated instructions; running out returns status error. """ import os import time from common import Reject, exact_keys, want_int, want_list, is_int, load_instance, short BOARD = "measured-mcz" CHECK_EVERY = 1024 clock = time.process_time ONE_Q = ("X", "Z", "H", "S", "SDG", "T", "TDG") TWO_Q = ("CX", "CZ") UNITARY = ONE_Q + TWO_Q MEAS = ("MX", "MZ") PHASE_K = {"Z": 4, "S": 2, "SDG": 6, "T": 1, "TDG": 7} # multiply by z^k when the wire is 1 INSTANCE_KEYS = ["board", "instance", "mode", "cpu_seconds", "mem_mb", "n_data", "ancillas_max", "measurements_max", "ops_max"] class Budget(Exception): pass # ---- exact arithmetic in Z[z]/(z^4 + 1): 4-tuples, multiplication by z^k is a signed rotation ---- def _neg(v): return (-v[0], -v[1], -v[2], -v[3]) ROT = { 1: lambda v: (-v[3], v[0], v[1], v[2]), 2: lambda v: (-v[2], -v[3], v[0], v[1]), 3: lambda v: (-v[1], -v[2], -v[3], v[0]), 4: _neg, 5: lambda v: (v[3], -v[0], -v[1], -v[2]), 6: lambda v: (v[2], v[3], -v[0], -v[1]), 7: lambda v: (v[1], v[2], v[3], -v[0]), } ZERO = (0, 0, 0, 0) ONE = (1, 0, 0, 0) def _norm_sq(c): """|c|^2 = P + sqrt(2) Q for c = a0 + a1 z + a2 z^2 + a3 z^3; returns (P, Q).""" a0, a1, a2, a3 = c return a0 * a0 + a1 * a1 + a2 * a2 + a3 * a3, a0 * a1 + a1 * a2 + a2 * a3 - a3 * a0 def _wire(q, W, what): if not is_int(q): raise Reject(f"{what}: wire must be an integer, got {short(q)}") if not 0 <= q < W: raise Reject(f"{what}: wire {q} is not declared (wires are 0..{W - 1})") return q def _unitary(g, W, live, where): """Parse one unitary instruction; returns a compiled tuple. Operands must be live.""" if not isinstance(g, list) or not g or not isinstance(g[0], str): raise Reject(f"{where}: an instruction is a non-empty array starting with a name") name = g[0] if name in ONE_Q: if len(g) != 2: raise Reject(f"{where}: {name} takes exactly 1 wire") q = _wire(g[1], W, where) if q not in live: raise Reject(f"{where}: {name} on dead wire {q}") m = 1 << q if name == "X": return ("X", m) if name == "H": return ("H", m) return ("P", PHASE_K[name], m, name) if name in TWO_Q: if len(g) != 3: raise Reject(f"{where}: {name} takes exactly 2 wires") a, b = _wire(g[1], W, where), _wire(g[2], W, where) if a == b: raise Reject(f"{where}: {name} needs two distinct wires") for q in (a, b): if q not in live: raise Reject(f"{where}: {name} on dead wire {q}") return (name, 1 << a, 1 << b) if name in MEAS or name in ("RZ", "IF"): raise Reject(f"{where}: {name} is not allowed here") raise Reject(f"{where}: unknown instruction {short(name, 20)}") def parse_ops(ops, N, a, mmax): """Static checks and compilation. Returns (compiled ops, number of records).""" W = N + a live = set(range(W)) nrec = 0 out = [] for i, op in enumerate(ops): where = f"op {i}" if not isinstance(op, list) or not op or not isinstance(op[0], str): raise Reject(f"{where}: an instruction is a non-empty array starting with a name") name = op[0] if name in MEAS: if len(op) != 2: raise Reject(f"{where}: {name} takes exactly 1 wire") q = _wire(op[1], W, where) if q < N: raise Reject(f"{where}: data wire {q} cannot be measured") if q not in live: raise Reject(f"{where}: {name} on dead wire {q}") if nrec >= mmax: raise Reject(f"{where}: more than {mmax} measurements") live.discard(q) out.append((name, 1 << q, nrec)) nrec += 1 elif name == "RZ": if len(op) != 2: raise Reject(f"{where}: RZ takes exactly 1 wire") q = _wire(op[1], W, where) if q < N: raise Reject(f"{where}: data wire {q} cannot be reset") if q in live: raise Reject(f"{where}: RZ on live wire {q} (RZ prepares dead workspace only)") live.add(q) out.append(("RZ",)) elif name == "IF": if len(op) != 4: raise Reject(f"{where}: IF takes [bits, value, gate]") bits = want_list(op[1], f"{where}: IF bits") prev = -1 mask = 0 for b in bits: if not is_int(b) or not prev < b < nrec: raise Reject(f"{where}: IF bits must be strictly increasing earlier record indices (records so far: " f"{nrec}), got {short(bits)}") prev = b mask |= 1 << b if not is_int(op[2]) or op[2] not in (0, 1): raise Reject(f"{where}: IF value must be the integer 0 or 1, got {short(op[2])}") out.append(("IF", mask, op[2], _unitary(op[3], W, live, where + " gate"))) else: out.append(_unitary(op, W, live, where)) return out, nrec def t_count(ops, rec): """Executed T/TDG gates on the record rec (bit k = record k).""" n = 0 for op in ops: if op[0] == "IF": if (bin(rec & op[1]).count("1") & 1) == op[2]: op = op[3] else: continue if op[0] == "P" and op[3] in ("T", "TDG"): n += 1 return n # ---- state updates: st is a dict basis index -> nonzero 4-tuple ---- def _apply_h(st, m): out = {} nm = ~m for b in {i & nm for i in st}: v0 = st.get(b) v1 = st.get(b | m) if v1 is None: out[b] = v0 out[b | m] = v0 elif v0 is None: out[b] = v1 out[b | m] = (-v1[0], -v1[1], -v1[2], -v1[3]) else: s = (v0[0] + v1[0], v0[1] + v1[1], v0[2] + v1[2], v0[3] + v1[3]) d = (v0[0] - v1[0], v0[1] - v1[1], v0[2] - v1[2], v0[3] - v1[3]) if s != ZERO: out[b] = s if d != ZERO: out[b | m] = d return out def _apply_mx(st, m, r): out = {} nm = ~m for b in {i & nm for i in st}: v0 = st.get(b) v1 = st.get(b | m) if v1 is None: w = v0 elif v0 is None: w = _neg(v1) if r else v1 elif r: w = (v0[0] - v1[0], v0[1] - v1[1], v0[2] - v1[2], v0[3] - v1[3]) else: w = (v0[0] + v1[0], v0[1] + v1[1], v0[2] + v1[2], v0[3] + v1[3]) if w != ZERO: out[b] = w return out def _apply_mz(st, m, r): if r: return {i ^ m: v for i, v in st.items() if i & m} return {i: v for i, v in st.items() if not i & m} def _apply_unitary(st, op): c = op[0] if c == "P": f, m = ROT[op[1]], op[2] return {i: (f(v) if i & m else v) for i, v in st.items()} if c == "CX": mc, mt = op[1], op[2] return {(i ^ mt if i & mc else i): v for i, v in st.items()} if c == "CZ": mm = op[1] | op[2] return {i: (_neg(v) if i & mm == mm else v) for i, v in st.items()} if c == "X": m = op[1] return {i ^ m: v for i, v in st.items()} return _apply_h(st, op[1]) class Run: def __init__(self, ops, N, nrec, limit): self.ops, self.N, self.nrec, self.limit = ops, N, nrec, limit self.steps = 0 self.next_check = CHECK_EVERY def tick(self): self.steps += 1 if self.steps >= self.next_check: self.next_check = self.steps + CHECK_EVERY if clock() > self.limit: raise Budget() def go(self, pc, st, h, rec, leaf): ops = self.ops n = len(ops) while pc < n: op = ops[pc] pc += 1 self.tick() c = op[0] if c == "MX" or c == "MZ": for r in (0, 1): ns = _apply_mx(st, op[1], r) if c == "MX" else _apply_mz(st, op[1], r) self.go(pc, ns, h + (c == "MX"), rec | (r << op[2]), leaf) return if c == "RZ": continue if c == "IF": if (bin(rec & op[1]).count("1") & 1) != op[2]: continue op = op[3] c = op[0] st = _apply_unitary(st, op) if c == "H": h += 1 leaf(rec, st, h) def _bits(rec, M): return "".join(str(rec >> k & 1) for k in range(M)) or "(none)" def check(I, art, time_budget=None): """verify() on an instance given as a loaded dict (tests pass small ad hoc instances this way).""" try: exact_keys(I, INSTANCE_KEYS, where="instance") N = want_int(I["n_data"], 1, 8, "instance n_data") amax = want_int(I["ancillas_max"], 0, 8, "instance ancillas_max") mmax = want_int(I["measurements_max"], 0, 8, "instance measurements_max") omax = want_int(I["ops_max"], 0, 4096, "instance ops_max") cpu = want_int(I["cpu_seconds"], 1, 10 ** 6, "instance cpu_seconds") if I["board"] != BOARD: raise Reject("instance board mismatch") except Reject as e: return {"status": "error", "detail": {"reason": f"invalid instance: {e}"}} if not isinstance(art, dict): raise Reject("artifact must be a JSON object") exact_keys(art, ["ancillas", "ops"]) a = want_int(art["ancillas"], 0, amax, "ancillas") ops = want_list(art["ops"], "ops", max_len=omax) cops, M = parse_ops(ops, N, a, mmax) # CPU budget in seconds from now (a fresh process per artifact in production; relative so that a long-lived caller, # such as the cross-check workers, is not charged for its earlier work) budget = float(os.environ.get("QARENA_CPU", cpu)) - 30.0 if time_budget is None else time_budget run = Run(cops, N, M, clock() + budget) NR = 1 << M full = (1 << N) - 1 cand = {} # record -> (h, scalar 4-tuple or None) def leaf0(rec, st, h): if len(st) > 1 or (st and 0 not in st): raise Reject(f"record {_bits(rec, M)}, input 0: the final state is not a multiple of |0...0>: data or live " "workspace was left changed or entangled (workspace must return to |0>)") cand[rec] = (h, st[0] if st else None) def make_leaf(x): sgn = x == full def leaf(rec, st, h): h0, c = cand[rec] if h != h0: raise RuntimeError("branch exponent depends on the input") if any(i != x for i in st): raise Reject(f"record {_bits(rec, M)}, input {x}: the final state has amplitude outside |x, 0...0>: the " "data basis state changed or live workspace is not |0>") v = st.get(x) if c is None: if v is not None: raise Reject(f"record {_bits(rec, M)}, input {x}: nonzero amplitude on a record whose input-0 " "amplitude is zero (the branch probability depends on the input)") elif v != (_neg(c) if sgn else c): if v is None or _norm_sq(v) != _norm_sq(c): raise Reject(f"record {_bits(rec, M)}, input {x}: the branch probability depends on the input " "(its amplitude has a different modulus than on input 0)") raise Reject(f"record {_bits(rec, M)}, input {x}: amplitude is not the branch scalar times " "(-1)^(x_0...x_{N-1}) (wrong phase: a residual data phase or a missing correction)") return leaf try: run.go(0, {0: ONE}, 0, 0, leaf0) if len(cand) != NR: raise RuntimeError("record enumeration incomplete") for x in range(1, 1 << N): run.go(0, {x: ONE}, 0, 0, make_leaf(x)) except Budget: return {"status": "error", "detail": {"reason": "verification did not finish within the CPU budget " "(not ranked)", "simulated_instructions": run.steps}} # invariant: sum |c_r|^2 = 1, every branch scaled to the largest exponent hmax = max(h for h, _ in cand.values()) P = Q = 0 for h, c in cand.values(): if c is not None: p, q = _norm_sq(c) P += p << (hmax - h) Q += q << (hmax - h) if P != 1 << hmax or Q != 0: raise RuntimeError("branch normalization invariant violated") nonzero = [rec for rec, (_, c) in cand.items() if c is not None] t = max(t_count(cops, rec) for rec in nonzero) return {"status": "accepted", "score": {"t": t, "gates": len(ops), "ancillas": a}, "detail": {"n_data": N, "t": t, "gates": len(ops), "ancillas": a, "measurements": M, "records": NR, "records_nonzero": len(nonzero), "simulated_instructions": run.steps, "check": f"all {1 << N} inputs x {NR} measurement records equal c_r U_N on data with live " "workspace |0>, sum |c_r|^2 = 1"}} def verify(inst, art, time_budget=None): return check(load_instance("mcz", inst), art, time_budget)