//! A program written for a task, against the service itself, its writer
//! process running a script in the model's place (`WAI_QSVC_WRITER_SCRIPT`):
//! a Bell pair's task, whose first program the engine finds wrong and whose
//! repair passes. The script is a FIFO (on Windows, a named pipe; [`Late`]), so
//! the writer holds at its start, with its model "loading", until the test writes
//! the script into it: until then every write stays queued, which is how the
//! per-client limit, the order and the attempts' events are seen as they happen.

use serde_json::{Value, json};
use std::io::{Read, Write};
use std::net::TcpStream;
use std::path::{Path, PathBuf};
use std::process::{Command, Stdio};
use std::time::{Duration, Instant};
use wai_quantum::engine::schema::Registry;

mod common;

use common::Running;

const BELL: &str = "```qasm\nOPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[2] q;\nbit[2] c;\nh q[0];\ncx q[0], q[1];\nc = measure q;\n```";
const NO_CNOT: &str = "```qasm\nOPENQASM 3.0;\ninclude \"stdgates.inc\";\nqubit[2] q;\nbit[2] c;\nh q[0];\nc = measure q;\n```";

fn bell_task() -> Value {
    json!({
        "schema": "wai.quantum.exercise/1",
        "title": "A Bell pair",
        "task": "Entangle two qubits so that their measured bits always agree.",
        "starter": ["OPENQASM 3.0;", "include \"stdgates.inc\";", "qubit[2] q;", "bit[2] c;", "c = measure q;"],
        "intent": { "distribution": { "00": 0.5, "11": 0.5 } }
    })
}

/// A body sent with chunked transfer coding, as it was meant.
fn dechunk(body: &str) -> String {
    let (mut out, mut rest) = (String::new(), body);
    while let Some((size, after)) = rest.split_once("\r\n") {
        let Ok(n) = usize::from_str_radix(size.trim(), 16) else { break };
        if n == 0 || after.len() < n {
            break;
        }
        out.push_str(&after[..n]);
        rest = after[n..].strip_prefix("\r\n").unwrap_or(&after[n..]);
    }
    out
}

/// One HTTP/1.1 exchange, with extra header fields: the status, the headers
/// (lowercased) and the body.
fn http(port: u16, method: &str, path: &str, body: Option<&Value>, extra: &[(&str, &str)]) -> (u16, String, String) {
    let mut s = TcpStream::connect(("127.0.0.1", port)).unwrap();
    s.set_read_timeout(Some(Duration::from_secs(180))).unwrap();
    let text = body.map(Value::to_string).unwrap_or_default();
    let fields: String = extra.iter().map(|(k, v)| format!("{k}: {v}\r\n")).collect();
    write!(s, "{method} {path} HTTP/1.1\r\nHost: 127.0.0.1\r\nContent-Type: application/json\r\nAccept: application/json, text/event-stream\r\n{fields}Content-Length: {}\r\nConnection: close\r\n\r\n{text}", text.len()).unwrap();
    let mut out = String::new();
    s.read_to_string(&mut out).unwrap();
    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_ascii_lowercase(), b.to_string())).unwrap_or_default();
    let body = if head.contains("transfer-encoding: chunked") { dechunk(&body) } else { body };
    (status, head, body)
}

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

fn rpc(port: u16, method: &str, params: Value) -> Value {
    let (code, v) = json_of(port, "POST", "/mcp", Some(&json!({ "jsonrpc": "2.0", "id": 1, "method": method, "params": params })), &[]);
    assert_eq!(code, 200, "{method}: {v}");
    v
}

/// The service, its writer running the script at `script` if any, once it answers: its
/// process and the port it bound.
fn start(dir: &Path, script: Option<&Path>) -> (Running, u16) {
    start_with(dir, script, 120)
}

/// [`start`], a write (and the writer's start) held to `write_secs`.
fn start_with(dir: &Path, script: Option<&Path>, write_secs: u64) -> (Running, u16) {
    start_env(dir, script, write_secs, &[])
}

