open-standards/wai/wai-quantum-service/tests/depth.rs

//! The service itself, started here, sent programs nested thousands deep in
//! every place a client can put one: the job's own program and an intent's
//! reference (`POST /v1/jobs`, `POST /v1/run`, a dry run), the MCP run tool,
//! the checks, and an error-correction circuit. A stack overflow in any of
//! them would take the whole process down, on a POST of a few kilobytes. Each
//! is refused with the reader's code, and the service still answers
//! `/v1/health` after each.
//!
//! And a request whose references would take long to read is refused
//! within one budget, in bounded time, and counts toward its client's
//! window as every refused request does.

use serde_json::{Value, json};
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::Path;
use std::process::{Child, Command, Stdio};
use std::time::{Duration, Instant};


/// One HTTP/1.1 exchange: the status, the headers and the body.
fn http(port: u16, method: &str, path: &str, body: Option<&Value>) -> (u16, String, String) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
    let text = body.map(Value::to_string).unwrap_or_default();
    write!(s, "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
    let mut out = String::new();
    let _ = s.read_to_string(&mut out);
    let status = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);
    let (head, body) = out.split_once("\r\n\r\n").map(|(h, b)| (h.to_string(), b.to_string())).unwrap_or_default();
    (status, head, body)
}

fn json_of(port: u16, method: &str, path: &str, body: Option<&Value>) -> (u16, Value) {
    let (code, _, text) = http(port, method, path, body);
    (code, serde_json::from_str(&text).unwrap_or_else(|e| panic!("{method} {path}: {code}: {e}: {text}")))
}

/// The service, its store at `dir`, `burst` submissions a window, once it
/// answers: its process and the port it bound.
fn start(dir: &Path, burst: u32) -> (Child, u16) {
    start_with(dir, burst, &[])
}

/// [`start`], with `env` set beside.
fn start_with(dir: &Path, burst: u32, env: &[(&str, &str)]) -> (Child, u16) {
    start_from(Path::new(env!("CARGO_BIN_EXE_wai-quantum-service")), dir, burst, env)
}

/// [`start_with`], the service started from the binary at `bin`. It binds a
/// port the system chooses, and says which in its log, so no two services
/// started at once (by these tests or anything else on the host) race for one.
fn start_from(bin: &Path, dir: &Path, burst: u32, env: &[(&str, &str)]) -> (Child, u16) {
    let mut cmd = Command::new(bin);
    cmd.arg("serve");
    launch(cmd, dir, burst, env)
}

/// [`start_with`], the service started by a shell that runs `limit` first
/// (`ulimit -n 256`, say).
#[cfg(unix)]
fn start_limited(dir: &Path, limit: &str, env: &[(&str, &str)]) -> (Child, u16) {
    let mut cmd = Command::new("sh");
    cmd.arg("-c").arg(format!("{limit} && exec \"$0\" serve")).arg(env!("CARGO_BIN_EXE_wai-quantum-service"));
    launch(cmd, dir, burst_of(env), env)
}

#[cfg(unix)]
fn burst_of(env: &[(&str, &str)]) -> u32 {
    env.iter().find(|(k, _)| *k == "WAI_QSVC_CLIENT_BURST").and_then(|(_, v)| v.parse().ok()).unwrap_or(1000)
}

/// Every service log written under `dir`.
fn logs(dir: &Path) -> String {
    let mut out = String::new();
    for e in std::fs::read_dir(dir).into_iter().flatten().flatten() {
        if e.file_name().to_string_lossy().starts_with("service-") {
            out += &std::fs::read_to_string(e.path()).unwrap_or_default();
        }
    }
    out
}

/// `cmd`, the service, started over the store at `dir`.
fn launch(mut cmd: Command, dir: &Path, burst: u32, env: &[(&str, &str)]) -> (Child, u16) {
    static STARTS: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
    let path = dir.join(format!("service-{}.log", STARTS.fetch_add(1, std::sync::atomic::Ordering::SeqCst)));
    cmd.envs(env.iter().copied())
        .env("WAI_QSVC_ADDR", "127.0.0.1:0")
        .env("WAI_QSVC_STORE", dir.join("store"))
        .env("WAI_QSVC_KEY", dir.join("signing.key"))
        .env("WAI_QSVC_JOB_GB", "2")
        .env("WAI_QSVC_BUDGET_GB", "4")
        .env("WAI_QSVC_WORKERS", "1")
        .env("WAI_QSVC_ENERGY", "0")
        .env("WAI_QSVC_CLIENT_BURST", burst.to_string())
        .stdout(Stdio::null())
        .stderr(std::fs::File::create(&path).map(Stdio::from).unwrap_or_else(|_| Stdio::null()));
    // A binary this test process has just written (a copy to replace) can
    // be busy for a moment: a child another test forked between the write's
    // open and its close holds the file open for writing until it execs
    // (`ETXTBSY`, Linux), whatever close-on-exec says. Started again shortly.
    let tried = Instant::now();
    let mut child = loop {
        match cmd.spawn() {
            Err(e) if e.kind() == std::io::ErrorKind::ExecutableFileBusy && tried.elapsed() < Duration::from_secs(10) => std::thread::sleep(Duration::from_millis(50)),
            started => break started.expect("the service starts"),
        }
    };
    let log = || std::fs::read_to_string(&path).unwrap_or_default();
    let t = Instant::now();
    let port = loop {
        if let Some(p) = log().lines().find_map(|l| l.split(" on 127.0.0.1:").nth(1)?.split(',').next()?.trim().parse::<u16>().ok()) {
            break p;
        }
        if let Ok(Some(status)) = child.try_wait() {
            panic!("the service ended as it started ({status}): {}", log());
        }
        assert!(t.elapsed() < Duration::from_secs(120), "the service did not start in 120 s: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    };
    while TcpStream::connect(("127.0.0.1", port)).is_err() || http(port, "GET", "/v1/health", None).0 != 200 {
        if let Ok(Some(status)) = child.try_wait() {
            panic!("the service ended ({status}): {}", log());
        }
        assert!(t.elapsed() < Duration::from_secs(120), "the service did not answer in 120 s: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    }
    (child, port)
}

/// The service is up and answering, and its process has not ended.
fn alive(port: u16, svc: &mut Child, after: &str) {
    assert!(svc.try_wait().unwrap().is_none(), "the service ended after {after}");
    let (code, h) = json_of(port, "GET", "/v1/health", None);
    assert_eq!((code, h["ok"].as_bool()), (200, Some(true)), "after {after}: {h}");
}

fn dir(name: &str) -> std::path::PathBuf {
    let d = std::env::temp_dir().join(format!("wai-qsvc-{name}-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&d);
    std::fs::create_dir_all(&d).unwrap();
    d
}

const HEAD: &str = "OPENQASM 3.0;\nqubit[1] q;\nbit[1] c;\n";

/// `rz(((…0.5…))) q[0];`, `n` parentheses deep: about 2n bytes.
fn parens(n: usize) -> String {
    format!("{HEAD}rz({}0.5{}) q[0];\nc = measure q;\n", "(".repeat(n), ")".repeat(n))
}

#[test]
fn programs_nested_thousands_deep_are_refused_and_the_service_runs_on() {
    let d = dir("depth");
    let (mut svc, port) = start(&d, 1000);
    let deep = parens(5_000);
    let blocks = format!("{HEAD}{}x q[0];{}\nc = measure q;\n", "{ ".repeat(5_000), " }".repeat(5_000));
    let power = format!("{HEAD}rz(1{}) q[0];\nc = measure q;\n", " ** 1".repeat(50_000));
    let refused = |code: u16, body: &Value, what: &str| {
        assert!((400..500).contains(&code), "{what}: {code} {body}");
        let text = body.to_string();
        assert!(text.contains("Q006"), "{what}: {body}");
    };

    // The job's own program, through both submissions and a dry run.
    for (path, qasm, what) in [("/v1/jobs", &deep, "parentheses"), ("/v1/run", &blocks, "blocks"), ("/v1/jobs", &power, "a power")] {
        let (code, b) = json_of(port, "POST", path, Some(&json!({ "qasm": qasm, "shots": 10 })));
        refused(code, &b, what);
        alive(port, &mut svc, what);
    }
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": deep, "shots": 10, "dry_run": true })));
    assert_eq!(code, 200, "{b}");
    assert_eq!(b["accepted"], false, "{b}");
    refused(400, &b["refusal"], "a dry run");
    alive(port, &mut svc, "a dry run");

    // An intent's reference: `rz(((((…0…))))) q[0];`.
    let reference = format!("{HEAD}rz({}0{}) q[0];\n", "(".repeat(5_000), ")".repeat(5_000));
    let bell = format!("{HEAD}h q[0];\nc = measure q;\n");
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": bell, "shots": 10, "intent": { "equivalent": reference } })));
    refused(code, &b, "a reference");
    alive(port, &mut svc, "a reference");
    // A set's items, and a sweep's generator.
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "programs": [bell, deep], "shots": 10 })));
    refused(code, &b, "a set");
    alive(port, &mut svc, "a set");
    let generator = format!("OPENQASM 3.0;\ninput int n;\nconst int a = {}n{};\nqubit[1] q;\nx q[0];\n", "(".repeat(5_000), ")".repeat(5_000));
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": generator, "sweep": { "n": [1, 2] }, "shots": 10 })));
    refused(code, &b, "a sweep");
    alive(port, &mut svc, "a sweep");

    // The MCP run tool, and a run as a task.
    let call = |args: Value, task: bool| {
        let mut params = json!({ "name": "run_quantum_circuit", "arguments": args });
        if task {
            params["task"] = json!({ "ttl": 60000 });
        }
        json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": params })))
    };
    let (code, b) = call(json!({ "qasm": deep, "shots": 10 }), false);
    assert_eq!((code, b["result"]["isError"].as_bool()), (200, Some(true)), "{b}");
    assert!(b.to_string().contains("Q006"), "{b}");
    alive(port, &mut svc, "the MCP run tool");
    let (code, b) = call(json!({ "qasm": bell, "shots": 10, "intent": { "equivalent": reference } }), true);
    assert_eq!(code, 200, "{b}");
    assert!(b["error"].is_object() && b.to_string().contains("Q006"), "{b}");
    alive(port, &mut svc, "an MCP task");

    // The checks: each reads its programs, past its count, off the runtime.
    for (path, body) in [
        ("/v1/critique", json!({ "qasm": deep })),
        ("/v1/equivalent", json!({ "a": deep, "b": bell })),
        ("/v1/compile", json!({ "qasm": deep })),
        ("/v1/estimate", json!({ "qasm": deep })),
        ("/v1/shortcuts", json!({ "qasm": blocks })),
    ] {
        let (code, b) = json_of(port, "POST", path, Some(&body));
        assert!(code < 500, "{path}: {code} {b}");
        assert!(b.to_string().contains("Q006"), "{path}: {b}");
        alive(port, &mut svc, path);
    }

    // An error-correction circuit's blocks, nested fifty thousand deep (as
    // deep as a body of a megabyte holds).
    let circuit = format!("{}H 0\n{}", "REPEAT 1 {\n".repeat(50_000), "}\n".repeat(50_000));
    let (code, b) = json_of(port, "POST", "/v1/qec/decode", Some(&json!({ "circuit": circuit, "shots": 10 })));
    assert!((400..500).contains(&code) && b.to_string().contains("nested deeper"), "{code} {b}");
    alive(port, &mut svc, "a circuit's blocks");

    // Nested to what people write, a job still runs.
    let (code, b) = json_of(port, "POST", "/v1/run", Some(&json!({ "qasm": parens(30), "shots": 10, "seed": 1 })));
    assert_eq!((code, b["status"].as_str()), (200, Some("done")), "{b}");

    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// References that each take long to read, many of them: refused within one
/// budget for the whole request, each read once, in bounded time, as too
/// large for the service, and counted toward the client's window, so a
/// client cannot send such requests without end.
#[test]
fn a_request_with_heavy_references_is_refused_in_bounded_time_and_counted() {
    let d = dir("budget");
    let (mut svc, port) = start(&d, 4);
    // Each reference loops five million times on reading, and each differs,
    // so none is read from another's reading.
    let heavy = |k: usize| format!("OPENQASM 3.0;\nqubit[1] q;\nint a = 0;\nfor int i in [0:4999999] {{ a += 1; }}\n// {k}\nx q[0];\n");
    let job = json!({
        "qasm": format!("{HEAD}x q[0];\n"),
        "shots": 10,
        "intent": (0..16).map(|k| json!({ "equivalent": heavy(k) })).collect::<Vec<_>>(),
    });
    let t = Instant::now();
    let (code, _, body) = http(port, "POST", "/v1/jobs", Some(&job));
    let took = t.elapsed();
    assert_eq!(code, 422, "{body}");
    assert!(body.contains("J031") && body.contains("steps"), "{body}");
    // One budget is about as long as one program the reader takes at its
    // bound; sixteen references read three or four times each took minutes.
    assert!(took < Duration::from_secs(60), "{took:?}");
    alive(port, &mut svc, "heavy references");

    // Too many intents are refused before any is read.
    let many = json!({ "qasm": format!("{HEAD}x q[0];\n"), "intent": (0..17).map(|k| json!({ "equivalent": heavy(k) })).collect::<Vec<_>>() });
    let t = Instant::now();
    let (code, _, body) = http(port, "POST", "/v1/jobs", Some(&many));
    assert!(code == 400 && body.contains("J020") && body.contains("at most 16"), "{code} {body}");
    assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed());

    // Two refused requests and two deep ones count: the fourth is the
    // window's last, and the fifth is refused for the window.
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": parens(5_000), "shots": 10 })));
    assert!(code == 400 && b.to_string().contains("Q006"), "{code} {b}");
    let (code, head, _) = http(port, "POST", "/v1/jobs", Some(&json!({ "qasm": parens(5_000), "shots": 10 })));
    assert_eq!(code, 400, "{head}");
    assert!(head.to_ascii_lowercase().contains("ratelimit: \"submissions\";r=0"), "{head}");
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": format!("{HEAD}x q[0];\n"), "shots": 10 })));
    assert!(code == 429 && b.to_string().contains("J030"), "{code} {b}");
    alive(port, &mut svc, "the window");

    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// Memory, through every door that reads a program: a declaration of a
/// hundred bytes that asks for 137 GB, an array of 10^10 booleans, a bit
/// register of 4·10^9 bits, resets of a wide register in a loop. Each ended
/// the process (an allocation that fails aborts it) or grew it without end.
/// Now each is refused (`Q130`) and the service answers `/v1/health` after.
/// A bit string wider than an integer is a refusal, not a 500; a run whose
/// MCP arguments are not an object, a refusal, not a dropped connection.
#[test]
fn programs_that_would_hold_without_limit_are_refused_and_the_service_runs_on() {
    let d = dir("memory");
    let (mut svc, port) = start(&d, 1000);
    let program = |body: &str| format!("OPENQASM 3.0;\nqubit[1] q;\n{body}x q[0];\n");
    for body in [
        "array[int[8], 4294967295] a;\n",
        "array[bool, 100000, 100000] a;\n",
        "bit[4000000000] c;\n",
        "qubit[60000] r;\nfor int i in [0:999999] reset r;\n",
    ] {
        let qasm = program(body);
        for (path, req) in [("/v1/jobs", json!({ "qasm": qasm, "dry_run": true })), ("/v1/jobs", json!({ "qasm": qasm, "shots": 10 })), ("/v1/critique", json!({ "qasm": qasm }))] {
            let t = Instant::now();
            let (code, b) = json_of(port, "POST", path, Some(&req));
            assert!(code < 500 && b.to_string().contains("Q130"), "{path} {body}: {code} {b}");
            assert!(t.elapsed() < Duration::from_secs(60), "{path} {body}: {:?}", t.elapsed());
            alive(port, &mut svc, body);
        }
    }
    let wide = program(&format!("int[32] x = \"{}\";\n", "1".repeat(200)));
    for (path, req) in [("/v1/jobs", json!({ "qasm": wide, "dry_run": true })), ("/v1/jobs", json!({ "qasm": wide, "shots": 10 })), ("/v1/critique", json!({ "qasm": wide }))] {
        let (code, b) = json_of(port, "POST", path, Some(&req));
        assert!(code < 500 && b.to_string().contains("Q112"), "{path}: {code} {b}");
    }
    for args in [json!([1, 2]), json!(5), json!("x")] {
        let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "arguments": args } })));
        assert_eq!(code, 200, "{b}");
        assert!(b["result"]["isError"] == true || b["error"].is_object(), "{b}");
        alive(port, &mut svc, "non-object arguments");
    }
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The reading child's own caps, what holds where a limit of the reader's
/// might not: given 100 MiB, a set whose programs each read within the
/// reader's limits but together hold more is refused for its memory, as is an error-correction circuit that unrolls past it;
/// the service runs on.
#[test]
fn a_reading_past_its_memory_is_stopped_and_refused_and_the_service_runs_on() {
    let d = dir("readcap");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_GB", "0.1"), ("WAI_QSVC_READ_SECS", "120")]);
    let big = |k: u32| format!("OPENQASM 3.0;\nqubit[2] q;\nfor int i in [0:999998] {{ x q[0]; x q[1]; }}\n// {k}\n");
    let t = Instant::now();
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "programs": (0..2).map(big).collect::<Vec<_>>(), "shots": 10 })));
    assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("MiB"), "{code} {b}");
    assert!(t.elapsed() < Duration::from_secs(120), "{:?}", t.elapsed());
    alive(port, &mut svc, "a set past the reading's memory");
    let circuit = "H 0 1 2 3 4 5 6 7\nREPEAT 100000000 {\n    X_ERROR(0.1) 0 1 2 3 4 5 6 7\n}\nM 0\nDETECTOR rec[-1]\n";
    let t = Instant::now();
    let (code, b) = json_of(port, "POST", "/v1/qec/decode", Some(&json!({ "circuit": circuit, "shots": 10 })));
    assert!(code == 422 && b.to_string().contains("J031"), "{code} {b}");
    assert!(t.elapsed() < Duration::from_secs(150), "{:?}", t.elapsed());
    alive(port, &mut svc, "a circuit past the reading's memory");
    svc.kill().unwrap();
    svc.wait().unwrap();
    // A sweep written out comes back into the service: held too, however
    // small its request, as that memory is the one every job shares.
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_ANSWER_MIB", "1")]);
    let body = "x q;\n".repeat(500);
    let sweep = json!({ "qasm": format!("OPENQASM 3.0;\ninput int n;\nqubit q;\nfor int i in [1:n] {{\n{body}}}\n"), "sweep": { "n": (0..100).map(|k| 100 + k).collect::<Vec<_>>() }, "shots": 1 });
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep));
    assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("gave more than"), "{code} {b}");
    alive(port, &mut svc, "a sweep past what a reading gives");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A client told to try again shortly spends nothing by doing so: refused
