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

# A client may submit `q` jobs in a window (RateLimit-Policy), and the script
# makes more than a default window's 30. So before each POST it reads where it
# stands (RateLimit: `r` left, `t` seconds until one more is) from GET /v1/health,
# which submits nothing, and with none left it waits for one. CI starts the
# service with a larger WAI_QSVC_CLIENT_BURST, so it never waits there; against
# production it takes longer, and checks the same things. A dry run counts as a
# submission and then answers whether a real one would be taken now, so before
# one the script waits for two (`pace 2`).
pace() {
  local need=${1:-1} h r t
  for _ in $(seq 1 80); do
    h=$(curl -sS --max-time 30 -D - -o /dev/null "$U/v1/health" | tr -d '\r' | sed -n 's/^[Rr]ate[Ll]imit: //p')
    r=$(printf '%s' "$h" | sed -n 's/.*r=\([0-9]*\).*/\1/p')
    t=$(printf '%s' "$h" | sed -n 's/.*t=\([0-9]*\).*/\1/p')
    if [ -z "$r" ] || [ "$r" -ge "$need" ]; then return 0; fi
    echo "pace: $r of the $need submissions needed left in this client's window; waiting ${t:-10} s" >&2
    sleep "$(( ${t:-10} < 1 ? 1 : ${t:-10} ))"
  done
  echo "FAIL: no submission came free in this client's window" >&2
  exit 1
}
post() { pace; 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
v=d["versions"]
assert v["schema"]=="wai.quantum.versions/1" and v["host"]["name"]=="wai-quantum-service" and v["schemas"]["job_result"]=="wai.quantum.job-result/1", v
assert d["store"]["addresses"].startswith("BLAKE3") and d["limits"]["set_items"] >= 1 and d["receipt_keys"].endswith("/.well-known/wai-quantum-receipt-keys"), d
print("describe: ok,", len(ms), "methods, receipt key", d["receipt_key"][:16]+"…,", "store", d["store"]["kind"], "durable" if d["store"]["durable"] else "in memory,", "energy", d["energy"]["meter"])' "$tmp/describe.json"

# The receipt keys at the well-known path: the current key, with its window.
curl -sS --max-time 30 "$U/.well-known/wai-quantum-receipt-keys" > "$tmp/keys.json"
py 'import json,sys
k=json.load(open(sys.argv[1])); d=json.load(open(sys.argv[2]))
cur=[x for x in k["keys"] if x.get("current")]
assert k["format"]=="wai.quantum.receipt-keys/1" and len(cur)==1 and cur[0]["key"]==d["receipt_key"], k
print("receipt keys: ok,", len(k["keys"]), "key(s), the current one", cur[0]["key"][:16]+"…")' "$tmp/keys.json" "$tmp/describe.json"

# Every answer carries the client's submission window as rate-limit fields.
curl -sS --max-time 30 -D "$tmp/headers.txt" -o /dev/null "$U/v1/health"
py 'import sys
h={l.split(":",1)[0].strip().lower(): l.split(":",1)[1].strip() for l in open(sys.argv[1]) if ":" in l}
assert h["ratelimit-policy"].startswith("\"submissions\";q=") and ";w=" in h["ratelimit-policy"], h
assert h["ratelimit"].startswith("\"submissions\";r="), h
print("rate-limit fields: ok,", h["ratelimit-policy"], "/", h["ratelimit"])' "$tmp/headers.txt"

# 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
assert r["schema"]=="wai.quantum.job-result/1" and r["engine"]["name"]=="wai-quantum", 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, its scope (the worker's
# own process on macOS; the processor packages over the job's window on
# Linux, with no attribution to the job) and its declared uncertainty, or why
# it is unmetered, never zero. SMOKE_EXPECT_ENERGY=1 (a host with a readable
# counter) 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
    assert e["attribution"] in ("process", "none") and (e["attribution"] == "process" or e["concurrent_jobs"] >= 0), e
    print("energy: %.6f J, %s, %s coverage, attribution %s" % (uj / 1e6, e["energy_provenance"], e["coverage"], e["attribution"]))
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 exact values: every qubit starts in |+⟩, which the rx and the
# cx leave alone, so ⟨Z0 Z11⟩ = 0 and ⟨X5⟩ = cos 1.1. One run, so the script
# stays within one client's window there too.
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"
  run "$tmp/gpu.json" "$tmp/gpu.out"
  py 'import json,math,sys
g=json.load(open(sys.argv[1])); assert g.get("status")=="done", g
gv=[e["value"] for e in g["result"]["expectations"]]; cv=[0.0, math.cos(1.1)]
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"
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. The run states its verdict, so the
# request it seals (the job's own `request`) asks for one.
mcp() { pace; 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"
pace
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","run_quantum_set","dry_run_quantum_job","get_quantum_job","verify_quantum_receipt","quantum_service_info","quantum_circuit_examples","critique_qasm","check_equivalent","improve_qasm","classical_shortcuts","compile_qasm","estimate_resources","qec_code","qec_decode","qec_threshold","gradient_qasm","optimize_qasm","vqe_ground_state","qaoa_maxcut","build_algorithm","pulse_schedule","pulse_check","list_components"} <= 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 and "verdict: executed" in text, text
req=json.load(open(sys.argv[2]))["examples"][0]["job"]
assert job["request"]==dict(req, verdict=True) and job["verdict"]["level"]=="executed" and job["verdict"]==job["result"]["verdict"], job
req=job["request"]
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"
pace
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"
# A host that draws views: the session id that says so, the view's own tool kept from the model, and a page.
pace
curl -sS -D "$tmp/mcp-views.h" --max-time 60 -X POST "$U/mcp" -H 'content-type: application/json' -H 'accept: application/json, text/event-stream' \
  --data-binary '{"jsonrpc":"2.0","id":1,"method":"initialize","params":{"protocolVersion":"2025-11-25","capabilities":{"extensions":{"io.modelcontextprotocol/ui":{"mimeTypes":["text/html;profile=mcp-app"]}}}}}' > /dev/null
sid=$(tr -d '\r' < "$tmp/mcp-views.h" | awk 'tolower($1)=="mcp-session-id:" {print $2}')
case "$sid" in wq-views-*) ;; *) echo "no views session id: $sid"; exit 1 ;; esac
pace
curl -sS --max-time 60 -X POST "$U/mcp" -H 'content-type: application/json' -H "mcp-session-id: $sid" --data-binary '{"jsonrpc":"2.0","id":2,"method":"tools/list"}' > "$tmp/mcp-views-tools.json"
pace
curl -sS --max-time 60 -X POST "$U/mcp" -H 'content-type: application/json' -H "mcp-session-id: $sid" --data-binary '{"jsonrpc":"2.0","id":3,"method":"resources/read","params":{"uri":"ui://wai-quantum/result"}}' > "$tmp/mcp-views-page.json"
py 'import json,sys
t={x["name"]: x for x in json.load(open(sys.argv[1]))["result"]["tools"]}
assert t["poll_quantum_job"]["_meta"]["ui"]["visibility"]==["app"] and t["run_quantum_circuit"]["_meta"]["ui"]["resourceUri"]=="ui://wai-quantum/result", t["poll_quantum_job"]
c=json.load(open(sys.argv[2]))["result"]["contents"][0]
assert c["mimeType"]=="text/html;profile=mcp-app" and len(c["text"].encode())<=1<<20, c["mimeType"]
print("mcp views: ok,", len(c["text"].encode()), "bytes")' "$tmp/mcp-views-tools.json" "$tmp/mcp-views-page.json"
# The engine's resources and prompts, as the command line's server serves them.
mcp '{"jsonrpc":"2.0","id":5,"method":"resources/read","params":{"uri":"wai-quantum://docs/diagnostics"}}' > "$tmp/mcp-resource.json"
mcp '{"jsonrpc":"2.0","id":6,"method":"prompts/get","params":{"name":"explain_result","arguments":{"job":"{}"}}}' > "$tmp/mcp-prompt.json"
py 'import json,sys
r=json.load(open(sys.argv[1]))["result"]["contents"][0]; d=json.loads(r["text"])
assert r["mimeType"]=="application/json" and len(d["codes"])>=50 and len(d["examples"])>=10 and any(c["code"]=="J011" for c in d["job_codes"]), r
m=json.load(open(sys.argv[2]))["result"]["messages"]; assert "verify_quantum_receipt" in m[1]["content"]["text"], m
print("mcp resources + prompts: ok,", len(d["codes"]), "reader codes,", len(d["job_codes"]), "job codes,", len(d["examples"]), "worked refusals")' "$tmp/mcp-resource.json" "$tmp/mcp-prompt.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 GHZ state as a device would give it: placed on the synthetic heavy-hexagon
# target's most reliable qubits and simulated under its noise, the target's
# hash (as GET /v1/targets gives it) in the receipt's result.
curl -sS --max-time 30 "$U/v1/targets" > "$tmp/targets.json"
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0; qubit[3] q; bit[3] c; h q[0]; cx q[0], q[1]; cx q[1], q[2]; c = measure q;", "shots": 2000, "seed": 5, "target": "synthetic-heavy-hex-27"}))' > "$tmp/device.json"
run "$tmp/device.json" "$tmp/device.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
ts={t["name"]: t["hash"] for t in json.load(open(sys.argv[2]))["targets"]}
assert d["status"]=="done" and r["method"]=="density_matrix", d
assert r["target"]["hash"]==ts["synthetic-heavy-hex-27"], (r["target"]["hash"], ts)
good=r["counts"].get("000",0)+r["counts"].get("111",0)
assert 1700 < good < 2000 and "receipt" in r, r["counts"]
p=r["target"]["placement"]
print("device target: ok, placed on", p["layout"], "expected success", round(p["expected_success"],4), "distance from ideal", round(r["ideal"]["total_variation_distance"],4))' "$tmp/device.out" "$tmp/targets.json"

# A refusal says why, what to write instead, and ends with its code.
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"
pace
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 d["code"]=="Q310" and "@wai.max_iterations" 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, and a thorough search of the classical
# shortcuts sent to the command line (the shortcuts themselves are the MCP
# server's tests', as classical_shortcuts).
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"
# A program carried to QIR and back, each conversion checked, and a job
# that carries its program as QIR.
py 'import json; print(json.dumps({"program": "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n", "to": "qir"}))' > "$tmp/convert.json"
post /v1/convert "$tmp/convert.json" > "$tmp/convert.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); assert d["ok"] and d["to"]=="qir" and d["written"]["profile"]=="base" and d["check"]["verdict"]["verdict"]=="equivalent", d; print("convert: ok, to QIR,", d["check"]["verdict"]["how"])' "$tmp/convert.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); print(json.dumps({"source": {"format": "qir", "program": d["program"]}, "shots": 50}))' "$tmp/convert.out" > "$tmp/fromqir.json"
post /v1/run "$tmp/fromqir.json" > "$tmp/fromqir.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); r=d["result"]; assert d["status"]=="done" and set(r["counts"]) <= {"00","11"} and d["request"]["source"]["format"]=="qir", d; print("a job from QIR: ok,", r["method"])' "$tmp/fromqir.out"
py 'import json; print(json.dumps({"qasm": "qubit[1] q;", "thorough": True}))' > "$tmp/thorough.json"
pace
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"