/// [`start_with`], with more of the service's variables. The service binds a port the
/// system chooses and says which in its log, as tests/contract.rs's and tests/depth.rs's
/// services do. A port found free here (a listener bound to port 0, then closed) and handed
/// to it raced: a child another test forks in that moment (its `mkfifo`, its service)
/// inherits the listener until its exec, so this test's first connection reached that copy
/// and was reset, before the service had bound the port (CI, run 28969 on macOS and
/// 04db5802a's on Linux, each 2 ms into the binary).
fn start_env(dir: &Path, script: Option<&Path>, write_secs: u64, env: &[(&str, &str)]) -> (Running, u16) {
    let path = dir.join("service.log");
    let mut cmd = Command::new(env!("CARGO_BIN_EXE_wai-quantum-service"));
    cmd.arg("serve")
        .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_WRITE_SECS", write_secs.to_string())
        .env_remove("WAI_QSVC_MODEL")
        .env_remove("WAI_QSVC_WRITER_SCRIPT")
        .stdout(Stdio::null())
        .stderr(std::fs::File::create(&path).expect("the service's log"));
    if let Some(s) = script {
        cmd.env("WAI_QSVC_WRITER_SCRIPT", s);
    }
    for (k, v) in env {
        cmd.env(k, v);
    }
    let mut child = Running::new(cmd.spawn().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(60), "the service did not start in 60 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(60), "the service did not answer: {}", log());
        std::thread::sleep(Duration::from_millis(50));
    }
    (child, port)
}

/// Waits until a job's status is `want`.
fn until(port: u16, id: &str, want: &str) -> Value {
    let t = Instant::now();
    loop {
        let (_, j) = json_of(port, "GET", &format!("/v1/jobs/{id}"), None, &[]);
        if j["status"] == want {
            return j;
        }
        assert!(t.elapsed() < Duration::from_secs(120), "{id} never {want}: {j}");
        std::thread::sleep(Duration::from_millis(25));
    }
}

/// The submissions left in this client's window, from a write's answer.
fn left(head: &str) -> u64 {
    head.lines().find_map(|l| l.strip_prefix("ratelimit: ")).and_then(|v| v.split("r=").nth(1)).and_then(|v| v.split(';').next()).and_then(|v| v.trim().parse().ok()).unwrap()
}

/// A document's canonical text (RFC 8785), which a receipt binds: what "the
/// same" means for two documents, whichever reader carried them.
fn canonical(v: &Value) -> String {
    String::from_utf8(wai_quantum::engine::receipt::canonical(v)).unwrap()
}

/// The code a refusal ends with.
fn code_of(v: &Value) -> &str {
    v["code"].as_str().unwrap_or_else(|| panic!("no code: {v}"))
}

/// A writer's script that reaches it late: a path whose read waits until the test
/// gives the script ([`Late::give`]), so the writer's start, its model "loading",
/// lasts as long as the test needs. Unix: a FIFO, whose open waits for a writer.
/// Windows' file system has no FIFO: there it is a named pipe (`\\.\pipe\...`) this
/// test serves, every connection held unanswered until the script is given, which
/// the writer's read waits on as it would on a FIFO.
struct Late {
    path: PathBuf,
    /// The pipe's connections, in the order the writers opened them.
    #[cfg(windows)]
    held: std::sync::mpsc::Receiver<std::fs::File>,
}

impl Late {
    #[cfg(unix)]
    fn new(dir: &Path) -> Late {
        let path = dir.join("script.json");
        assert!(Command::new("mkfifo").arg(&path).status().unwrap().success());
        Late { path }
    }

    /// The script, to whichever writer reads it now.
    #[cfg(unix)]
    fn give(&self, script: &str) {
        std::fs::write(&self.path, script).unwrap();
    }

    /// A pipe with a connection waiting before the service starts, as a FIFO is
    /// there before it, and another as each is taken.
    #[cfg(windows)]
    fn new(_: &Path) -> Late {
        use std::os::windows::ffi::OsStrExt;
        use std::os::windows::io::AsRawHandle;
        static N: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
        let name = format!(r"\\.\pipe\wai-qsvc-write-{}-{}", std::process::id(), N.fetch_add(1, std::sync::atomic::Ordering::SeqCst));
        let wide: Vec<u16> = std::ffi::OsStr::new(&name).encode_wide().chain([0]).collect();
        let mut next = pipe::listen(&wide);
        let (hold, held) = std::sync::mpsc::channel();
        std::thread::spawn(move || {
            loop {
                let connected = pipe::connect(next.as_raw_handle());
                let taken = std::mem::replace(&mut next, pipe::listen(&wide));
                if connected && hold.send(taken).is_err() {
                    return;
                }
            }
        });
        Late { path: PathBuf::from(name), held }
    }