/// for its jobs at once, before anything is counted or read, its window
/// stays whole however often it obeys `Retry-After`; and a job sent again
/// under its key, its window spent, is still its job, given as OpenQASM or
/// from another format.
#[test]
fn a_client_told_to_retry_spends_nothing_and_its_key_s_job_comes_back_past_the_window() {
    let d = dir("retry");
    let (mut svc, port) = start_with(&d, 4, &[("WAI_QSVC_CLIENT_ACTIVE", "1")]);
    let left = |head: &str| head.to_ascii_lowercase().lines().find_map(|l| l.strip_prefix("ratelimit: \"submissions\";r=").map(|r| r.split(';').next().unwrap_or("").to_string())).unwrap_or_default();
    let post = |path: &str, body: &Value, key: Option<&str>| {
        let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
        s.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
        let text = body.to_string();
        let k = key.map(|k| format!("Idempotency-Key: {k}\r\n")).unwrap_or_default();
        write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\n{k}Content-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
        let mut out = String::new();
        let _ = s.read_to_string(&mut out);
        let code: u16 = out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0);
        let (head, body) = out.split_once("\r\n\r\n").map(|(h, b)| (h.to_string(), b.to_string())).unwrap_or_default();
        (code, head, body)
    };
    // A job that runs long, its answer lost: sent under a key, given from a
    // source in OpenQASM.
    let mut long = String::from("OPENQASM 3.0;\nqubit[24] q;\nh q;\n");
    for k in 0..6000 {
        long += &format!("rx(0.{}) q[{}];\ncx q[{}], q[{}];\n", k % 9 + 1, k % 24, k % 24, (k + 1) % 24);
    }
    let job = json!({ "source": { "format": "qasm", "program": long }, "observables": ["Z0"] });
    let (code, head, first) = post("/v1/jobs", &job, Some("lost-answer"));
    assert_eq!(code, 202, "{first}");
    assert_eq!(left(&head), "3", "{head}");
    let id = serde_json::from_str::<Value>(&first).unwrap()["id"].as_str().unwrap().to_string();
    // Told twenty times to wait for it: nothing spent.
    let bell = json!({ "qasm": "OPENQASM 3.0; qubit q; h q;", "shots": 10 });
    for _ in 0..20 {
        let (code, head, body) = post("/v1/jobs", &bell, None);
        assert!(code == 429 && body.contains("wait for one to finish"), "{code} {body}");
        assert_eq!(left(&head), "3", "{head}");
        // Sent again at that pace, what counts would never spend the window.
        assert!(head.to_ascii_lowercase().contains("retry-after: 150"), "{head}");
    }
    // Its window spent by refusals that do count,
    for _ in 0..3 {
        let (code, _, body) = post("/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; foo q;" }), None);
        assert_eq!(code, 200, "{body}");
    }
    let (code, _, _) = post("/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }), None);
    assert_eq!(code, 429);
    // its job, sent again under its key, still comes back to it.
    let (code, _, again) = post("/v1/jobs", &job, Some("lost-answer"));
    assert_eq!(code, 202, "{again}");
    assert_eq!(serde_json::from_str::<Value>(&again).unwrap()["id"], id.as_str(), "{again}");
    let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None);
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// An index in a few bytes of text would make what is as long as the index:
/// a decoder's table for 10^14 detectors, in 79 bytes. Each way of naming one
/// is refused for what it is, and the service runs on.
#[test]
fn an_index_past_what_a_model_holds_is_refused_and_the_service_runs_on() {
    let d = dir("indices");
    let (mut svc, port) = start(&d, 1000);
    for body in [
        json!({ "model": "shift_detectors 100000000000000\ndetector(1) D0\n", "detectors": [] }),
        json!({ "model": "shift_detectors 100000000000000\ndetector D0\n", "detectors": [] }),
        json!({ "model": "detector(1) D4294967294\n", "detectors": [] }),
        json!({ "circuit": "M 4000000000\nDETECTOR rec[-1]\n", "shots": 10 }),
    ] {
        let (code, b) = json_of(port, "POST", "/v1/qec/decode", Some(&body));
        assert!((400..500).contains(&code) && b.to_string().contains("past the"), "{body}: {code} {b}");
        alive(port, &mut svc, "an index");
    }
    let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "qec_decode", "arguments": { "model": "shift_detectors 100000000000000\ndetector(1) D0\n", "detectors": [] } } })));
    assert!(code == 200 && b.to_string().contains("past the"), "{code} {b}");
    alive(port, &mut svc, "an index through MCP");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A program a reader takes seconds over, each differing, so none is read
/// from another's reading.
fn slow(k: usize) -> String {
    format!("OPENQASM 3.0;\nqubit[1] q;\nint a = 0;\nfor int i in [0:4999999] {{ a += 1; }}\n// {k}\nx q[0];\n")
}

/// `n` requests at once, each as `send` makes it: the seconds each took.
fn at_once(n: usize, send: impl Fn(usize) + Sync) -> Vec<f64> {
    std::thread::scope(|t| {
        let handles: Vec<_> = (0..n)
            .map(|i| {
                let send = &send;
                t.spawn(move || {
                    let began = Instant::now();
                    send(i);
                    began.elapsed().as_secs_f64()
                })
            })
            .collect();
        handles.into_iter().map(|h| h.join().unwrap()).collect()
    })
}

/// How long `/v1/health` takes while `during` runs beside it, at its worst.
fn health_during(port: u16, during: impl FnOnce() + Send) -> f64 {
    let done = std::sync::atomic::AtomicBool::new(false);
    std::thread::scope(|t| {
        let probe = t.spawn(|| {
            let mut worst = 0f64;
            while !done.load(std::sync::atomic::Ordering::SeqCst) {
                let began = Instant::now();
                assert_eq!(http(port, "GET", "/v1/health", None).0, 200);
                worst = worst.max(began.elapsed().as_secs_f64());
                std::thread::sleep(Duration::from_millis(20));
            }
            worst
        });
        // Set on the way out however `during` ends, so a failure in it fails here.
        struct Done<'a>(&'a std::sync::atomic::AtomicBool);
        impl Drop for Done<'_> {
            fn drop(&mut self) {
                self.0.store(true, std::sync::atomic::Ordering::SeqCst);
            }
        }
        {
            let _done = Done(&done);
            during();
        }
        probe.join().unwrap()
    })
}

/// A refused MCP task used to read its program on a runtime worker, with no
/// budget and no count: four of them stalled every route. A task refused
/// before anything was counted reads nothing, and spends nothing.
#[test]
fn a_refused_mcp_task_reads_nothing_and_holds_no_worker() {
    let d = dir("tasks");
    let (mut svc, port) = start(&d, 1000);
    let worst = health_during(port, || {
        let took = at_once(4, |k| {
            let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "task": {}, "arguments": { "qasm": slow(k), "idempotency_key": 5 } } })));
            assert!(code == 200 && b["error"]["data"]["verdict"]["level"] == "refused", "{b}");
        });
        assert!(took.iter().all(|&t| t < 1.0), "{took:?}");
    });
    assert!(worst < 1.0, "health took {worst} s");
    alive(port, &mut svc, "refused tasks");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// Every check counts, refused or not, before its bounds read anything; a