# A compile: Clifford+T within the budget, each pass recorded, the certificate
# sealed with the service's key and checked by POST /v1/verify; a job that
# compiles first runs bit-exact; one too fine is sent to the command line.
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\nrz(0.3) q[0];\ncx q[0], q[1];\nry(1.1) q[1];\n", "epsilon": 1e-4, "level": 2}))' > "$tmp/compile.json"
post /v1/compile "$tmp/compile.json" > "$tmp/compile.out"
py 'import json,sys
b=json.load(open(sys.argv[1])); c=b["result"]["certificate"]
assert b["result"]["ok"] and c["verdict"]=="approximate" and 0 < c["epsilon"]["bound"] <= 1e-4 and c["output"]["counts"]["rotations"]==0, c
assert [p["pass"] for p in c["passes"]][0]=="simplify" and "rz(" not in b["result"]["qasm"], c["passes"]
print("compile: ok,", len(c["passes"]), "passes, bound", c["epsilon"]["bound"], "T", c["output"]["counts"]["t_count"])' "$tmp/compile.out"
post /v1/verify "$tmp/compile.out" > "$tmp/compile-verify.out"
py 'import json,sys; v=json.load(open(sys.argv[1])); assert v["valid"] is True, v; print("compile certificate verifies: ok")' "$tmp/compile-verify.out"
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[1] q;\nrx(0.4) q[0];\n", "shots": 200, "compile": {"epsilon": 1e-4}}))' > "$tmp/compiled-job.json"
run "$tmp/compiled-job.json" "$tmp/compiled-job.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and d["method"]=="exact" and r["accuracy"]["class"]=="bit-exact" and r["compiled"]["distance"] <= 1e-4, d
print("compiled job: ok, bit-exact, within", r["compiled"]["distance"])' "$tmp/compiled-job.out"
py 'import json; print(json.dumps({"qasm": "qubit[1] q;\nrz(0.3) q[0];\nh q[0];\nrz(0.4) q[0];\n", "epsilon": 1e-13}))' > "$tmp/fine.json"
pace
code=$(curl -sS -o "$tmp/fine.out" -w '%{http_code}' -X POST "$U/v1/compile" -H 'content-type: application/json' --data-binary "@$tmp/fine.json")
[ "$code" = 422 ] || { echo "a too-fine compile got HTTP $code"; exit 1; }
py 'import json,sys; d=json.load(open(sys.argv[1])); assert "wai-quantum compile" in d["error"], d; print("fine compile: ok, sent to the command line")' "$tmp/fine.out"

