open-standards/wai/wai-quantum-service/smoke.sh

#!/usr/bin/env bash
# smoke.sh BASE_URL: the service's promises, checked over HTTP against a running
# instance (CI starts one locally; after a deploy, run it against production).
#
#   bash smoke.sh http://127.0.0.1:8241
#   bash smoke.sh https://<public base>
#
# Each check prints one line and the script fails on the first that does not
# hold. Needs curl and python3.
set -euo pipefail
U=${1:?usage: smoke.sh BASE_URL}
U=${U%/}
tmp=$(mktemp -d)
trap 'rm -rf "$tmp"' EXIT

post() { curl -sS --max-time 120 -X POST "$U$1" -H 'content-type: application/json' --data-binary "@$2"; }
py() { python3 -I -c "$@"; }
# POST /v1/run, then follow a job still queued or running past the service's
# wait through GET /v1/jobs/{id}, as a client would: on a loaded host a job
# can outlast the wait without anything being wrong. At most ten minutes.
run() {
  post /v1/run "$1" > "$2"
  for _ in $(seq 1 120); do
    case "$(py 'import json,sys; print(json.load(open(sys.argv[1])).get("status", ""))' "$2")" in
      queued|running) sleep 5; curl -sS --max-time 30 "$U/v1/jobs/$(py 'import json,sys; print(json.load(open(sys.argv[1]))["id"])' "$2")" > "$2.next" && mv "$2.next" "$2" ;;
      *) return 0 ;;
    esac
  done
}

# The service describes itself, with its methods and its receipt key.
curl -sS --max-time 30 "$U/v1" > "$tmp/describe.json"
py 'import json,sys
d=json.load(open(sys.argv[1]))
assert d["service"]=="wai-quantum-service", d
ms={m["method"] for m in d["methods"]}
assert {"exact","statevector","stabilizer","pauli","tensor_network","cutrank","density_matrix","trajectories"} <= ms, ms
assert len(d["receipt_key"])==64
print("describe: ok,", len(ms), "methods, receipt key", d["receipt_key"][:16]+"…")' "$tmp/describe.json"

# A 40-qubit GHZ state: Clifford, so the stabilizer method, and only the two
# all-equal outcomes.
py 'import json
n=40
q="OPENQASM 3.0;\nqubit[%d] q;\nh q[0];\n" % n + "".join("cx q[%d], q[%d];\n" % (i,i+1) for i in range(n-1))
print(json.dumps({"qasm": q, "shots": 1000, "seed": 7}))' > "$tmp/ghz.json"
run "$tmp/ghz.json" "$tmp/ghz.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and r["method"]=="stabilizer", d
c=r["counts"]; assert set(c) <= {"0"*40, "1"*40} and sum(c.values())==1000, c
assert "receipt" in r
print("ghz-40: ok,", r["method"], c)' "$tmp/ghz.out"

# Its receipt verifies, and a changed result does not.
py 'import json,sys
req=json.load(open(sys.argv[1])); res=json.load(open(sys.argv[2]))["result"]
json.dump({"request": req, "result": res}, open(sys.argv[3],"w"))
res["counts"]["0"*40] = res["counts"].get("0"*40, 0) + 1
json.dump({"request": req, "result": res}, open(sys.argv[4],"w"))' "$tmp/ghz.json" "$tmp/ghz.out" "$tmp/v.json" "$tmp/t.json"
post /v1/verify "$tmp/v.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True, d; print("receipt: verifies")'
post /v1/verify "$tmp/t.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is False, d; print("receipt: a changed result does not verify")'

# The job carries its request as submitted, so whoever holds only the id can
# check the receipt: verify from the service's answer alone.
py 'import json,sys
req=json.load(open(sys.argv[1])); job=json.load(open(sys.argv[2]))
assert job["request"]==req, job.get("request")
json.dump({"request": job["request"], "result": job["result"]}, open(sys.argv[3],"w"))' "$tmp/ghz.json" "$tmp/ghz.out" "$tmp/vj.json"
post /v1/verify "$tmp/vj.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True, d; print("receipt: verifies from the job alone")'

# The job's energy: a counter figure with its class and declared uncertainty,
# or why it is unmetered, never zero. SMOKE_EXPECT_ENERGY=1 (a macOS host)
# requires the figure. A changed figure breaks the receipt: it is sealed.
export SMOKE_EXPECT_ENERGY="${SMOKE_EXPECT_ENERGY:-0}"
py 'import json,os,sys
e=json.load(open(sys.argv[1]))["result"]["energy"]
if "joules_micro" in e:
    uj=e["joules_micro"]; u=e["energy_uncertainty"]
    assert uj > 0 and e["energy_provenance"] == "OnChipCounter", e
    assert u["relative_ppm"] > 0 and u["window_us"] > 0 and uj * u["relative_ppm"] >= 288676, e
    assert e["energy_counter_filtering"] in ("Unknown", "Off", "On") and e["coverage"] in ("whole", "partial"), e
    print("energy: %.6f J of CPU, %s, %s" % (uj / 1e6, e["coverage"], e["energy_provenance"]))