/// set's receipt is read as a check is, as its sweep is written out again.
#[test]
fn a_check_and_a_set_s_receipt_count_before_they_read() {
    let d = dir("counted");
    let (mut svc, port) = start(&d, 5);
    let left = |head: &str| head.to_ascii_lowercase().lines().find_map(|l| l.strip_prefix("ratelimit: \"submissions\";r=").map(|r| r.split(';').next().unwrap_or("").to_string())).unwrap_or_default();
    let wide = format!("OPENQASM 3.0;\nqubit[22] q;\nh q;\n{}", "rz(0.3) q;\n".repeat(200));
    let (code, head, body) = http(port, "POST", "/v1/equivalent", Some(&json!({ "a": wide, "b": wide })));
    assert!(code == 422 && body.contains("J031"), "{code} {body}");
    assert_eq!(left(&head), "4", "{head}");
    let (code, head, body) = http(port, "POST", "/v1/critique", Some(&json!({ "qasm": format!("OPENQASM 3.0;\nqubit q;\nrz({}0.5{}) q;\n", "(".repeat(5_000), ")".repeat(5_000)) })));
    assert!(code == 200 && body.contains("Q006"), "{code} {body}");
    assert_eq!(left(&head), "3", "{head}");
    let set = json!({ "request": { "qasm": "input int n;\nqubit q;\nx q;\n", "sweep": { "n": [1, 2] } }, "result": { "format": "wai.quantum.set/1", "count": 2, "items": [], "receipt": {} } });
    let (code, head, body) = http(port, "POST", "/v1/verify", Some(&set));
    assert!(code == 200 && body.contains("\"valid\":false"), "{code} {body}");
    assert_eq!(left(&head), "2", "{head}");
    let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "verify_quantum_receipt", "arguments": set } })));
    assert!(code == 200 && b["result"]["structuredContent"]["valid"] == false, "{b}");
    let (_, head, _) = http(port, "GET", "/v1", None);
    assert_eq!(left(&head), "1", "{head}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// Reading runs off the request threads: eight clients' heavy requests at
/// once leave `/v1/health` answering, as a job's reading used to stall it.
/// A client has its share of the turns, told at once past it.
#[test]
fn reading_leaves_the_runtime_free_and_a_client_has_its_share() {
    let d = dir("runtime");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]);
    let from = |who: &str, path: &str, body: &Value| {
        let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
        s.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
        let text = body.to_string();
        write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nX-Forwarded-For: {who}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
        let mut out = String::new();
        let _ = s.read_to_string(&mut out);
        out
    };
    let worst = health_during(port, || {
        at_once(8, |k| {
            let job = json!({ "qasm": format!("{HEAD}x q[0];\n"), "shots": 10, "intent": (0..16).map(|i| json!({ "equivalent": slow(100 * k + i) })).collect::<Vec<_>>() });
            let out = from(&format!("10.0.0.{k}"), "/v1/jobs", &job);
            assert!(out.contains("J031") || out.contains("J032"), "{out}");
        });
    });
    assert!(worst < 1.0, "health took {worst} s");
    // One client's parallel calls, as an agent makes them: served one after
    // another behind its share, every one answered.
    at_once(5, |_| {
        let out = from("10.0.1.1", "/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }));
        assert!(out.starts_with("HTTP/1.1 200"), "{out}");
    });
    // Past its share, a lent turn and the requests it may have waiting, told
    // at once.
    let answers = std::sync::Mutex::new(Vec::new());
    at_once(11, |k| {
        if k == 10 {
            std::thread::sleep(Duration::from_millis(300));
        }
        let began = Instant::now();
        let out = from("10.0.1.2", "/v1/critique", &json!({ "qasm": slow(1000 + k) }));
        answers.lock().unwrap().push((out.contains("J032"), began.elapsed().as_secs_f64()));
    });
    let answers = answers.into_inner().unwrap();
    assert!(answers.iter().any(|&(busy, took)| busy && took < 1.0), "{answers:?}");
    alive(port, &mut svc, "the turns");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// Steps bound work, not time: a reading past its deadline is stopped, in its
/// child, and refused, and the turn it held goes to the next. A client that
/// hangs up ends its reading, and its turn comes back at once.
#[test]
fn a_reading_past_its_time_is_stopped_and_a_hang_up_ends_it() {
    let d = dir("deadline");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_SECS", "1"), ("WAI_QSVC_CLIENT_READS", "1")]);
    let long = "OPENQASM 3.0;\nqubit q;\nint a = 0;\nfor int i in [0:9899999] { a += 1; }\nx q;\n";
    let t = Instant::now();
    let (code, b) = json_of(port, "POST", "/v1/critique", Some(&json!({ "qasm": long })));
    assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("1 s"), "{code} {b}");
    assert!(t.elapsed() < Duration::from_secs(10), "{:?}", t.elapsed());
    alive(port, &mut svc, "a reading past its time");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_CLIENT_READS", "1")]);
    let text = json!({ "qasm": long }).to_string();
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    write!(s, "POST /v1/critique HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\n\r\n{text}", text.len()).unwrap();
    std::thread::sleep(Duration::from_millis(300));
    drop(s);
    let began = Instant::now();
    loop {
        let (code, _, body) = http(port, "POST", "/v1/critique", Some(&json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })));
        if code == 200 {
            break;
        }
        assert!(code == 503 && began.elapsed() < Duration::from_secs(2), "{code} {body} after {:?}", began.elapsed());
        std::thread::sleep(Duration::from_millis(50));
    }
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A check's answer reads within the check's budget: a repair of forty fixes
/// of a program a read takes seconds over ran for minutes. Each pass of a
/// repair is a new text, so it ends at the budget, refused.
#[test]
fn a_check_s_answer_is_held_to_its_budget() {
    let d = dir("answer");
    let (mut svc, port) = start(&d, 1000);
    let program = format!("OPENQASM 3.0;\nqubit[1] q;\nbit[1] c;\nint x = 0;\nfor int i in [0:9899999] {{ x = i; }}\n{}", "measure q[0];\n".repeat(40));
    let t = Instant::now();
    let (code, b) = json_of(port, "POST", "/v1/repair", Some(&json!({ "qasm": program })));
    assert!(code == 422 && b.to_string().contains("J031"), "{code} {b}");
    assert!(t.elapsed() < Duration::from_secs(60), "{:?}", t.elapsed());
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A program within the reader's own bounds is taken here: its statements and
/// its gates each near their bound, past what one count of both would allow.
#[test]
fn a_job_within_the_reader_s_bounds_reads_here() {
    let d = dir("own");
    let (mut svc, port) = start(&d, 1000);
    let program = "OPENQASM 3.0;\nqubit q;\nint a = 0;\nfor int i in [0:8499999] { a += 1; }\nfor int i in [0:1399999] { x q; }\n";
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&json!({ "qasm": program, "shots": 1, "dry_run": true })));
    assert!(code == 200 && b["accepted"] == true, "{code} {b}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A client that stops sending holds its connection for no longer than the
/// limits say: a header section left open is closed, a body left open is
/// refused `408`, and the service answers beside them.
#[test]
fn a_stuck_connection_is_let_go() {
    let d = dir("stuck");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_HEADER_SECS", "1"), ("WAI_QSVC_BODY_SECS", "1")]);
    let open = |text: &str| {
        let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
        s.set_read_timeout(Some(Duration::from_secs(10))).unwrap();
        s.write_all(text.as_bytes()).unwrap();
        s
    };
    let mut header = open("POST /v1/jobs HTTP/1.1\r\nHost: 127.0.0.1\r\nX-Half: ");
    let mut body = open("POST /v1/jobs HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 100\r\n\r\n{\"qasm\"");
    alive(port, &mut svc, "stuck connections");
    let t = Instant::now();
    let mut out = String::new();
    let _ = header.read_to_string(&mut out);
    assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed());
    out.clear();
    let _ = body.read_to_string(&mut out);
    assert!(out.starts_with("HTTP/1.1 408"), "{out}");
    assert!(t.elapsed() < Duration::from_secs(5), "{:?}", t.elapsed());
    alive(port, &mut svc, "connections let go");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// `body` sent to `path` as the client `who` (behind the trusted proxy):
/// the status and the whole answer.
fn post_as(port: u16, who: &str, path: &str, body: &Value) -> (u16, String) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(150))).unwrap();
    let text = body.to_string();
    write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nX-Forwarded-For: {who}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
    let mut out = String::new();
    let _ = s.read_to_string(&mut out);
    (out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0), out)
}

/// An error-correction threshold at the service's bounds whose shots run for
/// minutes: what one client used to hold every check turn with.
fn heavy_threshold() -> Value {
    json!({ "family": "surface", "distances": [7, 7, 7, 7], "p": [0.04, 0.045, 0.05, 0.055], "shots": 12_500 })
}

