b9af554c7c
Error correction and mitigation join the version contract. An error-correction answer (wai.quantum.qec/1: a code, a decode or a threshold), a code's distance certificate (wai.quantum.distance/1) and a job's mitigation (wai.quantum.mitigation/1) are engine schemas; every answer engine::qec::call gives names the engine and its schema, inside what a host seals. The job request names mitigate, and a compile's options and certificate twirl and dd. qec_code, qec_decode and qec_threshold answer to output schemas on both servers; the strict tests call each, a mitigated run and a compile with both passes, the engine's test holds every decoder's answer and each mitigation method to its schema, the contract workflow requires it by name, and check_schemas.py holds the command line's answers too. Refusal codes: the set branch and this one had each taken J007. A set that does not read keeps J007; mitigation's refusals are J008 and error correction's J009, in the table, its page, the tests, the docs and the changelog. The basis rename reaches the new passes: compile::Options carries twirl and dd beside basis, and the command line's compile, both servers' compile options and their tests name the gate set basis. The service smoke test made 32 submissions with both branches' checks, past the 30 a client may make in a window (it is also run against production after a deploy). It now makes 28 (29 on a host with a GPU, whose single-precision run is held to the exact values instead of a second run): the code-capacity decode against its error, the changed estimate, the counts document and the classical shortcuts are left to the MCP servers' tests, which hold each. The agent docs are regenerated. Co-Authored-By: Claude Opus 5.5 <[email protected]>
546 lines
26 KiB
Python
546 lines
26 KiB
Python
"""The engine in-process: each function, its answers and its refusals."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import math
|
||
import pickle
|
||
import xml.etree.ElementTree as ET
|
||
|
||
import pytest
|
||
|
||
import wai_quantum as wq
|
||
from conftest import BELL, REPO, bare
|
||
|
||
DEMO = wq.Key.demo()
|
||
|
||
|
||
# -- run ----------------------------------------------------------------------
|
||
|
||
|
||
def test_a_bell_pair_runs_on_the_stabilizer_method_and_its_receipt_verifies():
|
||
job = wq.run(BELL, shots=1000, seed=7, key=DEMO)
|
||
assert isinstance(job, wq.Job) and isinstance(job, dict)
|
||
assert job.status == "done" and job.method == "stabilizer" and job.accuracy == "exact"
|
||
assert set(job.counts) <= {"00", "11"} and job.shots == 1000
|
||
assert job["request"] == {"qasm": BELL, "shots": 1000, "seed": 7}
|
||
assert job.result["energy"] == {"unmetered": wq.UNMETERED}
|
||
v = wq.verify(job)
|
||
assert v and v.valid and v.key == DEMO.public_key and v.signer == DEMO.did
|
||
# The job is plain JSON: written and read back, it still verifies.
|
||
again = json.loads(json.dumps(job))
|
||
assert wq.verify(again).valid and job.verify().valid
|
||
|
||
|
||
def test_the_same_job_gives_the_same_answer():
|
||
a = wq.run(BELL, shots=500, seed=3, key=DEMO)
|
||
b = wq.run({"qasm": BELL, "shots": 500, "seed": 3}, key=DEMO)
|
||
assert bare(a["result"]) == bare(b["result"])
|
||
assert {k: v for k, v in a.items() if k != "result"} == {k: v for k, v in b.items() if k != "result"}
|
||
# Another seed samples other counts from the same distribution.
|
||
c = wq.run(BELL, shots=500, seed=4, key=DEMO)
|
||
assert c.counts != a.counts and set(c.counts) <= {"00", "11"}
|
||
|
||
|
||
def test_helpers_read_counts_frequencies_expectations_and_amplitudes():
|
||
src = "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];"
|
||
job = wq.run(src, shots=2000, seed=5, observables=["Z0 Z1", {"pauli": "X0 X1", "coeff": 0.5}, "Z0"], amplitudes=["00", "11", "01"], key=DEMO)
|
||
assert job.shots == 2000
|
||
p = job.probabilities()
|
||
assert math.isclose(sum(p.values()), 1.0) and set(p) <= {"00", "11"}
|
||
e = job.expectations
|
||
assert e["Z0 Z1"] == pytest.approx(1.0) and e["X0 X1"] == pytest.approx(0.5) and abs(e["Z0"]) < 1e-12
|
||
a = job.amplitudes
|
||
assert abs(a["00"] - 2 ** -0.5) < 1e-6 and abs(a["11"] - 2 ** -0.5) < 1e-6 and a["01"] == 0
|
||
assert "sealed" in job.summary()
|
||
|
||
|
||
def test_an_arbitrary_angle_runs_on_an_f64_method_and_says_so():
|
||
# 0.3 rad is no dyadic multiple of 2π: the bit-exact method refuses it,
|
||
# and the planner sends it to the double-precision state vector.
|
||
src = "OPENQASM 3.0; qubit[1] q; ry(0.3) q[0];"
|
||
job = wq.run(src, observables=["Z0", "X0"], key=DEMO)
|
||
assert job.method == "statevector" and job.accuracy == "exact"
|
||
assert job.expectations["Z0"] == pytest.approx(math.cos(0.3), abs=1e-12)
|
||
assert job.expectations["X0"] == pytest.approx(math.sin(0.3), abs=1e-12)
|
||
with pytest.raises(wq.EngineError):
|
||
wq.run(src, observables=["Z0"], method="exact", key=DEMO)
|
||
|
||
|
||
def test_a_dyadic_circuit_runs_bit_exact_when_asked():
|
||
job = wq.run("OPENQASM 3.0; qubit[2] q; h q[0]; t q[0]; cx q[0], q[1];", method="exact", shots=100, key=DEMO)
|
||
assert job.method == "exact" and job.accuracy == "bit-exact" and len(job.result["state_hash"]) == 64
|
||
|
||
|
||
def test_a_noisy_run_names_its_method_and_its_noise_lowers_the_correlation():
|
||
src = "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];"
|
||
clean = wq.run(src, observables=["Z0 Z1"], key=DEMO)
|
||
noisy = wq.run(src, observables=["Z0 Z1"], noise={"depolarizing_2q": 0.1}, method="density_matrix", key=DEMO)
|
||
assert noisy.method == "density_matrix" and noisy.accuracy == "exact"
|
||
# A two-qubit depolarizing channel of strength p after the CX scales ⟨Z0 Z1⟩ by 1 − 16p/15.
|
||
assert noisy.expectations["Z0 Z1"] == pytest.approx(clean.expectations["Z0 Z1"] * (1 - 16 * 0.1 / 15), abs=1e-9)
|
||
|
||
|
||
def test_a_dynamic_program_runs_branch_by_branch():
|
||
src = "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; c[0] = measure q[0]; if (c[0]) x q[1]; c[1] = measure q[1];"
|
||
job = wq.run(src, shots=400, seed=2, key=DEMO)
|
||
assert set(job.counts) <= {"00", "11"} and job.shots == 400
|
||
|
||
|
||
def test_a_key_may_be_given_as_a_key_a_seed_or_hex_and_defaults_to_the_users():
|
||
k = wq.Key.generate()
|
||
for given in (k, k.seed, k.seed.hex()):
|
||
assert wq.verify(wq.run(BELL, shots=10, key=given)).key == k.public_key
|
||
u = wq.run(BELL, shots=10)
|
||
assert wq.verify(u).key == wq.Key.user().public_key
|
||
with pytest.raises(ValueError):
|
||
wq.run(BELL, key=b"short")
|
||
with pytest.raises(TypeError):
|
||
wq.run(BELL, key=3.0)
|
||
|
||
|
||
def test_a_program_may_be_a_path(tmp_path):
|
||
f = tmp_path / "bell.qasm"
|
||
f.write_text(BELL)
|
||
assert wq.run(f, shots=10, seed=1, key=DEMO).counts == wq.run(BELL, shots=10, seed=1, key=DEMO).counts
|
||
assert wq.lint(f)["ok"] is True
|
||
|
||
|
||
# -- refusals -----------------------------------------------------------------
|
||
|
||
|
||
def test_a_refusal_raises_the_engines_message_with_its_line():
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", key=DEMO)
|
||
assert e.value.line == 3 and "foo" in e.value.message and e.value.code == "Q103"
|
||
assert str(e.value) == e.value.message
|
||
# It survives pickling (a process pool carries it back).
|
||
back = pickle.loads(pickle.dumps(e.value))
|
||
assert (back.message, back.line, back.code) == (e.value.message, 3, "Q103")
|
||
with pytest.raises(wq.EngineError, match="asks for nothing"):
|
||
wq.run(BELL, shots=0, key=DEMO)
|
||
with pytest.raises(wq.EngineError, match="no such channel"):
|
||
wq.run(BELL, noise={"depolarising": 0.1}, key=DEMO)
|
||
|
||
|
||
def test_values_json_cannot_carry_are_refused_not_guessed():
|
||
with pytest.raises(ValueError, match="NaN"):
|
||
wq.run(BELL, options={"x": float("nan")}, key=DEMO)
|
||
with pytest.raises(ValueError, match="64 bits"):
|
||
wq.run(BELL, seed=2 ** 70, key=DEMO)
|
||
# Past 2^53 a double cannot hold it: the engine refuses it by its path.
|
||
with pytest.raises(wq.EngineError, match="seed"):
|
||
wq.run(BELL, seed=2 ** 53 + 1, key=DEMO)
|
||
with pytest.raises(TypeError, match="keys are strings"):
|
||
wq.run({"qasm": BELL, 1: 2}, key=DEMO)
|
||
with pytest.raises(TypeError, match="not a JSON value"):
|
||
wq.run(BELL, options={"x": object()}, key=DEMO)
|
||
|
||
|
||
# -- plan, lint, draw, layout -----------------------------------------------------
|
||
|
||
|
||
def test_plan_names_the_method_and_why_without_running():
|
||
p = wq.plan(BELL)
|
||
assert p["method"] == "stabilizer" and p["qubits"] == 2 and "Clifford" in p["why"]
|
||
assert set(p) == {"method", "why", "memory_bytes", "work", "qubits"}
|
||
assert wq.plan(BELL, method="statevector")["method"] == "statevector"
|
||
with pytest.raises(wq.EngineError):
|
||
wq.plan("qubit[1] q; bad q[0];")
|
||
|
||
|
||
def test_lint_gives_the_shape_or_the_problem_and_its_line():
|
||
assert wq.lint(BELL) == {"ok": True, "qubits": 2, "gates": 2, "bits": 2, "measured": 2, "dynamic_steps": None}
|
||
bad = wq.lint("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n")
|
||
assert bad["ok"] is False and bad["line"] == 3
|
||
|
||
|
||
def test_draw_writes_the_text_drawing_and_the_svg():
|
||
t = wq.draw(BELL)
|
||
assert t.splitlines()[0].startswith("q0: ") and "H" in t and "⊕" in t and "M1" in t
|
||
svg = wq.draw(BELL, format="svg")
|
||
root = ET.fromstring(svg)
|
||
assert root.tag == "{http://www.w3.org/2000/svg}svg"
|
||
items = [g for g in root.iter("{http://www.w3.org/2000/svg}g") if "qc-item" in g.get("class", "").split()]
|
||
assert len(items) == 4 and [g.get("data-op") for g in items if g.get("data-op")] == ["0", "1"]
|
||
assert wq.draw(BELL, format="svg") == svg
|
||
with pytest.raises(ValueError):
|
||
wq.draw(BELL, format="png")
|
||
with pytest.raises(ValueError):
|
||
wq.draw(BELL, hands=["person"])
|
||
with pytest.raises(wq.EngineError):
|
||
wq.draw("qubit[1] q; nope q[0];")
|
||
|
||
|
||
def test_svg_marks_each_gate_with_its_hand():
|
||
svg = wq.draw("OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", format="svg", hands=["origin", "assistant"])
|
||
assert 'class="qc-item qc-by-origin" data-op="0"' in svg and 'class="qc-item qc-by-assistant" data-op="1"' in svg
|
||
assert ">the assistant: 1 gate</text>" in svg
|
||
|
||
|
||
def test_layout_is_the_drawing_as_data():
|
||
lay = wq.layout(BELL)
|
||
assert lay["qubits"] == 2 and lay["columns"][1][0] == {"cells": [[0, "●"], [1, "⊕"]], "joined": True, "op": 1}
|
||
|
||
|
||
# -- critique, improve, shortcuts, equivalent ------------------------------------
|
||
|
||
|
||
def test_critique_gives_measures_and_smells_and_refuses_what_does_not_read():
|
||
# A fix is offered for gates on lines of their own.
|
||
c = wq.critique("OPENQASM 3.0;\nqubit[1] q;\nh q[0];\nh q[0];\nx q[0];\n")
|
||
assert c["gates"] == 3 and [(s["kind"], s["lines"], s["fixable"]) for s in c["smells"]] == [("cancels", [3, 4], True)]
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.critique("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];")
|
||
assert e.value.line == 3
|
||
|
||
|
||
def test_improve_offers_only_variants_proved_equivalent():
|
||
v = wq.improve("OPENQASM 3.0;\nqubit[1] q;\nh q[0];\nh q[0];\nx q[0];\n")
|
||
best = v["candidates"][0]
|
||
assert best["measures"]["gates"] < v["base"]["gates"] and best["equivalence"]["verdict"] == "equivalent"
|
||
assert wq.equivalent(best["qasm"], "OPENQASM 3.0; qubit[1] q; x q[0];")["verdict"] == "equivalent"
|
||
|
||
|
||
def test_shortcuts_names_a_cheap_route_for_a_clifford_circuit():
|
||
s = wq.shortcuts(BELL)
|
||
assert s["ok"] is True and s["cheapest"]
|
||
|
||
|
||
def test_equivalent_decides_and_says_how():
|
||
assert wq.equivalent("qubit[1] q; h q[0]; h q[0];", "qubit[1] q;")["how"] == "tableau"
|
||
ex = wq.equivalent("qubit[1] q; t q[0]; t q[0];", "qubit[1] q; s q[0];")
|
||
assert ex["verdict"] == "equivalent" and ex["how"] == "exact"
|
||
assert wq.equivalent("qubit[1] q; x q[0];", "qubit[1] q; z q[0];")["verdict"] == "different"
|
||
assert wq.equivalent("qubit[1] q; x q[0];", "qubit[2] q;")["how"] == "shape"
|
||
assert wq.equivalent("qubit[1] q; nope q[0];", "qubit[1] q;")["verdict"] == "unchecked"
|
||
|
||
|
||
# -- receipts -------------------------------------------------------------------
|
||
|
||
|
||
def test_any_change_to_a_sealed_job_breaks_its_receipt():
|
||
job = wq.run(BELL, shots=100, seed=1, key=DEMO)
|
||
for path, value in ((("result", "counts", "00"), 1), (("request", "shots"), 99), (("result", "method"), "statevector"), (("result", "energy", "unmetered"), "x")):
|
||
forged = json.loads(json.dumps(job))
|
||
d = forged
|
||
for k in path[:-1]:
|
||
d = d[k]
|
||
d[path[-1]] = value
|
||
assert not wq.verify(forged), path
|
||
no_receipt = json.loads(json.dumps(job))
|
||
del no_receipt["result"]["receipt"]
|
||
with pytest.raises(wq.EngineError, match="no receipt"):
|
||
wq.verify(no_receipt)
|
||
|
||
|
||
def test_seal_and_verify_round_trip_and_the_digest_is_the_receipts():
|
||
k = wq.Key.generate()
|
||
req = {"qasm": BELL, "shots": 10}
|
||
res = {"method": "stabilizer", "counts": {"00": 6, "11": 4}, "value": 0.1}
|
||
r = wq.seal(req, res, key=k)
|
||
assert r["alg"] == "ed25519" and r["canon"] == "rfc8785" and r["key"] == k.public_key
|
||
assert r["digest"] == wq.digest(req, res)
|
||
sealed = {"request": req, "result": dict(res, receipt=r)}
|
||
assert wq.verify(sealed).valid
|
||
# A receipt already on the result is not part of what is sealed.
|
||
assert wq.seal(req, dict(res, receipt={"old": 1}), key=k) == r
|
||
|
||
|
||
def test_verify_reads_a_dict_json_text_or_a_file(tmp_path):
|
||
job = wq.run(BELL, shots=10, key=DEMO)
|
||
f = tmp_path / "job.json"
|
||
f.write_text(json.dumps(job))
|
||
assert wq.verify(job) and wq.verify(json.dumps(job)) and wq.verify(f) and wq.verify(str(f))
|
||
|
||
|
||
def test_canonical_json_is_rfc_8785():
|
||
# Integers and strings: the standard library writes the same text when it
|
||
# sorts keys and drops whitespace.
|
||
v = {"z": [3, -2, {"b": True, "a": None}], "a": "x\n\u0001\"\\é", "m": 9007199254740991}
|
||
assert wq.canonical(v) == json.dumps(v, sort_keys=True, separators=(",", ":"), ensure_ascii=False).encode()
|
||
# Numbers as ECMAScript writes them; keys by UTF-16 code units.
|
||
assert wq.canonical({"b": 1.0, "a": [1e21, 1e-7, 0.000001, -0.0, 123e-20]}) == b'{"a":[1e+21,1e-7,0.000001,0,1.23e-18],"b":1}'
|
||
assert wq.canonical({"": 1, "\U0001f600": 2}) == '{"\U0001f600":2,"":1}'.encode()
|
||
|
||
|
||
def test_every_number_reaches_the_engine_as_the_double_it_is():
|
||
"""The engine's ECMAScript number vectors, each sent from Python as a
|
||
float: the canonical text is the vector's, so the float crossed into the
|
||
engine bit for bit."""
|
||
path = REPO / "wai" / "wai-quantum" / "src" / "engine" / "testdata" / "es-number-tostring.txt"
|
||
n = 0
|
||
for line in path.read_text().splitlines():
|
||
bits, want = line.split(" ", 1)
|
||
x = _from_bits(int(bits, 16))
|
||
if math.isnan(x) or math.isinf(x):
|
||
continue
|
||
assert wq.canonical(x) == want.encode(), bits
|
||
n += 1
|
||
assert n > 3000
|
||
|
||
|
||
def _from_bits(b: int) -> float:
|
||
import struct
|
||
|
||
return struct.unpack(">d", b.to_bytes(8, "big"))[0]
|
||
|
||
|
||
# -- keys -------------------------------------------------------------------------
|
||
|
||
|
||
def test_keys_name_their_signer_and_files_are_the_command_lines_form(tmp_path):
|
||
assert DEMO.is_demo and DEMO.did.startswith("did:key:z6Mk") and len(DEMO.public_key) == 64
|
||
k = wq.Key.generate()
|
||
p = k.save(tmp_path / "k")
|
||
assert p.read_text() == k.seed.hex() + "\n" and (p.stat().st_mode & 0o777) == 0o600 or __import__("os").name == "nt"
|
||
assert wq.Key.from_file(p) == k and wq.Key.from_hex(k.seed.hex()) == k
|
||
with pytest.raises(FileExistsError):
|
||
k.save(p)
|
||
(tmp_path / "bad").write_text("not a key")
|
||
with pytest.raises(ValueError):
|
||
wq.Key.from_file(tmp_path / "bad")
|
||
|
||
|
||
def test_the_users_key_is_made_once_and_a_named_file_must_exist(tmp_path, monkeypatch):
|
||
a = wq.Key.user()
|
||
assert a.path is not None and a.path.exists()
|
||
assert wq.Key.user() == a
|
||
monkeypatch.setenv("WAI_QUANTUM_KEY", str(tmp_path / "missing"))
|
||
with pytest.raises(FileNotFoundError):
|
||
wq.Key.user()
|
||
|
||
|
||
# -- the catalogue, the history, Bloch vectors --------------------------------------
|
||
|
||
|
||
def test_methods_and_accuracy_classes_are_the_catalogues():
|
||
ms = {m["method"]: m for m in wq.methods()}
|
||
assert {"exact", "statevector", "stabilizer", "pauli", "mps", "density_matrix", "trajectories"} <= set(ms)
|
||
assert ms["exact"]["accuracy"] == "bit-exact"
|
||
assert set(wq.accuracy_classes()) >= {"bit-exact", "exact", "bounded", "estimate", "heuristic"}
|
||
|
||
|
||
def test_a_history_records_versions_and_blames_each_gate():
|
||
bell = "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n"
|
||
k = wq.Key.generate()
|
||
h = wq.history("start", {"source": bell, "by": {"kind": "origin", "name": "example"}, "at": 1}, key=k)
|
||
h = wq.history("propose", {"history": h["history"], "parent": "v0", "proposed": bell + "x q[1];\n", "accepted": [0], "by": {"kind": "assistant", "name": "m"}, "at": 2}, key=k)
|
||
b = wq.history("blame", {"history": h["history"], "version": "v1"})
|
||
assert b["gates"] == ["origin", "origin", "assistant"]
|
||
log = wq.history("log", {"history": h["history"]})
|
||
assert log["check"]["ok"] is True
|
||
with pytest.raises(wq.EngineError):
|
||
wq.history("nope", {})
|
||
|
||
|
||
def test_bloch_vectors_are_the_engines_expectation_values():
|
||
cases = [
|
||
("qubit[1] q;", (0, 0, 1)),
|
||
("qubit[1] q; x q[0];", (0, 0, -1)),
|
||
("qubit[1] q; h q[0];", (1, 0, 0)),
|
||
("qubit[1] q; h q[0]; s q[0];", (0, 1, 0)),
|
||
("qubit[1] q; ry(0.7) q[0];", (math.sin(0.7), 0, math.cos(0.7))),
|
||
("qubit[1] q; rx(1.1) q[0];", (0, -math.sin(1.1), math.cos(1.1))),
|
||
]
|
||
for src, want in cases:
|
||
b = wq.bloch("OPENQASM 3.0; " + src, key=DEMO)
|
||
(q,) = b["qubits"]
|
||
assert (q["x"], q["y"], q["z"]) == pytest.approx(want, abs=1e-12), src
|
||
assert q["length"] == pytest.approx(1.0, abs=1e-12)
|
||
# Entangled: each qubit alone is maximally mixed.
|
||
ent = wq.bloch("OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", key=DEMO)
|
||
assert [q["length"] for q in ent["qubits"]] == pytest.approx([0, 0], abs=1e-12)
|
||
assert ent["method"] == "stabilizer" and ent["job"].verify().valid
|
||
# Noise shortens the vector: dephasing p scales x by 1 − 2p.
|
||
noisy = wq.bloch("OPENQASM 3.0; qubit[1] q; h q[0];", noise={"dephasing": 0.1}, method="density_matrix", key=DEMO)
|
||
assert noisy["qubits"][0]["x"] == pytest.approx(1 - 2 * 0.1, abs=1e-9)
|
||
# The method bounds its values: each vector carries the bounds.
|
||
bounded = wq.bloch("OPENQASM 3.0; qubit[1] q; ry(0.4) q[0];", method="pauli", key=DEMO)
|
||
assert bounded["accuracy"] == "bounded" and len(bounded["qubits"][0]["bound"]) == 3
|
||
assert [q["qubit"] for q in wq.bloch("OPENQASM 3.0; qubit[3] q;", qubits=[2], key=DEMO)["qubits"]] == [2]
|
||
|
||
|
||
# -- estimate -----------------------------------------------------------------
|
||
|
||
LADDER = "OPENQASM 3.0;\nqubit[4] q;\nh q;\nctrl(3) @ x q[0], q[1], q[2], q[3];\nrz(0.3) q[0];\nt q[1];\n"
|
||
|
||
|
||
def test_an_estimate_counts_the_program_seals_every_figure_and_re_runs():
|
||
e = wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=64, key=DEMO)
|
||
r = e["result"]
|
||
assert r["format"] == "wai.quantum.estimate/1"
|
||
c = r["counts"]
|
||
# A gate under three controls: an ancilla and three Toffolis.
|
||
assert (c["logical_qubits"], c["ancillas"], c["toffolis"], c["t_gates"], c["rotations"], c["measurements"]) == (5, 1, 3, 1, 1, 4)
|
||
assert r["epsilon"]["each"] == pytest.approx(1e-4 / 3) and r["epsilon"]["bound"] <= r["epsilon"]["synthesis"]
|
||
assert r["estimate"]["receipt"]["kind"] == "quantum-resource" and r["calibration"]["name"] == "exact_matching"
|
||
f = r["frontier"]
|
||
assert len(f) >= 2 and all(a["physical_qubits"] > b["physical_qubits"] for a, b in zip(f, f[1:]))
|
||
assert sum(1 for p in f if p.get("headline")) == 1
|
||
# Its seal checks as every job's does; the estimate check also re-runs it.
|
||
assert wq.verify(e).valid
|
||
v = wq.verify_estimate(e)
|
||
assert v["valid"] and v["signed"] and v["receipts"] and v["reproduced"] and v["signer"] == DEMO.did
|
||
forged = json.loads(json.dumps(e))
|
||
forged["result"]["estimate"]["physical_qubits"] += 1
|
||
assert not wq.verify_estimate(forged)["valid"]
|
||
# The same request, the same answer.
|
||
assert wq.estimate(LADDER, qubit="us-e3", budget=1e-4, max_slowdown=64, key=DEMO) == e
|
||
|
||
|
||
def test_an_estimate_reads_and_writes_the_interchange_documents():
|
||
doc = wq.estimate(LADDER, export="counts", key=DEMO)["result"]["export"]
|
||
assert doc == {"numQubits": 5, "tCount": 1, "rotationCount": 1, "rotationDepth": 1, "cczCount": 3, "ccixCount": 0, "measurementCount": 4}
|
||
back = wq.estimate(counts=doc, key=DEMO)
|
||
assert back["result"]["program"]["origin"] == "counts" and wq.verify_estimate(back)["valid"]
|
||
routine = wq.estimate(LADDER, export="routine", name="ladder", key=DEMO)["result"]["export"]
|
||
assert routine["version"] == "v1" and routine["program"]["name"] == "ladder"
|
||
assert wq.estimate(routine=json.dumps(routine), key=DEMO)["result"]["program"]["origin"] == "routine"
|
||
|
||
|
||
def test_an_estimate_refuses_with_its_code():
|
||
with pytest.raises(wq.EngineError, match=r"\[QE08\]"):
|
||
wq.estimate(LADDER, qubit="nowhere", key=DEMO)
|
||
with pytest.raises(wq.EngineError, match=r"\[QE07\]"):
|
||
wq.estimate(counts={"tCount": 3}, key=DEMO)
|
||
with pytest.raises(TypeError):
|
||
wq.estimate(key=DEMO)
|
||
|
||
|
||
# -- verdicts and repairs -------------------------------------------------------
|
||
|
||
|
||
def test_a_run_with_an_intent_carries_its_verdict_sealed():
|
||
job = wq.run(BELL, shots=100, intent={"distribution": {"00": 0.5, "11": 0.5}}, key=DEMO)
|
||
v = job.verdict
|
||
assert (v["level"], v["holds"], v["intent"][0]["how"]) == ("proof", True, "statevector")
|
||
assert v["statement"].startswith("Checked as proof")
|
||
assert wq.verify(job).valid
|
||
forged = json.loads(json.dumps(job))
|
||
forged["result"]["verdict"]["level"] = "evidence"
|
||
assert not wq.verify(forged).valid
|
||
# The same verdict read back; another intent worked out here, not sealed.
|
||
again = wq.verdict(job)
|
||
assert again["level"] == "proof" and again["sealed"] is True
|
||
other = wq.verdict(job, {"distribution": {"00": 1.0}})
|
||
assert (other["level"], other["holds"], other["sealed"]) == ("proof", False, False)
|
||
# A run that asks for no verdict carries none.
|
||
assert wq.run(BELL, shots=10, key=DEMO).verdict is None
|
||
assert wq.run(BELL, shots=10, verdict=True, key=DEMO).verdict["level"] == "executed"
|
||
|
||
|
||
def test_a_refusal_carries_its_code_and_a_refused_job_its_verdict():
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run("OPENQASM 3.0; qubit[1] q; rx(0.3) q[0];", method="stabilizer", key=DEMO)
|
||
assert e.value.code == "J011"
|
||
v = wq.verdict({"request": {"qasm": "OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n"}, "error": "line 3: unknown gate foo [Q103]"})
|
||
assert (v["level"], v["refusal"]["code"]) == ("refused", "Q103")
|
||
|
||
|
||
def test_repair_fixes_the_flagged_line_and_rejects_a_change_to_another():
|
||
r = wq.repair("OPENQASM 3.0;\nqubit[2] q\nh q[0];\n")
|
||
assert r["qasm"] == "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\n" and r["repairs"][0]["code"] == "Q004"
|
||
assert r["verdict"]["level"] == "parsed"
|
||
w = "OPENQASM 3.0;\nqubit[1] q;\nbit[1] c;\nc[0] = measure q[0];\nwhile (c[0]) {\n x q[0];\n c[0] = measure q[0];\n}\n"
|
||
assert wq.repair(w)["needs_person"][0]["code"] == "Q310"
|
||
assert wq.repair(w, max_iterations=3)["ok"] is True
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.repair("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", proposed="OPENQASM 3.0;\nqubit[2] q;\nh q[0];\n")
|
||
assert e.value.code == "J021"
|
||
taken = wq.repair("OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", proposed="OPENQASM 3.0;\nqubit[1] q;\nh q[0];\n")
|
||
assert taken["ok"] is True and taken["repairs"][0]["by"] == "proposal"
|
||
|
||
|
||
def test_a_sampled_intent_is_evidence_and_never_proof():
|
||
job = wq.run(BELL, shots=10, seed=7, intent={"sampled": {"00": 0.5, "11": 0.5}, "test": "g_test", "alpha": 0.01, "shots": 2000}, key=DEMO)
|
||
v = job.verdict
|
||
assert (v["level"], v["holds"], v["intent"][0]["strength"], v["intent"][0]["how"]) == ("evidence", True, "evidence", "g_test")
|
||
with pytest.raises(wq.EngineError) as e:
|
||
wq.run(BELL, shots=10, intent={"sampled": {"00": 0.5, "11": 0.5}, "test": "g_test", "shots": 2000}, key=DEMO)
|
||
assert e.value.code == "J020" and "alpha" in e.value.message
|
||
|
||
|
||
# -- error correction and mitigation ------------------------------------------
|
||
|
||
|
||
def test_a_code_comes_with_its_distance_certified_and_sealed():
|
||
c = wq.qec_code("color:5", key=DEMO)
|
||
r = c["result"]
|
||
assert (r["code"]["n"], r["code"]["k"]) == (19, 1)
|
||
cert = r["certificate"]
|
||
assert cert["distance"] == 5 and cert["x"]["distance"] == cert["z"]["distance"] == 5
|
||
assert len(cert["z"]["witness"]) == 5 and "color" in r["code"]["decoders"]
|
||
assert wq.verify(c).valid
|
||
# The certificate checks against its code, as a sealed answer or alone; a
|
||
# changed one does not.
|
||
for given in (c, cert, json.dumps(cert)):
|
||
check = wq.qec_code("color:5", certificate=given, key=DEMO)["result"]["check"]
|
||
assert check["verified"] and check["witnesses"] and check["search_reproduced"], check
|
||
forged = json.loads(json.dumps(cert))
|
||
forged["z"]["witness"] = forged["z"]["witness"][:-1]
|
||
assert not wq.qec_code("color:5", certificate=forged, key=DEMO)["result"]["check"]["verified"]
|
||
# The same request, the same answer; a search stopped short says so.
|
||
assert wq.qec_code("color:5", key=DEMO) == c
|
||
assert wq.qec_code("surface:5", certify=False, key=DEMO)["result"].get("certificate") is None
|
||
|
||
|
||
def test_every_error_below_half_the_distance_leaves_the_logical_state_unharmed():
|
||
# The distance-5 colour code: every error of weight 1 or 2, decoded by
|
||
# restriction to matching, leaves a product of stabilizers.
|
||
n = 19
|
||
errors = [[q] for q in range(n)] + [[a, b] for a in range(n) for b in range(a + 1, n)]
|
||
for e in errors:
|
||
r = wq.qec_decode("color:5", error=e, key=DEMO)["result"]
|
||
assert r["explains"] and not r["residual"]["logical"], (e, r["residual"])
|
||
# A syndrome alone decodes to a correction that explains it.
|
||
r = wq.qec_decode("surface:3", syndrome=[1], decoder="matching", key=DEMO)["result"]
|
||
assert r["explains"] and r["syndrome"] == [1]
|
||
|
||
|
||
def test_a_circuits_detectors_are_decoded_and_its_shots_counted():
|
||
one = wq.qec_decode(memory=(3, 3, 0.004), shots=2000, seed=3, decoder="correlated", key=DEMO)
|
||
r = one["result"]
|
||
assert r["shots"] == 2000 and 0 <= r["failures"] < 200
|
||
assert r["stderr"] == pytest.approx(math.sqrt(r["rate"] * (1 - r["rate"]) / 2000))
|
||
assert wq.verify(one).valid
|
||
assert wq.qec_decode(memory={"distance": 3, "rounds": 3, "p": 0.004}, shots=2000, seed=3, decoder="correlated", key=DEMO) == one
|
||
# A window over the rounds reads the same model.
|
||
w = wq.qec_decode(memory=(3, 4, 0.001), detectors=[], decoder="window", window=(2, 2), key=DEMO)["result"]
|
||
assert w["predicted"] == [] and len(w["detail"]["windows"]) >= 2
|
||
|
||
|
||
def test_a_threshold_measures_rates_with_their_errors():
|
||
t = wq.qec_threshold("color", distances=[3, 5, 7], p=[0.02], shots=2000, key=DEMO)
|
||
r = t["result"]
|
||
assert [pt["distance"] for pt in r["points"]] == [3, 5, 7]
|
||
assert all(pt["shots"] == 2000 and "stderr" in pt for pt in r["points"])
|
||
assert r["below_threshold"][0]["distance_helps"] is True
|
||
assert wq.verify(t).valid
|
||
|
||
|
||
def test_error_correction_refuses_with_its_code():
|
||
with pytest.raises(wq.EngineError, match=r"\[J009\]"):
|
||
wq.qec_code("hexagon:5", key=DEMO)
|
||
with pytest.raises(wq.EngineError, match=r"\[J009\]"):
|
||
wq.qec_threshold(shots=5_000_000, key=DEMO)
|
||
with pytest.raises(TypeError):
|
||
wq.qec_decode(memory=(3, 3), key=DEMO)
|
||
|
||
|
||
def test_a_mitigated_run_carries_each_value_raw_and_mitigated_with_its_assumptions():
|
||
ghz = "OPENQASM 3.0;\nqubit[3] q;\nh q[0];\ncx q[0], q[1];\ncx q[1], q[2];\nrz(0.3) q[2];\nrx(0.2) q[0];\n"
|
||
noise = {"depolarizing_2q": 0.02, "depolarizing_1q": 0.004}
|
||
job = wq.run(ghz, observables=["Z0 Z1"], noise=noise, mitigate={"method": "zne", "shots": 4000}, key=DEMO)
|
||
m = job["result"]["mitigation"]
|
||
assert m["method"] == "zne" and m["assumptions"]["bias"] and m["assumptions"]["variance"]
|
||
row = m["observables"][0]
|
||
assert abs(row["mitigated"] - row["ideal"]) < abs(row["raw"] - row["ideal"]) and row["stderr"] > 0
|
||
assert job.verify().valid
|
||
forged = json.loads(json.dumps(job))
|
||
forged["result"]["mitigation"]["observables"][0]["mitigated"] = 1.0
|
||
assert not wq.verify(forged).valid
|
||
# A method's name alone takes its defaults; mitigation without noise is refused.
|
||
assert wq.run(ghz, observables=["Z0 Z1"], noise=noise, mitigate="vd", key=DEMO)["result"]["mitigation"]["method"] == "vd"
|
||
with pytest.raises(wq.EngineError, match=r"\[J008\]"):
|
||
wq.run(ghz, observables=["Z0"], mitigate="zne", key=DEMO)
|