open-standards/wai/wai-quantum-cli/tests/providers.rs
Transaction Science 0f78a99ea4
All checks were successful
standards workspaces build + test / which workspaces this change touches (pull_request) Successful in 8s
WAI energy receipts / does this change touch it (pull_request) Successful in 7s
WAI breaking changes and the conformance runner / WAI breaking · does this change touch WAI (pull_request) Successful in 10s
wai-quantum command line / does this change touch it (pull_request) Successful in 9s
wai-quantum contract / does this change touch it (pull_request) Successful in 8s
wai-quantum docs, course and locales / does this change touch it (pull_request) Successful in 8s
wai-quantum editors / does this change touch it (pull_request) Successful in 7s
wai-quantum for Python / does this change touch it (pull_request) Successful in 7s
wai-quantum service / does this change touch it (pull_request) Successful in 8s
wai-quantum studio and desktop / does this change touch it (pull_request) Successful in 8s
wai-quantum views in agent hosts / does this change touch it (pull_request) Successful in 7s
wai-quantum studio and desktop / desktop app · windows (pull_request) Successful in 10m19s
wai-quantum service / service energy · linux packages (pull_request) Successful in 1m39s
repository checks / lockfiles, advisories, tests in CI, workflow hygiene, spec and section references, public text (pull_request) Successful in 1m37s
WAI breaking changes and the conformance runner / WAI breaking · wire, record and API against main · linux (pull_request) Successful in 39s
WAI breaking changes and the conformance runner / WAI breaking and conform · their tests, and the independent verifiers through wai conform · linux (pull_request) Successful in 2m39s
wai-quantum contract / the policy is linked, and the deprecations page reads (pull_request) Successful in 6s
wai-quantum command line / command line · linux (pull_request) Successful in 3m22s
wai-quantum docs, course and locales / the Studio's words in every locale · linux (pull_request) Successful in 13s
wai-quantum contract / API compatibility · linux (pull_request) Successful in 4m36s
wai-quantum docs, course and locales / the course and the command line's languages · linux (pull_request) Successful in 2m16s
wai-quantum for Python / python · linux (pull_request) Successful in 3m13s
wai-quantum contract / schemas and the committed tool lists · linux (pull_request) Successful in 6m46s
wai-quantum editors / editors · linux (pull_request) Successful in 5m25s
wai-quantum service / service · linux (pull_request) Successful in 4m49s
wai-quantum studio and desktop / studio · browser (pull_request) Successful in 4m11s
wai-quantum views in agent hosts / views · the page and its engine (pull_request) Successful in 3m11s
standards workspaces build + test / macOS energy backends (pull_request) Successful in 6m41s
wai-quantum command line / command line · macos (pull_request) Successful in 1m28s
wai-quantum views in agent hosts / views · a scripted host (pull_request) Successful in 3m32s
wai-quantum studio and desktop / desktop app · linux (pull_request) Successful in 5m57s
wai-quantum for Python / python · macos (pull_request) Successful in 2m4s
WAI energy receipts / energy receipts · linux (pull_request) Successful in 14m46s
wai-quantum service / service · macos (pull_request) Successful in 3m46s
wai-quantum studio and desktop / desktop app · macos (pull_request) Successful in 3m45s
wai-quantum editors / editors · macos (pull_request) Successful in 6m36s
WAI energy receipts / energy receipts · macos (pull_request) Successful in 15m28s
wai-quantum docs, course and locales / every command and snippet in the docs · linux (pull_request) Successful in 20m4s
standards workspaces build + test / joule-code-rs (joule-code/joule-code-rs, joule-code-rs, linux-arm64, true) (pull_request) Successful in 13m3s
WAI energy receipts / energy receipts · linux-arm64 (pull_request) Successful in 29m23s
wai-quantum-cli: a job on a provider is an MCP task
A client that speaks MCP 2025-11-25 can run run_on_provider as a task.
initialize declares tasks (list, cancel, tools/call) from that version
on, and run_on_provider declares execution.taskSupport "optional".
tools/call with task records and sends the job and answers once the
provider has taken it, the job's handle its taskId; a run that is not
free and not confirmed makes no task and sends nothing, its plan in the
error's data.

Every job recorded here is a task, whatever sent it. tasks/get asks the
provider once; tasks/list reads the records and asks no one; tasks/cancel
cancels at the provider and refuses a task that has ended. The ttl is
null, as records are kept.

tasks/result asked before the job has ended is answered when it ends, by
a thread of its own asking at 2 s doubling to a minute, so the stdio
server answers everything else meanwhile. Replies go through one shared
writer. A tasks/cancel here wakes the wait at once; a request the client
cancels (notifications/cancelled) is never answered; at most 64 wait at
once. A process-wide lock keeps a tool call and a wait from both sealing
one job.

Tested against the reference provider over HTTP: the result of a short
queue verifies; ping answers while a long one's result waits; a cancel
ends that wait within 3 s once it asks every 8 s; a cancelled request
gets no reply; tasks outlive the session. The CLI workflow names the new
tests, so a run that skips one fails.