/// One client sending checks that each hold a turn for as long as a reading
/// may take holds no more than its share of the checks: every other client's
/// check is answered beside them, where it waited and was refused.
#[test]
fn one_client_cannot_hold_every_check_turn() {
    let d = dir("checkshare");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "60"), ("WAI_QSVC_READ_SECS", "40")]);
    std::thread::scope(|t| {
        for _ in 0..2 {
            t.spawn(|| post_as(port, "198.51.100.66", "/v1/qec/threshold", &heavy_threshold()));
        }
        std::thread::sleep(Duration::from_millis(1500));
        for (path, body) in [("/v1/critique", json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })), ("/v1/qec/code", json!({ "code": "surface:3", "certify": false }))] {
            let began = Instant::now();
            let (code, out) = post_as(port, "203.0.113.7", path, &body);
            assert!(code == 200 && began.elapsed() < Duration::from_secs(5), "{path}: {code} after {:?}: {out}", began.elapsed());
        }
        alive(port, &mut svc, "one client's checks");
        svc.kill().unwrap();
    });
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// An error-correction call at the bounds whose shots would run for minutes
/// ends well inside the reading's time: refused as soon as its first shots
/// say the rest would pass the call's time, with the command that runs it
/// locally. One that fits is answered.
#[test]
fn a_qec_call_at_its_bounds_ends_well_inside_the_reading_s_time() {
    let d = dir("qectime");
    let (mut svc, port) = start(&d, 1000);
    for (path, body) in [("/v1/qec/threshold", heavy_threshold()), ("/v1/qec/decode", json!({ "memory": { "distance": 7, "rounds": 7, "p": 0.05 }, "shots": 200_000 }))] {
        let began = Instant::now();
        let (code, b) = json_of(port, "POST", path, Some(&body));
        assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("wai-quantum qec"), "{path}: {code} {b}");
        assert!(began.elapsed() < Duration::from_secs(30), "{path}: {:?}", began.elapsed());
        alive(port, &mut svc, path);
    }
    let (code, b) = json_of(port, "POST", "/v1/qec/threshold", Some(&json!({ "distances": [3], "p": [0.001], "shots": 1000 })));
    assert_eq!(code, 200, "{b}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The service's own peak memory, KiB, where the platform reports it.
fn peak_kib(svc: &Child) -> Option<u64> {
    let s = std::fs::read_to_string(format!("/proc/{}/status", svc.id())).ok()?;
    s.lines().find(|l| l.starts_with("VmHWM:"))?.split_whitespace().nth(1)?.parse().ok()
}

/// A reading's answer comes back into the service, whose memory every job
/// shares: a sweep whose hundred items each copy a large field, a few hundred
/// kilobytes sent, is refused for what its answer takes once read, before the
/// service reads it, and the service's own memory stays small. A dry run
/// takes back the plans and the points alone, and is held to what its
/// submission's answer would take: it says the same sweep is refused.
#[test]
fn a_reading_s_answer_is_held_to_what_it_takes_once_read() {
    let d = dir("answer-size");
    let (mut svc, port) = start(&d, 1000);
    let sweep = |pad: usize, dry: bool| json!({ "qasm": "OPENQASM 3.0;\ninput int n;\nqubit q;\nx q;\n", "sweep": { "n": (0..100).collect::<Vec<_>>() }, "shots": 1, "dry_run": dry, "pad": vec![0; pad] });
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(0, true)));
    assert!(code == 200 && b["accepted"] == true && b["items"].as_array().is_some_and(|i| i.len() == 100), "{code} {b}");
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(100_000, true)));
    assert!(code == 200 && b["accepted"] == false && b["refusal"]["code"] == "J031" && b.to_string().contains("once read"), "{code} {b}");
    // A job keeps its items: past what the service takes back, refused.
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(100_000, false)));
    assert!(code == 422 && b.to_string().contains("J031") && b.to_string().contains("once read"), "{code} {b}");
    if let Some(kib) = peak_kib(&svc) {
        assert!(kib < 400 << 10, "the service held {kib} KiB");
    }
    alive(port, &mut svc, "a large answer");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The binary replaced before the restart, by a rename (Linux, where a copy
/// over a running binary is refused) or a copy over it (elsewhere): the
/// running service goes on reading and running jobs with its own until it is
/// restarted, where every reading failed.
#[test]
fn the_binary_can_be_replaced_before_the_restart() {
    let d = dir("replace");
    let bin = d.join("app/wai-quantum-service");
    std::fs::create_dir_all(bin.parent().unwrap()).unwrap();
    std::fs::copy(env!("CARGO_BIN_EXE_wai-quantum-service"), &bin).unwrap();
    let (mut svc, port) = start_from(&bin, &d, 1000, &[]);
    let stub = d.join("app/stub");
    std::fs::write(&stub, "#!/bin/sh\necho 'not this service' >&2\nexit 2\n").unwrap();
    #[cfg(unix)]
    {
        use std::os::unix::fs::PermissionsExt;
        std::fs::set_permissions(&stub, std::fs::Permissions::from_mode(0o755)).unwrap();
    }
    if cfg!(target_os = "linux") {
        std::fs::rename(&stub, &bin).unwrap();
    } else {
        std::fs::copy(&stub, &bin).unwrap();
    }
    let (code, b) = json_of(port, "POST", "/v1/critique", Some(&json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" })));
    assert_eq!(code, 200, "{b}");
    let (code, b) = json_of(port, "POST", "/v1/run", Some(&json!({ "qasm": "OPENQASM 3.0; qubit q; bit c; h q; c = measure q;", "shots": 10 })));
    assert!(code == 200 && b["status"] == "done", "{code} {b}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The service's memory now, KiB: resident and swapped on Linux, resident
/// elsewhere.
fn mem_kib(svc: &Child) -> u64 {
    if let Ok(s) = std::fs::read_to_string(format!("/proc/{}/status", svc.id())) {
        let kb = |f: &str| s.lines().find(|l| l.starts_with(f)).and_then(|l| l.split_whitespace().nth(1)?.parse::<u64>().ok()).unwrap_or(0);
        return kb("VmRSS:") + kb("VmSwap:");
    }
    let o = Command::new("ps").args(["-o", "rss=", "-p", &svc.id().to_string()]).output().unwrap();
    String::from_utf8_lossy(&o.stdout).trim().parse().unwrap_or(0)
}

/// What ended jobs keep: thirty jobs from one client, each carrying a field
/// of a few hundred thousand numbers (a parsed request took tens of MiB, kept
/// for a day and reloaded on restart), keep their requests' text alone, and
/// the service stays small, after a restart too.
#[test]
fn ended_jobs_keep_little_and_a_restart_keeps_it_little() {
    let d = dir("kept");
    let (mut svc, port) = start(&d, 1000);
    let job = json!({ "qasm": "OPENQASM 3.0; qubit q; bit c; h q; c = measure q;", "shots": 8, "pad": vec![0; 400_000] });
    let mut last = String::new();
    for _ in 0..30 {
        let (code, b) = json_of(port, "POST", "/v1/run", Some(&job));
        assert!(code == 200 && b["status"] == "done", "{code} {b}");
        last = b["id"].as_str().unwrap().to_string();
    }
    let kib = mem_kib(&svc);
    assert!(kib < 200 << 10, "after thirty ended jobs the service holds {kib} KiB");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let (mut svc, port) = start(&d, 1000);
    let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{last}"), None);
    assert!(code == 200 && b["request"]["pad"].as_array().is_some_and(|p| p.len() == 400_000), "{code}");
    let kib = mem_kib(&svc);
    assert!(kib < 200 << 10, "after a restart the service holds {kib} KiB");
    alive(port, &mut svc, "the jobs kept");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A program that runs for minutes.
fn long_job() -> Value {
    let mut long = String::from("OPENQASM 3.0;\nqubit[24] q;\nh q;\n");
    for k in 0..6000 {
        long += &format!("rx(0.{}) q[{}];\ncx q[{}], q[{}];\n", k % 9 + 1, k % 24, k % 24, (k + 1) % 24);
    }
    json!({ "qasm": long, "observables": ["Z0"] })
}

/// Every answer comes within the answer's time from the request's arrival:
/// a run whose job is still running answers with the job to poll, by REST
/// and by MCP, where it waited 25 s and 50 s and passed a proxy's wait.
#[test]
fn a_run_s_answer_comes_within_the_answer_s_time() {
    let d = dir("answer-time");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_ANSWER_SECS", "15")]);
    let began = Instant::now();
    let (code, b) = json_of(port, "POST", "/v1/run", Some(&long_job()));
    assert!(code == 202 && b["id"].is_string() && began.elapsed() < Duration::from_secs(20), "{code} after {:?}", began.elapsed());
    let first = b["id"].as_str().unwrap().to_string();
    let began = Instant::now();
    let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "arguments": long_job() } })));
    let sc = &b["result"]["structuredContent"];
    assert!(code == 200 && sc["id"].is_string() && began.elapsed() < Duration::from_secs(20), "{code} after {:?}: {}", began.elapsed(), sc["status"]);
    assert!(sc["verdict"].is_object(), "{sc}");
    for id in [first.as_str(), sc["id"].as_str().unwrap()] {
        let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None);
    }
    alive(port, &mut svc, "runs answered in time");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// One client's quick calls pass its own slow, in-bounds check, on a turn
/// lent beside it, and its burst of eight quick calls is answered in full,
/// where they waited behind it and were refused.
#[test]
fn a_client_s_quick_calls_pass_its_own_slow_one_and_its_burst_is_answered() {
    let d = dir("ownqueue");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "60"), ("WAI_QSVC_READ_SECS", "40")]);
    let me = "198.51.100.23";
    std::thread::scope(|t| {
        t.spawn(|| post_as(port, me, "/v1/qec/threshold", &heavy_threshold()));
        std::thread::sleep(Duration::from_millis(1500));
        let began = Instant::now();
        let (code, out) = post_as(port, me, "/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }));
        // Well inside the slow call's 40 s, and inside the lent turn's 5 s.
        assert!(code == 200 && began.elapsed() < Duration::from_secs(5), "{code} after {:?}: {out}", began.elapsed());
        let answers = at_once(8, |_| {
            let (code, out) = post_as(port, me, "/v1/critique", &json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }));
            assert_eq!(code, 200, "{out}");
        });
        assert!(answers.iter().all(|&s| s < 10.0), "{answers:?}");
        alive(port, &mut svc, "a client's own queue");
        svc.kill().unwrap();
    });
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The caps stop a reading when no new process can be started: the kill is a
/// signal the service sends itself. Linux, where a process's limits can be
/// lowered while it runs (`prlimit`).
#[cfg(target_os = "linux")]
#[test]
fn a_reading_is_stopped_when_no_process_can_be_started() {
    if Command::new("prlimit").arg("--version").output().is_err() {
        eprintln!("no prlimit here: not run");
        return;
    }
    let d = dir("nproc");
    // No time of its own for the call: only the reading's cap stops it.
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_READ_SECS", "4"), ("WAI_QSVC_QEC_SECS", "100000")]);
    let pid = svc.id().to_string();
    let answer = std::thread::spawn(move || {
        let began = Instant::now();
        let (code, b) = json_of(port, "POST", "/v1/qec/threshold", Some(&heavy_threshold()));
        (code, b, began.elapsed())
    });
    std::thread::sleep(Duration::from_millis(1500));
    let ok = Command::new("prlimit").args(["--pid", &pid, "--nproc=1:1"]).status().unwrap().success();
    assert!(ok, "prlimit");
    let (code, b, took) = answer.join().unwrap();
    assert!(code == 422 && b.to_string().contains("J031") && took < Duration::from_secs(15), "{code} after {took:?}: {b}");
    assert!(svc.try_wait().unwrap().is_none(), "the service ended");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// This test process's own open-file limit raised to `n`, as far as its hard
/// limit lets it: it holds hundreds of connections to the service.
#[cfg(unix)]
fn room_for(n: u64) {
    // SAFETY: getrlimit and setrlimit read and write the struct given.
    unsafe {
        let mut l = libc::rlimit { rlim_cur: 0, rlim_max: 0 };
        if libc::getrlimit(libc::RLIMIT_NOFILE, &mut l) == 0 && l.rlim_cur < n as libc::rlim_t {
            l.rlim_cur = (n as libc::rlim_t).min(l.rlim_max);
            libc::setrlimit(libc::RLIMIT_NOFILE, &l);
        }
    }
}

/// How many connections this test process can hold open beside what it
/// needs for itself and its health checks: `want`, or fewer where its own
/// open-file limit is low (a whole suite run under `ulimit -n 256`).
#[cfg(unix)]
fn can_hold(want: usize) -> usize {
    // SAFETY: getrlimit writes the struct given.
    let soft = unsafe {
        let mut l = libc::rlimit { rlim_cur: 0, rlim_max: 0 };
        if libc::getrlimit(libc::RLIMIT_NOFILE, &mut l) == 0 { l.rlim_cur as usize } else { 256 }
    };
    // Under a low limit of its own the suite's tests share it, four at a time.
    if soft < 1024 { want.min(soft / 4) } else { want }
}

/// Whether this test process's hard open-file limit lets a child raise its
/// soft one past 256.
#[cfg(unix)]
fn can_raise() -> bool {
    // SAFETY: getrlimit writes the struct given.
    unsafe {
        let mut l = libc::rlimit { rlim_cur: 0, rlim_max: 0 };
        libc::getrlimit(libc::RLIMIT_NOFILE, &mut l) == 0 && l.rlim_max > 256
    }
}

/// Whether a health check on a new connection is answered `200` within 3 s.
fn healthy(port: u16) -> bool {
    let Ok(mut s) = TcpStream::connect_timeout(&std::net::SocketAddr::from(([127, 0, 0, 1], port)), Duration::from_secs(3)) else { return false };
    let _ = s.set_read_timeout(Some(Duration::from_secs(3)));
    let mut out = String::new();
    write!(s, "GET /v1/health HTTP/1.1\r\nHost: 127.0.0.1\r\nConnection: close\r\n\r\n").is_ok() && s.read_to_string(&mut out).is_ok() && out.split_whitespace().nth(1) == Some("200")
}

/// Connections held open, from one client and from many, partway through a
/// header section or a body, take no more of the service's open files than it
/// has, and a health check is answered throughout: where its limit is 256 and
/// cannot be raised (a launchd service's), and where only its soft limit is
/// 256, which the service raises. Past a client's open requests, the rest
/// are answered `503` at once, unread.
#[cfg(unix)]
#[test]
fn held_connections_leave_room_and_health_answers() {
    room_for(8192);
    for (name, limit) in [("hard", "ulimit -Sn 256 && ulimit -Hn 256"), ("soft", "ulimit -Sn 256")] {
        let d = dir(&format!("files-{name}"));
        let (mut svc, port) = start_limited(&d, limit, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]);
        let answered = |what: &str| {
            let ok = (0..15).filter(|_| healthy(port)).count();
            assert_eq!(ok, 15, "{name}: health answered {ok} of 15 {what}: {}", logs(&d));
        };
        // One client, 400 header sections that never end.
        let n = can_hold(400);
        let held: Vec<TcpStream> = (0..n)
            .filter_map(|_| {
                let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?;
                let _ = s.write_all(b"GET /v1/health HTTP/1.1\r\nHost: 127.0.0.1\r\n");
                Some(s)
            })
            .collect();
        std::thread::sleep(Duration::from_secs(1));
        answered("beside 400 header sections held open");
        drop(held);
        // 400 clients, a body each that stops arriving.
        let held: Vec<TcpStream> = (0..n)
            .filter_map(|i| {
                let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?;
                let _ = write!(s, "POST /v1/run HTTP/1.1\r\nHost: 127.0.0.1\r\nX-Forwarded-For: 198.51.{}.{}\r\nContent-Type: application/json\r\nContent-Length: 1048576\r\n\r\n{{\"qasm\": \"", i / 200, i % 200 + 1);
                Some(s)
            })
            .collect();
        std::thread::sleep(Duration::from_millis(1500));
        answered("beside 400 clients' bodies held open");
        drop(held);
        // One client, 400 bodies that stop arriving: past its open requests,
        // refused at once.
        let mut held: Vec<TcpStream> = (0..n)
            .filter_map(|_| {
                let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?;
                let _ = write!(s, "POST /v1/run HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: 1048576\r\n\r\n{{\"qasm\": \"");
                Some(s)
            })
            .collect();
        answered("beside one client's 400 bodies held open");
        let refused = held
            .iter_mut()
            .rev()
            .take(n.saturating_sub(40).min(50))
            .map(|s| {
                let _ = s.set_read_timeout(Some(Duration::from_secs(2)));
                let mut out = [0u8; 512];
                let n = s.read(&mut out).unwrap_or(0);
                let text = String::from_utf8_lossy(&out[..n]);
                n == 0 || (text.starts_with("HTTP/1.1 503") && text.contains("J032"))
            })
            .filter(|refused| *refused)
            .count();
        assert_eq!(refused, n.saturating_sub(40).min(50), "{name}: a client's requests past its open ones are refused or closed at once");
        drop(held);
        assert!(!logs(&d).contains("Too many open files"), "{name}: {}", logs(&d));
        // The limit, raised where it can be, and logged.
        if name == "soft" && can_raise() {
            assert!(logs(&d).contains("(raised from 256"), "{}", logs(&d));
        } else {
            assert!(logs(&d).contains("open files: 256"), "{}", logs(&d));
        }
        alive(port, &mut svc, "connections held open");
        svc.kill().unwrap();
        svc.wait().unwrap();
        let _ = std::fs::remove_dir_all(&d);
    }
}