# Error correction: a code's distance certified and sealed with the service's
# key, POST /v1/verify checking it; a decode sampled at circuit level; a
# threshold at code capacity, decoding each shot there; a certification too
# long for a call here sent to the command line; and a job that mitigates its
# values. (A decode judged against its error is the MCP server's tests'.)
py 'import json; print(json.dumps({"code": "color:5"}))' > "$tmp/qec-code.json"
post /v1/qec/code "$tmp/qec-code.json" > "$tmp/qec-code.out"
py 'import json,sys; b=json.load(open(sys.argv[1])); c=b["result"]["certificate"]; assert c["distance"]==5 and b["result"]["code"]["n"]==19, c; print("qec code: ok, distance", c["distance"])' "$tmp/qec-code.out"
post /v1/verify "$tmp/qec-code.out" > "$tmp/qec-verify.out"
py 'import json,sys; v=json.load(open(sys.argv[1])); assert v["valid"] is True, v; print("qec certificate verifies: ok")' "$tmp/qec-verify.out"
py 'import json; print(json.dumps({"memory": {"distance": 3, "rounds": 3, "p": 0.004}, "shots": 2000, "decoder": "correlated"}))' > "$tmp/qec-shots.json"
post /v1/qec/decode "$tmp/qec-shots.json" > "$tmp/qec-shots.out"
py 'import json,sys; r=json.load(open(sys.argv[1]))["result"]; assert r["shots"]==2000 and 0 < r["failures"] < 200, r; print("qec circuit-level decode: ok,", r["failures"], "failures")' "$tmp/qec-shots.out"
py 'import json; print(json.dumps({"family": "color", "distances": [3, 5], "p": [0.03], "shots": 1000}))' > "$tmp/qec-threshold.json"
post /v1/qec/threshold "$tmp/qec-threshold.json" > "$tmp/qec-threshold.out"
py 'import json,sys; r=json.load(open(sys.argv[1]))["result"]; assert len(r["points"])==2 and r["below_threshold"][0]["distance_helps"], r; print("qec threshold: ok")' "$tmp/qec-threshold.out"
py 'import json; print(json.dumps({"code": "bb:288"}))' > "$tmp/qec-far.json"
pace
code=$(curl -sS -o "$tmp/qec-far.out" -w '%{http_code}' -X POST "$U/v1/qec/code" -H 'content-type: application/json' --data-binary "@$tmp/qec-far.json")
[ "$code" = 422 ] || { echo "a long certification got HTTP $code"; exit 1; }
py 'import json,sys; d=json.load(open(sys.argv[1])); assert "wai-quantum qec code bb:288" in d["error"], d; print("long certification: ok, sent to the command line")' "$tmp/qec-far.out"
# The variational and algorithm library, in the two submissions a client's
# window has left here (every call above counts): an optimisation of a
# program's parameter by Adam, each step's gradient and value a link of the
# chain the receipt seals, and a program built and checked against its
# reference. A run past the service's bounds is refused before it counts.
ansatz='OPENQASM 3.0;\ninput float t;\nqubit[2] q;\nh q[0];\ncrx(t) q[0], q[1];\n'
py "import json; print(json.dumps({'qasm': '$ansatz', 'params': {'t': 0.4}, 'hamiltonian': ['Z1'], 'optimizer': 'adam', 'steps': 10, 'step_size': 0.2}))" > "$tmp/opt.json"
post /v1/optimize "$tmp/opt.json" > "$tmp/opt.out"
py 'import json,sys
b=json.load(open(sys.argv[1])); r=b["result"]; s=r["steps"]
assert r["schema"]=="wai.quantum.algorithms/1" and len(s)==11 and "gradient" in s[0] and r["chain"]==s[-1]["digest"] and r["start"] != r["chain"], r
assert s[0]["params"]=={"t": 0.4} and r["best"]["value"] < s[0]["value"] and r["precision"]["arithmetic"]=="f64" and len(b["result"]["receipt"]["signature"])==128, r["best"]
print("optimize: ok, E", round(s[0]["value"], 6), "->", round(r["best"]["value"], 6), "chain", r["chain"][:16]+"…")' "$tmp/opt.out"
py 'import json; print(json.dumps({"algorithm": "qpe", "counting": 4, "phase": 0.3125}))' > "$tmp/build.json"
post /v1/build "$tmp/build.json" > "$tmp/build.out"
py 'import json,sys; r=json.load(open(sys.argv[1]))["result"]; assert r["check"]["verdict"]=="equivalent" and r["check"]["how"]=="exact" and r["expected"]["most_likely"][0]["k"]==5, r; print("build: ok, phase estimation checked", r["check"]["how"])' "$tmp/build.out"
py "import json; print(json.dumps({'qasm': '$ansatz', 'params': {'t': 0.4}, 'hamiltonian': ['Z1'], 'steps': 100000}))" > "$tmp/opt-far.json"
code=$(curl -sS -o "$tmp/opt-far.out" -w '%{http_code}' -X POST "$U/v1/optimize" -H 'content-type: application/json' --data-binary "@$tmp/opt-far.json")
[ "$code" = 422 ] || { echo "a long optimisation got HTTP $code"; exit 1; }
py 'import json,sys; d=json.load(open(sys.argv[1])); assert "wai-quantum optimize" in d["error"] and d["code"]=="J031", d; print("long optimisation: ok, sent to the command line")' "$tmp/opt-far.out"
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\nrz(0.3) q[1];\n", "observables": ["Z0 Z1"], "noise": {"depolarizing_2q": 0.02}, "mitigate": "zne"}))' > "$tmp/mitigated.json"
run "$tmp/mitigated.json" "$tmp/mitigated.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); m=d["result"]["mitigation"]; o=m["observables"][0]
assert d["status"]=="done" and m["method"]=="zne" and abs(o["mitigated"]-o["ideal"]) < abs(o["raw"]-o["ideal"]), m
print("mitigated job: ok, raw", round(o["raw"], 4), "mitigated", round(o["mitigated"], 4), "ideal", round(o["ideal"], 4))' "$tmp/mitigated.out"