else:
    assert os.environ["SMOKE_EXPECT_ENERGY"] != "1", e
    assert e["unmetered"], e
    print("energy: unmetered,", e["unmetered"])' "$tmp/ghz.out"
py 'import json,sys
d=json.load(open(sys.argv[1]))
e=d["result"]["energy"]
if "joules_micro" in e: e["joules_micro"] += 1
else: e["unmetered"] += "."
json.dump(d, open(sys.argv[2],"w"))' "$tmp/vj.json" "$tmp/ve.json"
post /v1/verify "$tmp/ve.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is False, d; print("energy: a changed figure does not verify")'

# As a browser sends it back: whole doubles without a fraction, keys reversed.
py 'import json,sys
def js(v):
    if isinstance(v, float) and v.is_integer(): return int(v)
    if isinstance(v, dict): return {k: js(v[k]) for k in reversed(list(v))}
    if isinstance(v, list): return [js(x) for x in v]
    return v
v=json.load(open(sys.argv[1])); json.dump(js(v), open(sys.argv[2],"w"))' "$tmp/vj.json" "$tmp/vjs.json"
post /v1/verify "$tmp/vjs.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True, d; print("receipt: verifies after a browser-style round trip")'

# Continuous angles on a small circuit: the dense state vector, and the
# probabilities sum to the shots.
py 'import json; print(json.dumps({"qasm": "qubit[3] q; rx(0.3) q[0]; ry(1.2) q[1]; cx q[0], q[2]; crz(0.7) q[1], q[2];", "shots": 500, "observables": ["Z0", "Z1 Z2"]}))' > "$tmp/rot.json"
run "$tmp/rot.json" "$tmp/rot.out"
py 'import json,sys,math
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and r["method"]=="statevector", d
assert sum(r["counts"].values())==500
z0=r["expectations"][0]["value"]; assert abs(z0-math.cos(0.3))<1e-12, z0
print("rotations: ok, <Z0> =", z0)' "$tmp/rot.out"

# A 100-qubit circuit's expectation values: sparse Pauli dynamics, each value
# with its bound.
py 'import json
n=100
q="qubit[%d] q;\n" % n + "".join("rx(0.3) q[%d];\n" % i for i in range(n)) + "".join("cx q[%d], q[%d];\n" % (i,i+1) for i in range(n-1))
print(json.dumps({"qasm": q, "observables": ["Z0", "Z50 Z51"]}))' > "$tmp/wide.json"
run "$tmp/wide.json" "$tmp/wide.out"
py 'import json,sys,math
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and r["method"]=="pauli", d
e=r["expectations"]; assert all("bound" in x for x in e), e
assert abs(e[0]["value"]-math.cos(0.3)) <= e[0]["bound"]+1e-9, e
print("pauli-100: ok,", [(x["observable"], round(x["value"],6), x["bound"]) for x in e])' "$tmp/wide.out"

# The pauli answer carries its threshold ladder, and the smaller threshold
# never drops more.
py 'import json,sys
d=json.load(open(sys.argv[1])); e=d["result"]["expectations"][0]
l=e["ladder"]; b=[x["bound"] for x in l]
assert len(l)>=2 and l[-1]["value"]==e["value"] and all(y<=x+1e-12 for x,y in zip(b,b[1:])), e
print("pauli ladder: ok,", [(x["threshold"], round(x["value"],6)) for x in l])' "$tmp/wide.out"

# Noisy shots: branching trajectories, every shot counted, and shots that
# share their errors sharing a state.
py 'import json
n=14
q="qubit[%d] q;\nbit[%d] c;\n" % (n,n) + "".join("ry(0.4) q[%d];\n" % i for i in range(n)) + "".join("cz q[%d], q[%d];\n" % (i,i+1) for i in range(n-1)) + "c = measure q;\n"
print(json.dumps({"qasm": q, "method": "trajectories", "shots": 2000, "noise": {"depolarizing_1q": 0.0005, "depolarizing_2q": 0.002, "readout": 0.01}}))' > "$tmp/noisy.json"
run "$tmp/noisy.json" "$tmp/noisy.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); assert d.get("status")=="done", d; r=d["result"]
assert r["method"]=="trajectories", d
assert sum(r["counts"].values())==2000, r
h=r["accuracy"]["histories"]; assert 1 <= h < 2000, r["accuracy"]
print("noisy-14: ok,", r["method"], h, "histories for 2000 shots")' "$tmp/noisy.out"