/// A sweep whose result is more than its client's ended jobs may keep,
/// sent within every published limit, is read in full: by a poll, from
/// `/v1/run` and from the MCP set tool. It is kept until it has been read,
/// where it was dropped as it ended and the client was told no such job.
#[test]
fn a_result_larger_than_its_client_keeps_is_kept_until_it_is_read() {
    let d = dir("large-result");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_CLIENT_KEPT_MIB", "1")]);
    let sweep = |points: usize| json!({ "qasm": "OPENQASM 3.0;\ninput float theta;\nqubit[16] q;\nbit[16] c;\nh q;\nrz(theta) q[0];\nc = measure q;\n", "sweep": { "theta": (0..points).map(|i| 0.01 * (i + 1) as f64).collect::<Vec<_>>() }, "shots": 20000, "seed": 7 });
    let done = |id: &str| {
        let t = Instant::now();
        loop {
            let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None);
            assert_eq!(code, 200, "{id} after {:?}: {b}", t.elapsed());
            if b["status"] == "done" {
                return b;
            }
            assert!(b["status"] == "queued" || b["status"] == "running", "{b}");
            assert!(t.elapsed() < Duration::from_secs(120), "{id}");
        }
    };
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(10)));
    assert_eq!(code, 202, "{b}");
    let job = done(b["id"].as_str().unwrap());
    assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(10));
    let (code, b) = json_of(port, "POST", "/v1/run", Some(&sweep(10)));
    assert!(code == 200 || code == 202, "{code}: {b}");
    let job = done(b["id"].as_str().unwrap());
    assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(10));
    let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_set", "arguments": sweep(10) } })));
    let sc = &b["result"]["structuredContent"];
    assert_eq!(code, 200, "{b}");
    assert!(b["result"]["isError"] == false && sc["id"].is_string(), "{}", b["result"]["content"]);
    let job = done(sc["id"].as_str().unwrap());
    assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(10));
    alive(port, &mut svc, "large results");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A job's report is the same bytes on every route that serves it (a poll,
/// its events, the MCP job tool) and after a restart: one RFC 8785 text,
/// its receipt's keys in their order as well as its request's.
#[test]
fn a_job_reads_the_same_on_every_route_and_after_a_restart() {
    let d = dir("same-bytes");
    let (mut svc, port) = start(&d, 1000);
    let request = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n", "shots": 200, "seed": 9, "noise": { "depolarizing_1q": 0.000001, "readout": 1e-7, "depolarizing_2q": 0.0 }, "observables": ["Z0", "0.5 Z1"] });
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&request));
    assert_eq!(code, 202, "{b}");
    let id = b["id"].as_str().unwrap().to_string();
    let get = || {
        let t = Instant::now();
        loop {
            let (code, _, text) = http(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None);
            assert_eq!(code, 200, "{text}");
            if text.contains("\"status\":\"done\"") {
                return text;
            }
            assert!(t.elapsed() < Duration::from_secs(60), "{text}");
        }
    };
    let text = get();
    let v: Value = serde_json::from_str(&text).unwrap();
    assert_eq!(text.as_bytes(), wai_quantum::engine::receipt::canonical(&v), "a report is RFC 8785");
    let (_, _, events) = http(port, "GET", &format!("/v1/jobs/{id}/events"), None);
    let event = events.lines().find_map(|l| l.strip_prefix("data: ").filter(|d| d.contains("\"result\""))).unwrap_or_else(|| panic!("{events}"));
    assert_eq!(event, text, "the job event");
    let (_, _, mcp) = http(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "get_quantum_job", "arguments": { "id": id } } })));
    // The job in the MCP answer is the report's own bytes, what MCP adds to
    // it (its verdict) after them.
    assert!(mcp.contains(&text[..text.len() - 1]), "the MCP job tool: {mcp}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let (mut svc, port) = start(&d, 1000);
    let (code, _, again) = http(port, "GET", &format!("/v1/jobs/{id}"), None);
    assert_eq!(code, 200, "{again}");
    assert_eq!(again, text, "after a restart");
    alive(port, &mut svc, "a report on every route");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// An MCP message over 64 KiB from a client past its window is answered
/// under its own id, as a tool result that says why; a resend under its key
/// of a job it holds is answered with that job.
#[test]
fn a_large_mcp_message_past_the_window_is_answered_under_its_own_id() {
    let d = dir("mcp-large");
    let (mut svc, port) = start(&d, 2);
    let big = format!("OPENQASM 3.0;\nqubit[2] q;\n{}", "h q[0];\n".repeat(9000));
    let call = |id: u32, name: &str, args: Value| json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": id, "method": "tools/call", "params": { "name": name, "arguments": args } })));
    let (code, first) = call(1, "run_quantum_circuit", json!({ "qasm": big, "shots": 10, "idempotency_key": "large-1" }));
    assert_eq!(code, 200, "{first}");
    let job = first["result"]["structuredContent"]["id"].as_str().unwrap_or_else(|| panic!("{first}")).to_string();
    let (code, b) = call(2, "critique_qasm", json!({ "qasm": "OPENQASM 3.0; qubit q; h q;" }));
    assert_eq!(code, 200, "{b}");
    let (code, b) = call(3, "critique_qasm", json!({ "qasm": big }));
    assert_eq!((code, b["id"].as_u64(), b["result"]["isError"].as_bool()), (200, Some(3), Some(true)), "{b}");
    assert!(b["result"]["content"][0]["text"].as_str().is_some_and(|t| t.contains("J030")), "{b}");
    let (code, b) = call(4, "run_quantum_circuit", json!({ "qasm": big, "shots": 10, "idempotency_key": "large-1" }));
    assert_eq!((code, b["id"].as_u64()), (200, Some(4)), "{b}");
    assert_eq!(b["result"]["structuredContent"]["id"].as_str(), Some(job.as_str()), "{b}");
    alive(port, &mut svc, "a large MCP message past the window");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// Past its window a client's body is not read, whatever key it carries
/// where the client holds no job under that key; a receipt check has a
/// window of its own, asked before its body is read.
#[test]
fn a_body_is_read_only_once_its_window_or_its_key_allows() {
    let d = dir("unread-key");
    let (mut svc, port) = start_with(&d, 0, &[("WAI_QSVC_CLIENT_VERIFIES", "2")]);
    let raw = |path: &str, extra: &str, body: &str| {
        let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
        s.set_read_timeout(Some(Duration::from_secs(30))).unwrap();
        write!(s, "POST {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\n{extra}Content-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap();
        let mut out = String::new();
        let _ = s.read_to_string(&mut out);
        out.split_whitespace().nth(1).and_then(|c| c.parse::<u16>().ok()).unwrap_or(0)
    };
    for path in ["/v1/jobs", "/v1/run"] {
        assert_eq!(raw(path, "Idempotency-Key: fresh-key\r\n", "{not json"), 429, "{path}: a fresh key past the window");
    }
    // Every body gets a verdict (this one `unreadable`), within verify's own window.
    let codes: Vec<u16> = (0..3).map(|_| raw("/v1/verify", "", "{not json")).collect();
    assert_eq!(codes, [200, 200, 429], "verify's own window");
    alive(port, &mut svc, "bodies left unread");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// `tasks/result` holds its answer within the answer's time from arrival,
/// as every other route does, where it held a fixed 50 s.
#[test]
fn a_task_s_result_is_held_within_the_answer_s_time() {
    let d = dir("task-wait");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_ANSWER_SECS", "10"), ("WAI_QSVC_MIN_FREE_PCT", "100")]);
    let program = format!("OPENQASM 3.0;\nqubit[25] q;\nh q;\n{}", (0..25).map(|i| format!("t q[{i}];\n")).collect::<String>());
    let (code, made) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_circuit", "task": {}, "arguments": { "qasm": program, "shots": 10 } } })));
    assert_eq!(code, 200, "{made}");
    let task = made["result"]["task"]["taskId"].as_str().unwrap_or_else(|| panic!("{made}")).to_string();
    let began = Instant::now();
    let (code, b) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 2, "method": "tasks/result", "params": { "taskId": task } })));
    let took = began.elapsed();
    assert_eq!(code, 200, "{b}");
    assert!(took < Duration::from_secs(12), "answered after {took:?}: {b}");
    assert!(b["error"]["message"].as_str().is_some_and(|m| m.contains("still working")), "{b}");
    let _ = http(port, "DELETE", &format!("/v1/jobs/{task}"), None);
    alive(port, &mut svc, "a task's result held");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A documented QEC call sent beside another check from the same client is
/// answered: a lent turn moves onto the client's own as that comes free, and
/// runs past its 5 s while no one waits, where it was stopped at 5 s.
#[test]
fn a_documented_qec_call_beside_another_check_from_its_client_is_answered() {
    let d = dir("lent-qec");
    // The QEC, reading and answer times raised, so a slow or loaded host
    // runs the decode, however long it takes there, rather than refusing it.
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_QEC_SECS", "100"), ("WAI_QSVC_READ_SECS", "100"), ("WAI_QSVC_ANSWER_SECS", "110")]);
    let me = "198.51.100.41";
    let decode = json!({ "memory": { "distance": 5, "rounds": 5, "p": 0.005 }, "shots": 200_000, "seed": 1 });
    let critique = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n" });
    for pair in [[("/v1/critique", &critique), ("/v1/qec/decode", &decode)], [("/v1/qec/decode", &decode), ("/v1/qec/decode", &decode)]] {
        let answers: Vec<(u16, String, f64)> = std::thread::scope(|t| {
            let calls: Vec<_> = pair
                .iter()
                .enumerate()
                .map(|(i, (path, body))| {
                    t.spawn(move || {
                        std::thread::sleep(Duration::from_millis(30 * i as u64));
                        let began = Instant::now();
                        let (code, out) = post_as(port, me, path, body);
                        (code, out, began.elapsed().as_secs_f64())
                    })
                })
                .collect();
            calls.into_iter().map(|c| c.join().unwrap()).collect()
        });
        for ((path, _), (code, out, took)) in pair.iter().zip(&answers) {
            assert_eq!(*code, 200, "{path} after {took:.1} s: {}", out.lines().last().unwrap_or(""));
        }
    }
    alive(port, &mut svc, "QEC calls beside others");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The program the round-5 verifier kept the service busy with: a layer of
/// T gates and a CX ladder, thirty times over, one shot. It runs for tens of
/// seconds (the verifier's 22 qubits took about 35 s on its host; 24 here, so
/// a faster host still runs it past every wait below), so a `?wait=25` poll
/// on it holds its whole wait.
fn pump() -> Value {
    let n = 24;
    let mut q = format!("OPENQASM 3.0;\nqubit[{n}] q;\nbit[{n}] c;\n");
    for _ in 0..30 {
        q += "t q;\n";
        for i in 0..n - 1 {
            q += &format!("cx q[{i}], q[{}];\n", i + 1);
        }
    }
    q += "c = measure q;\n";
    json!({ "qasm": q, "method": "auto", "shots": 1 })
}

/// Whether a connection is still open, and what has been answered on it so
/// far, read without waiting.
fn so_far(s: &mut TcpStream) -> (bool, Vec<u8>) {
    let _ = s.set_nonblocking(true);
    let (mut got, mut b) = (Vec::new(), [0u8; 4096]);
    let open = loop {
        match s.read(&mut b) {
            Ok(0) => break false,
            Ok(n) => got.extend_from_slice(&b[..n]),
            Err(e) if e.kind() == std::io::ErrorKind::WouldBlock => break true,
            Err(_) => break false,
        }
    };
    let _ = s.set_nonblocking(false);
    (open, got)
}

/// The round-5 verifier's flood, as its probe sent it: where the service's
/// open-file limit is 256 and cannot be raised (192 connections), 210 long
/// polls (`?wait=25`) on a running job, each from an address of its own
/// (`10.0.k.m`), opened 3 ms apart, then health checked 25 times. They held
/// every connection, none could be closed for room, and every health check
/// was answered `503`. Long polls are held to a share of the connections,
/// past it answered at once (`503`, with `Retry-After`), and four connections past
/// the cap are kept for health checks: each is answered `200`. The same for
/// event streams, which held their connections as long polls did. Where this
/// test process holds fewer connections (a suite run under `ulimit -n 256`),
/// the service's cap is lowered as far, so they still outnumber it.
#[cfg(unix)]
#[test]
fn long_polls_and_streams_leave_a_health_check_answered() {
    room_for(8192);
    let n = can_hold(210);
    let cap = (n * 192 / 210).max(16).to_string();
    let mut env = vec![("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")];
    if n < 210 {
        env.push(("WAI_QSVC_CONNECTIONS", cap.as_str()));
    }
    let d = dir("long-polls");
    let (mut svc, port) = start_limited(&d, "ulimit -Sn 256 && ulimit -Hn 256", &env);
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&pump()));
    assert_eq!(code, 202, "{b}");
    let id = b["id"].as_str().unwrap().to_string();
    for (what, path) in [("long polls", format!("/v1/jobs/{id}?wait=25")), ("event streams", format!("/v1/jobs/{id}/events"))] {
        let mut held: Vec<TcpStream> = (0..n)
            .filter_map(|i| {
                let mut s = TcpStream::connect(("127.0.0.1", port)).ok()?;
                let _ = write!(s, "GET {path} HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: 10.0.{}.{}\r\n\r\n", i % 8, i / 8 + 1);
                std::thread::sleep(Duration::from_millis(3));
                Some(s)
            })
            .collect();
        std::thread::sleep(Duration::from_secs(3));
        let ok = (0..25).filter(|_| healthy(port)).count();
        // The flood was real: the job still runs, and a long poll is held
        // with nothing answered, a stream with its answer begun.
        let waiting = held
            .iter_mut()
            .map(so_far)
            .filter(|(open, got)| *open && if what == "long polls" { got.is_empty() } else { got.starts_with(b"HTTP/1.1 200") })
            .count();
        let sent = held.len();
        drop(held);
        std::thread::sleep(Duration::from_millis(500));
        let (_, now) = json_of(port, "GET", &format!("/v1/jobs/{id}"), None);
        assert!(now["status"] == "queued" || now["status"] == "running", "{what}: the job ended before the checks, so nothing was held for it: {now}");
        assert!(waiting > 0, "{what}: none of {sent} held");
        assert_eq!(ok, 25, "{what}: health answered {ok} of 25 beside {sent} {what}, {waiting} of them held: {}", logs(&d));
    }
    let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None);
    alive(port, &mut svc, "long polls and streams");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// [`post_as`] for the MCP endpoint: a tool call, and whether it was answered