# An estimate: the program's counts, the frontier, each estimate sealed with
# the service's key in the resource-estimate receipt family, POST /v1/verify
# checking the seals and re-running it, and one too fine sent to the command
# line. (A changed figure failing, and a counts document, are the MCP server's
# tests'.)
py 'import json; print(json.dumps({"qasm": "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", "budget": 1e-3, "qubit": "us-e3", "max_slowdown": 64}))' > "$tmp/estimate.json"
post /v1/estimate "$tmp/estimate.json" > "$tmp/estimate.out"
py 'import json,sys
b=json.load(open(sys.argv[1])); r=b["result"]; c=r["counts"]
assert r["format"]=="wai.quantum.estimate/1" and (c["logical_qubits"], c["toffolis"], c["t_gates"], c["rotations"])==(5, 3, 1, 1), c
assert r["estimate"]["receipt"]["kind"]=="quantum-resource" and r["calibration"]["name"]=="exact_matching", r["calibration"]
f=r["frontier"]; assert all(f[i]["physical_qubits"] > f[i+1]["physical_qubits"] for i in range(len(f)-1)), f
print("estimate: ok,", r["estimate"]["physical_qubits"], "physical qubits,", len(f), "on the frontier")' "$tmp/estimate.out"
post /v1/verify "$tmp/estimate.out" > "$tmp/estimate-verify.out"
py 'import json,sys; v=json.load(open(sys.argv[1])); assert v["valid"] is True and v["reproduced"] is True and v["receipts"] is True, v; print("estimate verifies and re-runs: ok,", v["estimates"], "receipts")' "$tmp/estimate-verify.out"
py 'import json; print(json.dumps({"counts": {"numQubits": 4, "rotationCount": 1000000, "rotationDepth": 10}, "budget": 1e-6}))' > "$tmp/fine-est.json"
pace
code=$(curl -sS -o "$tmp/fine-est.out" -w '%{http_code}' -X POST "$U/v1/estimate" -H 'content-type: application/json' --data-binary "@$tmp/fine-est.json")
[ "$code" = 422 ] || { echo "a too-fine estimate got HTTP $code"; exit 1; }
py 'import json,sys; d=json.load(open(sys.argv[1])); assert "wai-quantum estimate" in d["error"], d; print("fine estimate: ok, sent to the command line")' "$tmp/fine-est.out"
# A job with an intent: its verdict in the result, sealed by the receipt; a
# job that asks to be repaired runs the repair; a repair holds a proposal to
# the flagged lines; a refusal ends with its code.
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n", "shots": 100, "intent": {"distribution": {"00": 0.5, "11": 0.5}}}))' > "$tmp/intent.json"
run "$tmp/intent.json" "$tmp/intent.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); v=d["result"]["verdict"]
assert d["status"]=="done" and v["level"]=="proof" and v["holds"] is True and v["intent"][0]["how"]=="statevector", d
print("intent: ok,", v["level"], "-", v["statement"])' "$tmp/intent.out"
py 'import json,sys; d=json.load(open(sys.argv[1])); json.dump({"request": d["request"], "result": d["result"]}, open(sys.argv[2],"w"))' "$tmp/intent.out" "$tmp/vi.json"
post /v1/verify "$tmp/vi.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True, d; print("intent: the receipt seals the verdict")'
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[2] q\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nmeasure q;\n", "shots": 50, "repair": True, "verdict": True}))' > "$tmp/repair-job.json"
run "$tmp/repair-job.json" "$tmp/repair-job.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and [x["code"] for x in r["verdict"]["repairs"]]==["Q004","Q320"] and "c = measure q;" in r["repaired"]["qasm"], d
print("repaired job: ok,", [x["code"] for x in r["verdict"]["repairs"]])' "$tmp/repair-job.out"
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[1] q;\nfoo q[0];\n", "proposed": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\n"}))' > "$tmp/proposal.json"
post /v1/repair "$tmp/proposal.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["ok"] is False and d["refusal"]["code"]=="J021", d; print("repair: ok, a proposal off the flagged line rejected (J021)")'
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0;\nqubit[1] q;\nrx(0.3) q[0];\n", "method": "stabilizer"}))' > "$tmp/refused.json"
post /v1/run "$tmp/refused.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["error"].endswith("[J011]"), d; print("refusal: ok, ends with its code")'