# A wide noisy Clifford circuit: Pauli frames on the stabilizer tableau.
py 'import json
n=500
q="qubit[%d] q;\n" % n + "h q[0];\n" + "".join("cx q[%d], q[%d];\n" % (i,i+1) for i in range(n-1))
print(json.dumps({"qasm": q, "shots": 1000, "noise": {"depolarizing_2q": 0.0005, "readout": 0.001}, "observables": ["Z0 Z499"]}))' > "$tmp/nghz.json"
run "$tmp/nghz.json" "$tmp/nghz.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); assert d.get("status")=="done", d; r=d["result"]
assert r["method"]=="stabilizer" and r["accuracy"]["class"]=="estimate", d
assert sum(r["counts"].values())==1000, r
e=r["expectations"][0]; assert 0.5 < e["value"] < 1.0 and e["stderr"] > 0, e
print("noisy-ghz-500: ok,", r["method"], "<Z0 Z499> =", round(e["value"],4), "±", round(e["stderr"],4))' "$tmp/nghz.out"

# The single-precision state vector on the GPU, where the service has one,
# against the double-precision one.
if py 'import json,sys; d=json.load(open(sys.argv[1])); sys.exit(0 if any(m["method"]=="statevector_single" and m.get("available") for m in d["methods"]) else 1)' "$tmp/describe.json"; then
  py 'import json; print(json.dumps({"qasm": "qubit[12] q; h q; rx(0.3) q[0]; cx q[0], q[11]; crz(0.7) q[3], q[9]; ry(1.1) q[5];", "observables": ["Z0 Z11", "X5"], "method": "statevector_single"}))' > "$tmp/gpu.json"
  py 'import json,sys; d=json.load(open(sys.argv[1])); d["method"]="statevector"; json.dump(d, open(sys.argv[2],"w"))' "$tmp/gpu.json" "$tmp/cpu.json"
  run "$tmp/gpu.json" "$tmp/gpu.out"
  run "$tmp/cpu.json" "$tmp/cpu.out"
  py 'import json,sys
g=json.load(open(sys.argv[1])); c=json.load(open(sys.argv[2])); assert g.get("status")=="done" and c.get("status")=="done", (g, c)
gv=[e["value"] for e in g["result"]["expectations"]]; cv=[e["value"] for e in c["result"]["expectations"]]
assert g["result"]["accuracy"]["class"]=="single" and all(abs(a-b)<1e-5 for a,b in zip(gv,cv)), (gv, cv)
e=g["result"]["energy"]
# The counter reads the CPU only, so a GPU job never claims a whole figure.
assert "joules_micro" not in e or (e["coverage"]=="partial" and "GPU" in e["excludes"][0]), e
print("gpu: ok,", g["result"]["accuracy"]["device"], gv, "energy", e.get("coverage", "unmetered"))' "$tmp/gpu.out" "$tmp/cpu.out"
else
  echo "gpu: this instance has none"
fi

# What a program finds first: the index, the examples, the OpenAPI document.
curl -sS --max-time 30 -H 'accept: application/json' "$U/" > "$tmp/index.json"
py 'import json,sys; d=json.load(open(sys.argv[1])); assert d["mcp"].endswith("/mcp") and d["openapi"].endswith("/v1/openapi.json"), d; print("index: ok,", d["mcp"])' "$tmp/index.json"
curl -sS --max-time 30 "$U/v1/examples" > "$tmp/examples.json"
curl -sS --max-time 30 "$U/v1/openapi.json" > "$tmp/openapi.json"
py 'import json,sys
e=json.load(open(sys.argv[1]))["examples"]; o=json.load(open(sys.argv[2]))
assert len(e)>=5 and all("job" in x and "qasm" in x["job"] for x in e), e
assert o["openapi"].startswith("3.1") and "/v1/run" in o["paths"] and "/mcp" in o["paths"], list(o["paths"])
print("examples and openapi: ok,", len(e), "examples,", len(o["paths"]), "paths")' "$tmp/examples.json" "$tmp/openapi.json"