Co-Authored-By: Claude Opus 5.5 <[email protected]>
2026-11-03 17:49:13 -05:00

585 lines
33 KiB
Rust

//! `wai-quantum providers …`, `run --provider` and the MCP tools, as a person
//! and an agent use them, against the reference provider that this binary
//! serves itself (`wai-quantum providers reference`), over a local process
//! and over HTTP. Each run has its own configuration directory, and its
//! credentials go to files there (`WAI_QUANTUM_CREDENTIALS=file`), so no
//! keychain is touched. Nothing leaves this machine and nothing spends money:
//! the priced provider is the reference with a price in its configuration.
#![cfg(feature = "providers")]
use serde_json::{Value, json};
use std::io::{BufRead, BufReader, Write};
use std::path::{Path, PathBuf};
use std::process::{Child, Command, Output, Stdio};
const SECRET: &str = "wai-cli-test-credential-93ad1b";
const BELL: &str = "OPENQASM 3.0;\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n";
fn bin() -> &'static str {
env!("CARGO_BIN_EXE_wai-quantum")
}
/// A configuration directory of its own, and everything printed in it.
struct Home {
dir: PathBuf,
seen: std::cell::RefCell<String>,
}
impl Home {
fn new(tag: &str) -> Home {
let dir = std::env::temp_dir().join(format!("wai-quantum-cli-byoq-{tag}-{}", std::process::id()));
let _ = std::fs::remove_dir_all(&dir);
std::fs::create_dir_all(&dir).unwrap();
Home { dir, seen: Default::default() }
}
fn cmd(&self) -> Command {
let mut c = Command::new(bin());
c.env("XDG_CONFIG_HOME", self.dir.join("config")).env("HOME", &self.dir).env("WAI_QUANTUM_CREDENTIALS", "file").env_remove("WAI_QUANTUM_PROVIDERS").env_remove("WAI_QUANTUM_KEY");
c
}
/// Runs `args` with `stdin`, keeping what it printed.
fn run(&self, args: &[&str], stdin: &str) -> Output {
let mut child = self.cmd().args(args).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::piped()).spawn().unwrap();
child.stdin.take().unwrap().write_all(stdin.as_bytes()).unwrap();
let out = child.wait_with_output().unwrap();
let mut seen = self.seen.borrow_mut();
seen.push_str(&String::from_utf8_lossy(&out.stdout));
seen.push_str(&String::from_utf8_lossy(&out.stderr));
out
}
fn ok(&self, args: &[&str], stdin: &str) -> String {
let out = self.run(args, stdin);
assert!(out.status.success(), "{args:?}: {}\n{}", String::from_utf8_lossy(&out.stdout), String::from_utf8_lossy(&out.stderr));
String::from_utf8(out.stdout).unwrap()
}
fn program(&self) -> String {
let p = self.dir.join("bell.qasm");
std::fs::write(&p, BELL).unwrap();
p.display().to_string()
}
/// Every file under the directory, as text.
fn files(&self) -> String {
fn walk(p: &Path, out: &mut String) {
for e in std::fs::read_dir(p).into_iter().flatten().flatten() {
let p = e.path();
if p.is_dir() {
walk(&p, out);
} else if let Ok(t) = std::fs::read_to_string(&p) {
out.push_str(&format!("== {}\n{t}\n", p.display()));
}
}
}
let mut s = String::new();
walk(&self.dir, &mut s);
s
}
}
impl Drop for Home {
fn drop(&mut self) {
let _ = std::fs::remove_dir_all(&self.dir);
}
}
fn reference_command(extra: &[&str]) -> Vec<String> {
let mut c = vec![bin().to_string(), "providers".into(), "reference".into(), "--stdio".into()];
c.extend(extra.iter().map(|s| s.to_string()));
c
}
fn add_process(h: &Home, name: &str, extra: &[&str], credential: bool) -> String {
let mut args = vec!["providers", "add", name, "--kind", "process"];
if credential {
args.push("--credential-stdin");
}
args.push("--");
let command = reference_command(extra);
let refs: Vec<&str> = command.iter().map(String::as_str).collect();
args.extend(refs);
h.ok(&args, if credential { SECRET } else { "" })
}
#[test]
fn a_person_adds_tests_plans_and_runs_on_a_provider_and_the_receipt_verifies() {
let h = Home::new("person");
let prog = h.program();
assert!(h.ok(&["providers", "list"], "").contains("no providers yet"));
add_process(&h, "lab", &[], false);
let list = h.ok(&["providers", "list", "--json"], "");
let v: Value = serde_json::from_str(&list).unwrap();
assert_eq!((v["providers"][0]["name"].as_str(), v["providers"][0]["protocol"].as_str()), (Some("lab"), Some("wai.quantum.provider/1")));
let d = h.ok(&["providers", "describe", "lab"], "");
assert!(d.contains("reference:line-5, 5 qubits") && d.contains("billing free"), "{d}");
let t = h.ok(&["providers", "test", "lab"], "");
assert!(t.contains("31 passed, 0 failed, 0 skipped"), "{t}");
let free = h.ok(&["providers", "free"], "");
assert!(free.contains("exact") && free.contains("quantum-jobs") && free.contains("lab") && free.contains("free, as it states"), "{free}");
// The dry run: the compile, its check and the estimate, nothing sent.
let dry = h.ok(&["run", &prog, "--shots", "300", "--provider", "lab", "--dry-run"], "");
assert!(dry.contains("compile 2 qubits, 300 shots") && dry.contains("estimate free") && dry.contains("nothing has been sent"), "{dry}");
// The run, sealed with this user's key, and checked by verify.
let body = h.ok(&["run", &prog, "--shots", "300", "--seed", "4", "--provider", "lab", "--json"], "");
let job: Value = serde_json::from_str(&body).unwrap();
assert_eq!((job["method"].as_str(), job["result"]["accuracy"]["class"].as_str()), (Some("provider"), Some("measured")));
assert_eq!(job["request"]["provider"], "lab");
let saved = h.dir.join("job.json");
std::fs::write(&saved, &body).unwrap();
let check = h.ok(&["verify", saved.to_str().unwrap()], "");
assert!(check.contains("signature VALID"), "{check}");
// As a person reads it.
let text = h.ok(&["run", &prog, "--provider", "lab"], "");
assert!(text.contains("accuracy measured") && text.contains("ideal total variation distance") && text.contains("cost free") && text.contains("receipt verifies"), "{text}");
// A job the engine's methods would take and a device cannot is refused.
let out = h.run(&["run", &prog, "--provider", "lab", "-o", "Z0"], "");
assert!(!out.status.success() && String::from_utf8_lossy(&out.stderr).contains("counts"));
assert!(h.ok(&["providers", "remove", "lab"], "").contains("removed lab"));
assert!(!h.run(&["providers", "describe", "lab"], "").status.success());
}
#[test]
fn a_run_that_is_not_free_is_refused_until_the_person_confirms_it() {
let h = Home::new("priced");
let prog = h.program();
add_process(&h, "paid", &["--price", "0.0003", "--currency", "USD"], false);
let out = h.run(&["run", &prog, "--shots", "1000", "--provider", "paid"], "");
assert_eq!(out.status.code(), Some(3), "{}", String::from_utf8_lossy(&out.stderr));
let (o, e) = (String::from_utf8_lossy(&out.stdout), String::from_utf8_lossy(&out.stderr));
assert!(o.contains("estimate 0.3 USD (billed by the shot)") && e.contains("--yes"), "{o}\n{e}");
let body: Value = serde_json::from_str(&h.ok(&["run", &prog, "--shots", "1000", "--provider", "paid", "--yes", "--json"], "")).unwrap();
assert_eq!((body["result"]["cost"]["confirmed"].as_bool(), body["result"]["cost"]["charged"]["amount"].as_str()), (Some(true), Some("0.3")));
// Its conformance check runs no job without leave.
let t = h.ok(&["providers", "test", "paid"], "");
assert!(t.contains("skip the programs run: not every run is free"), "{t}");
}
#[test]
fn a_credential_is_kept_in_a_file_only_its_owner_reads_and_appears_in_nothing_printed() {
let h = Home::new("secret");
let prog = h.program();
let hash = wai_quantum_qpu::reference::credential_hash(SECRET);
let added = add_process(&h, "lab", &["--credential-blake3", &hash], true);
assert!(added.contains("kept in a file only you can read"), "{added}");
let list = h.ok(&["providers", "list"], "");
assert!(list.contains("credential: file (readable by you only; there is no keychain here)"), "{list}");
let run = h.ok(&["run", &prog, "--provider", "lab", "--json"], "");
assert!(run.contains("\"receipt\""));
h.ok(&["providers", "describe", "lab"], "");
h.ok(&["providers", "test", "lab", "--json"], "");
h.ok(&["run", &prog, "--provider", "lab", "--dry-run", "--json"], "");
// The credential is in its own file, readable by its owner only, and in
// nothing else: not the list, not a receipt, not anything printed.
let file = h.dir.join("config/wai/quantum-provider-credentials/lab");
assert_eq!(std::fs::read_to_string(&file).unwrap(), SECRET);
#[cfg(unix)]
{
use std::os::unix::fs::PermissionsExt;
assert_eq!(std::fs::metadata(&file).unwrap().permissions().mode() & 0o777, 0o600);
}
std::fs::write(h.dir.join("run.json"), &run).unwrap();
let everything = h.files().replace(&format!("== {}\n{SECRET}\n", file.display()), "");
assert!(!everything.contains(SECRET), "the credential is in a file other than its own");
assert!(!h.seen.borrow().contains(SECRET), "the credential was printed");
// A credential named by an environment variable is read from it, and kept nowhere.
h.ok(&["providers", "add", "env", "--kind", "process", "--credential-env", "MY_PROVIDER_TOKEN", "--", bin(), "providers", "reference", "--stdio", "--credential-blake3", &hash], "");
let mut c = h.cmd();
let out = c.env("MY_PROVIDER_TOKEN", SECRET).args(["run", &prog, "--provider", "env", "--json"]).output().unwrap();
assert!(out.status.success(), "{}", String::from_utf8_lossy(&out.stderr));
assert!(!String::from_utf8_lossy(&out.stdout).contains(SECRET) && !String::from_utf8_lossy(&out.stderr).contains(SECRET));
// Without it, the provider refuses.
let out = h.run(&["run", &prog, "--provider", "env"], "");
assert!(!out.status.success() && String::from_utf8_lossy(&out.stderr).contains("unauthorized"), "{}", String::from_utf8_lossy(&out.stderr));
// A credential is never a setting.
let out = h.run(&["providers", "add", "bad", "--kind", "http", "--endpoint", "https://example.invalid", "--set", "api_token=x"], "");
assert!(!out.status.success() && String::from_utf8_lossy(&out.stderr).contains("looks like a credential"));
}
/// The reference served over HTTP by this binary, and its URL.
fn serve_http(h: &Home, extra: &[&str]) -> (Child, String) {
let mut child = h.cmd().args(["providers", "reference", "--http", "127.0.0.1:0"]).args(extra).stdout(Stdio::piped()).spawn().unwrap();
let mut line = String::new();
BufReader::new(child.stdout.take().unwrap()).read_line(&mut line).unwrap();
let url = line.trim().strip_prefix("listening ").unwrap_or_else(|| panic!("{line}")).to_string();
(child, url)
}
#[test]
fn a_provider_over_http_runs_with_its_credential_in_the_header() {
if Command::new("curl").arg("--version").output().is_err() {
println!("no curl on this host: the HTTP transport runs where curl is");
return;
}
let h = Home::new("http");
let prog = h.program();
let hash = wai_quantum_qpu::reference::credential_hash(SECRET);
let (mut child, url) = serve_http(&h, &["--credential-blake3", &hash, "--basis", "prxcz", "--all-to-all"]);
h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url, "--credential-stdin"], SECRET);
let t = h.ok(&["providers", "test", "web"], "");
let body = h.ok(&["run", &prog, "--provider", "web", "--json"], "");
let _ = child.kill();
let _ = child.wait();
assert!(t.contains("31 passed, 0 failed"), "{t}");
let job: Value = serde_json::from_str(&body).unwrap();
assert_eq!(job["result"]["compiled"]["basis"], "prxcz");
assert!(!h.seen.borrow().contains(SECRET));
// A credential over plain http to another machine is refused.
h.ok(&["providers", "add", "far", "--kind", "http", "--endpoint", "http://192.0.2.1/q", "--credential-stdin"], SECRET);
let out = h.run(&["run", &prog, "--provider", "far", "--dry-run"], "");
assert!(!out.status.success() && String::from_utf8_lossy(&out.stderr).contains("only over https"), "{}", String::from_utf8_lossy(&out.stderr));
}
/// One MCP message and its answer.
fn rpc(stdin: &mut impl Write, stdout: &mut impl BufRead, id: u64, method: &str, params: Value) -> Value {
writeln!(stdin, "{}", json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })).unwrap();
stdin.flush().unwrap();
let mut line = String::new();
stdout.read_line(&mut line).unwrap();
serde_json::from_str(&line).unwrap_or_else(|e| panic!("{e}: {line}"))
}
#[test]
fn an_agent_lists_estimates_is_refused_and_runs_only_with_confirm_cost() {
let h = Home::new("agent");
add_process(&h, "free-lab", &[], false);
add_process(&h, "paid", &["--price", "0.25", "--currency", "EUR", "--per", "job"], false);
let mut child = h.cmd().args(["mcp", "--demo-key"]).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
rpc(&mut stdin, &mut stdout, 1, "initialize", json!({ "protocolVersion": "2025-06-18" }));
let tools = rpc(&mut stdin, &mut stdout, 2, "tools/list", json!({}));
let names: Vec<String> = tools["result"]["tools"].as_array().unwrap().iter().map(|t| t["name"].as_str().unwrap().to_string()).collect();
for n in ["providers_list", "provider_estimate", "run_on_provider"] {
assert!(names.iter().any(|x| x == n), "{n} not in {names:?}");
}
let call = |stdin: &mut std::process::ChildStdin, stdout: &mut BufReader<std::process::ChildStdout>, id: u64, name: &str, args: Value| rpc(stdin, stdout, id, "tools/call", json!({ "name": name, "arguments": args }))["result"].clone();
let list = call(&mut stdin, &mut stdout, 3, "providers_list", json!({ "free": true }));
assert_eq!(list["isError"], false, "{list}");
assert_eq!(list["structuredContent"]["providers"].as_array().unwrap().len(), 2);
assert_eq!(list["structuredContent"]["free"]["providers"][0]["name"], "free-lab", "{list}");
let est = call(&mut stdin, &mut stdout, 4, "provider_estimate", json!({ "provider": "paid", "qasm": BELL, "shots": 100 }));
assert_eq!((est["structuredContent"]["free"].as_bool(), est["structuredContent"]["estimate"]["amount"].as_str()), (Some(false), Some("0.25")), "{est}");
let refused = call(&mut stdin, &mut stdout, 5, "run_on_provider", json!({ "provider": "paid", "qasm": BELL, "shots": 100 }));
assert_eq!(refused["isError"], true, "{refused}");
assert!(refused["content"][0]["text"].as_str().unwrap().contains("confirm_cost: true"), "{refused}");
let ran = call(&mut stdin, &mut stdout, 6, "run_on_provider", json!({ "provider": "paid", "qasm": BELL, "shots": 100, "confirm_cost": true }));
assert_eq!(ran["isError"], false, "{ran}");
let job = ran["structuredContent"].clone();
let free = call(&mut stdin, &mut stdout, 7, "run_on_provider", json!({ "provider": "free-lab", "qasm": BELL }));
assert_eq!(free["isError"], false, "a free run needs no confirmation: {free}");
let checked = call(&mut stdin, &mut stdout, 8, "verify_quantum_result", json!({ "job": job }));
assert_eq!(checked["structuredContent"]["valid"], true, "{checked}");
// The engine's own run tool does not stand in for a provider.
let stand_in = call(&mut stdin, &mut stdout, 9, "run_quantum_circuit", json!({ "qasm": BELL, "provider": "paid" }));
assert_eq!(stand_in["isError"], true, "{stand_in}");
drop(stdin);
let _ = child.wait();
}
fn jobs(h: &Home) -> Vec<Value> {
let v: Value = serde_json::from_str(&h.ok(&["providers", "jobs", "--json"], "")).unwrap();
v["jobs"].as_array().unwrap().clone()
}
/// What a job's record is for: the command that sent it can stop, the wait can
/// run out, and whoever comes back later collects the job by its handle.
#[test]
fn a_job_outlives_the_command_that_sent_it_and_is_collected_by_its_handle() {
if Command::new("curl").arg("--version").output().is_err() {
println!("no curl on this host: the HTTP transport runs where curl is");
return;
}
let h = Home::new("collect");
let prog = h.program();
let hash = wai_quantum_qpu::reference::credential_hash(SECRET);
let (mut queue, url) = serve_http(&h, &["--polls", "3", "--credential-blake3", &hash]);
h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url, "--credential-stdin"], SECRET);
// Sent, and answered with its handle: no one is waiting.
let sent: Value = serde_json::from_str(&h.ok(&["run", &prog, "--provider", "web", "--shots", "200", "--no-wait", "--json"], "")).unwrap();
assert_eq!((sent["status"].as_str(), sent["method"].as_str()), (Some("running"), Some("provider")), "{sent}");
let id = sent["id"].as_str().unwrap().to_string();
assert!(id.starts_with("pj-"), "{id}");
let listed = jobs(&h);
assert_eq!((listed.len(), listed[0]["id"].as_str(), listed[0]["status"].as_str(), listed[0]["provider"].as_str()), (1, Some(id.as_str()), Some("running"), Some("web")), "{listed:?}");
// Each question is its own process. Still at the provider is exit 2, and says how to come back.
for _ in 0..3 {
let out = h.run(&["providers", "collect", &id], "");
assert_eq!(out.status.code(), Some(2), "{}", String::from_utf8_lossy(&out.stderr));
assert!(String::from_utf8_lossy(&out.stdout).contains(&format!("wai-quantum providers collect {id}")));
}
let body = h.ok(&["providers", "collect", &id, "--json"], "");
let job: Value = serde_json::from_str(&body).unwrap();
assert_eq!((job["status"].as_str(), job["result"]["job"]["polls"].as_u64()), (Some("done"), Some(4)), "{job}");
let saved = h.dir.join("collected.json");
std::fs::write(&saved, &body).unwrap();
let check = h.ok(&["verify", saved.to_str().unwrap()], "");
assert!(check.contains("signature VALID"), "{check}");
// Done is kept: the same job, once the provider and its queue are gone.
let _ = queue.kill();
let _ = queue.wait();
h.ok(&["providers", "remove", "web"], "");
assert_eq!(h.ok(&["providers", "collect", &id, "--json"], "").trim(), body.trim());
// The command is killed while it waits, and the job goes on.
let (mut queue, url) = serve_http(&h, &["--polls", "1000000"]);
h.ok(&["providers", "add", "slow", "--kind", "http", "--endpoint", &url], "");
let mut waiting = h.cmd().args(["run", &prog, "--provider", "slow"]).stdout(Stdio::null()).stderr(Stdio::null()).spawn().unwrap();
let mut again = None;
for _ in 0..300 {
again = jobs(&h).iter().find(|j| j["provider"] == "slow" && j["status"] == "running").cloned();
if again.is_some() {
break;
}
std::thread::sleep(std::time::Duration::from_millis(100));
}
let _ = waiting.kill();
let _ = waiting.wait();
let again = again.expect("the job is recorded while the command waits");
let id = again["id"].as_str().unwrap();
let out = h.run(&["providers", "collect", id], "");
assert_eq!(out.status.code(), Some(2), "{}", String::from_utf8_lossy(&out.stderr));
// A provider that cannot be reached says nothing of the job: exit 75, and it is still recorded.
let _ = queue.kill();
let _ = queue.wait();
let out = h.run(&["providers", "collect", id], "");
assert_eq!(out.status.code(), Some(75), "{}", String::from_utf8_lossy(&out.stderr));
assert!(String::from_utf8_lossy(&out.stderr).contains("nothing about the job has changed"));
assert!(jobs(&h).iter().any(|j| j["id"] == id && j["status"] == "running"));
// A handle that is not one reaches nothing.
let out = h.run(&["providers", "collect", "../quantum-providers"], "");
assert_eq!(out.status.code(), Some(1));
assert!(String::from_utf8_lossy(&out.stderr).contains("is not a job's handle"));
// No credential is in any record, and none was printed.
for e in std::fs::read_dir(h.dir.join("config/wai/provider-jobs")).unwrap().flatten() {
assert!(!std::fs::read_to_string(e.path()).unwrap().contains(SECRET), "{}", e.path().display());
}
assert!(!h.seen.borrow().contains(SECRET));
}
/// The same for an agent: a job it did not wait for, asked after and cancelled by its id.
#[test]
fn an_agent_comes_back_to_a_job_by_its_id_and_cancels_only_by_asking() {
if Command::new("curl").arg("--version").output().is_err() {
println!("no curl on this host: the HTTP transport runs where curl is");
return;
}
let h = Home::new("agent-jobs");
let prog = h.program();
let (mut queue, url) = serve_http(&h, &["--polls", "1"]);
h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url], "");
let (mut long, long_url) = serve_http(&h, &["--polls", "1000000"]);
h.ok(&["providers", "add", "long", "--kind", "http", "--endpoint", &long_url], "");
let first: Value = serde_json::from_str(&h.ok(&["run", &prog, "--provider", "long", "--no-wait", "--json"], "")).unwrap();
let id = first["id"].as_str().unwrap().to_string();
let mut child = h.cmd().args(["mcp", "--demo-key"]).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap();
let mut stdin = child.stdin.take().unwrap();
let mut stdout = BufReader::new(child.stdout.take().unwrap());
rpc(&mut stdin, &mut stdout, 1, "initialize", json!({ "protocolVersion": "2025-06-18" }));
let call = |stdin: &mut std::process::ChildStdin, stdout: &mut BufReader<std::process::ChildStdout>, n: u64, name: &str, args: Value| rpc(stdin, stdout, n, "tools/call", json!({ "name": name, "arguments": args }))["result"].clone();
let list = call(&mut stdin, &mut stdout, 2, "list_provider_jobs", json!({}));
assert_eq!(list["structuredContent"]["jobs"][0]["id"].as_str(), Some(id.as_str()), "{list}");
let asked = call(&mut stdin, &mut stdout, 3, "get_provider_job", json!({ "id": id }));
assert_eq!((asked["isError"].as_bool(), asked["structuredContent"]["status"].as_str()), (Some(false), Some("running")), "{asked}");
let cancelled = call(&mut stdin, &mut stdout, 4, "cancel_provider_job", json!({ "id": id }));
assert_eq!(cancelled["structuredContent"]["status"], "cancelled", "{cancelled}");
// A short queue finishes within the call, and its receipt verifies.
let ran = call(&mut stdin, &mut stdout, 5, "run_on_provider", json!({ "provider": "web", "qasm": BELL }));
assert_eq!(ran["structuredContent"]["status"], "done", "{ran}");
let job = ran["structuredContent"].clone();
let checked = call(&mut stdin, &mut stdout, 6, "verify_quantum_result", json!({ "job": job }));
assert_eq!(checked["structuredContent"]["valid"], true, "{checked}");
let same = call(&mut stdin, &mut stdout, 7, "get_provider_job", json!({ "id": job["id"] }));
assert_eq!(same["structuredContent"], job);
let bad = call(&mut stdin, &mut stdout, 8, "get_provider_job", json!({ "id": "../../secret" }));
assert_eq!(bad["isError"], true, "{bad}");
drop(stdin);
let _ = child.wait();
for c in [&mut queue, &mut long] {
let _ = c.kill();
let _ = c.wait();
}
}
/// `wai-quantum mcp` over standard input and output, its replies in any
/// order: a task's result is answered when its job ends, later than what was
/// asked after it.
struct Mcp {
child: Child,
stdin: Option<std::process::ChildStdin>,
replies: std::sync::mpsc::Receiver<Value>,
early: Vec<Value>,
}
impl Mcp {
fn start(h: &Home) -> Mcp {
let mut child = h.cmd().args(["mcp", "--demo-key"]).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap();
let stdin = child.stdin.take();
let out = BufReader::new(child.stdout.take().unwrap());
let (tx, replies) = std::sync::mpsc::channel();
std::thread::spawn(move || {
for line in out.lines().map_while(Result::ok) {
if tx.send(serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("{e}: {line}"))).is_err() {
break;
}
}
});
Mcp { child, stdin, replies, early: Vec::new() }
}
fn send(&mut self, msg: Value) {
let stdin = self.stdin.as_mut().unwrap();
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
}
fn ask(&mut self, id: u64, method: &str, params: Value) {
self.send(json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }));
}
/// The reply to `id`, if it comes within `within`; others are kept.
fn reply(&mut self, id: u64, within: std::time::Duration) -> Option<Value> {
if let Some(i) = self.early.iter().position(|r| r["id"] == id) {
return Some(self.early.remove(i));
}
let until = std::time::Instant::now() + within;
while let Some(left) = until.checked_duration_since(std::time::Instant::now()) {
match self.replies.recv_timeout(left) {
Ok(r) if r["id"] == id => return Some(r),
Ok(r) => self.early.push(r),
Err(_) => break,
}
}
None
}
fn call(&mut self, id: u64, method: &str, params: Value) -> Value {
self.ask(id, method, params);
self.reply(id, std::time::Duration::from_secs(120)).unwrap_or_else(|| panic!("no reply to {id} ({method})"))
}
/// Closes standard input, and waits for the server to end.
fn close(mut self) {
drop(self.stdin.take());
let _ = self.child.wait();
}
}
/// A test that fails part way leaves no server running.
impl Drop for Mcp {
fn drop(&mut self) {
drop(self.stdin.take());
let _ = self.child.kill();
let _ = self.child.wait();
}
}
/// A child process ended with its test, however the test ends.
struct Reaped(Child);
impl Drop for Reaped {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
/// A job on a provider as an MCP task: made at once, its id the job's handle;
/// its result waited for while the server answers everything else; cancelled
/// by asking, which ends the wait; a result the client gave up on never
/// answered; and the task still there for the next session.
#[test]
fn an_agent_runs_on_a_provider_as_a_task_and_the_server_answers_while_it_waits() {
if Command::new("curl").arg("--version").output().is_err() {
println!("no curl on this host: the HTTP transport runs where curl is");
return;
}
use std::time::{Duration, Instant};
let h = Home::new("tasks");
let (web, url) = serve_http(&h, &["--polls", "2"]);
let _web = Reaped(web);
h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url], "");
let (long, long_url) = serve_http(&h, &["--polls", "1000000"]);
let _long = Reaped(long);
h.ok(&["providers", "add", "long", "--kind", "http", "--endpoint", &long_url], "");
add_process(&h, "paid", &["--price", "0.25", "--currency", "EUR", "--per", "job"], false);
let mut m = Mcp::start(&h);
// Tasks are declared to a client whose protocol has them.
let before = m.call(1, "initialize", json!({ "protocolVersion": "2025-06-18" }));
assert!(before["result"]["capabilities"].get("tasks").is_none(), "{before}");
let init = m.call(2, "initialize", json!({ "protocolVersion": "2025-11-25" }));
assert_eq!(init["result"]["capabilities"]["tasks"], json!({ "list": {}, "cancel": {}, "requests": { "tools": { "call": {} } } }), "{init}");
let tools = m.call(3, "tools/list", json!({}));
for t in tools["result"]["tools"].as_array().unwrap() {
let want = if t["name"] == "run_on_provider" { json!({ "taskSupport": "optional" }) } else { Value::Null };
assert_eq!(t["execution"], want, "{}", t["name"]);
}
let task = |m: &mut Mcp, id: u64, provider: &str| {
let made = m.call(id, "tools/call", json!({ "name": "run_on_provider", "arguments": { "provider": provider, "qasm": BELL, "shots": 100 }, "task": { "ttl": 60000 } }));
let t = made["result"]["task"].clone();
assert_eq!((t["status"].as_str(), t["ttl"].is_null()), (Some("working"), true), "{made}");
let id = t["taskId"].as_str().unwrap_or_else(|| panic!("{made}")).to_string();
assert!(id.starts_with("pj-"), "{id}");
id
};
// A short queue: the result is the job once it has ended, naming its task, and its receipt verifies.
let quick = task(&mut m, 4, "web");
m.ask(5, "tasks/result", json!({ "taskId": quick }));
let r = m.reply(5, Duration::from_secs(120)).expect("the result comes when the job ends");
let r = &r["result"];
assert_eq!((r["isError"].as_bool(), r["structuredContent"]["status"].as_str(), r["structuredContent"]["id"].as_str()), (Some(false), Some("done"), Some(quick.as_str())), "{r}");
assert_eq!(r["_meta"]["io.modelcontextprotocol/related-task"], json!({ "taskId": quick }));
let checked = m.call(6, "tools/call", json!({ "name": "verify_quantum_result", "arguments": { "job": r["structuredContent"] } }));
assert_eq!(checked["result"]["structuredContent"]["valid"], true, "{checked}");
assert_eq!(m.call(7, "tasks/get", json!({ "taskId": quick }))["result"]["status"], "completed");
let again = m.call(8, "tasks/cancel", json!({ "taskId": quick }));
assert_eq!(again["error"]["code"], -32602, "a task that has ended is not cancelled: {again}");
// A long queue: while its result waits, the server answers everything else.
let slow = task(&mut m, 10, "long");
m.ask(11, "tasks/result", json!({ "taskId": slow }));
let asked = Instant::now();
assert_eq!(m.call(12, "ping", json!({})), json!({ "jsonrpc": "2.0", "id": 12, "result": {} }));
assert!(asked.elapsed() < Duration::from_secs(20), "the waiting result held the server for {:?}", asked.elapsed());
assert_eq!(m.call(13, "tasks/get", json!({ "taskId": slow }))["result"]["status"], "working");
assert!(m.reply(11, Duration::from_millis(300)).is_none(), "the result came before its job ended");
// Cancelled by asking: the waiting result ends at once, with the cancelled
// job. By 7 s the wait asks every 8 s, so an answer within 3 s is the cancel's.
std::thread::sleep(Duration::from_secs(7).saturating_sub(asked.elapsed()));
let c = m.call(14, "tasks/cancel", json!({ "taskId": slow }));
assert_eq!(c["result"]["status"], "cancelled", "{c}");
let woke = Instant::now();
let r = m.reply(11, Duration::from_secs(30)).expect("a cancel ends the wait");
assert!(woke.elapsed() < Duration::from_secs(3), "the wait slept through the cancel: {:?}", woke.elapsed());
assert_eq!((r["result"]["isError"].as_bool(), r["result"]["structuredContent"]["status"].as_str()), (Some(false), Some("cancelled")), "{r}");
// A result the client gave up on is never answered.
let dropped = task(&mut m, 20, "long");
m.ask(21, "tasks/result", json!({ "taskId": dropped }));
m.send(json!({ "jsonrpc": "2.0", "method": "notifications/cancelled", "params": { "requestId": 21, "reason": "the agent moved on" } }));
assert_eq!(m.call(22, "tasks/cancel", json!({ "taskId": dropped }))["result"]["status"], "cancelled");
m.call(23, "ping", json!({}));
assert!(m.reply(21, Duration::from_secs(3)).is_none(), "a cancelled request was answered");
// Every job recorded here is a task, the oldest first.
let list = m.call(24, "tasks/list", json!({}));
let seen: Vec<(String, String)> = list["result"]["tasks"].as_array().unwrap().iter().map(|t| (t["taskId"].as_str().unwrap().to_string(), t["status"].as_str().unwrap().to_string())).collect();
assert_eq!(seen, [(quick.clone(), "completed".to_string()), (slow.clone(), "cancelled".to_string()), (dropped.clone(), "cancelled".to_string())], "{list}");
// What makes no task: a tool that is not a run, a run not yet agreed to, a task that is not one.
assert_eq!(m.call(30, "tools/call", json!({ "name": "critique_qasm", "arguments": { "qasm": BELL }, "task": {} }))["error"]["code"], -32601);
let refused = m.call(31, "tools/call", json!({ "name": "run_on_provider", "arguments": { "provider": "paid", "qasm": BELL }, "task": {} }));
assert_eq!(refused["error"]["code"], -32602, "{refused}");
assert!(refused["error"]["message"].as_str().unwrap().contains("confirm_cost: true"), "{refused}");
assert_eq!(refused["error"]["data"]["plan"]["estimate"]["amount"], "0.25", "{refused}");
assert_eq!(m.call(32, "tasks/list", json!({}))["result"]["tasks"].as_array().unwrap().len(), 3, "a refused run is not recorded");
for bad in ["pj-0000000000000000", "../quantum-providers", ""] {
assert_eq!(m.call(33, "tasks/get", json!({ "taskId": bad }))["error"]["code"], -32602, "{bad:?}");
}
assert_eq!(m.call(34, "tasks/list", json!({ "cursor": "2" }))["error"]["code"], -32602);
// A task outlives the session that made it.
let kept = task(&mut m, 40, "long");
m.close();
let mut m = Mcp::start(&h);
m.call(1, "initialize", json!({ "protocolVersion": "2025-11-25" }));
assert_eq!(m.call(2, "tasks/get", json!({ "taskId": kept }))["result"]["status"], "working");
assert_eq!(m.call(3, "tasks/cancel", json!({ "taskId": kept }))["result"]["status"], "cancelled");
m.close();
}