# The script makes more submissions than a default window's 30; `pace` (above)
# keeps it within whatever window the service gives this client.
#
# A dry run: the plan, its estimates with their basis, the bounds, and whether
# the service would take it; nothing runs.
py 'import json; print(json.dumps({"qasm": "OPENQASM 3.0; qubit[30] q; h q[0]; cx q[0], q[29];", "shots": 100, "dry_run": True}))' > "$tmp/dry.json"
pace 2
post /v1/run "$tmp/dry.json" > "$tmp/dry.out"
py 'import json,sys
d=json.load(open(sys.argv[1]))
assert d["dry_run"] is True and d["ran"] is False and d["accepted"] is True and d["method"]=="stabilizer", d
assert "id" not in d and d["time"]["basis"] and d["bounds"]["job_seconds"] > 0 and ("joules_micro_estimate" in d["energy"] or d["energy"]["unestimated"]), d
print("dry run: ok,", d["method"], "%.1f MiB," % (d["memory_bytes"] / 2**20), "time", d["time"]["seconds"], "-", d["time"]["basis"][:48] + "…")' "$tmp/dry.out"

# A set of programs: one result and receipt per item, one receipt over the
# Merkle root of them all, which POST /v1/verify checks; a changed item fails.
# Its third item is a pulse program run as a job of the method pulse (its
# error bounded, its receipt binding its program), which POST /v1/pulse/check
# then runs again and holds to its result.
py 'import json
pulse = "OPENQASM 3.0;\ndefcalgrammar \"openpulse\";\ncal {\n  extern port d0;\n  extern port a0;\n  frame q0 = newframe(d0, 5.0e9, 0);\n  frame c0 = newframe(a0, 5.0e9, 0);\n}\ndefcal x $0 { play(q0, gaussian(0.4, 64dt, 16dt)); }\ndefcal measure $0 -> bit { return capture_v2(c0, 100dt); }\nbit c;\nx $0;\nc = measure $0;\n"
print(json.dumps({"programs": ["OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; cx q[0], q[1]; c = measure q;", {"qasm": "OPENQASM 3.0; qubit q; rx(0.3) q;", "observables": ["Z0"], "shots": 0}, {"qasm": pulse, "method": "pulse"}], "shots": 100, "seed": 5}))' > "$tmp/set.json"
run "$tmp/set.json" "$tmp/set.out"
py 'import json,sys
d=json.load(open(sys.argv[1])); r=d["result"]
assert d["status"]=="done" and d["method"]=="set" and r["format"]=="wai.quantum.set/1" and (r["count"], r["done"])==(3, 3), d
p=r["items"][2]; a=p["result"]["accuracy"]
assert p["method"]=="pulse" and a["class"]=="bounded" and a["bound"] <= 1e-6 and len(p["result"]["pulse"]["calibration"]["program"])==64, p
json.dump({"request": p["request"], "result": p["result"]}, open(sys.argv[4],"w"))
assert r["items"][0]["method"]=="stabilizer" and abs(r["items"][1]["result"]["expectations"][0]["value"] - 0.955336489125606) < 1e-12, r["items"]
assert r["items"][0]["result"]["provenance"]["software"]["host"].startswith("wai-quantum-service"), r["items"][0]["result"]
json.dump({"request": d["request"], "result": r}, open(sys.argv[2],"w"))
r["items"][0]["result"]["counts"]["00"] = r["items"][0]["result"]["counts"].get("00", 0) + 1
json.dump({"request": d["request"], "result": r}, open(sys.argv[3],"w"))
print("set: ok,", r["count"], "items, root", r["root"][:16]+"…, the pulse item P(1)", round(p["result"]["pulse"]["distribution"]["1"], 6), "within", a["bound"])' "$tmp/set.out" "$tmp/vset.json" "$tmp/tset.json" "$tmp/pulse-check.json"
post /v1/verify "$tmp/vset.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True and d["items"]==3, d; print("set receipt: verifies, every item sealed")'
post /v1/verify "$tmp/tset.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is False, d; print("set receipt: a changed item does not verify")'
post /v1/pulse/check "$tmp/pulse-check.json" > "$tmp/pulse-check.out"
py 'import json,sys; r=json.load(open(sys.argv[1]))["result"]; assert r["verified"] is True and r["reproduced"] is True and r["receipt"], r; print("pulse check: ok, the set item run again here, bit for bit")' "$tmp/pulse-check.out"

