diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index 7296c90ea..2ec0e532f 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -145,6 +145,26 @@ with the command line's records of jobs on a provider, and asks after it by its The cask the release writes had three findings: no blank line between each disk image's checksum and its address, and no `depends_on :macos`. It now has both, so the tap carries it as the release wrote it with no finding. +### The job service: a job sent again under its key is the same job + +A client that lost an answer could not safely ask again: a second submission was a second job, counted against its +limits. The three MCP tools that run also declared `idempotentHint: true`, which was not so. + +- **wai-quantum-service:** + - `POST /v1/jobs` and `POST /v1/run` take an `Idempotency-Key`, and `run_quantum_circuit`, `run_quantum_set` and + `run_on_target` take `idempotency_key`. Sent again with the same key and request, a job is answered with the + key's job, before any limit and counting as no submission. With another request, the key is refused (`J034`). + - A key names its job while the job is kept, whatever its status, but for a job the service interrupted: that key + runs the request again. A record keeps the key's digest with the client's name, never the key, and the key names + its job after a restart. + - The three tools' `idempotentHint` is now false: without a key, each call is a new job. + - `GET /v1` states `idempotency`, and the OpenAPI document describes the header. +- **wai-quantum:** the refusal code `J034`. +- **wai-quantum-cli:** `submit` and `task` send each job under a key of their own. Where no answer came back, they + send it again under that key, twice at most, if the service says it keeps keys. +- **wai-quantum for Python:** `Service.submit` and `Service.run` do the same, and take `idempotency_key`. A job the + service interrupted now ends the wait; before, the wait went on to its deadline. + ### The receipt ledger holds a decode to the frames it delivered, as one animated PNG - **Tariff and invoice version 5** ([extensions/receipt-ledger.md](extensions/receipt-ledger.md) §4.8): version 4's diff --git a/wai/wai-quantum-cli/docs/llms-full.txt b/wai/wai-quantum-cli/docs/llms-full.txt index 4ddbd24b1..16cb8bc47 100644 --- a/wai/wai-quantum-cli/docs/llms-full.txt +++ b/wai/wai-quantum-cli/docs/llms-full.txt @@ -622,7 +622,7 @@ Streamable HTTP, no account and no key. ### `run_quantum_circuit`: Run a quantum circuit -Run an OpenQASM program and return measurement counts, Pauli expectation values and/or amplitudes. The service chooses the method (bit-exact or double-precision state vector up to 32 qubits, single precision on a GPU to 33, stabilizer tableau for Clifford circuits up to 32768 qubits, matrix product states up to 4096, sparse Pauli dynamics for expectation values up to 1024 with an error bound, an exact density matrix or trajectories for noise), reports why and how accurate, and signs the result. Give `intent` (an exact distribution, or a reference program it must equal or whose state it must prepare) and the engine checks it with the run. Every answer carries `verdict`: the level the engine checked to (proof, evidence, executed, parsed, refused), whether the intent holds and a plain statement, sealed in the receipt. Waits up to 50 s; a longer job answers with status running and an id for get_quantum_job. +Run an OpenQASM program and return measurement counts, Pauli expectation values and/or amplitudes. The service chooses the method (bit-exact or double-precision state vector up to 32 qubits, single precision on a GPU to 33, stabilizer tableau for Clifford circuits up to 32768 qubits, matrix product states up to 4096, sparse Pauli dynamics for expectation values up to 1024 with an error bound, an exact density matrix or trajectories for noise), reports why and how accurate, and signs the result. Give `intent` (an exact distribution, or a reference program it must equal or whose state it must prepare) and the engine checks it with the run. Every answer carries `verdict`: the level the engine checked to (proof, evidence, executed, parsed, refused), whether the intent holds and a plain statement, sealed in the receipt. Waits up to 50 s; a longer job answers with status running and an id for get_quantum_job. Each call is a new job unless it gives idempotency_key, which makes sending it again safe. Parameters: - `qasm` (string): The program, OpenQASM 3 (or 2), read as the whole language: registers, aliases and slices; the standard gates, `U` and `gphase`; user `gate` definitions and `def` subroutines, inlined; every modifier (`ctrl @`, `negctrl @`, `inv @`, `pow(k) @`) on any gate, lowered exactly; classical types (int, uint, float, angle, complex, bool, bit, arrays), their arithmetic and built-in functions, computed on reading; `for`, `while`, `switch`, `break`, `continue`; measurement and `reset` anywhere; classical control on measured bits (`if (c[0]) x q[1];`, `if (c >= 2) { … } else { … }`, `&&`, `||`); a `while` on measured bits takes `@wai.max_iterations N` before it. `output` declarations come back in the result; `delay`, `box` and `stretch` are carried and labelled as not applied; `defcal` and `cal` are kept, not run. A program that measures in the middle or acts on what it measured runs on statevector or stabilizer: counts split their shots at each measurement; expectation values are exact, of the state before the final measurements. What the reader refuses comes back with a code, as `[Q103]`; qasm_diagnostics says what each code means. @@ -642,10 +642,11 @@ Parameters: - `pulse` (object): A pulse run's options (method "pulse"): the sample period, the levels kept, the tolerance on the stated error, the steps a sample is divided into, each qubit's parameters, static couplings, and ports beyond the names d0, m0, a0. - `source` (object): The program in another format, in place of qasm: read to the job's qasm before the job is read (a module that does not read is refused with the reader's code, Q700 to Q721), the request keeping its source. - `params` (object): A value for each of the program's parameters (input float theta;, input angle phi;), as {"theta": 0.3}: bound exactly before the program is read; a parameter left without one is refused (J050). In a sweep, the parameters it fixes. +- `idempotency_key` (string): A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job. ### `run_quantum_set`: Run a set of quantum programs, or one over a grid -Run several OpenQASM programs as one set (`programs`: each a program's text or a job), or one program over a grid of its inputs (`qasm` with `sweep`, as {"theta": [0.1, 0.2], "n": [2, 3]}: every combination; integer inputs written out, float and angle inputs bound to the double given). The set's other fields (shots, observables, seed, intent, …) apply to every item. Each item gets its own result and receipt, each stating its verdict, and one receipt over the set binds every item through a Merkle root; verify_quantum_receipt checks it whole. At most 100 items, run one after another in one job. Waits up to 50 s; a longer set answers with status running and an id for get_quantum_job. +Run several OpenQASM programs as one set (`programs`: each a program's text or a job), or one program over a grid of its inputs (`qasm` with `sweep`, as {"theta": [0.1, 0.2], "n": [2, 3]}: every combination; integer inputs written out, float and angle inputs bound to the double given). The set's other fields (shots, observables, seed, intent, …) apply to every item. Each item gets its own result and receipt, each stating its verdict, and one receipt over the set binds every item through a Merkle root; verify_quantum_receipt checks it whole. At most 100 items, run one after another in one job. Waits up to 50 s; a longer set answers with status running and an id for get_quantum_job. Each call is a new job unless it gives idempotency_key, which makes sending it again safe. Parameters: - `qasm` (string): With sweep: the program, which declares the inputs the sweep gives values (`input uint n;`, `input float theta;`, `input angle phi;`). @@ -666,6 +667,7 @@ Parameters: - `params` (object): A value for each of the program's parameters (input float theta;, input angle phi;), as {"theta": 0.3}: bound exactly before the program is read; a parameter left without one is refused (J050). In a sweep, the parameters it fixes. - `programs` (array): The set's programs: each a program's text, or a job ({"qasm", …}) whose own fields win over the set's. - `sweep` (object): Each input of qasm and its values: every combination runs, the names in sorted order, the first varying slowest; an integer input's values are whole numbers. +- `idempotency_key` (string): A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job. ### `dry_run_quantum_job`: Plan a quantum job or set without running it @@ -815,7 +817,7 @@ Parameters: ### `run_on_target`: Run a program as a device would -Run an OpenQASM program as it would run on a device target: compiled exactly to its gates, placed on its most reliable qubits and routes (the expected success given), checked against the circuit, and simulated under the noise its calibration gives, on the density matrix or by trajectories. Counts and expectation values with their distance from the ideal distribution, the noise derived, and a receipt that binds the target's hash. A model of the device, not a run on it. Every angle must be a dyadic multiple of 2π. Waits up to 50 s, as run_quantum_circuit does. +Run an OpenQASM program as it would run on a device target: compiled exactly to its gates, placed on its most reliable qubits and routes (the expected success given), checked against the circuit, and simulated under the noise its calibration gives, on the density matrix or by trajectories. Counts and expectation values with their distance from the ideal distribution, the noise derived, and a receipt that binds the target's hash. A model of the device, not a run on it. Every angle must be a dyadic multiple of 2π. Waits up to 50 s, as run_quantum_circuit does, and like it, each call is a new job unless it gives idempotency_key. Parameters: - `qasm` (string): The program, OpenQASM 3 (or 2), read as the whole language: registers, aliases and slices; the standard gates, `U` and `gphase`; user `gate` definitions and `def` subroutines, inlined; every modifier (`ctrl @`, `negctrl @`, `inv @`, `pow(k) @`) on any gate, lowered exactly; classical types (int, uint, float, angle, complex, bool, bit, arrays), their arithmetic and built-in functions, computed on reading; `for`, `while`, `switch`, `break`, `continue`; measurement and `reset` anywhere; classical control on measured bits (`if (c[0]) x q[1];`, `if (c >= 2) { … } else { … }`, `&&`, `||`); a `while` on measured bits takes `@wai.max_iterations N` before it. `output` declarations come back in the result; `delay`, `box` and `stretch` are carried and labelled as not applied; `defcal` and `cal` are kept, not run. A program that measures in the middle or acts on what it measured runs on statevector or stabilizer: counts split their shots at each measurement; expectation values are exact, of the state before the final measurements. What the reader refuses comes back with a code, as `[Q103]`; qasm_diagnostics says what each code means. @@ -834,6 +836,7 @@ Parameters: - `mitigate` (any): Mitigate the job's expectation values (or, for readout, its counts) under its own noise: zne (the whole program folded U(U†U)^k and extrapolated), pea (the noise learned as a sparse Pauli–Lindblad model and amplified), pec (the learned channels inverted and sampled), cdr (a line fitted over near-Clifford training circuits), vd (two copies), readout (the counts' flips inverted). Needs noise and observables (shots for readout), at most 8 qubits. The result's `mitigation` gives raw and mitigated values with standard errors, the ideal value (known because this is a simulation), the method's bias and variance assumptions and its cost. A method's name, or an object. - `pulse` (object): A pulse run's options (method "pulse"): the sample period, the levels kept, the tolerance on the stated error, the steps a sample is divided into, each qubit's parameters, static couplings, and ports beyond the names d0, m0, a0. - `source` (object): The program in another format, in place of qasm: read to the job's qasm before the job is read (a module that does not read is refused with the reader's code, Q700 to Q721), the request keeping its source. +- `idempotency_key` (string): A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job. ### `quantum_circuit_examples`: Ready-to-run quantum examples diff --git a/wai/wai-quantum-cli/docs/service-mcp.json b/wai/wai-quantum-cli/docs/service-mcp.json index bde046afe..83d3690ff 100644 --- a/wai/wai-quantum-cli/docs/service-mcp.json +++ b/wai/wai-quantum-cli/docs/service-mcp.json @@ -4,7 +4,7 @@ { "name": "run_quantum_circuit", "title": "Run a quantum circuit", - "description": "Run an OpenQASM program and return measurement counts, Pauli expectation values and/or amplitudes. The service chooses the method (bit-exact or double-precision state vector up to 32 qubits, single precision on a GPU to 33, stabilizer tableau for Clifford circuits up to 32768 qubits, matrix product states up to 4096, sparse Pauli dynamics for expectation values up to 1024 with an error bound, an exact density matrix or trajectories for noise), reports why and how accurate, and signs the result. Give `intent` (an exact distribution, or a reference program it must equal or whose state it must prepare) and the engine checks it with the run. Every answer carries `verdict`: the level the engine checked to (proof, evidence, executed, parsed, refused), whether the intent holds and a plain statement, sealed in the receipt. Waits up to 50 s; a longer job answers with status running and an id for get_quantum_job.", + "description": "Run an OpenQASM program and return measurement counts, Pauli expectation values and/or amplitudes. The service chooses the method (bit-exact or double-precision state vector up to 32 qubits, single precision on a GPU to 33, stabilizer tableau for Clifford circuits up to 32768 qubits, matrix product states up to 4096, sparse Pauli dynamics for expectation values up to 1024 with an error bound, an exact density matrix or trajectories for noise), reports why and how accurate, and signs the result. Give `intent` (an exact distribution, or a reference program it must equal or whose state it must prepare) and the engine checks it with the run. Every answer carries `verdict`: the level the engine checked to (proof, evidence, executed, parsed, refused), whether the intent holds and a plain statement, sealed in the receipt. Waits up to 50 s; a longer job answers with status running and an id for get_quantum_job. Each call is a new job unless it gives idempotency_key, which makes sending it again safe.", "inputSchema": { "type": "object", "anyOf": [ @@ -529,12 +529,19 @@ "type": "number" }, "description": "A value for each of the program's parameters (input float theta;, input angle phi;), as {\"theta\": 0.3}: bound exactly before the program is read; a parameter left without one is refused (J050). In a sweep, the parameters it fixes." + }, + "idempotency_key": { + "type": "string", + "minLength": 1, + "maxLength": 255, + "pattern": "^[!#-\\[\\]-~]+$", + "description": "A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job." } } }, "annotations": { "readOnlyHint": false, - "idempotentHint": true, + "idempotentHint": false, "openWorldHint": false }, "outputSchema": { @@ -9822,7 +9829,7 @@ { "name": "run_quantum_set", "title": "Run a set of quantum programs, or one over a grid", - "description": "Run several OpenQASM programs as one set (`programs`: each a program's text or a job), or one program over a grid of its inputs (`qasm` with `sweep`, as {\"theta\": [0.1, 0.2], \"n\": [2, 3]}: every combination; integer inputs written out, float and angle inputs bound to the double given). The set's other fields (shots, observables, seed, intent, …) apply to every item. Each item gets its own result and receipt, each stating its verdict, and one receipt over the set binds every item through a Merkle root; verify_quantum_receipt checks it whole. At most 100 items, run one after another in one job. Waits up to 50 s; a longer set answers with status running and an id for get_quantum_job.", + "description": "Run several OpenQASM programs as one set (`programs`: each a program's text or a job), or one program over a grid of its inputs (`qasm` with `sweep`, as {\"theta\": [0.1, 0.2], \"n\": [2, 3]}: every combination; integer inputs written out, float and angle inputs bound to the double given). The set's other fields (shots, observables, seed, intent, …) apply to every item. Each item gets its own result and receipt, each stating its verdict, and one receipt over the set binds every item through a Merkle root; verify_quantum_receipt checks it whole. At most 100 items, run one after another in one job. Waits up to 50 s; a longer set answers with status running and an id for get_quantum_job. Each call is a new job unless it gives idempotency_key, which makes sending it again safe.", "inputSchema": { "type": "object", "properties": { @@ -10339,12 +10346,19 @@ } }, "description": "Each input of qasm and its values: every combination runs, the names in sorted order, the first varying slowest; an integer input's values are whole numbers." + }, + "idempotency_key": { + "type": "string", + "minLength": 1, + "maxLength": 255, + "pattern": "^[!#-\\[\\]-~]+$", + "description": "A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job." } } }, "annotations": { "readOnlyHint": false, - "idempotentHint": true, + "idempotentHint": false, "openWorldHint": false }, "outputSchema": { @@ -43066,7 +43080,7 @@ { "name": "run_on_target", "title": "Run a program as a device would", - "description": "Run an OpenQASM program as it would run on a device target: compiled exactly to its gates, placed on its most reliable qubits and routes (the expected success given), checked against the circuit, and simulated under the noise its calibration gives, on the density matrix or by trajectories. Counts and expectation values with their distance from the ideal distribution, the noise derived, and a receipt that binds the target's hash. A model of the device, not a run on it. Every angle must be a dyadic multiple of 2π. Waits up to 50 s, as run_quantum_circuit does.", + "description": "Run an OpenQASM program as it would run on a device target: compiled exactly to its gates, placed on its most reliable qubits and routes (the expected success given), checked against the circuit, and simulated under the noise its calibration gives, on the density matrix or by trajectories. Counts and expectation values with their distance from the ideal distribution, the noise derived, and a receipt that binds the target's hash. A model of the device, not a run on it. Every angle must be a dyadic multiple of 2π. Waits up to 50 s, as run_quantum_circuit does, and like it, each call is a new job unless it gives idempotency_key.", "inputSchema": { "type": "object", "required": [ @@ -43574,12 +43588,19 @@ "description": "The program's bytes as base64: jeff." } } + }, + "idempotency_key": { + "type": "string", + "minLength": 1, + "maxLength": 255, + "pattern": "^[!#-\\[\\]-~]+$", + "description": "A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job." } } }, "annotations": { "readOnlyHint": false, - "idempotentHint": true, + "idempotentHint": false, "openWorldHint": false }, "outputSchema": { diff --git a/wai/wai-quantum-cli/src/service.rs b/wai/wai-quantum-cli/src/service.rs index d76f24dcb..14e9ba94c 100644 --- a/wai/wai-quantum-cli/src/service.rs +++ b/wai/wai-quantum-cli/src/service.rs @@ -69,15 +69,38 @@ impl Client { v["error"].as_str().map(str::to_string).unwrap_or_else(|| format!("HTTP {code}: {v}")) } - /// Submits `request` (a job or a set): the job as admitted. + /// Submits `request` (a job or a set): the job as admitted. It goes under + /// an `Idempotency-Key` of its own. Where no answer comes back, it is sent + /// again under that key, twice at most, if the service says it keeps keys: + /// a job the service took is then answered, not taken again. pub fn submit(&self, request: &Value) -> Result { - let (code, v) = self.post("/v1/jobs", request)?; + let key = fresh_key(); + let url = format!("{}/v1/jobs", self.base); + let mut sent = 0; + let (code, v) = loop { + sent += 1; + match Self::read(self.auth(self.agent.post(&url)).header("idempotency-key", &key).send_json(request), &url) { + Ok(r) => break r, + Err(e) if sent < 3 && self.keeps_keys() => { + eprintln!("{e}; sending the job again under its key"); + std::thread::sleep(Duration::from_secs(sent)); + } + Err(e) if sent > 1 => return Err(format!("{e} (sent {sent} times under one key)")), + Err(e) => return Err(format!("{e}; the service may have taken the job, and it was not sent again, since the service does not say it keeps keys")), + } + }; if code >= 400 { return Err(Self::refused(code, &v)); } Ok(v) } + /// Whether the service keeps the key a job is sent under, so that the job + /// sent again is answered rather than taken twice. + fn keeps_keys(&self) -> bool { + self.get("/v1").is_ok_and(|(code, v)| code < 400 && v["idempotency"]["on"].as_array().is_some_and(|on| on.iter().any(|e| e == "POST /v1/jobs"))) + } + /// Whether the service answers dry runs and sets: one that keeps a job /// store says so in its description. An older one would run a dry run's /// job, so it is never sent one. @@ -189,3 +212,90 @@ impl Client { None } } + +/// A key for one submission: 128 random bits, in hex. +fn fresh_key() -> String { + let mut b = [0u8; 16]; + getrandom::fill(&mut b).expect("the operating system's random source"); + b.iter().map(|x| format!("{x:02x}")).collect() +} + +#[cfg(test)] +mod tests { + use super::*; + use std::io::{Read, Write}; + use std::net::{TcpListener, TcpStream}; + + /// One request read off `s`, body and all: its first line and its + /// headers, their names in lower case. + fn request(s: &mut TcpStream) -> (String, Vec<(String, String)>) { + let (mut head, mut byte) = (Vec::new(), [0u8; 1]); + while !head.ends_with(b"\r\n\r\n") && s.read(&mut byte).unwrap() == 1 { + head.push(byte[0]); + } + let text = String::from_utf8_lossy(&head).to_string(); + let mut lines = text.split("\r\n"); + let first = lines.next().unwrap_or("").to_string(); + let headers: Vec<(String, String)> = lines.filter_map(|l| l.split_once(':')).map(|(k, v)| (k.trim().to_ascii_lowercase(), v.trim().to_string())).collect(); + let n = headers.iter().find(|(k, _)| k == "content-length").and_then(|(_, v)| v.parse().ok()).unwrap_or(0); + let mut body = vec![0u8; n]; + s.read_exact(&mut body).unwrap(); + (first, headers) + } + + fn header(headers: &[(String, String)], name: &str) -> String { + headers.iter().find(|(k, _)| k == name).map(|(_, v)| v.clone()).unwrap_or_default() + } + + fn answer(s: &mut TcpStream, code: u16, body: &Value) { + let b = body.to_string(); + write!(s, "HTTP/1.1 {code} X\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{b}", b.len()).unwrap(); + } + + /// A service that takes the job and drops the connection before its + /// answer, then describes itself with `description`, then answers. + /// The keys each job came with. + fn lossy(description: Value, connections: usize) -> (String, std::thread::JoinHandle>) { + let l = TcpListener::bind("127.0.0.1:0").unwrap(); + let base = format!("http://{}", l.local_addr().unwrap()); + let server = std::thread::spawn(move || { + let mut keys = Vec::new(); + for (n, s) in l.incoming().take(connections).enumerate() { + let mut s = s.unwrap(); + let (line, headers) = request(&mut s); + if line.starts_with("GET /v1 ") { + answer(&mut s, 200, &description); + continue; + } + assert!(line.starts_with("POST /v1/jobs "), "{line}"); + keys.push(header(&headers, "idempotency-key")); + if n > 0 { + answer(&mut s, 202, &json!({ "id": "j1", "status": "queued" })); + } + } + keys + }); + (base, server) + } + + #[test] + fn a_job_whose_answer_is_lost_is_sent_again_under_its_key() { + let (base, server) = lossy(json!({ "idempotency": { "on": ["POST /v1/jobs", "POST /v1/run"] } }), 3); + let job = Client::new(&base, None).submit(&json!({ "qasm": "OPENQASM 3.0; qubit q;", "shots": 1 })).unwrap(); + assert_eq!(job["id"], "j1"); + let keys = server.join().unwrap(); + assert_eq!(keys.len(), 2, "{keys:?}"); + assert_eq!(keys[0], keys[1], "sent again under the same key"); + assert!(keys[0].len() == 32 && keys[0].bytes().all(|b| b.is_ascii_hexdigit()), "{keys:?}"); + // Each submission has its own key. + assert_ne!(fresh_key(), fresh_key()); + } + + #[test] + fn a_job_is_not_sent_again_to_a_service_that_keeps_no_keys() { + let (base, server) = lossy(json!({ "service": "wai-quantum-service" }), 2); + let e = Client::new(&base, None).submit(&json!({ "qasm": "OPENQASM 3.0; qubit q;", "shots": 1 })).unwrap_err(); + assert!(e.contains("may have taken the job"), "{e}"); + assert_eq!(server.join().unwrap().len(), 1, "sent once"); + } +} diff --git a/wai/wai-quantum-py/python/wai_quantum/service.py b/wai/wai-quantum-py/python/wai_quantum/service.py index c0e13b553..72e1cc139 100644 --- a/wai/wai-quantum-py/python/wai_quantum/service.py +++ b/wai/wai-quantum-py/python/wai_quantum/service.py @@ -10,6 +10,10 @@ Routes (``/v1/openapi.json`` describes them): ``GET /v1`` the service, ``GET /v1/methods``, ``POST /v1/jobs`` submit, ``GET /v1/jobs/{id}`` a job's status and result, ``DELETE /v1/jobs/{id}`` cancel, ``POST /v1/run`` submit and wait. + +A job goes under an ``Idempotency-Key``. Where no answer comes back, the +client sends it again under that key, twice at most, if the service says it +keeps keys: a job the service took is then answered, not taken again. """ from __future__ import annotations @@ -20,6 +24,7 @@ import time import urllib.error import urllib.parse import urllib.request +import uuid from typing import Any, Dict, Mapping, Optional, Tuple, Union from . import _engine @@ -37,7 +42,7 @@ DEFAULT_SERVICE = "https://wai.transaction.science/quantum-jobs" #: and its version. USER_AGENT = f"wai-quantum-python/{_engine.__version__}" -_FINISHED = ("done", "failed", "cancelled") +_FINISHED = ("done", "failed", "cancelled", "interrupted") class Service: @@ -61,12 +66,13 @@ class Service: # -- HTTP --------------------------------------------------------------- - def _call(self, method: str, path: str, body: Any = None) -> Tuple[int, Any]: + def _call(self, method: str, path: str, body: Any = None, extra: Optional[Mapping[str, str]] = None) -> Tuple[int, Any]: url = f"{self.base}/v1{path}" data = None if body is None else json.dumps(body, allow_nan=False).encode() headers = {"User-Agent": self.user_agent, "Accept": "application/json"} if data is not None: headers["Content-Type"] = "application/json" + headers.update(extra or {}) req = urllib.request.Request(url, data=data, method=method, headers=headers) try: with urllib.request.urlopen(req, timeout=self.timeout) as r: @@ -102,11 +108,16 @@ class Service: # -- jobs ------------------------------------------------------------- - def submit(self, program: Union[Program, Mapping[str, Any]], **job: Any) -> Dict[str, Any]: + def submit(self, program: Union[Program, Mapping[str, Any]], *, idempotency_key: Optional[str] = None, **job: Any) -> Dict[str, Any]: """Submit a job and return at once: its record, with its ``id`` and - ``status`` (``queued`` or ``running``). The keywords are - :func:`wai_quantum.job_request`'s.""" - return self._call("POST", "/jobs", job_request(program, **job))[1] + ``status`` (``queued`` or ``running``). The other keywords are + :func:`wai_quantum.job_request`'s. + + ``idempotency_key`` names the submission (1 to 255 visible ASCII + characters, no quote or backslash; make it unguessable): the same + key with the same job is answered with the same job, and with + another job refused (``J034``). Without one, a fresh key is made.""" + return self._post("/jobs", job_request(program, **job), idempotency_key) def status(self, job_id: str) -> Dict[str, Any]: """A job's record as it stands: its status, its plan, the request as @@ -124,10 +135,32 @@ class Service: service's reason; one still running after ``wait`` raises too.""" return self._finish(self.status(job_id), wait, poll) - def run(self, program: Union[Program, Mapping[str, Any]], *, wait: float = 600.0, poll: float = 1.0, **job: Any) -> Job: + def run(self, program: Union[Program, Mapping[str, Any]], *, wait: float = 600.0, poll: float = 1.0, idempotency_key: Optional[str] = None, **job: Any) -> Job: """Run a job on the service and wait for its result (``POST /v1/run``, - then the job's record while it is still running).""" - return self._finish(self._call("POST", "/run", job_request(program, **job))[1], wait, poll) + then the job's record while it is still running). ``idempotency_key`` + is :meth:`submit`'s.""" + return self._finish(self._post("/run", job_request(program, **job), idempotency_key), wait, poll) + + def _post(self, path: str, request: Dict[str, Any], key: Optional[str]) -> Dict[str, Any]: + """A job sent under its key, and sent again under it where no answer + came back and the service keeps keys.""" + key = key or uuid.uuid4().hex + sent = 0 + while True: + sent += 1 + try: + return self._call("POST", path, request, {"Idempotency-Key": key})[1] + except ServiceError as e: + if e.status is not None or sent == 3 or not self._keeps_keys(): + raise + time.sleep(sent) + + def _keeps_keys(self) -> bool: + """Whether the service keeps the key a job is sent under.""" + try: + return "POST /v1/jobs" in ((self.describe() or {}).get("idempotency") or {}).get("on", []) + except ServiceError: + return False def _finish(self, record: Dict[str, Any], wait: float, poll: float) -> Job: deadline = time.monotonic() + wait diff --git a/wai/wai-quantum-py/tests/test_service.py b/wai/wai-quantum-py/tests/test_service.py index 37a986825..8afd9222e 100644 --- a/wai/wai-quantum-py/tests/test_service.py +++ b/wai/wai-quantum-py/tests/test_service.py @@ -17,6 +17,9 @@ from conftest import BELL, bare DEMO = wq.Key.demo() +#: A stub's answer that is none: the connection closes once the request is read. +DROP = (None, None) + # -- a stub, for what the client sends ------------------------------------------ @@ -38,6 +41,8 @@ class Stub: body = self.rfile.read(n) if n else b"" stub.seen.append({"method": self.command, "path": self.path, "headers": dict(self.headers), "body": json.loads(body) if body else None}) status, answer = stub.answers.pop(0) if stub.answers else (500, {"error": "the stub ran out of answers"}) + if status is None: + return raw = answer if isinstance(answer, bytes) else json.dumps(answer).encode() self.send_response(status) self.send_header("content-type", "application/json") @@ -115,6 +120,32 @@ def test_a_failed_or_slow_job_raises(stub): svc.result("k", wait=0.05, poll=0.1) +def test_a_job_whose_answer_is_lost_is_sent_again_under_its_key(stub): + s = stub([DROP, (200, {"idempotency": {"on": ["POST /v1/jobs", "POST /v1/run"]}}), (202, {"id": "j1", "status": "queued"})]) + assert wq.Service(s.url, timeout=10).submit(BELL, shots=1)["id"] == "j1" + assert [(r["method"], r["path"]) for r in s.seen] == [("POST", "/quantum-jobs/v1/jobs"), ("GET", "/quantum-jobs/v1"), ("POST", "/quantum-jobs/v1/jobs")] + first, again = s.seen[0]["headers"]["Idempotency-Key"], s.seen[2]["headers"]["Idempotency-Key"] + assert first == again and len(first) == 32 + # A key given is the key sent. + s = stub([(202, {"id": "j2", "status": "queued"})]) + wq.Service(s.url).submit(BELL, shots=1, idempotency_key="mine-1") + assert s.seen[0]["headers"]["Idempotency-Key"] == "mine-1" and s.seen[0]["body"] == {"qasm": BELL, "shots": 1} + + +def test_a_job_is_not_sent_again_to_a_service_that_keeps_no_keys(stub): + s = stub([DROP, (200, {"service": "wai-quantum-service"})]) + with pytest.raises(wq.ServiceError) as e: + wq.Service(s.url, timeout=10).submit(BELL, shots=1) + assert e.value.status is None and [r["method"] for r in s.seen] == ["POST", "GET"] + + +def test_an_interrupted_job_ends_the_wait(stub): + s = stub([(200, {"id": "j", "status": "interrupted", "error": "interrupted: the service stopped while this job ran"})]) + with pytest.raises(wq.ServiceError, match="interrupted"): + wq.Service(s.url).result("j", wait=30, poll=0.01) + assert len(s.seen) == 1 + + def test_an_unreachable_service_raises_without_a_status(): with pytest.raises(wq.ServiceError) as e: wq.Service("http://127.0.0.1:9", timeout=2).describe() @@ -187,6 +218,21 @@ def test_cancel_a_queued_job_and_a_running_one(service_url): assert e.value.status == 409 +def test_a_job_sent_again_under_its_key_is_the_same_job(service_url): + svc = wq.Service(service_url) + a = svc.submit(BELL, shots=10, idempotency_key="py-key-1") + b = svc.submit(BELL, shots=10, idempotency_key="py-key-1") + assert a["id"] == b["id"] + assert svc.run(BELL, shots=10, idempotency_key="py-key-1", wait=120, poll=0.1).status == "done" + assert svc.submit(BELL, shots=10)["id"] != a["id"], "a fresh key is a new job" + with pytest.raises(wq.ServiceError) as e: + svc.submit(BELL, shots=11, idempotency_key="py-key-1") + assert e.value.status == 422 and e.value.message.endswith("[J034]") + with pytest.raises(wq.ServiceError) as e: + svc.submit(BELL, shots=10, idempotency_key="has space") + assert e.value.status == 400 and e.value.message.endswith("[J034]") + + def test_the_service_refuses_what_the_engine_refuses_with_its_words(service_url): svc = wq.Service(service_url) with pytest.raises(wq.ServiceError) as e: diff --git a/wai/wai-quantum-service/README.md b/wai/wai-quantum-service/README.md index 691ee365c..753cb3a4e 100644 --- a/wai/wai-quantum-service/README.md +++ b/wai/wai-quantum-service/README.md @@ -175,7 +175,7 @@ The three error-correction calls and the five of the variational and algorithm l A job may give a `target`: a built-in target's name, or a target document (`wai.quantum.target/1`, its schema in `GET /v1/openapi.json`). The program is compiled to the target's gates, placed on the qubits and routes most likely to succeed, and simulated under the noise its calibration gives. The result's `target` names the layout, its expected success, the noise and the target's hash, inside the receipt. The service reads the document it is given and fetches nothing. -Every refusal says why and ends with its code, which the body also carries as `code` ([the job layer's codes](../wai-quantum/docs/job-refusals.md), beside the reader's): `400` for a program or job that does not read (a set that does not read: `J007`), `422` for one larger than a job may be (`J031`), `429` past the per-client limits (`J030`), `503` when the queue is full (`J032`), and on a private deployment `401` without one of its API keys (`J033`). `GET /v1` states the limits. +Every refusal says why and ends with its code, which the body also carries as `code` ([the job layer's codes](../wai-quantum/docs/job-refusals.md), beside the reader's): `400` for a program or job that does not read (a set that does not read: `J007`), `422` for one larger than a job may be (`J031`), `429` past the per-client limits (`J030`), `503` when the queue is full (`J032`), on a private deployment `401` without one of its API keys (`J033`), and an `Idempotency-Key` that does not read (`400`) or was sent before with another request (`422`): `J034`. `GET /v1` states the limits. ## Status and limits @@ -183,6 +183,17 @@ A job is `queued`, `running`, `done`, `failed`, `cancelled`, or `interrupted`: t Every answer carries the client's submission window in the rate-limit fields of the IETF HTTP API working group: `RateLimit-Policy: "submissions";q=30;w=600` (at most 30 submissions in 600 s) and `RateLimit: "submissions";r=12;t=34` (12 left, one more in 34 s). A refusal past the limits carries `Retry-After`. +## Sending a job again + +A connection can fail after the service took a job and before its answer came back. To send the job again safely, give `POST /v1/jobs` or `POST /v1/run` an `Idempotency-Key` (the same working group's field): 1 to 255 visible ASCII characters, with no quote or backslash, as a structured field's string or bare. Make it unguessable, a UUID or 128 random bits, since it is kept per client and a client is an address. + +- Sent again with the same key and the same request, the job is answered with the job the key names, before any limit, and counts as no submission. +- The same key with another request is refused, `422` and `J034`. +- A key names its job for as long as the job is kept, whatever its status, with one exception: a job the service interrupted. That job's key runs the request again. +- The record keeps a digest of the key with the client's name, never the key. + +The MCP tools that run (`run_quantum_circuit`, `run_quantum_set`, `run_on_target`) take the same key as `idempotency_key`. Without one, each call is a new job, which is why their `idempotentHint` is false. The command line (`wai-quantum submit`) and the Python client send each job under a key of their own. Where no answer comes back, they send it again under that key, twice at most, and only to a service whose `GET /v1` says it keeps keys (`idempotency`). + ## Kept across restarts Every job is in the service's store before its submission is answered, and again at each change of its status. Programs, requests, results and receipts are kept under the BLAKE3 digests of their bytes; a job's `digests` names them. When the service starts, it reloads the store: diff --git a/wai/wai-quantum-service/src/discovery.rs b/wai/wai-quantum-service/src/discovery.rs index f275a4112..165d3295c 100644 --- a/wai/wai-quantum-service/src/discovery.rs +++ b/wai/wai-quantum-service/src/discovery.rs @@ -230,6 +230,7 @@ pub fn openapi(server: &str) -> Value { "components": { "schemas": { "JobRequest": job_request, "Job": job, "Error": err, "CompileOptions": compile_options, "Target": wai_quantum::engine::targets::schema(), "EstimateRequest": estimate_request, "DryRun": dry_run, "SetResult": set_result }, "headers": rate, + "parameters": { "IdempotencyKey": { "name": "Idempotency-Key", "in": "header", "required": false, "schema": { "type": "string", "minLength": 1, "maxLength": 255 }, "description": "A key of the client's making for this submission, 1 to 255 visible ASCII characters with no quote or backslash, as a structured field's string or bare; make it unguessable (a UUID, or 128 random bits). Sent again with the same key and the same request, the job is answered with the key's job, before any limit, and counts as no submission. The same key with another request is refused (422, J034). A key names its job while the job is kept, whatever its status, but for a job the service interrupted, which the key runs again." } }, "securitySchemes": { "apiKey": { "type": "http", "scheme": "bearer", "description": "Only on a private deployment that turns API keys on (WAI_QSVC_API_KEYS); the public service takes no key." } } }, "paths": { @@ -240,12 +241,13 @@ pub fn openapi(server: &str) -> Value { "/.well-known/wai-quantum-receipt-keys": { "get": { "summary": "Every key receipts are signed with: each with its window (not_before_ms, not_after_ms), the current one marked, and a retired one's transition to its successor signed by both", "responses": { "200": { "description": "The key set (wai.quantum.receipt-keys/1)" } } } }, "/v1/run": { "post": { "summary": "Submit a job or a set and wait up to 25 s for it (with dry_run, the plan)", + "parameters": [{ "$ref": "#/components/parameters/IdempotencyKey" }], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/JobRequest" } } } }, "responses": { "200": { "description": "The finished job, or with dry_run the plan (DryRun)", "headers": { "RateLimit": { "$ref": "#/components/headers/RateLimit" }, "RateLimit-Policy": { "$ref": "#/components/headers/RateLimit-Policy" } }, "content": { "application/json": { "schema": { "oneOf": [{ "$ref": "#/components/schemas/Job" }, { "$ref": "#/components/schemas/DryRun" }] } } } }, "202": { "description": "Still running: poll GET /v1/jobs/{id}", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Job" } } } }, - "400": { "description": "A program or job that does not read", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Error" } } } }, - "422": { "description": "Larger than one job may be" }, + "400": { "description": "A program, job or Idempotency-Key that does not read", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Error" } } } }, + "422": { "description": "Larger than one job may be, or an Idempotency-Key sent before with another request (J034)" }, "401": { "description": "A private deployment's key missing or not one it holds (J033)" }, "429": { "description": "Past the per-client limits (with Retry-After)" }, "503": { "description": "The queue is full" } @@ -253,10 +255,16 @@ pub fn openapi(server: &str) -> Value { } }, "/v1/jobs": { "post": { "summary": "Submit a job or a set (with dry_run, the plan)", + "parameters": [{ "$ref": "#/components/parameters/IdempotencyKey" }], "requestBody": { "required": true, "content": { "application/json": { "schema": { "$ref": "#/components/schemas/JobRequest" } } } }, "responses": { - "202": { "description": "Accepted, and kept in the service's store before this answer", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Job" } } } }, - "200": { "description": "With dry_run: the plan, nothing run", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/DryRun" } } } } + "202": { "description": "Accepted, and kept in the service's store before this answer; sent again with its Idempotency-Key, the key's job", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Job" } } } }, + "200": { "description": "With dry_run: the plan, nothing run", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/DryRun" } } } }, + "400": { "description": "A program, job or Idempotency-Key that does not read", "content": { "application/json": { "schema": { "$ref": "#/components/schemas/Error" } } } }, + "422": { "description": "Larger than one job may be, or an Idempotency-Key sent before with another request (J034)" }, + "401": { "description": "A private deployment's key missing or not one it holds (J033)" }, + "429": { "description": "Past the per-client limits (with Retry-After)" }, + "503": { "description": "The queue is full" } } } }, "/v1/jobs/{id}": { diff --git a/wai/wai-quantum-service/src/http.rs b/wai/wai-quantum-service/src/http.rs index bc56b252e..65165e0eb 100644 --- a/wai/wai-quantum-service/src/http.rs +++ b/wai/wai-quantum-service/src/http.rs @@ -66,6 +66,12 @@ //! group's fields), and a refusal past it `Retry-After`. Where the service //! takes API keys (`auth`), every request but the open ones carries one. //! +//! `POST /v1/jobs` and `POST /v1/run` take an `Idempotency-Key` (the same +//! working group's field): a job sent again with its key is answered with +//! the key's job, before any limit, and counts as no submission, so a client +//! that lost an answer can send the job again safely. The same key with +//! another request is refused, 422 (`J034`). See [`crate::jobs`]. +//! //! A job is `{"qasm": "", "method": "auto", "shots": N, //! "seed": S, "observables": ["Z0 Z1", ...], "options": {...}}`; see //! [`crate::catalog`]. A set gives `"programs"` (a list of programs or jobs) @@ -80,7 +86,7 @@ //! code (`engine::refusal`). use crate::catalog; -use crate::jobs::{Refused, Service, Work}; +use crate::jobs::{Key, Refused, Service, Work, check_key}; use crate::receipt::{self, Signer}; use axum::extract::{ConnectInfo, DefaultBodyLimit, Path, Query, State}; use axum::http::{HeaderMap, HeaderValue, Method, StatusCode}; @@ -220,6 +226,8 @@ async fn describe(State(app): State) -> Response { "receipt_key": app.signer.public_key(), "receipt_keys": format!("{}/.well-known/wai-quantum-receipt-keys", app.base), "api_keys": app.keys.is_some(), + // A client that lost an answer sends the job again under its key only where this says the key is kept. + "idempotency": { "header": "Idempotency-Key", "argument": "idempotency_key", "key_bytes": crate::jobs::KEY_MAX, "on": ["POST /v1/jobs", "POST /v1/run", "run_quantum_circuit", "run_quantum_set", "run_on_target"], "kept": "while its job is kept, whatever its status, but for a job the service interrupted, which the key runs again" }, "endpoints": ["GET /v1", "GET /v1/health", "GET /v1/methods", "GET /v1/examples", "GET /v1/targets", "GET /v1/openapi.json", "GET /.well-known/wai-quantum-receipt-keys", "POST /v1/jobs", "GET /v1/jobs/{id}", "GET /v1/jobs/{id}/events", "DELETE /v1/jobs/{id}", "POST /v1/run", "POST /v1/verify", "POST /v1/critique", "POST /v1/equivalent", "POST /v1/improve", "POST /v1/shortcuts", "POST /v1/compile", "POST /v1/estimate", "POST /v1/repair", "POST /v1/convert", "POST /v1/qec/code", "POST /v1/qec/decode", "POST /v1/qec/threshold", "POST /v1/gradient", "POST /v1/optimize", "POST /v1/vqe", "POST /v1/qaoa", "POST /v1/build", "POST /mcp"], "mcp": format!("{}/mcp", app.base), "console": app.console, @@ -280,11 +288,28 @@ pub(crate) fn plan_job(app: &App, request: &Value) -> Result<(catalog::Plan, Wor } /// Plans and admits `request` for `who`: the job's id, or the refusal's -/// status and reason. -pub(crate) fn admit(app: &App, who: &str, request: Value) -> Result { +/// status and reason. A request sent again with its key is the key's job, +/// whatever planning it now would give. +pub(crate) fn admit(app: &App, who: &str, request: Value, key: Option) -> Result { let request = with_source(request)?; + let refused = |Refused(code, why)| (code, why); + if let Some(k) = key + && let Some(id) = app.service.named(who, k, &request).map_err(refused)? + { + return Ok(id); + } let (plan, work) = plan_job(app, &request)?; - app.service.submit_work(who, request, plan, work).map_err(|Refused(code, why)| (code, why)) + app.service.submit_work(who, request, plan, work, key).map_err(refused) +} + +/// The `Idempotency-Key` a request gives, if it gives one: a structured +/// field's string (`"…"`), or the same characters bare ([`check_key`]). +fn idempotency_key(headers: &HeaderMap) -> Result, (u16, String)> { + let Some(v) = headers.get("idempotency-key") else { return Ok(None) }; + let text = v.to_str().map_err(|_| (400, tag(refusal::IDEMPOTENCY_KEY, "the Idempotency-Key is not visible ASCII")))?.trim(); + let key = text.strip_prefix('"').and_then(|t| t.strip_suffix('"')).unwrap_or(text); + check_key(key).map_err(|e| (400, e))?; + Ok(Some(key.to_string())) } /// A job that carries its program in another format (`"source"`), with its @@ -402,7 +427,11 @@ async fn submit(State(app): State, ConnectInfo(peer): ConnectInfo error(code, &why), }; } - match admit(&app, &who, request) { + let key = match idempotency_key(&headers) { + Ok(k) => k, + Err((code, why)) => return error(code, &why), + }; + match admit(&app, &who, request, key.as_deref().map(Key)) { Ok(id) => reply(202, app.service.get(&id).unwrap_or(json!({ "id": id }))), Err((code, why)) => error(code, &why), } @@ -420,7 +449,11 @@ async fn run(State(app): State, ConnectInfo(peer): ConnectInfo error(code, &why), }; } - let id = match admit(&app, &who, request) { + let key = match idempotency_key(&headers) { + Ok(k) => k, + Err((code, why)) => return error(code, &why), + }; + let id = match admit(&app, &who, request, key.as_deref().map(Key)) { Ok(id) => id, Err((code, why)) => return error(code, &why), }; @@ -753,7 +786,7 @@ async fn preflight() -> Response { let h = r.headers_mut(); h.insert("access-control-allow-origin", HeaderValue::from_static("*")); h.insert("access-control-allow-methods", HeaderValue::from_static("GET, POST, DELETE, OPTIONS")); - h.insert("access-control-allow-headers", HeaderValue::from_static("content-type, accept, authorization, mcp-protocol-version, mcp-session-id")); + h.insert("access-control-allow-headers", HeaderValue::from_static("content-type, accept, authorization, idempotency-key, mcp-protocol-version, mcp-session-id")); h.insert("access-control-max-age", HeaderValue::from_static("86400")); r } @@ -842,9 +875,26 @@ pub fn router(app: Shared) -> Router { #[cfg(test)] mod tests { - use super::identity; + use super::{identity, idempotency_key}; + use axum::http::{HeaderMap, HeaderValue}; use std::net::IpAddr; + #[test] + fn an_idempotency_key_reads_as_a_structured_string_or_bare() { + let read = |v: &str| { + let mut h = HeaderMap::new(); + h.insert("idempotency-key", HeaderValue::from_str(v).unwrap()); + idempotency_key(&h) + }; + assert_eq!(idempotency_key(&HeaderMap::new()), Ok(None)); + assert_eq!(read("\"8e03978e-40d5-43e8-bc93-6894a57f9324\""), Ok(Some("8e03978e-40d5-43e8-bc93-6894a57f9324".into()))); + assert_eq!(read("8e03978e40d543e8"), Ok(Some("8e03978e40d543e8".into()))); + for bad in ["\"\"", "\"two words\"", "a\"b", "\"open"] { + let (code, why) = read(bad).unwrap_err(); + assert!(code == 400 && why.ends_with("[J034]"), "{bad}: {why}"); + } + } + fn id(s: &str) -> String { identity(s.parse::().unwrap()) } diff --git a/wai/wai-quantum-service/src/jobs.rs b/wai/wai-quantum-service/src/jobs.rs index 87fab425c..f62456dc1 100644 --- a/wai/wai-quantum-service/src/jobs.rs +++ b/wai/wai-quantum-service/src/jobs.rs @@ -39,6 +39,14 @@ //! when the service stopped comes back `interrupted`, with the reason; a //! queued one is queued again, in its order. Finished jobs are kept for //! [`Config::retain`], then dropped with whatever no other job names. +//! - **Retries.** A submission may carry a key of the client's making +//! ([`Key`]): sent again with the same key and the same request, it is +//! answered with the job the key names, before any limit, and counts as +//! nothing, so a client that lost an answer can ask again without a second +//! job. The same key with another request is refused (`J034`). A key names +//! its job for as long as the job is kept, whatever its status, but for a +//! job the service interrupted: that key runs the request again. A record +//! keeps the key's digest with the client's name, never the key. use crate::catalog::Plan; use crate::energy; @@ -145,6 +153,30 @@ impl Status { pub fn active(self) -> bool { matches!(self, Status::Queued | Status::Running) } + + /// Whether a job in this status still answers for its key: every one but + /// a job the service interrupted, which never ran to its own end. + fn holds_key(self) -> bool { + self != Status::Interrupted + } +} + +/// What a client sends a submission again under: a key of its own making, +/// as `Idempotency-Key` or a tool's `idempotency_key` ([`check_key`]). +#[derive(Clone, Copy, Debug)] +pub struct Key<'a>(pub &'a str); + +/// The longest key, bytes. +pub const KEY_MAX: usize = 255; + +/// Whether `key` reads as a key: 1 to [`KEY_MAX`] visible ASCII characters, +/// with no quote or backslash, so it is the same as a structured field's +/// string and bare. +pub fn check_key(key: &str) -> Result<(), String> { + if key.is_empty() || key.len() > KEY_MAX || !key.bytes().all(|b| b.is_ascii_graphic() && b != b'"' && b != b'\\') { + return Err(tag(refusal::IDEMPOTENCY_KEY, format!("an idempotency key is 1 to {KEY_MAX} visible ASCII characters, with no quote or backslash, and should be unguessable (a UUID, or 128 random bits)"))); + } + Ok(()) } /// What a job runs: one program, or a set's items with their plans. @@ -180,6 +212,9 @@ pub struct Job { pub error: Option, pub peak_rss: u64, pub digests: Digests, + /// The digest of the key it was sent with, and its client's name + /// ([`Service::key_digest`]), if it was sent with one. + pub key: Option, child: Option, } @@ -263,6 +298,9 @@ impl Job { if self.peak_rss > 0 { r["peak_rss_bytes"] = json!(self.peak_rss); } + if let Some(k) = &self.key { + r["key"] = json!(k); + } if let Work::Set { plans, .. } = &self.work { r["set_plans"] = json!(plans); } @@ -316,6 +354,8 @@ pub struct Service { jobs: Mutex>, queue: Mutex>, clients: Mutex>, + /// The job each key's digest names. Taken before `jobs` or `queue`. + keys: Mutex>, wake: Notify, memory: Arc, workers: Arc, @@ -364,6 +404,7 @@ impl Service { jobs: Mutex::new(HashMap::new()), queue: Mutex::new(VecDeque::new()), clients: Mutex::new(HashMap::new()), + keys: Mutex::new(HashMap::new()), wake: Notify::new(), memory, workers, @@ -440,6 +481,11 @@ impl Service { } } } + if let Some(k) = &job.key + && job.status.holds_key() + { + s.keys.get_mut().unwrap().insert(k.clone(), id.clone()); + } s.jobs.get_mut().unwrap().insert(id, job); } queued.sort(); @@ -501,6 +547,7 @@ impl Service { result: s("result"), receipt: s("receipt"), }, + key: s("key"), child: None, }) } @@ -592,15 +639,55 @@ impl Service { None } + /// The digest a key is kept under: the client's name (as a record keeps + /// it) and the key, so one client's key never names another's job. + fn key_digest(name: &str, key: Key) -> String { + store::digest(format!("wai.quantum.job-key/1\0{name}\0{}", key.0).as_bytes()) + } + + /// The job `digest` names, if it still answers for it: its id when it + /// was sent with the request whose digest is `request`, or the refusal + /// of a key sent with another. + fn named_by(&self, keys: &HashMap, digest: &str, request: &str) -> Result, Refused> { + let Some(id) = keys.get(digest) else { return Ok(None) }; + let jobs = self.jobs.lock().unwrap(); + match jobs.get(id) { + Some(j) if j.status.holds_key() && j.digests.request.as_deref() == Some(request) => Ok(Some(id.clone())), + Some(j) if j.status.holds_key() => Err(Refused(422, tag(refusal::IDEMPOTENCY_KEY, "this idempotency key was sent with another request; a key names one job, so send this request with a new key"))), + _ => Ok(None), + } + } + + /// The job `key` names for `client` and `request`, if one does: what a + /// submission sent again is answered with, before it is planned. + pub fn named(&self, client: &str, key: Key, request: &Value) -> Result, Refused> { + let digest = Self::key_digest(&self.name(client), key); + let keys = self.keys.lock().unwrap(); + self.named_by(&keys, &digest, &store::digest(&canon::canonical(request))) + } + /// Admits a planned job for `client`, or says why not. Its request and /// programs, and its record, are in the store before it is answered. #[cfg(test)] pub fn submit(&self, client: &str, request: Value, plan: Plan) -> Result { - self.submit_work(client, request, plan, Work::One) + self.submit_work(client, request, plan, Work::One, None) } - /// [`Service::submit`] for one program or a set's items. - pub fn submit_work(&self, client: &str, request: Value, plan: Plan, work: Work) -> Result { + /// [`Service::submit`] for one program or a set's items, sent with a key + /// or without one: a request sent again with its key is answered with + /// the key's job, before any limit, and counts as nothing. + pub fn submit_work(&self, client: &str, request: Value, plan: Plan, work: Work, key: Option) -> Result { + let name = self.name(client); + let canonical = canon::canonical(&request); + let request_digest = store::digest(&canonical); + let key = key.map(|k| Self::key_digest(&name, k)); + // Held until the job is kept, so two sends of one key make one job. + let mut keys = key.as_ref().map(|_| self.keys.lock().unwrap()); + if let (Some(keys), Some(k)) = (&keys, &key) + && let Some(id) = self.named_by(keys, k, &request_digest)? + { + return Ok(id); + } // A full queue is checked again below, as the job is queued. if let Some(r) = self.would_refuse(client, &plan) && r.0 != 503 @@ -613,14 +700,14 @@ impl Service { Work::One => request.get("qasm").and_then(Value::as_str).into_iter().collect(), Work::Set { items, .. } => items.iter().filter_map(|i| i.request.get("qasm").and_then(Value::as_str)).collect(), }; - let mut digests = Digests { request: Some(self.store.put(&canon::canonical(&request)).map_err(unkept)?), ..Digests::default() }; + let mut digests = Digests { request: Some(self.store.put(&canonical).map_err(unkept)?), ..Digests::default() }; for p in programs { digests.programs.push(self.store.put(p.as_bytes()).map_err(unkept)?); } let id = job_id(); let job = Job { id: id.clone(), - client: self.name(client), + client: name, request, plan, work, @@ -632,6 +719,7 @@ impl Service { error: None, peak_rss: 0, digests, + key: key.clone(), child: None, }; { @@ -643,6 +731,10 @@ impl Service { queue.push_back(id.clone()); } self.jobs.lock().unwrap().insert(id.clone(), job); + if let (Some(keys), Some(k)) = (keys.as_mut(), key) { + keys.insert(k, id.clone()); + } + drop(keys); self.changed(&id); self.wake.notify_one(); Ok(id) @@ -783,6 +875,7 @@ impl Service { let _ = self.store.remove_record(id); } if !gone.is_empty() { + self.keys.lock().unwrap().retain(|_, id| !gone.contains(id)); self.sweep(); } } @@ -1249,6 +1342,105 @@ mod tests { s.jobs.lock().unwrap().values().filter(|j| j.client == name).map(|j| j.id.clone()).collect() } + /// A job sent again under its key is the key's job, before any limit and + /// counting as nothing; the same key from another client is that + /// client's own; with another request it is refused; a job cancelled + /// still answers for its key, and one the service interrupted does not. + #[test] + fn a_job_sent_again_under_its_key_is_the_same_job_and_counts_as_nothing() { + let s = Service::new(Config { per_client_active: 1, per_client_burst: 3, ..config() }); + let req = json!({ "qasm": "OPENQASM 3.0; qubit q; h q;", "shots": 5 }); + let submit = |client: &str, r: &Value, key: &str| s.submit_work(client, r.clone(), plan(1 << 20), Work::One, Some(Key(key))); + let id = submit("a", &req, "k-1").unwrap(); + // At its one active job and with two submissions left, the client + // sends it twice more and has the same two left. + assert_eq!(submit("a", &req, "k-1").unwrap(), id); + assert_eq!(submit("a", &req, "k-1").unwrap(), id); + assert_eq!(s.window("a").remaining, 2); + assert_eq!(s.named("a", Key("k-1"), &req).unwrap(), Some(id.clone())); + assert_eq!(s.named("a", Key("k-2"), &req).unwrap(), None); + // Without a key it is a new job, held to the limits. + assert_eq!(s.submit("a", req.clone(), plan(1 << 20)).unwrap_err().0, 429); + // Another client's key of the same text names its own job. + let theirs = submit("b", &req, "k-1").unwrap(); + assert_ne!(theirs, id); + // The same key with another request is refused, and nothing is made. + let other = json!({ "qasm": "OPENQASM 3.0; qubit q; x q;", "shots": 5 }); + let Err(Refused(code, why)) = submit("a", &other, "k-1") else { panic!("a key named two jobs") }; + assert_eq!(code, 422); + // It names no job: the id is what reads and cancels one. + assert!(why.ends_with("[J034]") && !why.contains(&id), "{why}"); + assert_eq!(ids_of(&s, "a"), vec![id.clone()]); + // A cancelled job answers for its key. + assert_eq!(s.cancel(&id), Some(true)); + assert_eq!(submit("a", &req, "k-1").unwrap(), id); + // One the service interrupted does not: the key runs it again. + s.jobs.lock().unwrap().get_mut(&id).unwrap().status = Status::Interrupted; + let again = submit("a", &req, "k-1").unwrap(); + assert_ne!(again, id); + assert_eq!(submit("a", &req, "k-1").unwrap(), again); + // The record keeps the key's digest, never the key. + let record = s.jobs.lock().unwrap()[&again].record().to_string(); + assert!(record.contains(&Service::key_digest(&s.name("a"), Key("k-1"))) && !record.contains("k-1"), "{record}"); + } + + #[test] + fn a_key_reads_as_one_to_255_visible_ascii_characters() { + let (most, past) = ("x".repeat(KEY_MAX), "x".repeat(KEY_MAX + 1)); + for k in ["a", "0123456789abcdef0123456789abcdef", "550e8400-e29b-41d4-a716-446655440000", most.as_str(), "!#$%&'()*+,-./:;<=>?@[]^_`{|}~"] { + assert!(check_key(k).is_ok(), "{k}"); + } + for k in ["", past.as_str(), "has space", "tab\t", "quote\"", "back\\slash", "é"] { + assert!(check_key(k).unwrap_err().ends_with("[J034]"), "{k}"); + } + } + + /// Sent at once from many threads under one key, a job is one job. + #[test] + fn a_key_sent_at_once_from_many_threads_makes_one_job() { + let s = Service::new(Config { per_client_active: 8, per_client_burst: 64, max_queue: 64, ..config() }); + let req = json!({ "qasm": "OPENQASM 3.0; qubit q; h q;", "shots": 5 }); + let ids: Vec = std::thread::scope(|t| { + let handles: Vec<_> = (0..16).map(|_| t.spawn(|| s.submit_work("a", req.clone(), plan(1 << 20), Work::One, Some(Key("same"))).unwrap())).collect(); + handles.into_iter().map(|h| h.join().unwrap()).collect() + }); + assert!(ids.iter().all(|i| *i == ids[0]), "{ids:?}"); + assert_eq!(ids_of(&s, "a").len(), 1); + assert_eq!(s.queue.lock().unwrap().len(), 1); + assert_eq!(s.window("a").remaining, 63); + } + + /// A key outlives the service: reopened, the store's job still answers + /// for it, but for the job that was running, now interrupted. + #[test] + fn a_key_names_its_job_across_a_restart() { + let dir = store_dir("keys"); + let req = json!({ "qasm": "OPENQASM 3.0; qubit q; h q;", "shots": 5 }); + let run = json!({ "qasm": "OPENQASM 3.0; qubit q; x q;", "shots": 5 }); + let cfg = || Config { per_client_active: 4, per_client_burst: 10, ..config() }; + let (queued, running); + { + let s = Service::with_store(cfg(), Box::new(store::Files::open(&dir.join("store")).unwrap())); + queued = s.submit_work("a", req.clone(), plan(1 << 20), Work::One, Some(Key("q-key"))).unwrap(); + running = s.submit_work("a", run.clone(), plan(1 << 20), Work::One, Some(Key("r-key"))).unwrap(); + let mut jobs = s.jobs.lock().unwrap(); + let j = jobs.get_mut(&running).unwrap(); + j.status = Status::Running; + s.keep(j).unwrap(); + } + let raw = std::fs::read_dir(dir.join("store/jobs")).unwrap().flatten().map(|e| std::fs::read_to_string(e.path()).unwrap()).collect::(); + assert!(!raw.contains("q-key") && !raw.contains("r-key"), "a record keeps no key"); + let s = Service::open(cfg(), Box::new(store::Files::open(&dir.join("store")).unwrap())).unwrap(); + assert_eq!(s.named("a", Key("q-key"), &req).unwrap(), Some(queued.clone())); + assert_eq!(s.submit_work("a", req.clone(), plan(1 << 20), Work::One, Some(Key("q-key"))).unwrap(), queued); + assert_eq!(s.get(&running).unwrap()["status"], "interrupted"); + assert_eq!(s.named("a", Key("r-key"), &run).unwrap(), None); + let again = s.submit_work("a", run.clone(), plan(1 << 20), Work::One, Some(Key("r-key"))).unwrap(); + assert_ne!(again, running); + drop(s); + let _ = std::fs::remove_dir_all(&dir); + } + #[test] fn a_client_holds_at_most_one_jobs_worth_of_memory_at_once() { let s = Service::new(Config { per_client_active: 5, per_client_burst: 10, ..config() }); @@ -1338,7 +1530,7 @@ mod tests { let key = wai_quantum::ed25519_dalek::SigningKey::from_bytes(&[19; 32]); let request = json!({ "qasm": "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", "shots": 20, "observables": ["Z0 Z1"] }); let (p, mut result) = wai_quantum::engine::job::plan_and_run(&request, 1 << 30).unwrap(); - let mut job = Job { id: job_id(), client: "test".into(), request: request.clone(), plan: p, work: Work::One, status: Status::Queued, submitted_ms: now_ms(), started_ms: None, finished_ms: None, result: None, error: None, peak_rss: 0, digests: Digests::default(), child: None }; + let mut job = Job { id: job_id(), client: "test".into(), request: request.clone(), plan: p, work: Work::One, status: Status::Queued, submitted_ms: now_ms(), started_ms: None, finished_ms: None, result: None, error: None, peak_rss: 0, digests: Digests::default(), key: None, child: None }; let r = crate::mcp::schemas(); let check = |j: &Job| { let v = j.to_json(); @@ -1368,7 +1560,7 @@ mod tests { let items = wai_quantum::engine::sets::expand(&set_request).unwrap(); let plans = wai_quantum::engine::sets::plan(&items, 1 << 30).unwrap(); let body = wai_quantum::engine::sets::run(&set_request, 1 << 30, &key, json!({ "unmetered": "a test" })).unwrap(); - let set = Job { id: job_id(), client: "test".into(), request: set_request, plan: wai_quantum::engine::sets::summary(&plans), work: Work::Set { items, plans }, status: Status::Done, submitted_ms: now_ms(), started_ms: Some(now_ms()), finished_ms: Some(now_ms()), result: Some(body["result"].clone()), error: None, peak_rss: 1 << 20, digests: Digests::default(), child: None }; + let set = Job { id: job_id(), client: "test".into(), request: set_request, plan: wai_quantum::engine::sets::summary(&plans), work: Work::Set { items, plans }, status: Status::Done, submitted_ms: now_ms(), started_ms: Some(now_ms()), finished_ms: Some(now_ms()), result: Some(body["result"].clone()), error: None, peak_rss: 1 << 20, digests: Digests::default(), key: None, child: None }; let done = check(&set); assert_eq!(wai_quantum::engine::sets::verify(&done)["valid"], true, "{done}"); } @@ -1475,7 +1667,7 @@ mod tests { assert!(t["seconds"].is_null() && t["at_most"] == 5, "{t}"); // Timed jobs here take over, and their metered power gives an energy estimate. for _ in 0..3 { - let job = Job { id: job_id(), client: "x".into(), request: json!({}), plan: Plan { work: 1_000_000_000, ..p.clone() }, work: Work::One, status: Status::Done, submitted_ms: 0, started_ms: None, finished_ms: Some(1), result: Some(json!({ "elapsed_ms": 500, "energy": { "joules_micro": 5_000_000, "energy_uncertainty": { "window_us": 500_000 } } })), error: None, peak_rss: 0, digests: Digests::default(), child: None }; + let job = Job { id: job_id(), client: "x".into(), request: json!({}), plan: Plan { work: 1_000_000_000, ..p.clone() }, work: Work::One, status: Status::Done, submitted_ms: 0, started_ms: None, finished_ms: Some(1), result: Some(json!({ "elapsed_ms": 500, "energy": { "joules_micro": 5_000_000, "energy_uncertainty": { "window_us": 500_000 } } })), error: None, peak_rss: 0, digests: Digests::default(), key: None, child: None }; s.time(&job); } let (t, e) = s.estimate(&p); diff --git a/wai/wai-quantum-service/src/mcp.rs b/wai/wai-quantum-service/src/mcp.rs index 5f0fb49a9..16c93e0dc 100644 --- a/wai/wai-quantum-service/src/mcp.rs +++ b/wai/wai-quantum-service/src/mcp.rs @@ -216,7 +216,7 @@ fn tools() -> Value { } fn tool_list() -> Value { - let job_props = json!({ + let mut job_props = json!({ "qasm": { "type": "string", "description": "The program, OpenQASM 3 (or 2), read as the whole language: registers, aliases and slices; the standard gates, `U` and `gphase`; user `gate` definitions and `def` subroutines, inlined; every modifier (`ctrl @`, `negctrl @`, `inv @`, `pow(k) @`) on any gate, lowered exactly; classical types (int, uint, float, angle, complex, bool, bit, arrays), their arithmetic and built-in functions, computed on reading; `for`, `while`, `switch`, `break`, `continue`; measurement and `reset` anywhere; classical control on measured bits (`if (c[0]) x q[1];`, `if (c >= 2) { … } else { … }`, `&&`, `||`); a `while` on measured bits takes `@wai.max_iterations N` before it. `output` declarations come back in the result; `delay`, `box` and `stretch` are carried and labelled as not applied; `defcal` and `cal` are kept, not run. A program that measures in the middle or acts on what it measured runs on statevector or stabilizer: counts split their shots at each measurement; expectation values are exact, of the state before the final measurements. What the reader refuses comes back with a code, as `[Q103]`; qasm_diagnostics says what each code means." }, "shots": { "type": "integer", "minimum": 0, "maximum": 1000000, "description": "Measurement shots. Default 1024 when no observables or amplitudes are asked for." }, "observables": { "type": "array", "items": { "type": "string" }, "description": "Pauli strings to take expectation values of, e.g. [\"Z0\", \"Z0 Z1\", \"X2 Y3\"]. Qubits count from 0." }, @@ -248,20 +248,25 @@ fn tool_list() -> Value { let mut plan_props = set_props.clone(); plan_props["qasm"] = job_props["qasm"].clone(); plan_props["source"] = job_props["source"].clone(); + // A run's key, which a plan does not take. + let key = json!({ "type": "string", "minLength": 1, "maxLength": crate::jobs::KEY_MAX, "pattern": "^[!#-\\[\\]-~]+$", "description": "A key of your making that names this run, so that a call sent again is not a second job: the same key with the same arguments answers with the job it names, for as long as the job is kept, rather than running another; the same key with other arguments is refused (J034). Make it unguessable (a UUID). A job the service interrupted is run again under its key. Without a key, every call is a new job." }); + for props in [&mut job_props, &mut set_props, &mut on_target["properties"]] { + props["idempotency_key"] = key.clone(); + } let mut list = json!([ { "name": "run_quantum_circuit", "title": "Run a quantum circuit", - "description": "Run an OpenQASM program and return measurement counts, Pauli expectation values and/or amplitudes. The service chooses the method (bit-exact or double-precision state vector up to 32 qubits, single precision on a GPU to 33, stabilizer tableau for Clifford circuits up to 32768 qubits, matrix product states up to 4096, sparse Pauli dynamics for expectation values up to 1024 with an error bound, an exact density matrix or trajectories for noise), reports why and how accurate, and signs the result. Give `intent` (an exact distribution, or a reference program it must equal or whose state it must prepare) and the engine checks it with the run. Every answer carries `verdict`: the level the engine checked to (proof, evidence, executed, parsed, refused), whether the intent holds and a plain statement, sealed in the receipt. Waits up to 50 s; a longer job answers with status running and an id for get_quantum_job.", + "description": "Run an OpenQASM program and return measurement counts, Pauli expectation values and/or amplitudes. The service chooses the method (bit-exact or double-precision state vector up to 32 qubits, single precision on a GPU to 33, stabilizer tableau for Clifford circuits up to 32768 qubits, matrix product states up to 4096, sparse Pauli dynamics for expectation values up to 1024 with an error bound, an exact density matrix or trajectories for noise), reports why and how accurate, and signs the result. Give `intent` (an exact distribution, or a reference program it must equal or whose state it must prepare) and the engine checks it with the run. Every answer carries `verdict`: the level the engine checked to (proof, evidence, executed, parsed, refused), whether the intent holds and a plain statement, sealed in the receipt. Waits up to 50 s; a longer job answers with status running and an id for get_quantum_job. Each call is a new job unless it gives idempotency_key, which makes sending it again safe.", "inputSchema": { "type": "object", "anyOf": wai_quantum::engine::formats::program_required(), "properties": job_props }, - "annotations": { "readOnlyHint": false, "idempotentHint": true, "openWorldHint": false } + "annotations": { "readOnlyHint": false, "idempotentHint": false, "openWorldHint": false } }, { "name": "run_quantum_set", "title": "Run a set of quantum programs, or one over a grid", - "description": format!("Run several OpenQASM programs as one set (`programs`: each a program's text or a job), or one program over a grid of its inputs (`qasm` with `sweep`, as {{\"theta\": [0.1, 0.2], \"n\": [2, 3]}}: every combination; integer inputs written out, float and angle inputs bound to the double given). The set's other fields (shots, observables, seed, intent, …) apply to every item. Each item gets its own result and receipt, each stating its verdict, and one receipt over the set binds every item through a Merkle root; verify_quantum_receipt checks it whole. At most {} items, run one after another in one job. Waits up to 50 s; a longer set answers with status running and an id for get_quantum_job.", wai_quantum::engine::sets::MAX_ITEMS), + "description": format!("Run several OpenQASM programs as one set (`programs`: each a program's text or a job), or one program over a grid of its inputs (`qasm` with `sweep`, as {{\"theta\": [0.1, 0.2], \"n\": [2, 3]}}: every combination; integer inputs written out, float and angle inputs bound to the double given). The set's other fields (shots, observables, seed, intent, …) apply to every item. Each item gets its own result and receipt, each stating its verdict, and one receipt over the set binds every item through a Merkle root; verify_quantum_receipt checks it whole. At most {} items, run one after another in one job. Waits up to 50 s; a longer set answers with status running and an id for get_quantum_job. Each call is a new job unless it gives idempotency_key, which makes sending it again safe.", wai_quantum::engine::sets::MAX_ITEMS), "inputSchema": { "type": "object", "properties": set_props }, - "annotations": { "readOnlyHint": false, "idempotentHint": true, "openWorldHint": false } + "annotations": { "readOnlyHint": false, "idempotentHint": false, "openWorldHint": false } }, { "name": "dry_run_quantum_job", @@ -379,9 +384,9 @@ fn tool_list() -> Value { { "name": "run_on_target", "title": "Run a program as a device would", - "description": "Run an OpenQASM program as it would run on a device target: compiled exactly to its gates, placed on its most reliable qubits and routes (the expected success given), checked against the circuit, and simulated under the noise its calibration gives, on the density matrix or by trajectories. Counts and expectation values with their distance from the ideal distribution, the noise derived, and a receipt that binds the target's hash. A model of the device, not a run on it. Every angle must be a dyadic multiple of 2π. Waits up to 50 s, as run_quantum_circuit does.", + "description": "Run an OpenQASM program as it would run on a device target: compiled exactly to its gates, placed on its most reliable qubits and routes (the expected success given), checked against the circuit, and simulated under the noise its calibration gives, on the density matrix or by trajectories. Counts and expectation values with their distance from the ideal distribution, the noise derived, and a receipt that binds the target's hash. A model of the device, not a run on it. Every angle must be a dyadic multiple of 2π. Waits up to 50 s, as run_quantum_circuit does, and like it, each call is a new job unless it gives idempotency_key.", "inputSchema": on_target, - "annotations": { "readOnlyHint": false, "idempotentHint": true, "openWorldHint": false } + "annotations": { "readOnlyHint": false, "idempotentHint": false, "openWorldHint": false } }, { "name": "quantum_circuit_examples", @@ -558,8 +563,29 @@ pub async fn handle(app: &App, who: &str, msg: &Value, base: &str, views: bool) }) } +/// A run's `idempotency_key`, taken out of its arguments and checked: it +/// names the job, and is no part of it or of what its receipt seals. +fn take_key(mut args: Value) -> Result<(Value, Option), String> { + let key = match args.as_object_mut().and_then(|o| o.remove("idempotency_key")) { + None | Some(Value::Null) => None, + Some(Value::String(k)) => { + crate::jobs::check_key(&k)?; + Some(k) + } + Some(_) => return Err(wai_quantum::engine::refusal::tag(wai_quantum::engine::refusal::IDEMPOTENCY_KEY, "idempotency_key is a string")), + }; + Ok((args, key)) +} + async fn call(app: &App, who: &str, name: &str, args: Value, base: &str) -> Value { use wai_quantum::engine::targets::{import, job::target_of, library}; + let (args, key) = match name { + "run_quantum_circuit" | "run_on_target" | "run_quantum_set" => match take_key(args) { + Ok(a) => a, + Err(why) => return tool_result(format!("refused: {why}"), json!({ "error": why }), true), + }, + _ => (args, None), + }; match name { "list_targets" => { let list = library::list(); @@ -596,7 +622,7 @@ async fn call(app: &App, who: &str, name: &str, args: Value, base: &str) -> Valu if !verdict::asked(&args) { args["verdict"] = json!(true); } - let id = match crate::http::admit(app, who, args) { + let id = match crate::http::admit(app, who, args, key.as_deref().map(crate::jobs::Key)) { Ok(id) => id, Err((_, why)) => return tool_result(format!("refused: {why}"), json!({ "error": why }), true), }; @@ -1194,4 +1220,49 @@ mod tests { assert_eq!(call("show_target", json!({ "target": "a-device-elsewhere" }))["isError"], true); assert_eq!(call("run_on_target", json!({ "qasm": "OPENQASM 3.0; qubit q;" }))["isError"], true); } + + /// A run sent again under its key is the key's job: the key is taken out + /// of the job it names, one that does not read or that was sent with + /// another program is refused, and the hint says a call without one is a + /// new job. + #[test] + fn a_run_sent_again_under_its_key_is_the_same_job() { + let dir = std::env::temp_dir().join(format!("wai-qsvc-mcp-key-{}", std::process::id())); + let signer = std::sync::Arc::new(crate::receipt::Signer::load_or_create(&dir.join("key")).unwrap()); + let app = App::bare(crate::jobs::Service::new(crate::jobs::Config::from_env()), signer, "http://127.0.0.1:1"); + let rt = tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap(); + let run = |args: Value| rt.block_on(call(&app, "test", "run_quantum_circuit", args, "http://127.0.0.1:1")); + let bell = "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; cx q[0], q[1]; c = measure q;"; + // The job a first call made, as the call admits it (with its verdict + // asked), cancelled so that no call waits for it here. + let id = crate::http::admit(&app, "test", json!({ "qasm": bell, "shots": 20, "verdict": true }), Some(crate::jobs::Key("agent-1"))).unwrap(); + assert_eq!(app.service.cancel(&id), Some(true)); + let again = run(json!({ "qasm": bell, "shots": 20, "idempotency_key": "agent-1" })); + assert_eq!(again["isError"], false, "{again}"); + assert_eq!(again["structuredContent"]["id"], id.as_str(), "{again}"); + assert!(again["structuredContent"]["request"].get("idempotency_key").is_none(), "the key is no part of the job"); + for (args, why) in [ + (json!({ "qasm": bell, "shots": 21, "idempotency_key": "agent-1" }), "another program"), + (json!({ "qasm": bell, "shots": 20, "idempotency_key": "has space" }), "a key that does not read"), + (json!({ "qasm": bell, "shots": 20, "idempotency_key": 7 }), "a key that is not a string"), + ] { + let r = run(args); + assert_eq!(r["isError"], true, "{why}: {r}"); + assert!(r["structuredContent"]["error"].as_str().unwrap().ends_with("[J034]"), "{why}: {r}"); + } + assert_eq!(take_key(json!({ "qasm": bell, "idempotency_key": "k" })).unwrap(), (json!({ "qasm": bell }), Some("k".to_string()))); + assert_eq!(take_key(json!({ "qasm": bell, "idempotency_key": null })).unwrap(), (json!({ "qasm": bell }), None)); + // The three runs take a key, say a call without one is a new job, and + // a plan takes none. + let all = tools(); + let tool = |n: &str| all.as_array().unwrap().iter().find(|t| t["name"] == n).unwrap().clone(); + for n in ["run_quantum_circuit", "run_quantum_set", "run_on_target"] { + let t = tool(n); + assert_eq!(t["annotations"]["idempotentHint"], false, "{n}"); + assert_eq!(t["inputSchema"]["properties"]["idempotency_key"]["maxLength"], crate::jobs::KEY_MAX, "{n}"); + assert!(t["description"].as_str().unwrap().contains("idempotency_key"), "{n}"); + } + assert!(tool("dry_run_quantum_job")["inputSchema"]["properties"].get("idempotency_key").is_none()); + let _ = std::fs::remove_dir_all(&dir); + } } diff --git a/wai/wai-quantum/docs/job-refusals.md b/wai/wai-quantum/docs/job-refusals.md index a0c7962dd..87384a3d3 100644 --- a/wai/wai-quantum/docs/job-refusals.md +++ b/wai/wai-quantum/docs/job-refusals.md @@ -72,6 +72,7 @@ browser engine. A test holds this page to the code's table. | `J031` | refused | A job or a check larger than this host takes: run it on your own machine, with the command line or its MCP server. | | `J032` | retry | The host is full (its queue, or the checks running at once): send it again later. | | `J033` | refused | A host that takes requests only with one of its API keys: give yours as `Authorization: Bearer `; a key it does not hold is refused the same way. | +| `J034` | refused | An idempotency key (`Idempotency-Key`, or a tool's `idempotency_key`) that does not read, or that this client sent before with another request: a key names one job, so another request needs another key. A key is 1 to 255 visible ASCII characters, and should be unguessable (a UUID, or 128 random bits). | | `J040` | refused | A pulse run that does not read or cannot be done: a field it does not take, options or device parameters that do not read, a target without the qubit, or a schedule outside the simulator's model (two tones on one qubit at once, a play after its qubit's capture, more levels and qubits than it holds, a frame far from its qubit); the message says which. | | `J041` | refused | A pulse run whose integration error cannot be stated within its tolerance: a step longer than the generator's norm allows, or a schedule so long that the bound or the rounding passes the tolerance; the message names the steps that would meet it, or says that none does. | | `J050` | refused | A job's `params` that do not read: not an object of finite numbers, a name the program does not declare as a parameter (`input float`, `input angle`), or a parameter of the program left without a value. | diff --git a/wai/wai-quantum/src/engine/refusal.rs b/wai/wai-quantum/src/engine/refusal.rs index 70adf6ac6..08b760a5a 100644 --- a/wai/wai-quantum/src/engine/refusal.rs +++ b/wai/wai-quantum/src/engine/refusal.rs @@ -99,6 +99,7 @@ table! { TOO_LARGE_HERE = "J031", Refused, "A job or a check larger than this host takes: run it on your own machine, with the command line or its MCP server."; BUSY = "J032", Retry, "The host is full (its queue, or the checks running at once): send it again later."; KEY_REQUIRED = "J033", Refused, "A host that takes requests only with one of its API keys: give yours as `Authorization: Bearer `; a key it does not hold is refused the same way."; + IDEMPOTENCY_KEY = "J034", Refused, "An idempotency key (`Idempotency-Key`, or a tool's `idempotency_key`) that does not read, or that this client sent before with another request: a key names one job, so another request needs another key. A key is 1 to 255 visible ASCII characters, and should be unguessable (a UUID, or 128 random bits)."; // Pulse runs (engine::pulse). PULSE = "J040", Refused, "A pulse run that does not read or cannot be done: a field it does not take, options or device parameters that do not read, a target without the qubit, or a schedule outside the simulator's model (two tones on one qubit at once, a play after its qubit's capture, more levels and qubits than it holds, a frame far from its qubit); the message says which.";