    /// The script, to the writer that has the pipe open now: given to the
    /// connections held, in the order they were opened, until one takes it (a
    /// writer stopped at its start leaves its own closed), and if none does, to
    /// the next.
    #[cfg(windows)]
    fn give(&self, script: &str) {
        let to = |mut c: std::fs::File| c.write_all(script.as_bytes()).and_then(|_| c.sync_all()).is_ok();
        let mut given = self.held.try_iter().any(to);
        while !given {
            given = to(self.held.recv_timeout(Duration::from_secs(60)).expect("a writer opened its script within 60 s"));
        }
    }
}

/// A named pipe's two calls the tests make on Windows ([`Late`]).
#[cfg(windows)]
mod pipe {
    use std::os::windows::io::{FromRawHandle, RawHandle};

    #[link(name = "kernel32")]
    unsafe extern "system" {
        fn CreateNamedPipeW(name: *const u16, open_mode: u32, pipe_mode: u32, instances: u32, out_size: u32, in_size: u32, timeout_ms: u32, security: *mut std::ffi::c_void) -> RawHandle;
        fn ConnectNamedPipe(pipe: RawHandle, overlapped: *mut std::ffi::c_void) -> i32;
    }
    /// `PIPE_ACCESS_OUTBOUND`: this end writes, the writer's reads.
    const OUTBOUND: u32 = 0x2;
    /// `PIPE_TYPE_BYTE | PIPE_READMODE_BYTE | PIPE_WAIT`.
    const BYTES_WAITING: u32 = 0;
    /// `PIPE_UNLIMITED_INSTANCES`.
    const UNLIMITED: u32 = 255;
    /// `ERROR_PIPE_CONNECTED`: a writer opened it before it was waited on.
    const CONNECTED_BEFORE: i32 = 535;

    /// A new instance of the pipe `name` (wide, ending in a nul), waiting for a writer.
    pub fn listen(name: &[u16]) -> std::fs::File {
        // SAFETY: the name ends in a nul; no security attributes are given.
        let h = unsafe { CreateNamedPipeW(name.as_ptr(), OUTBOUND, BYTES_WAITING, UNLIMITED, 1 << 16, 0, 0, std::ptr::null_mut()) };
        assert!(h as isize != -1, "a named pipe: {}", std::io::Error::last_os_error());
        // SAFETY: a handle just made, owned by the file from here.
        unsafe { std::fs::File::from_raw_handle(h) }
    }

    /// Waits for a writer to open the instance `h`: whether one did.
    pub fn connect(h: RawHandle) -> bool {
        // SAFETY: an instance made by [`listen`], waited on without overlapped I/O.
        (unsafe { ConnectNamedPipe(h, std::ptr::null_mut()) }) != 0 || std::io::Error::last_os_error().raw_os_error() == Some(CONNECTED_BEFORE)
    }
}