# A job's status as server-sent events, to its end; and a poll held until it changes.
id=$(py 'import json,sys; print(json.load(open(sys.argv[1]))["id"])' "$tmp/set.out")
curl -sS --max-time 30 -N "$U/v1/jobs/$id/events" > "$tmp/events.txt"
py 'import json,sys
t=open(sys.argv[1]).read()
assert "event: job" in t, t
data=[l[len("data: "):] for l in t.splitlines() if l.startswith("data: ")]
assert json.loads(data[-1])["status"]=="done", data
print("status events: ok, the stream ends with the whole job")' "$tmp/events.txt"
curl -sS --max-time 30 "$U/v1/jobs/$id?wait=2" | py 'import json,sys; d=json.load(sys.stdin); assert d["status"]=="done" and d["digests"]["algorithm"]=="blake3" and len(d["digests"]["result"])==64, d; print("poll: ok, its digests", d["digests"]["request"][:16]+"…")'

# A sweep over a parameter, by the MCP set tool: every point run, each the
# program written with its value, the set's verdict its weakest item's.
py 'import json; print(json.dumps({"jsonrpc":"2.0","id":8,"method":"tools/call","params":{"name":"run_quantum_set","arguments":{"qasm":"OPENQASM 3.0;\ninput float theta;\nqubit q;\nrx(theta) q;\n","sweep":{"theta":[0.0, 0.5, 1.0]},"observables":["Z0"]}}}))' > "$tmp/mcp-set.json"
mcp "@$tmp/mcp-set.json" > "$tmp/mcp-set.out"
py 'import json,math,sys
r=json.load(open(sys.argv[1]))["result"]; sc=r["structuredContent"]; items=sc["result"]["items"]
assert not r["isError"] and sc["status"]=="done" and sc["verdict"]["level"]=="executed", r
assert [i["point"]["theta"] for i in items]==[0.0, 0.5, 1.0] and all(abs(i["result"]["expectations"][0]["value"] - math.cos(i["point"]["theta"])) < 1e-12 for i in items), items
print("sweep (mcp run_quantum_set): ok,", [round(i["result"]["expectations"][0]["value"], 6) for i in items])' "$tmp/mcp-set.out"