/// with no error.
fn mcp_as(port: u16, who: &str, tool: &str, args: &Value) -> (bool, String) {
    let body = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool, "arguments": args } });
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(150))).unwrap();
    let text = body.to_string();
    write!(s, "POST /mcp HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json\r\nX-Forwarded-For: {who}\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
    let mut out = String::new();
    let _ = s.read_to_string(&mut out);
    let answer: Value = out.split_once("\r\n\r\n").and_then(|(_, b)| serde_json::from_str(b).ok()).unwrap_or(Value::Null);
    (out.starts_with("HTTP/1.1 200") && answer["result"].is_object() && answer["result"]["isError"] != true, out)
}

/// The README's threshold (`qec threshold --distances 3,5,7 --p
/// 0.002,0.004 --shots 20000`) made longer, as the round-5 verifier made it
/// longer (30,000 shots on its host, where that took about 15 s), until it
/// takes 11 s alone here, past the 10 s a request waits for a turn: its
/// shots raised half again at a time to the 200,000 a threshold here takes
/// in all, then its rounds a point raised from the distance to 9, 11, 13 (21
/// at most here), for as long as the service still takes it within the 25 s
/// an error-correction call may take. The longest call it took, and how
/// long that took.
///
/// On a loaded host the README's threshold alone comes near those 25 s, or
/// past them: the service refuses a call it projects past them (`J031`), from
/// a pilot of the call's own shots. CI's Linux runner projected 28 s and
/// refused it, and the test failed before it began; and one taken at 21 s
/// alone left no room for two at once, so the second was refused (`J031`) in
/// the burst. A call that took more than 14 s, or was refused, has its shots
/// scaled to the 12 s its time (or the refusal's projection) gives, at least
/// 1000, and is sent again, at most three times: a long check, past the 10 s
/// a request waits for a turn, with room under the 25 s for two at once.
fn slow_threshold(port: u16) -> (Value, f64) {
    let run = |t: &Value| -> Result<f64, String> {
        let began = Instant::now();
        let (code, out) = post_as(port, "203.0.113.99", "/v1/qec/threshold", t);
        let secs = began.elapsed().as_secs_f64();
        eprintln!("the threshold alone, {} shots, rounds {}: {code} in {secs:.1} s", t["shots"], t["rounds"]);
        match code {
            200 => Ok(secs),
            422 if out.contains("J031") => Err(out.lines().last().unwrap_or("").to_string()),
            _ => panic!("the threshold alone: {code}: {}", out.lines().last().unwrap_or("")),
        }
    };
    let mut t = json!({ "distances": [3, 5, 7], "p": [0.002, 0.004], "shots": 20000 });
    let mut first = run(&t);
    for _ in 0..3 {
        let secs = match &first {
            Ok(secs) if *secs <= 14.0 => break,
            Ok(secs) => *secs,
            Err(why) => {
                let projected = why.split("would take about ").nth(1).and_then(|r| r.split(' ').next()).and_then(|s| s.parse::<f64>().ok()).filter(|s| *s > 0.0);
                projected.unwrap_or_else(|| panic!("a refusal as too slow with no projected time: {why}"))
            }
        };
        let shots = t["shots"].as_u64().unwrap_or(20000);
        let fewer = ((shots as f64 * 12.0 / secs) as u64).max(1000);
        if fewer >= shots {
            break;
        }
        t["shots"] = json!(fewer);
        first = run(&t);
    }
    let mut best = (t.clone(), first.unwrap_or_else(|why| panic!("the threshold was refused as too slow here, at {} shots: {why}", t["shots"])));
    while best.1 < 11.0 {
        let shots = t["shots"].as_u64().unwrap_or(20000);
        let rounds = t["rounds"].as_u64().unwrap_or(7);
        if shots < 33_333 {
            t["shots"] = json!((shots * 3 / 2).min(33_333));
        } else if rounds < 13 {
            t["rounds"] = json!(rounds + 2);
        } else {
            break;
        }
        match run(&t) {
            Ok(secs) => best = (t.clone(), secs),
            Err(_) => break,
        }
    }
    best
}

/// The round-5 verifier's burst, at the service's own limits: one client
/// sends the README's decode (`--memory 5,5,0.003 --shots 20000 --decoder
/// correlated`), a threshold, and six critiques of a Bell pair, 30 ms apart,
/// over HTTP and then over MCP. The threshold, on a turn lent beside the
/// decode, moved onto the client's own turn when the decode ended and kept
/// the lent one, so no critique could have either: each waited out its 10 s
/// and was refused (`J032`), or, with the threshold under 10 s, was answered
/// only as it ended. A moved call gives its lent turn back: every call is
/// answered, each critique well before the threshold. On a host so loaded
/// (a whole suite beside it) that the threshold, measured beside the burst,
/// would pass the 25 s an error-correction call may take, it is refused at
/// once (`J031`) and nothing was tested: the threshold is measured again and
/// the burst sent again, three times at most.
#[test]
fn the_readme_s_burst_from_one_client_is_answered() {
    let d = dir("readme-burst");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]);
    let decode = json!({ "memory": { "distance": 5, "rounds": 5, "p": 0.003 }, "shots": 20000, "decoder": "correlated" });
    let critique = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n" });
    for (route, me) in [("HTTP", "198.51.100.77"), ("MCP", "198.51.100.78")] {
        let mut tries = 0;
        let (answers, at, alone) = loop {
            tries += 1;
            let (threshold, alone) = slow_threshold(port);
            let mut calls: Vec<(&str, &str, &Value)> = vec![("/v1/qec/decode", "qec_decode", &decode), ("/v1/qec/threshold", "qec_threshold", &threshold)];
            calls.extend(std::iter::repeat_n(("/v1/critique", "critique_qasm", &critique), 6));
            let began = Instant::now();
            let answers: Vec<(bool, String, f64)> = std::thread::scope(|t| {
                let each: Vec<_> = calls
                    .iter()
                    .enumerate()
                    .map(|(i, (path, tool, body))| {
                        let began = &began;
                        t.spawn(move || {
                            std::thread::sleep(Duration::from_millis(30 * i as u64));
                            let (ok, out) = if route == "HTTP" {
                                let (code, out) = post_as(port, me, path, body);
                                (code == 200, out)
                            } else {
                                mcp_as(port, me, tool, body)
                            };
                            (ok, out, began.elapsed().as_secs_f64())
                        })
                    })
                    .collect();
                each.into_iter().map(|c| c.join().unwrap()).collect()
            });
            let at: Vec<String> = calls.iter().zip(&answers).map(|((path, tool, _), (ok, _, s))| format!("{} {} at {s:.2} s", if route == "HTTP" { path } else { tool }, if *ok { "answered" } else { "refused" })).collect();
            eprintln!("{route}, try {tries}, the threshold alone {alone:.1} s: {}", at.join(", "));
            let (ok, out, _) = &answers[1];
            if !ok && out.contains("J031") && out.contains("would take about") && tries < 3 {
                eprintln!("{route}: the threshold, measured beside the burst, was past the time a call may take, so nothing was tested; again");
                continue;
            }
            break (answers, at, alone);
        };
        for ((ok, out, _), name) in answers.iter().zip(&at) {
            assert!(*ok, "{route}: {name} (the threshold alone {alone:.1} s): {}\n{at:#?}", out.lines().last().unwrap_or(""));
        }
        let threshold_at = answers[1].2;
        for (_, _, s) in &answers[2..] {
            assert!(*s + 1.0 < threshold_at, "{route}: a critique answered at {s:.2} s, the threshold at {threshold_at:.2} s (alone {alone:.1} s): the critiques waited for it\n{at:#?}");
        }
    }
    alive(port, &mut svc, "the README's burst");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A JSON-RPC id is echoed exactly, an integer past 2^53 included.