#[test]
fn a_write_passes_after_one_repair_its_attempts_stream_as_they_land_and_its_receipt_verifies() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-write-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let late = Late::new(&dir);
    let (mut svc, port) = start(&dir, Some(&late.path));

    // The service says it writes, its model still loading.
    let (_, d) = json_of(port, "GET", "/v1", None, &[]);
    let w = &d["write"];
    assert_eq!((w["available"].as_bool(), w["writer"].as_str(), w["model"].is_null()), (Some(true), Some("loading its model"), true), "{w}");
    assert_eq!((w["limits"]["writes_at_once"].as_u64(), w["limits"]["writes_per_client"].as_u64(), w["limits"]["write_seconds"].as_u64()), (Some(1), Some(1), Some(120)), "{w}");
    assert_eq!((w["limits"]["task_bytes"].as_u64(), w["limits"]["prompt_bytes"].as_u64(), w["limits"]["cases"].as_u64(), w["limits"]["intents"].as_u64()), (Some(65536), Some(8192), Some(16), Some(8)), "{w}");
    assert!(d["endpoints"].as_array().unwrap().iter().any(|e| e == "POST /v1/write"), "{d}");
    // The queue's limit is each kind's.
    assert_eq!(d["limits"]["queue"]["per_kind"].as_u64(), Some(256), "{d}");
    let (_, api) = json_of(port, "GET", "/v1/openapi.json", None, &[]);
    assert!(api["paths"]["/v1/write"]["post"]["responses"]["422"]["description"].as_str().unwrap().contains("J031"), "{}", api["paths"]["/v1/write"]);

    // A malformed task, and options out of range, each refused with its code.
    for (req, status, code) in [
        (json!({ "task": { "task": "anything" } }), 400, "J081"),
        (json!({ "task": { "cases": [{ "name": "n = 1", "inputs": {} }] } }), 400, "J081"),
        (json!({ "task": bell_task(), "rounds": 9 }), 400, "J082"),
        (json!({ "task": bell_task(), "max_new": 0 }), 400, "J082"),
        (json!({ "task": bell_task(), "qasm": "OPENQASM 3.0;" }), 400, "J082"),
    ] {
        let (c, v) = json_of(port, "POST", "/v1/write", Some(&req), &[]);
        assert_eq!((c, code_of(&v)), (status, code), "{req}: {v}");
    }

    // A write, queued while the writer loads; sent again under its key, the
    // same job, counting as no submission.
    let req = json!({ "task": bell_task() });
    let (c, head, body) = http(port, "POST", "/v1/write", Some(&req), &[("Idempotency-Key", "write-1")]);
    let a: Value = serde_json::from_str(&body).unwrap();
    assert_eq!((c, a["status"].as_str(), a["method"].as_str()), (202, Some("queued"), Some("write")), "{a}");
    let id = a["id"].as_str().unwrap().to_string();
    let (c, head2, body) = http(port, "POST", "/v1/write", Some(&req), &[("Idempotency-Key", "write-1")]);
    assert_eq!((c, serde_json::from_str::<Value>(&body).unwrap()["id"].as_str()), (202, Some(id.as_str())), "{body}");
    assert_eq!(left(&head2), left(&head), "a write sent again under its key counts as nothing");
    // The same key with another request is refused.
    let (c, v) = json_of(port, "POST", "/v1/write", Some(&json!({ "task": bell_task(), "rounds": 1 })), &[("Idempotency-Key", "write-1")]);
    assert_eq!((c, code_of(&v)), (422, "J034"), "{v}");
    // One write a client: a second, queued or running beside the first, is refused.
    let (c, v) = json_of(port, "POST", "/v1/write", Some(&req), &[]);
    assert_eq!((c, code_of(&v)), (429, "J030"), "{v}");
    assert!(v["error"].as_str().unwrap().contains("one at a time"), "{v}");
    // Another client's write waits behind it.
    let (c, b) = json_of(port, "POST", "/v1/write", Some(&req), &[("X-Forwarded-For", "203.0.113.9")]);
    assert_eq!((c, b["status"].as_str()), (202, Some("queued")), "{b}");
    let other = b["id"].as_str().unwrap().to_string();

    // The write's events, followed from while it is queued to its end.
    let events = {
        let path = format!("/v1/jobs/{id}/events");
        std::thread::spawn(move || http(port, "GET", &path, None, &[]).2)
    };
    std::thread::sleep(Duration::from_millis(300));
    // The writer's model loads: the script reaches it.
    let script = json!([NO_CNOT, BELL]);
    late.give(&script.to_string());

    let done = until(port, &id, "done");
    let t = &done["result"];
    // The transcript: the task, the model, the prompt, the decoding, each attempt and the pass.
    assert_eq!((t["schema"].as_str(), t["passed"].as_bool(), t["passed_round"].as_u64(), t["level"].as_str()), (Some("wai.quantum.write/1"), Some(true), Some(1), Some("proof")), "{t}");
    assert_eq!(canonical(&t["task"]), canonical(&bell_task()));
    assert_eq!(t["task_digest"], blake3::hash(&wai_quantum::engine::receipt::canonical(&bell_task())).to_hex().to_string());
    assert_eq!(t["model"], json!({ "name": "scripted", "licence": "none", "digest": wai_quantum_writer::digest(&script) }));
    assert_eq!(t["decoding"], json!({ "method": "greedy", "max_new": 3000, "think_max": 2000, "rounds": 4 }));
    // Each check's run was planned in one job's memory (WAI_QSVC_JOB_GB), and the transcript says so.
    assert_eq!(t["check_memory_bytes"].as_u64(), Some(2 << 30), "{t}");
    assert_eq!((t["prompt"]["version"].as_str(), t["engine"].as_str()), (Some(wai_quantum_writer::PROMPT_VERSION), Some(wai_quantum::engine::schema::ENGINE_VERSION)), "{t}");
    let attempts = t["attempts"].as_array().unwrap();
    assert_eq!(attempts.len(), 2, "{t}");
    assert_eq!((attempts[0]["passed"].as_bool(), attempts[0]["check"]["what_went_wrong"][0]["kind"].as_str()), (Some(false), Some("distribution")), "{t}");
    assert!(attempts[1]["passed"] == true && t["program"].as_str().unwrap().contains("cx q[0], q[1];"), "{t}");
    assert_eq!(t["program_digest"], blake3::hash(t["program"].as_str().unwrap().as_bytes()).to_hex().to_string());
    assert!(t["energy"]["unmetered"].is_string(), "the energy is inside the signed body: {t}");
    // The timing is beside the transcript, outside the signed body.
    let timing = done["timing"].as_array().unwrap();
    assert!(timing.len() == 2 && timing.iter().all(|x| x["reply_seconds"].as_f64().is_some() && x["check_seconds"].as_f64().is_some()), "{done}");
    assert!(t.get("timing").is_none() && done.get("progress").is_none(), "{done}");
    // The job follows its schema.
    let errs = wai_quantum_schemas().check("service-tools.v1.json#/$defs/write", &done);
    assert!(errs.is_empty(), "{}\n{done}", errs.join("; "));

    // Its receipt verifies against its request, and a changed transcript does not.
    let (_, v) = json_of(port, "POST", "/v1/verify", Some(&json!({ "request": done["request"], "result": done["result"] })), &[]);
    assert_eq!(v["valid"], true, "{v}");
    let mut forged = done["result"].clone();
    forged["passed_round"] = json!(0);
    let (_, v) = json_of(port, "POST", "/v1/verify", Some(&json!({ "request": done["request"], "result": forged })), &[]);
    assert_eq!(v["valid"], false, "{v}");

    // Its events: its status, then each attempt as it landed, then the job.
    let stream = events.join().unwrap();
    let mut seen: Vec<(String, Value)> = Vec::new();
    let mut name = String::new();
    for line in stream.lines() {
        if let Some(e) = line.strip_prefix("event:") {
            name = e.trim().to_string();
        } else if let Some(d) = line.strip_prefix("data:") {
            seen.push((name.clone(), serde_json::from_str(d.trim()).unwrap_or_else(|e| panic!("{e}: {d}"))));
        }
    }
    // The stream began while the write was queued; a scripted write runs in
    // milliseconds, so the stream may find it running or already past it.
    let statuses: Vec<&str> = seen.iter().filter(|(k, _)| k == "status").filter_map(|(_, v)| v["status"].as_str()).collect();
    assert!(statuses == ["queued", "running"] || statuses == ["queued"], "{stream}");
    let rest: Vec<&(String, Value)> = seen.iter().filter(|(k, _)| k != "status").collect();
    assert_eq!(rest.iter().map(|(k, _)| k.as_str()).collect::<Vec<_>>(), ["attempt", "attempt", "job"], "{stream}");
    for (n, (_, a)) in rest[..2].iter().enumerate() {
        assert_eq!((a["id"].as_str(), a["attempt"]["round"].as_u64(), a["attempt"]["passed"].as_bool()), (Some(id.as_str()), Some(n as u64), Some(n == 1)), "{a}");
        assert_eq!(canonical(&a["attempt"]), canonical(&attempts[n]));
        assert!(a["timing"]["reply_seconds"].is_number(), "{a}");
    }
    assert_eq!(rest[2].1["status"], "done");

    // The other client's write ran after it, one at a time, and the same
    // task gave the same transcript.
    let b = until(port, &other, "done");
    assert!(b["started_ms"].as_u64().unwrap() >= done["finished_ms"].as_u64().unwrap(), "two writes ran at once: {done}\n{b}");
    let strip = |r: &Value| {
        let mut r = r.clone();
        for k in ["energy", "receipt"] {
            r.as_object_mut().unwrap().remove(k);
        }
        r
    };
    assert_eq!(canonical(&strip(&b["result"])), canonical(&strip(&done["result"])));

    // The service names its model now, loaded.
    let (_, d) = json_of(port, "GET", "/v1", None, &[]);
    assert_eq!((d["write"]["writer"].as_str(), &d["write"]["model"]), (Some("ready"), &t["model"]), "{d}");

    // As an MCP task: made at once, followed to its end, its result the job,
    // with the verdict of the program it wrote, held to the tool's schema.
    let tools = rpc(port, "tools/list", json!({}));
    let tool = tools["result"]["tools"].as_array().unwrap().iter().find(|t| t["name"] == "write_circuit").cloned().unwrap();
    assert_eq!(tool["execution"]["taskSupport"], "optional", "{tool}");
    let made = rpc(port, "tools/call", json!({ "name": "write_circuit", "arguments": { "task": bell_task(), "rounds": 1 }, "task": { "ttl": 60000 } }));
    let task = made["result"]["task"]["taskId"].as_str().unwrap_or_else(|| panic!("{made}")).to_string();
    let t0 = Instant::now();
    while rpc(port, "tasks/get", json!({ "taskId": task }))["result"]["status"] == "working" {
        assert!(t0.elapsed() < Duration::from_secs(120), "the task never ended");
        std::thread::sleep(Duration::from_millis(50));
    }
    let r = rpc(port, "tasks/result", json!({ "taskId": task }));
    let sc = &r["result"]["structuredContent"];
    assert_eq!((r["result"]["isError"].as_bool(), sc["status"].as_str(), sc["result"]["passed_round"].as_u64()), (Some(false), Some("done"), Some(1)), "{r}");
    assert_eq!((sc["verdict"]["level"].as_str(), sc["verdict"]["holds"].as_bool()), (Some("proof"), Some(true)), "{r}");
    assert!(sc["verdict"]["statement"].as_str().unwrap().starts_with("the program written at round 1"), "{r}");
    let get = tools["result"]["tools"].as_array().unwrap().iter().find(|t| t["name"] == "get_quantum_job").cloned().unwrap();
    let errs = Registry::empty().check_value(&get["outputSchema"], sc);
    assert!(errs.is_empty(), "{}\n{sc}", errs.join("; "));
    assert!(r["result"]["content"][0]["text"].as_str().unwrap().contains("passed after 1 repair(s)"), "{r}");

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