# A dry run the service would refuse, by the MCP tool: nothing ran, and the
# refusal comes with its code and the verdict of what was refused (the
# program reads; the method named cannot take it).
py 'import json; print(json.dumps({"jsonrpc":"2.0","id":7,"method":"tools/call","params":{"name":"dry_run_quantum_job","arguments":{"qasm":"OPENQASM 3.0; qubit[40] q; rx(0.3) q;","amplitudes":["0"*40],"method":"statevector"}}}))' > "$tmp/mcp-dry.json"
pace 2
mcp "@$tmp/mcp-dry.json" | py 'import json,sys; r=json.load(sys.stdin)["result"]; sc=r["structuredContent"]; assert not r["isError"] and sc["accepted"] is False and sc["refusal"]["code"]=="J011" and sc["verdict"]["level"]=="parsed" and sc["verdict"]["refusal"]["code"]=="J011", r; print("mcp dry run refused: ok,", r["content"][0]["text"][:60]+"…")'

# Writing a program for a task. GET /v1 says under `write` whether this service
# has a model. Without one, a write is refused (503, J080), before it counts as
# a submission. With one, SMOKE_WRITE=1 has it write a Bell pair's program,
# which takes minutes with the real model: the write is followed to its end,
# its transcript passes, its receipt verifies and a changed transcript does
# not, and its events give each attempt before the job. Without SMOKE_WRITE=1
# the model is only named.
export SMOKE_WRITE="${SMOKE_WRITE:-0}"
curl -sS --max-time 30 "$U/v1" > "$tmp/write-describe.json"
avail=$(py 'import json,sys; print("1" if json.load(open(sys.argv[1]))["write"]["available"] else "0")' "$tmp/write-describe.json")
py 'import json; print(json.dumps({"task": {"schema": "wai.quantum.exercise/1", "title": "A Bell pair", "task": "Entangle two qubits so that their measured bits always agree.", "starter": ["OPENQASM 3.0;", "include \"stdgates.inc\";", "qubit[2] q;", "bit[2] c;", "c = measure q;"], "intent": {"distribution": {"00": 0.5, "11": 0.5}}}}))' > "$tmp/write.json"
if [ "$avail" = 0 ]; then
  [ "$SMOKE_WRITE" != 1 ] || { echo "FAIL: SMOKE_WRITE=1, but this service has no model to write with" >&2; exit 1; }
  code=$(curl -sS --max-time 30 -o "$tmp/write.out" -w '%{http_code}' -X POST "$U/v1/write" -H 'content-type: application/json' --data-binary "@$tmp/write.json")
  py 'import json,sys; d=json.load(open(sys.argv[1])); assert sys.argv[2]=="503" and d["code"]=="J080", (sys.argv[2], d); print("write: this service has no model, and a write is refused (503, J080)")' "$tmp/write.out" "$code"