# An agent over the Model Context Protocol: initialize, list the tools, run
# the first example, and check its receipt.
mcp() { curl -sS --max-time 120 -X POST "$U/mcp" -H 'content-type: application/json' -H 'accept: application/json, text/event-stream' --data-binary "$1"; }
mcp '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-06-18","capabilities":{},"clientInfo":{"name":"smoke","version":"1"}}}' > "$tmp/mcp-init.json"
code=$(curl -sS -o /dev/null -w '%{http_code}' --max-time 30 -X POST "$U/mcp" -H 'content-type: application/json' --data-binary '{"jsonrpc":"2.0","method":"notifications/initialized"}')
mcp '{"jsonrpc":"2.0","id":2,"method":"tools/list"}' > "$tmp/mcp-tools.json"
py 'import json,sys
i=json.load(open(sys.argv[1]))["result"]; t=json.load(open(sys.argv[2]))["result"]["tools"]
assert i["protocolVersion"]=="2025-06-18" and "tools" in i["capabilities"], i
names={x["name"] for x in t}; assert {"run_quantum_circuit","get_quantum_job","verify_quantum_receipt","quantum_service_info","quantum_circuit_examples","critique_qasm","check_equivalent","improve_qasm","classical_shortcuts"} <= names, names
assert sys.argv[3]=="202", sys.argv[3]
print("mcp: ok, initialized,", len(t), "tools")' "$tmp/mcp-init.json" "$tmp/mcp-tools.json" "$code"
py 'import json,sys
job=json.load(open(sys.argv[1]))["examples"][0]["job"]
print(json.dumps({"jsonrpc":"2.0","id":3,"method":"tools/call","params":{"name":"run_quantum_circuit","arguments":job}}))' "$tmp/examples.json" > "$tmp/mcp-call.json"
mcp "@$tmp/mcp-call.json" > "$tmp/mcp-run.json"
py 'import json,sys
r=json.load(open(sys.argv[1]))["result"]; job=r["structuredContent"]; text=r["content"][0]["text"]
assert not r["isError"] and job["status"]=="done" and "counts" in job["result"], r
assert "method:" in text and "counts" in text, text
req=json.load(open(sys.argv[2]))["examples"][0]["job"]
print(json.dumps({"jsonrpc":"2.0","id":4,"method":"tools/call","params":{"name":"verify_quantum_receipt","arguments":{"request":req,"result":job["result"]}}}))' "$tmp/mcp-run.json" "$tmp/examples.json" > "$tmp/mcp-verify.json"
curl -sS --max-time 60 -X POST "$U/mcp" -H 'content-type: application/json' --data-binary "@$tmp/mcp-verify.json" > "$tmp/mcp-verified.json"
py 'import json,sys
v=json.load(open(sys.argv[1]))["result"]; assert v["structuredContent"]["valid"] is True, v
print("mcp run + verify: ok,", json.load(open(sys.argv[2]))["result"]["content"][0]["text"].splitlines()[1])' "$tmp/mcp-verified.json" "$tmp/mcp-run.json"

# Grover's search on five qubits as the IDE writes it: its oracle and its
# diffusion are each a Z under four controls. Four rounds find the item.
py 'import json
n, marked = 5, 0b10110
def mcz(): return "ctrl(4) @ z q[0],q[1],q[2],q[3],q[4];\n"
q = "OPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[5] q;\n" + "".join("h q[%d];\n" % i for i in range(n))
for _ in range(4):
    flips = "".join("x q[%d];\n" % i for i in range(n) if not marked >> i & 1)
    q += flips + mcz() + flips
    q += "".join("h q[%d];\nx q[%d];\n" % (i, i) for i in range(n)) + mcz() + "".join("x q[%d];\nh q[%d];\n" % (i, i) for i in range(n))
print(json.dumps({"qasm": q, "shots": 1000}))' > "$tmp/grover.json"
run "$tmp/grover.json" "$tmp/grover.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
hits=r["counts"].get("10110", 0)
assert d["status"]=="done" and hits > 980, d
print("grover-5 (ctrl modifiers): ok,", r["method"], hits, "of 1000 on the marked item")' "$tmp/grover.out"

# A program that measures in the middle and acts on what it read: teleportation
# delivers ry(0.7)|0⟩ to q[2], so ⟨Z2⟩ = cos 0.7 and ⟨X2⟩ = sin 0.7 exactly.
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0; qubit[3] q; bit[3] c; ry(0.7) q[0]; h q[1]; cx q[1], q[2]; cx q[0], q[1]; h q[0]; c[0] = measure q[0]; c[1] = measure q[1]; if (c[1]) x q[2]; if (c[0]) z q[2]; c[2] = measure q[2];", "observables": ["Z2", "X2"], "shots": 2000}))' > "$tmp/tele.json"
run "$tmp/tele.json" "$tmp/tele.out"
py 'import json,math,sys
d=json.load(open(sys.argv[1])); r=d["result"]; e=[x["value"] for x in r["expectations"]]
assert d["status"]=="done" and abs(e[0]-math.cos(0.7))<1e-12 and abs(e[1]-math.sin(0.7))<1e-12, d
assert r["dynamic"]["branches_weighed"]==4 and sum(r["counts"].values())==2000, r
print("teleport (mid-circuit measurement, if): ok,", r["method"], e)' "$tmp/tele.out"