#[test]
fn a_service_with_no_model_refuses_a_write_and_says_so() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-write-none-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let (mut svc, port) = start(&dir, None);
    let (_, d) = json_of(port, "GET", "/v1", None, &[]);
    assert_eq!(d["write"]["available"], false, "{d}");
    assert!(d["write"]["why"].as_str().unwrap().contains("no model"), "{d}");
    let (c, v) = json_of(port, "POST", "/v1/write", Some(&json!({ "task": bell_task() })), &[]);
    assert_eq!((c, code_of(&v)), (503, "J080"), "{v}");
    let r = rpc(port, "tools/call", json!({ "name": "write_circuit", "arguments": { "task": bell_task() } }));
    assert_eq!(r["result"]["isError"], true, "{r}");
    assert!(r["result"]["structuredContent"]["error"].as_str().unwrap().ends_with("[J080]"), "{r}");
    let t = rpc(port, "tools/call", json!({ "name": "write_circuit", "arguments": { "task": bell_task() }, "task": {} }));
    assert_eq!((t["error"]["code"].as_i64(), t["error"]["data"]["code"].as_str()), (Some(-32602), Some("J080")), "{t}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&dir);
}

/// A write cancelled while the writer starts is not sent to it once it has
/// started: the cancel stands, and the writer never runs it. The script is a
/// FIFO nothing writes at first, so the writer's start at boot runs out of
/// time and fails; the write, taken next, then waits while the writer starts
/// again, and is cancelled there.
#[test]
fn a_write_cancelled_while_the_writer_starts_is_never_written() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-write-cancel-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let late = Late::new(&dir);
    let (mut svc, port) = start_with(&dir, Some(&late.path), 8);
    let (c, a) = json_of(port, "POST", "/v1/write", Some(&json!({ "task": bell_task() })), &[]);
    assert_eq!(c, 202, "{a}");
    let id = a["id"].as_str().unwrap().to_string();
    // Taken once the first start has failed, while the writer starts again.
    until(port, &id, "running");
    let (_, d) = json_of(port, "GET", "/v1", None, &[]);
    assert_eq!(d["write"]["writer"], "loading its model", "{d}");
    let (c, v) = json_of(port, "DELETE", &format!("/v1/jobs/{id}"), None, &[]);
    assert!(c == 200 && v["status"] == "cancelled", "{c} {v}");
    // The writer loads.
    late.give(&json!([BELL]).to_string());
    let t = Instant::now();
    while json_of(port, "GET", "/v1", None, &[]).1["write"]["writer"] != "ready" {
        assert!(t.elapsed() < Duration::from_secs(60), "the writer never loaded");
        std::thread::sleep(Duration::from_millis(25));
    }
    // A write after it runs, so whatever the writer was sent before it is done.
    let (c, b) = json_of(port, "POST", "/v1/write", Some(&json!({ "task": bell_task() })), &[]);
    assert_eq!(c, 202, "the cancelled write still counts against its client: {b}");
    let next = until(port, b["id"].as_str().unwrap(), "done");
    assert_eq!(next["result"]["passed"], true, "{next}");
    let (_, w) = json_of(port, "GET", &format!("/v1/jobs/{id}"), None, &[]);
    assert_eq!(w["status"], "cancelled", "{w}");
    assert!(w.get("peak_rss_bytes").is_none() && w.get("result").is_none(), "the writer ran the cancelled write: {w}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&dir);
}

