//! `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();
}