elif [ "$SMOKE_WRITE" != 1 ]; then
  py 'import json,sys; w=json.load(open(sys.argv[1]))["write"]; m=w.get("model") or {}; print("write: this service writes with", m.get("name", "a model still loading"), m.get("licence", ""), (m.get("digest") or "")[:16]+"…;", "SMOKE_WRITE=1 runs one")' "$tmp/write-describe.json"
else
  pace
  code=$(curl -sS --max-time 60 -o "$tmp/write.out" -w '%{http_code}' -X POST "$U/v1/write" -H 'content-type: application/json' --data-binary "@$tmp/write.json")
  [ "$code" = 202 ] || { echo "FAIL: POST /v1/write answered $code: $(cat "$tmp/write.out")" >&2; exit 1; }
  wid=$(py 'import json,sys; print(json.load(open(sys.argv[1]))["id"])' "$tmp/write.out")
  # The events, followed in the background from now to the write's end.
  curl -sS -N --max-time 7200 "$U/v1/jobs/$wid/events" > "$tmp/write-events.txt" &
  epid=$!
  secs=$(py 'import json,sys; print(json.load(open(sys.argv[1]))["write"]["limits"]["write_seconds"])' "$tmp/write-describe.json")
  t0=$(date +%s)
  while :; do
    curl -sS --max-time 60 "$U/v1/jobs/$wid?wait=25" > "$tmp/write.job"
    case "$(py 'import json,sys; print(json.load(open(sys.argv[1]))["status"])' "$tmp/write.job")" in
      queued|running) [ $(( $(date +%s) - t0 )) -lt $(( secs * 2 + 600 )) ] || { echo "FAIL: the write $wid did not end" >&2; exit 1; } ;;
      *) break ;;
    esac
  done
  wait "$epid" || true
  py 'import json,sys
j=json.load(open(sys.argv[1])); w=json.load(open(sys.argv[2]))["write"]
assert j["status"]=="done", j
t=j["result"]
assert t["schema"]=="wai.quantum.write/1" and t["passed"] is True and t["program"] and t["model"]["digest"]==w["model"]["digest"], t
assert len(j["timing"])==len(t["attempts"]) and "timing" not in t and "energy" in t and "receipt" in t, j
json.dump({"request": j["request"], "result": t}, open(sys.argv[3],"w"))
json.dump({"request": j["request"], "result": dict(t, passed_round=99)}, open(sys.argv[4],"w"))
print("write", j["id"]+": ok,", t["model"]["name"], "passed after", t["passed_round"], "repair(s) at the level", t["level"]+",", len(t["attempts"]), "attempt(s) in", round(sum(x["reply_seconds"]+x["check_seconds"] for x in j["timing"])), "s")' "$tmp/write.job" "$tmp/write-describe.json" "$tmp/vwrite.json" "$tmp/twrite.json"
  post /v1/verify "$tmp/vwrite.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is True, d; print("write receipt: verifies")'
  post /v1/verify "$tmp/twrite.json" | py 'import json,sys; d=json.load(sys.stdin); assert d["valid"] is False, d; print("write receipt: a changed transcript does not verify")'
  py 'import json,sys
t=open(sys.argv[1]).read(); j=json.load(open(sys.argv[2]))
kinds=[l[len("event: "):] for l in t.splitlines() if l.startswith("event: ")]
assert kinds.count("attempt")==len(j["result"]["attempts"]) and kinds[-1]=="job", kinds
print("write events: ok,", kinds.count("attempt"), "attempt event(s), then the job")' "$tmp/write-events.txt" "$tmp/write.job"
fi

echo "smoke: every check held at $U"