#[test]
fn an_mcp_id_is_echoed_exactly() {
    let d = dir("mcp-ids");
    let (mut svc, port) = start(&d, 1000);
    for id in ["9007199254740993", "-9007199254740993", "9223372036854775807", "12345678901234567890", "1730000000123456789"] {
        let body: Value = serde_json::from_str(&format!("{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"ping\"}}")).unwrap();
        let (code, _, text) = http(port, "POST", "/mcp", Some(&body));
        assert_eq!(code, 200, "{text}");
        assert!(text.contains(&format!("\"id\":{id},")) || text.contains(&format!("\"id\":{id}}}")), "{id}: {text}");
    }
    alive(port, &mut svc, "ids echoed");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A job's result is held to `WAI_QSVC_RESULT_MIB` before it runs, where a
/// result past it was run and then failed: a sweep whose result's bound
/// passes the cap is refused at once (`422`, `J031`), naming its bound and
/// the cap, on every route that takes a job, and its dry run says so; the
/// largest sweep under the cap is taken and ends done, its result within its
/// bound; a pulse run past the cap is refused at once as well, by its
/// schedule. At the defaults, the published set at its most, 100 items of
/// 65,536 outcomes each (about 150 MB as served), is refused, and 100
/// points of 30,000 shots of 16 bits each are taken.
#[test]
fn a_result_past_what_a_job_may_give_is_refused_before_it_runs() {
    let sweep = |points: usize, shots: u64| json!({ "qasm": "OPENQASM 3.0;\ninput float theta;\nqubit[16] q;\nbit[16] c;\nh q;\nrz(theta) q[0];\nc = measure q;\n", "sweep": { "theta": (0..points).map(|i| 0.01 * (i + 1) as f64).collect::<Vec<_>>() }, "shots": shots, "seed": 7 });
    let dry = |port: u16, v: &Value| {
        let mut v = v.clone();
        v["dry_run"] = json!(true);
        let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&v));
        assert_eq!(code, 200, "{b}");
        b
    };
    let d = dir("result-cap");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_RESULT_MIB", "1")]);
    let cap = 1u64 << 20;
    assert_eq!(dry(port, &sweep(1, 2000))["bounds"]["result_bytes"].as_u64(), Some(cap));
    // The most points of 2,000 shots a point under the cap, by the dry runs' bounds.
    let bound = |n: usize| dry(port, &sweep(n, 2000))["result_bytes"].as_u64().unwrap();
    let (one, two) = (bound(1), bound(2));
    let mut points = ((cap - (2 * one - two)) / (two - one)) as usize;
    while bound(points) > cap {
        points -= 1;
    }
    while bound(points + 1) <= cap {
        points += 1;
    }
    let past = dry(port, &sweep(points + 1, 2000));
    assert!(past["accepted"] == false && past["status"] == 422 && past["refusal"]["code"] == "J031", "{past}");
    // Refused at once, on every route that takes a job.
    let mcp = |args: Value| json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": "run_quantum_set", "arguments": args } });
    for (path, body) in [("/v1/jobs", sweep(points + 1, 2000)), ("/v1/run", sweep(points + 1, 2000)), ("/mcp", mcp(sweep(points + 1, 2000)))] {
        let began = Instant::now();
        let (code, b) = json_of(port, "POST", path, Some(&body));
        let why = if path == "/mcp" { b["result"]["content"][0]["text"].as_str().unwrap_or("").to_string() } else { b["error"].as_str().unwrap_or("").to_string() };
        assert!(if path == "/mcp" { code == 200 && b["result"]["isError"] == true } else { code == 422 }, "{path}: {code} {b}");
        assert!(why.contains("J031") && why.contains("could take up to") && why.contains("the 1 MiB"), "{path}: {why}");
        assert!(began.elapsed() < Duration::from_secs(10), "{path}: answered after {:?}", began.elapsed());
    }
    // At the bound: taken, and done, its result within its bound.
    let at = bound(points);
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep(points, 2000)));
    assert_eq!(code, 202, "{b}");
    let id = b["id"].as_str().unwrap().to_string();
    let t = Instant::now();
    let job = loop {
        let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None);
        assert_eq!(code, 200, "{b}");
        if b["status"] != "queued" && b["status"] != "running" {
            break b;
        }
        assert!(t.elapsed() < Duration::from_secs(120));
    };
    assert_eq!(job["status"], "done", "{}", job["error"]);
    let served = wai_quantum::engine::receipt::canonical(&job["result"]).len() as u64;
    eprintln!("{points} points of 2,000 shots: bound {at} bytes, served {served}");
    assert!(served <= at && at <= cap, "{points} points: served {served} bytes, bound {at}");
    // A pulse run is bounded from its schedule as well: 100,000 bits
    // declared (in registers of 64, as the pulse reader takes them) and one
    // captured, each bit named and every outcome's key as wide (about 1.4 MB
    // as served), are refused at once, where the run had been taken and
    // failed only as it ended; 100 bits are taken.
    let pulse = |n: usize| json!({ "method": "pulse", "shots": 100, "qasm": format!("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, 7.1e9, 0);\n}}\ndefcal x $0 {{ play(q0, gaussian(0.4, 64dt, 16dt)); }}\ndefcal measure $0 -> bit {{ return capture_v2(c0, 4dt); }}\n{}x $0;\nb{}[{}] = measure $0;\n", (0..n.div_ceil(64)).map(|r| format!("bit[{}] b{r};\n", 64.min(n - 64 * r))).collect::<String>(), (n - 1) / 64, (n - 1) % 64) });
    let began = Instant::now();
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&pulse(100_000)));
    if code == 202 {
        let id = b["id"].as_str().unwrap_or_default().to_string();
        let job = loop {
            let (_, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=25"), None);
            if b["status"] != "queued" && b["status"] != "running" {
                break b;
            }
            assert!(began.elapsed() < Duration::from_secs(300));
        };
        panic!("a pulse run of 100,000 bits was taken, and then after {:?}: {} {}", began.elapsed(), job["status"], job["error"]);
    }
    let why = b["error"].as_str().unwrap_or_default();
    assert!(code == 422 && why.contains("J031") && why.contains("could take up to") && why.contains("the 1 MiB"), "a pulse run past the cap: {code} {b}");
    assert!(began.elapsed() < Duration::from_secs(10), "answered after {:?}", began.elapsed());
    let small = dry(port, &pulse(100));
    assert!(small["accepted"] == true && small["result_bytes"].as_u64().is_some_and(|b| b < cap), "{small}");
    alive(port, &mut svc, "results held to their cap");
    svc.kill().unwrap();
    svc.wait().unwrap();
    // The defaults.
    let (mut svc, port) = start(&d, 1000);
    let most = dry(port, &sweep(100, 1_000_000));
    let bytes = most["result_bytes"].as_u64().unwrap_or(0);
    eprintln!("100 points of 65,536 outcomes: bound {bytes} bytes");
    assert!(most["accepted"] == false && most["refusal"]["code"] == "J031" && (130_000_000..180_000_000).contains(&bytes), "{most}");
    let taken = dry(port, &sweep(100, 30_000));
    eprintln!("100 points of 30,000 shots: bound {} bytes", taken["result_bytes"]);
    assert!(taken["accepted"] == true, "{taken}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A job's report once it has ended, polled.
fn ended(port: u16, id: &str, within: Duration) -> Value {
    let t = Instant::now();
    loop {
        let (code, b) = json_of(port, "GET", &format!("/v1/jobs/{id}?wait=5"), None);
        assert_eq!(code, 200, "{b}");
        if b["status"] != "queued" && b["status"] != "running" {
            return b;
        }
        assert!(t.elapsed() < within, "{id} still {} after {within:?}", b["status"]);
    }
}

/// [`json_of`] as the client `who` (through a trusted proxy).
fn json_as(port: u16, who: &str, path: &str, body: &Value) -> (u16, Value) {
    let (code, out) = post_as(port, who, path, body);
    let text = out.split_once("\r\n\r\n").map(|(_, b)| b).unwrap_or("");
    (code, serde_json::from_str(text).unwrap_or(Value::Null))
}

/// An MCP tool's answer: its result's `structuredContent`, and whether it
/// was an error.
fn mcp_tool(port: u16, tool: &str, args: &Value) -> (bool, Value) {
    let body = json!({ "jsonrpc": "2.0", "id": 1, "method": "tools/call", "params": { "name": tool, "arguments": args } });
    let (code, b) = json_of(port, "POST", "/mcp", Some(&body));
    assert_eq!(code, 200, "{tool}: {b}");
    (b["result"]["isError"] == true, b["result"]["structuredContent"].clone())
}

/// A 16-qubit sweep: `points` angles, `shots` a point, 16 measured bits.
fn sweep16(points: usize, shots: u64) -> Value {
    json!({ "qasm": "OPENQASM 3.0;\ninput float theta;\nqubit[16] q;\nbit[16] c;\nh q;\nrz(theta) q[0];\nc = measure q;\n", "sweep": { "theta": (0..points).map(|i| 0.01 * (i + 1) as f64).collect::<Vec<_>>() }, "shots": shots, "seed": 7 })
}

/// An MCP dry run says what the MCP run tools take: it plans the job as they
/// run it, its verdict asked, so the most points of a sweep its dry run
/// accepts under a cap of 1 MiB are taken by `run_quantum_set`, where the
/// dry run planned the job without its verdict and the run then refused it
/// at once (`J031`).
#[test]
fn an_mcp_dry_run_says_what_the_run_tools_take() {
    let d = dir("mcp-dry");
    let (mut svc, port) = start_with(&d, 10_000, &[("WAI_QSVC_RESULT_MIB", "1")]);
    let accepted = |n: usize| -> bool {
        let (err, v) = mcp_tool(port, "dry_run_quantum_job", &sweep16(n, 2000));
        assert!(!err, "{v}");
        v["accepted"] == true
    };
    let mut n = 1;
    while accepted(n + 1) {
        n += 1;
        assert!(n < 100);
    }
    let (err, v) = mcp_tool(port, "run_quantum_set", &sweep16(n, 2000));
    let why = v["error"].as_str().unwrap_or("").to_string();
    assert!(!(err && why.contains("J031")), "{n} points, accepted by the MCP dry run, refused by the run: {why}");
    let id = v["id"].as_str().unwrap_or_default().to_string();
    let job = if v["status"] == "done" { v } else { ended(port, &id, Duration::from_secs(120)) };
    assert_eq!(job["status"], "done", "{}", job["error"]);
    let (err, v) = mcp_tool(port, "run_quantum_set", &sweep16(n + 1, 2000));
    assert!(err && v["error"].as_str().is_some_and(|w| w.contains("J031")), "{} points: {v}", n + 1);
    alive(port, &mut svc, "MCP dry runs");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A dry run that says a request is taken is taken: a sweep whose items
/// each carry a large shared field read, kept, past what the service takes
/// back of one reading (`J031`) is refused by its dry run too, as the dry
/// run's reading is held to what the submission's would take; one under it
/// is taken by both.
#[test]
fn a_dry_run_that_says_taken_is_taken_by_its_submission() {
    let d = dir("dry-weight");
    let (mut svc, port) = start(&d, 10_000);
    for pad in [12_000usize, 20_000] {
        let pads = vec!["1e20"; pad].join(",");
        let text = format!("{{\"qasm\":\"OPENQASM 3.0;\\ninput float theta;\\nqubit q;\\nbit c;\\nrx(theta) q;\\nc = measure q;\\n\",\"sweep\":{{\"theta\":[{}]}},\"shots\":10,\"pad\":[{pads}]}}", (0..100).map(|i| format!("0.{}", i + 1)).collect::<Vec<_>>().join(","));
        let send = |dry: bool| -> (u16, String) {
            let body = if dry { text.replacen("{", "{\"dry_run\":true,", 1) } else { text.clone() };
            let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
            s.set_read_timeout(Some(Duration::from_secs(120))).unwrap();
            write!(s, "POST /v1/jobs HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap();
            let mut out = String::new();
            let _ = s.read_to_string(&mut out);
            (out.split_whitespace().nth(1).and_then(|c| c.parse().ok()).unwrap_or(0), out.split_once("\r\n\r\n").map(|(_, b)| b.to_string()).unwrap_or_default())
        };
        let (code, dry) = send(true);
        assert_eq!(code, 200, "{dry}");
        let dry: Value = serde_json::from_str(&dry).unwrap();
        let (code, sent) = send(false);
        eprintln!("pad {pad}: dry run accepted {}, submission {code}", dry["accepted"]);
        assert_eq!(dry["accepted"] == true, code == 202, "pad {pad}: the dry run said {} ({}), the submission {code} {sent}", dry["accepted"], dry["refusal"]);
        if code == 202 {
            let id = serde_json::from_str::<Value>(&sent).unwrap()["id"].as_str().unwrap().to_string();
            let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None);
        } else {
            assert!(sent.contains("J031") && dry["refusal"]["code"] == "J031", "pad {pad}: {sent} / {dry}");
        }
    }
    alive(port, &mut svc, "dry runs held to the submission's reading");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A set's bound counts each item's request as it is served: a number RFC
/// 8785 writes longer than serde (`1e20`, 21 characters there) is counted at
/// its served length, so the result of a sweep sharing thousands of them is
/// within its dry run's `result_bytes`.
#[test]
fn a_set_s_bound_counts_its_items_requests_as_served() {
    let d = dir("set-numbers");
    let (mut svc, port) = start(&d, 10_000);
    let pads = vec!["1e20"; 20_000].join(",");
    let text = format!("{{\"qasm\":\"OPENQASM 3.0;\\ninput float theta;\\nqubit q;\\nbit c;\\nrx(theta) q;\\nc = measure q;\\n\",\"sweep\":{{\"theta\":[0.1,0.2,0.3]}},\"shots\":10,\"pad\":[{pads}]}}");
    let req: Value = serde_json::from_str(&text).unwrap();
    let mut dry = req.clone();
    dry["dry_run"] = json!(true);
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&dry));
    assert_eq!(code, 200, "{b}");
    let bound = b["result_bytes"].as_u64().unwrap();
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&req));
    assert_eq!(code, 202, "{b}");
    let job = ended(port, b["id"].as_str().unwrap(), Duration::from_secs(120));
    assert_eq!(job["status"], "done", "{}", job["error"]);
    let served = wai_quantum::engine::receipt::canonical(&job["result"]).len() as u64;
    eprintln!("3 points sharing 20,000 numbers: served {served} bytes, bound {bound}");
    assert!(served <= bound, "served {served} bytes, past the bound of {bound}");
    alive(port, &mut svc, "a set's numbers");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A GHZ state's sweep is bounded by the two outcomes the stabilizer method
/// can give, not by a distinct outcome a shot: 100 copies of a 24-qubit GHZ
/// state at 30,000 shots each, refused before at a bound of 89 MB for a
/// result of about 250 KB, are taken and run, their result within its bound.
#[test]
fn an_honest_clifford_set_is_taken_by_the_outcomes_it_can_give() {
    let d = dir("ghz-set");
    let (mut svc, port) = start(&d, 10_000);
    let mut ghz = "OPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[24] q;\nbit[24] c;\nh q[0];\n".to_string();
    for i in 1..24 {
        ghz += &format!("cx q[{}], q[{i}];\n", i - 1);
    }
    ghz += "c = measure q;\n";
    let set = json!({ "programs": vec![ghz; 100], "shots": 30_000, "seed": 3 });
    let mut dry = set.clone();
    dry["dry_run"] = json!(true);
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&dry));
    assert_eq!(code, 200, "{b}");
    assert!(b["accepted"] == true, "{}", b["refusal"]);
    let bound = b["result_bytes"].as_u64().unwrap();
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&set));
    assert_eq!(code, 202, "{b}");
    let job = ended(port, b["id"].as_str().unwrap(), Duration::from_secs(300));
    assert_eq!(job["status"], "done", "{}", job["error"]);
    let served = wai_quantum::engine::receipt::canonical(&job["result"]).len() as u64;
    eprintln!("100 GHZ24 at 30,000 shots: served {served} bytes, bound {bound}");
    assert!(served <= bound && bound < 8 << 20, "served {served}, bound {bound}");
    alive(port, &mut svc, "a GHZ set");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A run with its verdict is bounded by what its critique can say, not by a