/// A write is refused for what costs nothing to read before its client's
/// submission is counted, and its programs are read only after, each once and
/// all within one budget, as work that holds up no other request: four writes at
/// once, each with eight programs that run for millions of statements, are each
/// refused in moments, the service answers meanwhile, and the refusals count
/// against their client's window.
#[test]
fn a_write_s_programs_are_read_last_within_one_budget_and_hold_up_nothing() {
    let dir = std::env::temp_dir().join(format!("wai-qsvc-write-budget-{}", std::process::id()));
    let _ = std::fs::remove_dir_all(&dir);
    std::fs::create_dir_all(&dir).unwrap();
    let script = dir.join("script.json");
    std::fs::write(&script, json!([BELL]).to_string()).unwrap();
    let (mut svc, port) = start_env(&dir, Some(&script), 120, &[("WAI_QSVC_CLIENT_BURST", "3")]);
    let looping = |n: u64| format!("OPENQASM 3.0;\nqubit[1] q;\nint a = 0;\nfor int i in [0:{n}] {{ a += 1; }}\n");
    let body = json!({ "task": { "intent": (0..8).map(|k| json!({ "equivalent": looping(9_999_990 - k) })).collect::<Vec<_>>() } });
    let sent: Vec<_> = (0..4)
        .map(|_| {
            let b = body.clone();
            std::thread::spawn(move || {
                let t = Instant::now();
                let (c, v) = json_of(port, "POST", "/v1/write", Some(&b), &[]);
                (c, v, t.elapsed())
            })
        })
        .collect();
    std::thread::sleep(Duration::from_millis(200));
    let t = Instant::now();
    assert_eq!(http(port, "GET", "/v1/health", None, &[]).0, 200);
    let health = t.elapsed();
    let answers: Vec<(u16, Value, Duration)> = sent.into_iter().map(|h| h.join().unwrap()).collect();
    assert!(health < Duration::from_secs(2), "the service took {health:?} to answer while four writes were read");
    for (c, v, took) in &answers {
        assert!(*took < Duration::from_secs(10), "a write took {took:?} to be refused: {c} {v}");
    }
    // Three counted and refused for their programs, past their budget; the fourth past the client's window.
    let mut codes: Vec<(u16, String)> = answers.iter().map(|(c, v, _)| (*c, code_of(v).to_string())).collect();
    codes.sort();
    assert_eq!(codes, [(400, "J081".to_string()), (400, "J081".to_string()), (400, "J081".to_string()), (429, "J030".to_string())], "{answers:?}");
    assert!(answers.iter().filter(|(c, _, _)| *c == 400).all(|(_, v, _)| v["error"].as_str().unwrap().contains("budget")), "{answers:?}");
    svc.kill().unwrap();
    svc.wait().unwrap();
    let _ = std::fs::remove_dir_all(&dir);
}

/// The service's schemas, as its tools' answers are held to them.
fn wai_quantum_schemas() -> Registry {
    let mut r = Registry::engine();
    r.add("service-tools.v1.json", include_str!("../schemas/service-tools.v1.json")).unwrap();
    r
}