# A Clifford circuit on 60 qubits, then T and H on each: past the state
# vector, and not Clifford, so auto samples it exactly by cut-rank. Its graph
# (a ring with chords of length 5) keeps every cut narrow.
py 'import json
n=60; q="OPENQASM 3.0; qubit[%d] q; bit[%d] c; h q;" % (n, n)
q+="".join(" cz q[%d], q[%d]; cz q[%d], q[%d];" % (i, (i+1)%n, i, (i+5)%n) for i in range(n))
q+="".join(" t q[%d]; h q[%d];" % (i, i) for i in range(n)) + " c = measure q;"
print(json.dumps({"qasm": q, "shots": 100, "seed": 3}))' > "$tmp/cut.json"
run "$tmp/cut.json" "$tmp/cut.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and r["method"]=="cutrank" and sum(r["counts"].values())==100, d
assert r["accuracy"]["class"]=="exact" and r["accuracy"]["max_cut"]<=12, r["accuracy"]
print("cutrank (60 qubits, Clifford then T and H): ok, max cut", r["accuracy"]["max_cut"], "bits,", r["distinct_outcomes"], "distinct outcomes")' "$tmp/cut.out"

# A refusal says why.
py 'import json; print(json.dumps({"qasm": "qubit[1] q; bit c; c = measure q; while (c) { x q; c = measure q; }"}))' > "$tmp/bad.json"
code=$(curl -sS -o "$tmp/bad.out" -w '%{http_code}' -X POST "$U/v1/jobs" -H 'content-type: application/json' --data-binary "@$tmp/bad.json")
[ "$code" = 400 ] || { echo "a bad program got HTTP $code"; exit 1; }
py 'import json,sys; d=json.load(open(sys.argv[1])); assert "not supported" in d["error"], d; print("refusal: ok,", d["error"])' "$tmp/bad.out"

# The design checks answer at once: a critique, an improvement checked the same
# circuit, the equivalence check, the classical shortcuts, and a thorough search
# sent to the command line.
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\nh q[0];\ncx q[0], q[1];\n"}))' > "$tmp/design.json"
post /v1/critique "$tmp/design.json" > "$tmp/critique.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); assert d["ok"] and d["gates"]==3 and d["smells"][0]["kind"]=="cancels" and d["smells"][0]["lines"]==[3,4], d; print("critique: ok,", d["smells"][0]["message"])' "$tmp/critique.out"
post /v1/improve "$tmp/design.json" > "$tmp/improve.out"
py 'import json,sys; c=json.load(open(sys.argv[1]))["candidates"][0]; assert c["measures"]["gates"]==1 and c["equivalence"]["verdict"]=="equivalent", c; print("improve: ok,", c["equivalence"]["how"])' "$tmp/improve.out"
py 'import json; print(json.dumps({"a": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\nh q[0];\ncx q[0], q[1];\n", "b": "OPENQASM 3.0;\nqubit[2] q;\ncx q[0], q[1];\n"}))' > "$tmp/pair.json"
post /v1/equivalent "$tmp/pair.json" > "$tmp/equivalent.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); assert d["verdict"]=="equivalent" and d["how"]=="tableau", d; print("equivalent: ok, by tableau")' "$tmp/equivalent.out"
post /v1/shortcuts "$tmp/design.json" > "$tmp/shortcuts.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); assert d["summary"].startswith("Classically easy at any width"), d; print("shortcuts: ok,", d["summary"][:40] + "…")' "$tmp/shortcuts.out"
py 'import json; print(json.dumps({"qasm": "qubit[1] q;", "thorough": True}))' > "$tmp/thorough.json"
code=$(curl -sS -o "$tmp/thorough.out" -w '%{http_code}' -X POST "$U/v1/shortcuts" -H 'content-type: application/json' --data-binary "@$tmp/thorough.json")
[ "$code" = 422 ] || { echo "a thorough search got HTTP $code"; exit 1; }
py 'import json,sys; d=json.load(open(sys.argv[1])); assert "wai-quantum shortcuts --thorough" in d["error"], d; print("thorough: ok, sent to the command line")' "$tmp/thorough.out"

echo "smoke: every check held at $U"