/// remark on every gate: a loop of 124,700 gates with nothing to remark on is
/// run by an MCP run tool and by `POST /v1/run` with `"verdict": true` (the
/// Studio's request), where both were refused for a 64.7 MiB bound. A job
/// past the cap is refused naming its largest part and what makes it
/// smaller: a verdict's critique of thousands of gates that cancel, the
/// reader's notes, or the counts.
#[test]
fn a_long_loop_s_verdict_is_taken_and_a_refusal_names_its_part() {
    let lp = "OPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[8] q;\nbit[8] c;\nfor int s in [1:4300] {\n  cx q[0], q[1]; rz(0.1) q[1]; cx q[0], q[1];\n  cx q[1], q[2]; rz(0.2) q[2]; cx q[1], q[2];\n  cx q[2], q[3]; rz(0.3) q[3]; cx q[2], q[3];\n  cx q[3], q[4]; rz(0.4) q[4]; cx q[3], q[4];\n  cx q[4], q[5]; rz(0.5) q[5]; cx q[4], q[5];\n  cx q[5], q[6]; rz(0.6) q[6]; cx q[5], q[6];\n  cx q[6], q[7]; rz(0.7) q[7]; cx q[6], q[7];\n  rx(0.1) q[0]; rx(0.1) q[1]; rx(0.1) q[2]; rx(0.1) q[3]; rx(0.1) q[4]; rx(0.1) q[5]; rx(0.1) q[6]; rx(0.1) q[7];\n}\nc = measure q;\n";
    let d = dir("loop-verdict");
    let (mut svc, port) = start(&d, 10_000);
    let (err, v) = mcp_tool(port, "run_quantum_circuit", &json!({ "qasm": lp, "shots": 1000 }));
    assert!(!err, "the MCP run: {v}");
    let (code, b) = json_of(port, "POST", "/v1/run", Some(&json!({ "qasm": lp, "shots": 1000, "verdict": true })));
    assert!(code == 200 && b["status"] == "done", "{code} {}", b["error"]);
    svc.kill().unwrap();
    svc.wait().unwrap();
    // Past a cap of 1 MiB: each refusal names the part that passes it.
    let (mut svc, port) = start_with(&d, 10_000, &[("WAI_QSVC_RESULT_MIB", "1")]);
    let cancels = "OPENQASM 3.0;\nqubit q;\nbit c;\nfor int i in [1:10000] { h q; h q; }\nc = measure q;\n";
    let noted = format!("OPENQASM 3.0;\nqubit q;\nbit c;\n{}h q;\nc = measure q;\n", (0..7000).map(|i| format!("pragma note{i} {}\n", "x".repeat(120))).collect::<String>());
    for (what, job, says) in [
        ("a critique of 10,000 pairs", json!({ "qasm": cancels, "shots": 10, "verdict": true }), "its verdict"),
        ("7,000 notes", json!({ "qasm": noted, "shots": 10 }), "the reader's warnings and notes"),
        ("a sweep's counts", sweep16(40, 2000), "its counts"),
    ] {
        let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&job));
        let why = b["error"].as_str().unwrap_or("");
        eprintln!("{what}: {code} {why}");
        assert!(code == 422 && why.contains("J031") && why.contains(says), "{what}: {code} {why}");
    }
    alive(port, &mut svc, "verdicts bounded by their critique");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// The README's largest sweep, 100 points of 16 measured bits at 30,000
/// shots a point, is taken and ends done at the service's defaults: the set's
/// worker keeps each item's result as text as it ends, where it kept every
/// item's as a tree and passed the memory it was admitted with (on Linux
/// every time, `the job used 693 MiB, over the 258 MiB`).
#[test]
fn the_readme_s_largest_sweep_is_taken_and_ends_done() {
    let d = dir("readme-sweep");
    let (mut svc, port) = start(&d, 10_000);
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&sweep16(100, 30_000)));
    assert_eq!(code, 202, "{b}");
    let job = ended(port, b["id"].as_str().unwrap(), Duration::from_secs(600));
    assert_eq!(job["status"], "done", "{}", job["error"]);
    assert_eq!(job["result"]["items"].as_array().map(Vec::len), Some(100));
    alive(port, &mut svc, "the README's sweep");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A client's quick calls are never stuck behind two of its own long checks:
/// one client sends two thresholds that each run past the 10 s a request
/// waits for a turn, then four critiques. The first threshold holds the
/// client's own turn and the second a lent one; the lent turn goes back once
/// a critique of its own has waited 7 s, so every critique is answered (the
/// second threshold may be refused, `J032`, to send again), where every
/// critique waited out its 10 s and was refused.
#[test]
fn a_client_s_quick_calls_pass_two_of_its_own_long_checks() {
    let d = dir("two-long");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]);
    let me = "198.51.100.91";
    let critique = json!({ "qasm": "OPENQASM 3.0;\nqubit[2] q;\nh q[0];\ncx q[0], q[1];\n" });
    let (threshold, alone) = slow_threshold(port);
    if alone < 11.0 {
        eprintln!("the threshold took {alone:.1} s alone, under the 10 s a request waits for a turn: nothing to test on this host");
    }
    let began = Instant::now();
    let answers: Vec<(String, u16, String, f64)> = std::thread::scope(|t| {
        let mut calls = Vec::new();
        for (i, (name, path, body, after)) in [("threshold 1", "/v1/qec/threshold", &threshold, 0u64), ("threshold 2", "/v1/qec/threshold", &threshold, 30)].into_iter().chain((0..4).map(|k| ("critique", "/v1/critique", &critique, 4000 + 30 * k))).enumerate() {
            let began = &began;
            calls.push(t.spawn(move || {
                std::thread::sleep(Duration::from_millis(after));
                let (code, out) = post_as(port, me, path, body);
                (format!("{name} #{i}"), code, out.lines().last().unwrap_or("").to_string(), began.elapsed().as_secs_f64())
            }));
        }
        calls.into_iter().map(|c| c.join().unwrap()).collect()
    });
    eprintln!("{answers:#?}");
    for (name, code, out, at) in &answers {
        if name.starts_with("critique") || name.starts_with("threshold 1") {
            assert_eq!(*code, 200, "{name} at {at:.2} s: {out}");
        } else {
            assert!(*code == 200 || (*code == 503 && out.contains("lent")), "{name} at {at:.2} s: {code} {out}");
        }
    }
    alive(port, &mut svc, "two long checks and quick calls");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// Each JSON-RPC id is echoed as it was sent: a fraction or an exponent
/// (`1.0`, `1e3`), and an integer past 64 bits, which read as a number and
/// written again came back as `1`, `1000` and `1.2345678901234568e+29`.
#[test]
fn an_mcp_id_is_echoed_as_it_was_sent() {
    let d = dir("mcp-raw-ids");
    let (mut svc, port) = start(&d, 1000);
    for id in ["1.0", "1e3", "1E3", "-0.0", "1.50", "18446744073709551616", "123456789012345678901234567890", "\"1.0\"", "7"] {
        for method in ["ping", "tools/list"] {
            let body = format!("{{\"jsonrpc\":\"2.0\",\"id\":{id},\"method\":\"{method}\"}}");
            let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
            s.set_read_timeout(Some(Duration::from_secs(60))).unwrap();
            write!(s, "POST /mcp HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}", body.len()).unwrap();
            let mut out = String::new();
            let _ = s.read_to_string(&mut out);
            assert!(out.starts_with("HTTP/1.1 200"), "{id} {method}: {out}");
            assert!(out.contains(&format!("\"id\":{id},")) || out.contains(&format!("\"id\":{id}}}")), "{method}: sent the id {id}, answered {}", out.split_once("\r\n\r\n").map(|(_, b)| b).unwrap_or("").chars().take(120).collect::<String>());
        }
    }
    alive(port, &mut svc, "ids echoed as sent");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// A long poll past its client's holds is not held, and says so: `429`,
/// `J032`, `Retry-After` and the job's status, so a client polling in a
/// loop waits, where it was answered `200` at once, unchanged, and a client
/// that polled again at once spun. Another client's long poll is held.
#[test]
fn a_long_poll_past_its_client_s_holds_says_when_to_poll_again() {
    let d = dir("poll-again");
    let (mut svc, port) = start_with(&d, 1000, &[("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1")]);
    let (code, b) = json_of(port, "POST", "/v1/jobs", Some(&pump()));
    assert_eq!(code, 202, "{b}");
    let id = b["id"].as_str().unwrap().to_string();
    let me = "198.51.100.93";
    let streams: Vec<TcpStream> = (0..4)
        .map(|_| {
            let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
            write!(s, "GET /v1/jobs/{id}/events HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: {me}\r\n\r\n").unwrap();
            s
        })
        .collect();
    std::thread::sleep(Duration::from_millis(500));
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(60))).unwrap();
    let t = Instant::now();
    write!(s, "GET /v1/jobs/{id}?wait=25 HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: {me}\r\nConnection: close\r\n\r\n").unwrap();
    let mut out = String::new();
    let _ = s.read_to_string(&mut out);
    let took = t.elapsed();
    let (head, body) = out.split_once("\r\n\r\n").unwrap_or_default();
    let body: Value = serde_json::from_str(body).unwrap_or(Value::Null);
    assert!(head.starts_with("HTTP/1.1 429") && head.to_ascii_lowercase().contains("retry-after: 2"), "a fifth hold: {head}");
    assert!(body["code"] == "J032" && (body["status"] == "running" || body["status"] == "queued") && took < Duration::from_secs(5), "{body} after {took:?}");
    // Another client's long poll is held.
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(60))).unwrap();
    let t = Instant::now();
    write!(s, "GET /v1/jobs/{id}?wait=2 HTTP/1.1\r\nHost: x\r\nX-Forwarded-For: 198.51.100.94\r\nConnection: close\r\n\r\n").unwrap();
    let mut out = String::new();
    let _ = s.read_to_string(&mut out);
    assert!(out.starts_with("HTTP/1.1 200") && t.elapsed() >= Duration::from_millis(1900), "another client's poll: {:?} {}", t.elapsed(), out.lines().next().unwrap_or(""));
    drop(streams);
    let _ = http(port, "DELETE", &format!("/v1/jobs/{id}"), None);
    alive(port, &mut svc, "polls past the holds");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}

/// What ended jobs keep unread stays under `WAI_QSVC_UNREAD_MIB`: a job is
/// taken only while every result not yet read, those of the jobs queued and
/// running at their bounds among them, and its own bound stay under the cap.
/// Thirty clients each send a job of about 0.8 MiB while one job holds the
/// only worker; where each was taken because nothing had ended, so their
/// results passed the 2 MiB cap six times over once they ended, those past
/// the cap are refused, `503`, and what they all hold unread stays under it.
#[test]
fn ended_jobs_unread_stay_under_the_unread_cap() {
    let d = dir("unread-cap");
    let env = [("WAI_QSVC_TRUSTED_PROXIES", "127.0.0.1"), ("WAI_QSVC_UNREAD_MIB", "2"), ("WAI_QSVC_KEPT_MIB", "1"), ("WAI_QSVC_CLIENT_KEPT_MIB", "1"), ("WAI_QSVC_RESULT_MIB", "1"), ("WAI_QSVC_CLIENT_UNREAD_MIB", "4"), ("WAI_QSVC_MAX_QUEUE", "40")];
    let (mut svc, port) = start_with(&d, 1000, &env);
    let (code, b) = json_as(port, "198.51.100.200", "/v1/jobs", &pump());
    assert_eq!(code, 202, "{b}");
    let blocker = b["id"].as_str().unwrap().to_string();
    let job = json!({ "qasm": "OPENQASM 3.0;\nqubit[15] q;\nbit[15] c;\nh q;\nc = measure q;\n", "shots": 30_000, "seed": 1 });
    let mut dry = job.clone();
    dry["dry_run"] = json!(true);
    let (_, b) = json_as(port, "198.51.100.199", "/v1/jobs", &dry);
    let bound = b["result_bytes"].as_u64().unwrap();
    assert!(b["accepted"] == true && bound > 512 << 10 && bound < 1 << 20, "{b}");
    let mut taken = Vec::new();
    for k in 0..30 {
        let (code, b) = json_as(port, &format!("198.51.100.{}", k + 1), "/v1/jobs", &job);
        match code {
            202 => taken.push(b["id"].as_str().unwrap().to_string()),
            503 => assert!(b["error"].as_str().is_some_and(|w| w.contains("not yet read") && w.contains("J032")), "{b}"),
            _ => panic!("{code} {b}"),
        }
    }
    let _ = http(port, "DELETE", &format!("/v1/jobs/{blocker}"), None);
    // Each ended, none read until all have.
    let t = Instant::now();
    loop {
        let (_, v) = json_of(port, "GET", "/v1", None);
        if v["load"]["queued"] == 0 && v["load"]["running"] == 0 {
            break;
        }
        assert!(t.elapsed() < Duration::from_secs(120), "{}", v["load"]);
        std::thread::sleep(Duration::from_millis(200));
    }
    let held: u64 = taken.iter().map(|id| {
        let (_, j) = json_of(port, "GET", &format!("/v1/jobs/{id}"), None);
        assert_eq!(j["status"], "done", "{}", j["error"]);
        http(port, "GET", &format!("/v1/jobs/{id}"), None).2.len() as u64
    }).sum();
    eprintln!("{} of 30 taken, bound {bound} each; held unread at once: {held} bytes of jobs' reports", taken.len());
    assert!(!taken.is_empty() && taken.len() < 30, "{} of 30 taken", taken.len());
    assert!(taken.len() as u64 * bound <= 2 << 20, "{} jobs of {bound} bytes taken under a 2 MiB cap", taken.len());
    alive(port, &mut svc, "the unread cap");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&d);
}