diff --git a/.github/workflows/wai-quantum-cli.yml b/.github/workflows/wai-quantum-cli.yml index aa8273a92..d990712c0 100644 --- a/.github/workflows/wai-quantum-cli.yml +++ b/.github/workflows/wai-quantum-cli.yml @@ -78,12 +78,16 @@ jobs: cargo test --locked --release 2>&1 | tee "$RUNNER_TEMP/test.log" grep -qE '^test result: ok\. [1-9][0-9]* passed; 0 failed' "$RUNNER_TEMP/test.log" || { echo "::error::no test ran"; exit 1; } # Your own providers end to end (tests/providers.rs), against the - # reference provider this binary serves, each test by name. + # reference provider this binary serves, and their jobs as MCP + # tasks, each test by name. for t in a_person_adds_tests_plans_and_runs_on_a_provider_and_the_receipt_verifies \ a_run_that_is_not_free_is_refused_until_the_person_confirms_it \ a_credential_is_kept_in_a_file_only_its_owner_reads_and_appears_in_nothing_printed \ a_provider_over_http_runs_with_its_credential_in_the_header \ - an_agent_lists_estimates_is_refused_and_runs_only_with_confirm_cost; do + an_agent_lists_estimates_is_refused_and_runs_only_with_confirm_cost \ + an_agent_runs_on_a_provider_as_a_task_and_the_server_answers_while_it_waits \ + providers::tasks::tests::a_job_reads_as_its_task \ + providers::tasks::tests::a_cancelled_request_ends_its_wait_and_a_change_wakes_it; do grep -q "^test $t \.\.\. ok$" "$RUNNER_TEMP/test.log" || { echo "::error::$t did not run and pass"; exit 1; } done # The model harness and the views in an agent's host, each test by name. diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index e596ede3a..f865be243 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -117,6 +117,25 @@ counts could not be read. A device's queue can take days. - **`wai.quantum.provider/1`:** §5.7 says what a job outlasting its client asks of a provider and a client. Nothing on the wire changes. +### wai-quantum-cli: a job on a provider is an MCP task + +An agent ran a job on a person's provider and then had to remember to ask after it with `get_provider_job`. A client +that speaks MCP 2025-11-25 can now run `run_on_provider` as a task, the protocol's own way to wait. + +- **The task is the job.** `initialize` declares `tasks` (list, cancel, `tools/call`) from that version on, and + `run_on_provider` declares `execution.taskSupport` "optional". `tools/call` with `task` records and sends the job as + the call does and answers once the provider has taken it, the job's handle its `taskId`. A run that is not free and + not confirmed makes no task and sends nothing; the error's data carries the plan. +- **Every job recorded here is a task**, whatever sent it: `run --no-wait`, an earlier session, another agent. + `tasks/get` asks the provider once and says why when it could not; `tasks/list` reads the records and asks no one, + since they are one person's own; `tasks/cancel` cancels at the provider and refuses a task that has ended. A task's + time to live is unlimited, as the records are kept. +- **`tasks/result` waits without holding the server.** Asked before the job has ended, it is answered when the job ends, + by a thread that asks the provider at an interval growing from 2 s to a minute, while the server answers every other + message. A `tasks/cancel` here ends that wait at once, and a request the client cancels + (`notifications/cancelled`) is never answered. Nothing is lost if the server stops: the job is in its record. +- A tool call and a task's wait asking after one job no longer both seal it: this process asks one question at a time. + ### The route record states what a hop delivered by its count and root, so a hop may deliver any number of objects - **Route record version 3** ([extensions/route-record.md](extensions/route-record.md) §3.4, §5.2, §6, §6.1), on a diff --git a/wai/wai-quantum-cli/README.md b/wai/wai-quantum-cli/README.md index f3c0ba301..602cbf38d 100644 --- a/wai/wai-quantum-cli/README.md +++ b/wai/wai-quantum-cli/README.md @@ -430,6 +430,7 @@ wai-quantum verify sweep.json # the set's receipt, each item's, the r - `list_targets`, `show_target` and `run_on_target` (device targets, above); `run_quantum_circuit` and `plan_quantum_circuit` also take `target` - `providers_list`, `provider_estimate` and `run_on_provider` (your own hardware, above): `run_on_provider` says it can spend your money, and refuses a run that is not free without `confirm_cost: true`. A job not finished within the call comes back running, with its id - `get_provider_job`, `cancel_provider_job` and `list_provider_jobs`: the same jobs, from any session +- A client on MCP 2025-11-25 can run `run_on_provider` as a task. `tools/call` with `task` records and sends the job and answers at once, the job's handle its `taskId`. `tasks/get` asks the provider once, `tasks/list` lists every job recorded here, whatever sent it, `tasks/cancel` cancels at the provider, and `tasks/result` is the job once it has ended. A result asked before then is answered when the job ends, and the server answers everything else meanwhile; cancelling the job ends that wait at once A tool that takes a program also takes a generator, with its inputs' values in `inputs`. Runs happen on this machine and are signed with this user's key; nothing is sent anywhere. Every answer about a program carries its verdict (`check_equivalent`'s level is under `verification`, since its own `verdict` came first), every refusal its codes as data, and every tool declares a versioned output schema. diff --git a/wai/wai-quantum-cli/docs/local-mcp.json b/wai/wai-quantum-cli/docs/local-mcp.json index 3a4f60e27..b167d774c 100644 --- a/wai/wai-quantum-cli/docs/local-mcp.json +++ b/wai/wai-quantum-cli/docs/local-mcp.json @@ -49823,6 +49823,9 @@ } } }, + "execution": { + "taskSupport": "optional" + }, "outputSchema": { "$id": "wai-quantum://schemas/run_on_provider/1", "type": "object", diff --git a/wai/wai-quantum-cli/src/mcp.rs b/wai/wai-quantum-cli/src/mcp.rs index a589a1023..d685c6dbb 100644 --- a/wai/wai-quantum-cli/src/mcp.rs +++ b/wai/wai-quantum-cli/src/mcp.rs @@ -21,9 +21,16 @@ //! its `ui://` resource, the resources are listed, and each answer carries //! what the view draws in its `_meta`. Every host gets each such answer in //! plain words beside its JSON. +//! +//! A client that speaks MCP 2025-11-25 can run a job on one of the person's +//! providers as a task, its id the job's handle ([`crate::providers::tasks`]). +//! A task's result asked before its job has ended is answered when it ends, +//! by a thread of its own, so the server answers every other message +//! meanwhile. use serde_json::{Value, json}; use std::io::{BufRead, Write}; +use std::sync::{Arc, Mutex, PoisonError}; use wai_quantum::ed25519_dalek::SigningKey; use wai_quantum::engine::guide::{self, Host}; use wai_quantum::engine::schema::Registry; @@ -471,12 +478,39 @@ fn core_tools() -> Value { ]) } +/// Where the server's replies go, a whole line at a time: standard output, +/// shared with what answers later (a task's result, once its job has ended). +#[derive(Clone)] +pub(crate) struct Replies(Arc>>); + +impl Default for Replies { + fn default() -> Self { + Replies::to(std::io::sink()) + } +} + +impl Replies { + pub(crate) fn to(w: impl Write + Send + 'static) -> Replies { + Replies(Arc::new(Mutex::new(Box::new(w)))) + } + + /// Writes `reply`; false when it could not be written. + fn send(&self, reply: &Value) -> bool { + let mut w = self.0.lock().unwrap_or_else(PoisonError::into_inner); + writeln!(w, "{reply}").and_then(|_| w.flush()).is_ok() + } +} + /// What the server knows of the agent it serves: the name it gave in -/// `initialize`, and whether its host draws the engine's views. +/// `initialize`, whether its host draws the engine's views, where a later +/// reply goes, and the tasks' results waiting for their jobs. #[derive(Default)] pub struct Client { name: Option, views: bool, + replies: Replies, + #[cfg(feature = "providers")] + waits: crate::providers::tasks::Waits, } /// A tool's answer: its JSON, or an error the agent can read and act on. @@ -774,8 +808,15 @@ fn history(name: &str, args: &Value, key: &SigningKey, client: &Client) -> Resul /// The reply to one message, or `None` for a notification. fn handle(msg: &Value, memory_gb: u64, key: &SigningKey, client: &mut Client) -> Option { - let id = msg.get("id")?.clone(); let method = msg.get("method").and_then(Value::as_str).unwrap_or(""); + // A request the client gave up on: a task's result waiting for it is not answered. + #[cfg(feature = "providers")] + if method == "notifications/cancelled" + && let Some(request) = msg.pointer("/params/requestId") + { + client.waits.cancelled(request); + } + let id = msg.get("id")?.clone(); let params = msg.get("params").cloned().unwrap_or(json!({})); if method == "initialize" { client.name = params.pointer("/clientInfo/name").and_then(Value::as_str).filter(|n| !n.is_empty()).map(|n| n.chars().take(80).collect()); @@ -806,6 +847,10 @@ fn handle(msg: &Value, memory_gb: u64, key: &SigningKey, client: &mut Client) -> } let mut capabilities = guide::capabilities(); capabilities["tools"] = json!({}); + #[cfg(feature = "providers")] + if params.get("protocolVersion").and_then(Value::as_str).is_some_and(|v| v >= crate::providers::tasks::FROM) { + capabilities["tasks"] = crate::providers::tasks::capability(); + } Some(match method { "initialize" => ok(json!({ "protocolVersion": params.get("protocolVersion").and_then(Value::as_str).unwrap_or(PROTOCOL), @@ -822,52 +867,132 @@ fn handle(msg: &Value, memory_gb: u64, key: &SigningKey, client: &mut Client) -> } ok(json!({ "tools": t })) } + "tools/call" if params.get("task").is_some_and(|t| !t.is_null()) => { + let name = params.get("name").and_then(Value::as_str).unwrap_or(""); + #[cfg(feature = "providers")] + if name == "run_on_provider" { + let args = params.get("arguments").cloned().unwrap_or(json!({})); + return Some(match crate::providers::Books::load().map_err(|e| (-32603, e, Value::Null)).and_then(|b| crate::providers::tasks::create(&args, &b)) { + Ok(v) => ok(v), + Err(r) => refusal(&id, r), + }); + } + err(-32601, format!("{name} does not run as a task (its execution.taskSupport is forbidden): call it without one; run_on_provider does")) + } "tools/call" => { let name = params.get("name").and_then(Value::as_str).unwrap_or(""); let args = params.get("arguments").cloned().unwrap_or(json!({})); - match call(name, &args, memory_gb, key, client) { - Ok(mut v) => { - // The verdict, beside the answer, for every tool that reads a program. - let read = at_inputs(&args).unwrap_or_else(|_| args.clone()); - if let Some(x) = verdict::for_tool(name, &read, &v) - && v.is_object() - { - v[verdict::verdict_key(name)] = x; - } - // The answer in plain words beside its JSON; what its view draws, to a host that draws it. - let mut content = vec![json!({ "type": "text", "text": v.to_string() })]; - if let Some(t) = views::text(name, &read, &v) { - content.push(json!({ "type": "text", "text": t })); - } - let mut answer = json!({ "content": content, "structuredContent": v, "isError": false }); - if client.views - && let Some(d) = views::data(name, &read, &answer["structuredContent"]) - { - answer["_meta"] = Value::Object([(views::META.to_string(), d)].into_iter().collect()); - } - ok(answer) - } - Err(e) => { - // A refusal as data: its codes, and the verdict of what was refused. - let mut data = wai_quantum::engine::refusal::refusal_json(&e); - data["error"] = json!(e); - if verdict::reads_program(name) { - let read = at_inputs(&args).unwrap_or_else(|_| args.clone()); - data["verdict"] = verdict::of_refusal(&read, &e); - } - ok(json!({ "content": [{ "type": "text", "text": e }], "structuredContent": data, "isError": true })) - } - } + ok(answer(name, &args, call(name, &args, memory_gb, key, client), client.views)) } + #[cfg(feature = "providers")] + "tasks/get" | "tasks/list" | "tasks/cancel" | "tasks/result" => return task_method(method, ¶ms, &id, key, client), _ => err(-32601, format!("no method {method}")), }) } +/// A JSON-RPC error with its data. +#[cfg(feature = "providers")] +fn refusal(id: &Value, (code, message, data): (i64, String, Value)) -> Value { + let mut e = json!({ "code": code, "message": message }); + if !data.is_null() { + e["data"] = data; + } + json!({ "jsonrpc": "2.0", "id": id, "error": e }) +} + +/// A tool's outcome as its call answers it. +fn answer(name: &str, args: &Value, outcome: Result, views: bool) -> Value { + match outcome { + Ok(mut v) => { + // The verdict, beside the answer, for every tool that reads a program. + let read = at_inputs(args).unwrap_or_else(|_| args.clone()); + if let Some(x) = verdict::for_tool(name, &read, &v) + && v.is_object() + { + v[verdict::verdict_key(name)] = x; + } + // The answer in plain words beside its JSON; what its view draws, to a host that draws it. + let mut content = vec![json!({ "type": "text", "text": v.to_string() })]; + if let Some(t) = views::text(name, &read, &v) { + content.push(json!({ "type": "text", "text": t })); + } + let mut answer = json!({ "content": content, "structuredContent": v, "isError": false }); + if views + && let Some(d) = views::data(name, &read, &answer["structuredContent"]) + { + answer["_meta"] = Value::Object([(views::META.to_string(), d)].into_iter().collect()); + } + answer + } + Err(e) => { + // A refusal as data: its codes, and the verdict of what was refused. + let mut data = wai_quantum::engine::refusal::refusal_json(&e); + data["error"] = json!(e); + if verdict::reads_program(name) { + let read = at_inputs(args).unwrap_or_else(|_| args.clone()); + data["verdict"] = verdict::of_refusal(&read, &e); + } + json!({ "content": [{ "type": "text", "text": e }], "structuredContent": data, "isError": true }) + } + } +} + +/// A task's result: `get_provider_job`'s answer for the ended job, naming +/// the task. +#[cfg(feature = "providers")] +fn task_result(task: &str, job: Value, views: bool) -> Value { + let failed = job["status"] == "failed"; + let mut a = answer("get_provider_job", &json!({ "id": task }), Ok(job), views); + a["isError"] = json!(failed); + a["_meta"][crate::providers::tasks::RELATED_TASK] = json!({ "taskId": task }); + a +} + +/// `tasks/get`, `tasks/list`, `tasks/cancel` and `tasks/result`, over the +/// jobs on the person's providers. A result whose job is still at its +/// provider is answered later, through `client.replies`, and `None` now. +#[cfg(feature = "providers")] +fn task_method(method: &str, params: &Value, id: &Value, key: &SigningKey, client: &Client) -> Option { + use crate::providers::{Books, tasks}; + let reply = |r: Result| match r { + Ok(v) => json!({ "jsonrpc": "2.0", "id": id, "result": v }), + Err(r) => refusal(id, r), + }; + let books = match Books::load() { + Ok(b) => b, + Err(e) => return Some(refusal(id, (-32603, e, Value::Null))), + }; + if method == "tasks/list" { + if params.get("cursor").is_some_and(|c| !c.is_null()) { + return Some(refusal(id, (-32602, "no such cursor: every task is on the first page".into(), Value::Null))); + } + return Some(reply(Ok(tasks::list(&books)))); + } + let Some(task) = params.get("taskId").and_then(Value::as_str) else { return Some(refusal(id, (-32602, "give taskId".into(), Value::Null))) }; + match method { + "tasks/get" => Some(reply(tasks::get(task, key, &books))), + "tasks/cancel" => Some(reply(tasks::cancel(task, key, &books, &client.waits))), + _ => match tasks::ended(task, &books) { + Err(r) => Some(refusal(id, r)), + Ok(Some(job)) => Some(reply(Ok(task_result(task, job, client.views)))), + Ok(None) => { + let (replies, views, request, handle) = (client.replies.clone(), client.views, id.clone(), task.to_string()); + let later = move |ended: Result| { + replies.send(&match ended { + Ok(job) => json!({ "jsonrpc": "2.0", "id": request, "result": task_result(&handle, job, views) }), + Err(e) => refusal(&request, (-32603, e, Value::Null)), + }); + }; + tasks::wait_for(task, id, key, &client.waits, later).err().map(|r| refusal(id, r)) + } + }, + } +} + /// Serve until standard input closes. pub fn serve(memory_gb: u64, key: &SigningKey) { let stdin = std::io::stdin(); - let mut out = std::io::stdout().lock(); - let mut client = Client::default(); + let mut client = Client { replies: Replies::to(std::io::stdout()), ..Client::default() }; for line in stdin.lock().lines() { let Ok(line) = line else { break }; if line.trim().is_empty() { @@ -878,7 +1003,7 @@ pub fn serve(memory_gb: u64, key: &SigningKey) { Err(e) => Some(json!({ "jsonrpc": "2.0", "id": null, "error": { "code": -32700, "message": format!("not JSON: {e}") } })), }; if let Some(r) = reply - && writeln!(out, "{r}").and_then(|_| out.flush()).is_err() + && !client.replies.send(&r) { break; } diff --git a/wai/wai-quantum-cli/src/providers.rs b/wai/wai-quantum-cli/src/providers.rs index 4a07066c7..2637bd5bf 100644 --- a/wai/wai-quantum-cli/src/providers.rs +++ b/wai/wai-quantum-cli/src/providers.rs @@ -29,14 +29,15 @@ //! The same, for an agent, is MCP tools ([`mcp_tools`]): `providers_list`, //! `provider_estimate` and `run_on_provider`, which refuses a run that is not //! free without `confirm_cost: true`; and `get_provider_job`, -//! `cancel_provider_job` and `list_provider_jobs`. +//! `cancel_provider_job` and `list_provider_jobs`. A client that speaks MCP +//! 2025-11-25 can run `run_on_provider` as a task instead ([`tasks`]). use crate::{Job, KeyArgs, Output, die, job_request, load_key, print_job}; use clap::{Args, Subcommand}; use serde_json::{Map, Value, json}; use std::io::Read as _; -use std::sync::Arc; use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::{Arc, Mutex, MutexGuard, PoisonError}; use std::time::Duration; use wai_quantum::ed25519_dalek::SigningKey; use wai_quantum_qpu::conformance::{self, Report}; @@ -64,6 +65,30 @@ const WAIT: Wait = Wait { interval: Duration::from_secs(2), polls: 3600 }; /// job service's run does. const CALL_WAIT: Wait = Wait { interval: Duration::from_secs(2), polls: 25 }; +pub mod tasks; + +/// Held while this process asks after a recorded job, so that two questions +/// about one job, from a tool call and a task's wait, do not both seal it. +static ASKING: Mutex<()> = Mutex::new(()); + +fn asking() -> MutexGuard<'static, ()> { + ASKING.lock().unwrap_or_else(PoisonError::into_inner) +} + +/// The person's registry, their credential stores and their job records. +pub struct Books { + reg: Registry, + stores: Stores, + ledger: Ledger, +} + +impl Books { + /// The person's own. + pub fn load() -> Result { + Ok(Books { reg: Registry::load_default()?, stores: Stores::system()?, ledger: Ledger::load_default()? }) + } +} + /// `run`'s flags for a provider. #[derive(Args, Clone, Default)] pub struct RunArgs { @@ -616,6 +641,7 @@ pub fn mcp_tools() -> Vec { "title": "Run on a provider (can spend the person's money)", "description": "Runs a program on one of the person's own quantum providers: their device, their account or their simulator. THIS CAN SPEND THE PERSON'S MONEY OR QUOTA. A run that is not free, or whose cost is unknown, is refused unless confirm_cost is true; set it only after the person has seen the estimate (provider_estimate) and agreed to that run. The answer is a job in the service's form: counts keyed by the classical bits, measured on the device and not reproducible by recomputation, with the circuit's exact distribution beside them, sealed in a receipt with the compiled program's hash, the provider, its protocol and the device's calibration snapshot (verify_quantum_result checks it). A device's queue can be long: if the job has not finished within this call the answer has status running and an id, nothing is cancelled and nothing is held, and get_provider_job collects it later, from any session, for as long as the provider keeps it. Do not run it again to find out: that is a second job, and it can be a second bill.", "inputSchema": provider_schema(json!({ "confirm_cost": { "type": "boolean", "description": "true only when the person has agreed to this run's cost. A run that is not free is refused without it." } })), + "execution": { "taskSupport": "optional" }, }), json!({ "name": "get_provider_job", @@ -692,6 +718,7 @@ pub fn mcp_call_with(name: &str, args: &Value, key: &SigningKey, reg: &Registry, })(), "get_provider_job" | "cancel_provider_job" => (|| { let handle = args.get("id").and_then(Value::as_str).ok_or("give \"id\": the job's id, from run_on_provider or list_provider_jobs")?; + let _held = asking(); let mut ticket = ledger.get(handle)?; if !ticket.waiting() { return Ok(ticket.job()); @@ -711,19 +738,10 @@ pub fn mcp_call(name: &str, args: &Value, key: &SigningKey) -> Option r, - Err(e) => return Some(Err(e)), - }; - let stores = match Stores::system() { - Ok(s) => s, - Err(e) => return Some(Err(e)), - }; - let ledger = match Ledger::load_default() { - Ok(l) => l, - Err(e) => return Some(Err(e)), - }; - mcp_call_with(name, args, key, ®, &stores, &ledger, CALL_WAIT) + match Books::load() { + Ok(b) => mcp_call_with(name, args, key, &b.reg, &b.stores, &b.ledger, CALL_WAIT), + Err(e) => Some(Err(e)), + } } #[cfg(test)] diff --git a/wai/wai-quantum-cli/src/providers/tasks.rs b/wai/wai-quantum-cli/src/providers/tasks.rs new file mode 100644 index 000000000..7b6de1605 --- /dev/null +++ b/wai/wai-quantum-cli/src/providers/tasks.rs @@ -0,0 +1,328 @@ +//! A job on one of the person's providers as an MCP task (protocol +//! 2025-11-25). The task is the job, and its `taskId` the job's handle, so a +//! task an agent made, a job sent with `run --no-wait` and one left by an +//! earlier session are all asked after the same way. +//! +//! `run_on_provider` with `task` records the job and sends it, and answers as +//! soon as the provider has taken it. `tasks/get` asks the provider once; +//! `tasks/list` reads the records and asks no one; `tasks/cancel` cancels at +//! the provider, and a job that finished first keeps its result. +//! +//! `tasks/result` is the job once it has ended. Asked before then it waits, as +//! the protocol has it wait, but not in the loop that reads the next message: +//! a device's queue can be days long, and the server answers everything else +//! meanwhile, a `tasks/cancel` for that job among them. The wait asks the +//! provider at a growing interval, from 2 s to a minute, and looks again at +//! once when a cancel here may have ended the job; a request the client +//! cancels gets no answer. Nothing the wait holds is lost if the server stops: +//! the job is in its record, and a new session asks after it by its handle. + +use super::{Books, asking, entry, mcp_request, target_of}; +use serde_json::{Value, json}; +use std::collections::HashSet; +use std::sync::{Arc, Condvar, Mutex, MutexGuard, PoisonError}; +use std::time::{Duration, Instant}; +use wai_quantum::ed25519_dalek::SigningKey; +use wai_quantum_qpu::job::{self as pjob, JobError}; +use wai_quantum_qpu::registry; + +/// The protocol version from which the server declares tasks. +pub const FROM: &str = "2025-11-25"; + +/// Where an answer about a task names it. +pub const RELATED_TASK: &str = "io.modelcontextprotocol/related-task"; + +/// How often a client is asked to look, in milliseconds: a device's queue is +/// counted in minutes or hours, and each look asks the provider. +const POLL_MS: u64 = 10_000; + +/// A result's wait asks after its job this long after the first question, and +/// twice as long each time after that, up to [`MOST`]. +const FIRST: Duration = Duration::from_secs(2); +const MOST: Duration = Duration::from_secs(60); + +/// The most results waiting at once; another is told to ask `tasks/get`. +const MOST_WAITING: usize = 64; + +/// A refusal: its JSON-RPC code, what it says, and its data. +pub type Refusal = (i64, String, Value); + +/// What the server declares under `capabilities.tasks`. +pub fn capability() -> Value { + json!({ "list": {}, "cancel": {}, "requests": { "tools": { "call": {} } } }) +} + +/// `ms` since 1970 as the protocol writes a time: ISO 8601, in UTC. +fn iso(ms: u64) -> String { + let (days, rem) = ((ms / 86_400_000) as i64, ms % 86_400_000); + let z = days + 719_468; + let era = z.div_euclid(146_097); + let doe = z.rem_euclid(146_097); + let yoe = (doe - doe / 1460 + doe / 36_524 - doe / 146_096) / 365; + let doy = doe - (365 * yoe + yoe / 4 - yoe / 100); + let mp = (5 * doy + 2) / 153; + let d = doy - (153 * mp + 2) / 5 + 1; + let m = if mp < 10 { mp + 3 } else { mp - 9 }; + let y = yoe + era * 400 + i64::from(m <= 2); + format!("{y:04}-{m:02}-{d:02}T{:02}:{:02}:{:02}.{:03}Z", rem / 3_600_000, rem / 60_000 % 60, rem / 1000 % 60, rem % 1000) +} + +/// A job, in the form a record gives it, as a task. A job's record is kept, +/// so its time to live is unlimited. +pub fn task_of(job: &Value) -> Value { + let said = |k: &str, or: &str| job[k].as_str().unwrap_or(or).to_string(); + let (status, message) = match job["status"].as_str() { + Some("running") => ("working", said("why", "running")), + Some("done") => ("completed", "done".to_string()), + Some("cancelled") => ("cancelled", said("error", "cancelled")), + _ => ("failed", said("error", "it failed")), + }; + let created = job["submitted_ms"].as_u64().unwrap_or(0); + json!({ + "taskId": job["id"], + "status": status, + "statusMessage": message, + "createdAt": iso(created), + "lastUpdatedAt": iso(job["finished_ms"].as_u64().unwrap_or(created)), + "ttl": null, + "pollInterval": POLL_MS, + }) +} + +fn missing(e: String) -> Refusal { + (-32602, format!("Failed to retrieve task: {e}"), json!({ "error": e })) +} + +/// `run_on_provider` with `task`: the job recorded and sent, as a task. A run +/// that is not free and not confirmed makes no task and sends nothing, and +/// its refusal carries the plan to show the person. +pub fn create(args: &Value, books: &Books) -> Result { + let refused = |why: String| (-32602, format!("no task was made: {why}"), json!({ "error": why })); + let (provider, req) = mcp_request(args).map_err(refused)?; + let t = entry(&books.reg, &provider).and_then(|e| registry::open(e, &books.stores)).map_err(refused)?; + let confirmed = args.get("confirm_cost") == Some(&Value::Bool(true)); + match pjob::start(&t, &req, confirmed, &books.ledger) { + Ok(ticket) => Ok(json!({ "task": task_of(&ticket.job()) })), + Err(JobError::Unconfirmed { plan, message }) => Err((-32602, format!("no task was made: refused: {message}. Show the person this estimate; call again with confirm_cost: true only when they agree"), json!({ "error": message, "plan": plan }))), + Err(e) => Err(refused(e.to_string())), + } +} + +/// `tasks/get`: the job's task, its provider asked once. A question that +/// could not be asked says nothing of the job, so the task is as recorded, +/// and its message says why. +pub fn get(handle: &str, key: &SigningKey, books: &Books) -> Result { + let _held = asking(); + let mut ticket = books.ledger.get(handle).map_err(missing)?; + if !ticket.waiting() { + return Ok(task_of(&ticket.job())); + } + let asked = target_of(&books.reg, &books.stores, ticket.provider()).and_then(|t| pjob::check(&t, &mut ticket, key, &books.ledger).map(|p| p.job().clone()).map_err(|e| e.to_string())); + Ok(match asked { + Ok(job) => task_of(&job), + Err(e) => { + let mut task = task_of(&ticket.job()); + task["statusMessage"] = json!(format!("{} The last check failed, which says nothing of the job: {e}", task["statusMessage"].as_str().unwrap_or_default())); + task + } + }) +} + +/// `tasks/list`: every job recorded here, the oldest first, asking no one. +/// They are this person's own, on this machine, so every one is listed. +pub fn list(books: &Books) -> Value { + let (tickets, _) = books.ledger.list(); + json!({ "tasks": tickets.iter().map(|t| task_of(&t.job())).collect::>() }) +} + +/// `tasks/cancel`: the job cancelled at its provider, and every result +/// waiting here told to look again. A task that has ended is refused, as the +/// protocol has it; a job that finished before the cancel reached its +/// provider keeps its result. +pub fn cancel(handle: &str, key: &SigningKey, books: &Books, waits: &Waits) -> Result { + let _held = asking(); + let mut ticket = books.ledger.get(handle).map_err(missing)?; + if !ticket.waiting() { + let task = task_of(&ticket.job()); + return Err((-32602, format!("the task has ended ({}): a task that has ended is not cancelled", task["status"].as_str().unwrap_or_default()), json!({ "task": task }))); + } + let unasked = |e: String| (-32603, format!("could not cancel {handle}, and nothing about the job has changed; asking again is safe: {e}"), json!({ "error": e })); + let t = target_of(&books.reg, &books.stores, ticket.provider()).map_err(unasked)?; + let progress = pjob::cancel(&t, &mut ticket, key, &books.ledger).map_err(|e| unasked(e.to_string()))?; + waits.changed(); + Ok(task_of(progress.job())) +} + +/// The job of `handle` if it has ended; `None` if it is still at its +/// provider. +pub fn ended(handle: &str, books: &Books) -> Result, Refusal> { + let _held = asking(); + let ticket = books.ledger.get(handle).map_err(missing)?; + Ok((!ticket.waiting()).then(|| ticket.job())) +} + +/// The results waiting for their jobs here, shared with the threads that +/// wait. +#[derive(Clone, Default)] +pub struct Waits(Arc<(Mutex, Condvar)>); + +#[derive(Default)] +struct Waiting { + /// The requests waiting, by their JSON-RPC id. + requests: HashSet, + /// Counts what may have ended a job here: a wait that sees it move looks + /// again at once. + changes: u64, +} + +impl Waits { + fn state(&self) -> MutexGuard<'_, Waiting> { + self.0.0.lock().unwrap_or_else(PoisonError::into_inner) + } + + /// A job here may have ended. + fn changed(&self) { + self.state().changes += 1; + self.0.1.notify_all(); + } + + /// The client cancelled `request`: if it is a result still waiting, it + /// ends with no answer. + pub fn cancelled(&self, request: &Value) { + if self.state().requests.remove(&request.to_string()) { + self.0.1.notify_all(); + } + } + + /// Waits up to `d` for `request`, or until something here may have ended + /// a job since `seen`. False when the client cancelled the request. + fn sleep(&self, request: &str, seen: u64, d: Duration) -> bool { + let until = Instant::now() + d; + let mut s = self.state(); + loop { + if !s.requests.contains(request) { + return false; + } + let now = Instant::now(); + if s.changes != seen || now >= until { + return true; + } + s = self.0.1.wait_timeout(s, until - now).unwrap_or_else(PoisonError::into_inner).0; + } + } +} + +/// The job of `handle` if it has ended, its provider asked once if it had +/// not. A question that failed is no answer, and the wait asks again; a +/// provider that can no longer be opened ends the wait. +fn look(handle: &str, key: &SigningKey, books: &Books) -> Result, String> { + let _held = asking(); + let mut ticket = books.ledger.get(handle)?; + if ticket.waiting() { + let t = target_of(&books.reg, &books.stores, ticket.provider())?; + let _ = pjob::check(&t, &mut ticket, key, &books.ledger); + } + Ok((!ticket.waiting()).then(|| ticket.job())) +} + +/// `tasks/result` for a job still at its provider: a thread of its own asks +/// after it until it ends, and gives `deliver` the ended job, or why it can +/// no longer be asked after; nothing at all if the client cancels `request` +/// first. +pub fn wait_for(handle: &str, request: &Value, key: &SigningKey, waits: &Waits, deliver: impl FnOnce(Result) + Send + 'static) -> Result<(), Refusal> { + let id = request.to_string(); + { + let mut s = waits.state(); + if s.requests.len() >= MOST_WAITING { + return Err((-32603, format!("{MOST_WAITING} results are already waiting here: ask tasks/get until the task has ended, then tasks/result"), Value::Null)); + } + s.requests.insert(id.clone()); + } + let (handle, key, waits) = (handle.to_string(), key.clone(), waits.clone()); + std::thread::spawn(move || { + let follow = || -> Option> { + let books = match Books::load() { + Ok(b) => b, + Err(e) => return Some(Err(e)), + }; + let mut interval = FIRST; + loop { + let seen = waits.state().changes; + match look(&handle, &key, &books) { + Ok(Some(job)) => return Some(Ok(job)), + Ok(None) => {} + Err(e) => return Some(Err(format!("could not ask after {handle}, which is still recorded: {e}"))), + } + if !waits.sleep(&id, seen, interval) { + return None; + } + interval = (interval * 2).min(MOST); + } + }; + let ended = follow(); + // A cancel that comes after the job ended still means no answer. + if waits.state().requests.remove(&id) + && let Some(ended) = ended + { + deliver(ended); + } + }); + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn a_job_reads_as_its_task() { + assert_eq!(iso(0), "1970-01-01T00:00:00.000Z"); + assert_eq!(iso(951_782_400_000 + 3_723_004), "2000-02-29T01:02:03.004Z"); + let job = |status: &str, extra: Value| { + let mut j = json!({ "id": "pj-0123456789abcdef", "status": status, "submitted_ms": 1_000, "why": "at lab, queued or running" }); + j.as_object_mut().unwrap().extend(extra.as_object().unwrap().clone()); + task_of(&j) + }; + let t = job("running", json!({})); + assert_eq!((t["taskId"].as_str(), t["status"].as_str(), t["statusMessage"].as_str()), (Some("pj-0123456789abcdef"), Some("working"), Some("at lab, queued or running"))); + assert_eq!((t["createdAt"].as_str(), t["lastUpdatedAt"].as_str()), (Some("1970-01-01T00:00:01.000Z"), Some("1970-01-01T00:00:01.000Z"))); + assert!(t["ttl"].is_null() && t.get("ttl").is_some(), "{t}"); + assert_eq!(t["pollInterval"], POLL_MS); + let done = job("done", json!({ "finished_ms": 61_000 })); + assert_eq!((done["status"].as_str(), done["lastUpdatedAt"].as_str()), (Some("completed"), Some("1970-01-01T00:01:01.000Z"))); + assert_eq!(job("cancelled", json!({ "error": "cancelled at lab" }))["status"], "cancelled"); + let failed = job("failed", json!({ "error": "refused at lab: too wide" })); + assert_eq!((failed["status"].as_str(), failed["statusMessage"].as_str()), (Some("failed"), Some("refused at lab: too wide"))); + } + + #[test] + fn a_cancelled_request_ends_its_wait_and_a_change_wakes_it() { + let w = Waits::default(); + w.state().requests.insert("7".into()); + let seen = w.state().changes; + let started = Instant::now(); + let w2 = w.clone(); + let t = std::thread::spawn(move || w2.sleep("7", seen, Duration::from_secs(30))); + std::thread::sleep(Duration::from_millis(50)); + w.changed(); + assert!(t.join().unwrap(), "a change is no cancel"); + assert!(started.elapsed() < Duration::from_secs(10), "a change wakes the wait"); + let seen = w.state().changes; + let w2 = w.clone(); + let t = std::thread::spawn(move || w2.sleep("7", seen, Duration::from_secs(30))); + std::thread::sleep(Duration::from_millis(50)); + w.cancelled(&json!(7)); + assert!(!t.join().unwrap(), "the request was cancelled"); + assert!(started.elapsed() < Duration::from_secs(20)); + // A change made before the wait began is seen at once. + w.state().requests.insert("8".into()); + let seen = w.state().changes; + w.changed(); + let quick = Instant::now(); + assert!(w.sleep("8", seen, Duration::from_secs(30))); + assert!(quick.elapsed() < Duration::from_secs(5)); + // A cancel for a request that is not waiting is forgotten. + w.cancelled(&json!("nothing")); + assert_eq!(w.state().requests.len(), 1); + } +} diff --git a/wai/wai-quantum-cli/tests/providers.rs b/wai/wai-quantum-cli/tests/providers.rs index 51672815e..aba60057a 100644 --- a/wai/wai-quantum-cli/tests/providers.rs +++ b/wai/wai-quantum-cli/tests/providers.rs @@ -401,3 +401,185 @@ fn an_agent_comes_back_to_a_job_by_its_id_and_cancels_only_by_asking() { let _ = c.wait(); } } + +/// `wai-quantum mcp` over standard input and output, its replies in any +/// order: a task's result is answered when its job ends, later than what was +/// asked after it. +struct Mcp { + child: Child, + stdin: Option, + replies: std::sync::mpsc::Receiver, + early: Vec, +} + +impl Mcp { + fn start(h: &Home) -> Mcp { + let mut child = h.cmd().args(["mcp", "--demo-key"]).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap(); + let stdin = child.stdin.take(); + let out = BufReader::new(child.stdout.take().unwrap()); + let (tx, replies) = std::sync::mpsc::channel(); + std::thread::spawn(move || { + for line in out.lines().map_while(Result::ok) { + if tx.send(serde_json::from_str::(&line).unwrap_or_else(|e| panic!("{e}: {line}"))).is_err() { + break; + } + } + }); + Mcp { child, stdin, replies, early: Vec::new() } + } + + fn send(&mut self, msg: Value) { + let stdin = self.stdin.as_mut().unwrap(); + writeln!(stdin, "{msg}").unwrap(); + stdin.flush().unwrap(); + } + + fn ask(&mut self, id: u64, method: &str, params: Value) { + self.send(json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params })); + } + + /// The reply to `id`, if it comes within `within`; others are kept. + fn reply(&mut self, id: u64, within: std::time::Duration) -> Option { + if let Some(i) = self.early.iter().position(|r| r["id"] == id) { + return Some(self.early.remove(i)); + } + let until = std::time::Instant::now() + within; + while let Some(left) = until.checked_duration_since(std::time::Instant::now()) { + match self.replies.recv_timeout(left) { + Ok(r) if r["id"] == id => return Some(r), + Ok(r) => self.early.push(r), + Err(_) => break, + } + } + None + } + + fn call(&mut self, id: u64, method: &str, params: Value) -> Value { + self.ask(id, method, params); + self.reply(id, std::time::Duration::from_secs(120)).unwrap_or_else(|| panic!("no reply to {id} ({method})")) + } + + /// Closes standard input, and waits for the server to end. + fn close(mut self) { + drop(self.stdin.take()); + let _ = self.child.wait(); + } +} + +/// A test that fails part way leaves no server running. +impl Drop for Mcp { + fn drop(&mut self) { + drop(self.stdin.take()); + let _ = self.child.kill(); + let _ = self.child.wait(); + } +} + +/// A child process ended with its test, however the test ends. +struct Reaped(Child); + +impl Drop for Reaped { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +/// A job on a provider as an MCP task: made at once, its id the job's handle; +/// its result waited for while the server answers everything else; cancelled +/// by asking, which ends the wait; a result the client gave up on never +/// answered; and the task still there for the next session. +#[test] +fn an_agent_runs_on_a_provider_as_a_task_and_the_server_answers_while_it_waits() { + if Command::new("curl").arg("--version").output().is_err() { + println!("no curl on this host: the HTTP transport runs where curl is"); + return; + } + use std::time::{Duration, Instant}; + let h = Home::new("tasks"); + let (web, url) = serve_http(&h, &["--polls", "2"]); + let _web = Reaped(web); + h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url], ""); + let (long, long_url) = serve_http(&h, &["--polls", "1000000"]); + let _long = Reaped(long); + h.ok(&["providers", "add", "long", "--kind", "http", "--endpoint", &long_url], ""); + add_process(&h, "paid", &["--price", "0.25", "--currency", "EUR", "--per", "job"], false); + let mut m = Mcp::start(&h); + // Tasks are declared to a client whose protocol has them. + let before = m.call(1, "initialize", json!({ "protocolVersion": "2025-06-18" })); + assert!(before["result"]["capabilities"].get("tasks").is_none(), "{before}"); + let init = m.call(2, "initialize", json!({ "protocolVersion": "2025-11-25" })); + assert_eq!(init["result"]["capabilities"]["tasks"], json!({ "list": {}, "cancel": {}, "requests": { "tools": { "call": {} } } }), "{init}"); + let tools = m.call(3, "tools/list", json!({})); + for t in tools["result"]["tools"].as_array().unwrap() { + let want = if t["name"] == "run_on_provider" { json!({ "taskSupport": "optional" }) } else { Value::Null }; + assert_eq!(t["execution"], want, "{}", t["name"]); + } + let task = |m: &mut Mcp, id: u64, provider: &str| { + let made = m.call(id, "tools/call", json!({ "name": "run_on_provider", "arguments": { "provider": provider, "qasm": BELL, "shots": 100 }, "task": { "ttl": 60000 } })); + let t = made["result"]["task"].clone(); + assert_eq!((t["status"].as_str(), t["ttl"].is_null()), (Some("working"), true), "{made}"); + let id = t["taskId"].as_str().unwrap_or_else(|| panic!("{made}")).to_string(); + assert!(id.starts_with("pj-"), "{id}"); + id + }; + // A short queue: the result is the job once it has ended, naming its task, and its receipt verifies. + let quick = task(&mut m, 4, "web"); + m.ask(5, "tasks/result", json!({ "taskId": quick })); + let r = m.reply(5, Duration::from_secs(120)).expect("the result comes when the job ends"); + let r = &r["result"]; + assert_eq!((r["isError"].as_bool(), r["structuredContent"]["status"].as_str(), r["structuredContent"]["id"].as_str()), (Some(false), Some("done"), Some(quick.as_str())), "{r}"); + assert_eq!(r["_meta"]["io.modelcontextprotocol/related-task"], json!({ "taskId": quick })); + let checked = m.call(6, "tools/call", json!({ "name": "verify_quantum_result", "arguments": { "job": r["structuredContent"] } })); + assert_eq!(checked["result"]["structuredContent"]["valid"], true, "{checked}"); + assert_eq!(m.call(7, "tasks/get", json!({ "taskId": quick }))["result"]["status"], "completed"); + let again = m.call(8, "tasks/cancel", json!({ "taskId": quick })); + assert_eq!(again["error"]["code"], -32602, "a task that has ended is not cancelled: {again}"); + // A long queue: while its result waits, the server answers everything else. + let slow = task(&mut m, 10, "long"); + m.ask(11, "tasks/result", json!({ "taskId": slow })); + let asked = Instant::now(); + assert_eq!(m.call(12, "ping", json!({})), json!({ "jsonrpc": "2.0", "id": 12, "result": {} })); + assert!(asked.elapsed() < Duration::from_secs(20), "the waiting result held the server for {:?}", asked.elapsed()); + assert_eq!(m.call(13, "tasks/get", json!({ "taskId": slow }))["result"]["status"], "working"); + assert!(m.reply(11, Duration::from_millis(300)).is_none(), "the result came before its job ended"); + // Cancelled by asking: the waiting result ends at once, with the cancelled + // job. By 7 s the wait asks every 8 s, so an answer within 3 s is the cancel's. + std::thread::sleep(Duration::from_secs(7).saturating_sub(asked.elapsed())); + let c = m.call(14, "tasks/cancel", json!({ "taskId": slow })); + assert_eq!(c["result"]["status"], "cancelled", "{c}"); + let woke = Instant::now(); + let r = m.reply(11, Duration::from_secs(30)).expect("a cancel ends the wait"); + assert!(woke.elapsed() < Duration::from_secs(3), "the wait slept through the cancel: {:?}", woke.elapsed()); + assert_eq!((r["result"]["isError"].as_bool(), r["result"]["structuredContent"]["status"].as_str()), (Some(false), Some("cancelled")), "{r}"); + // A result the client gave up on is never answered. + let dropped = task(&mut m, 20, "long"); + m.ask(21, "tasks/result", json!({ "taskId": dropped })); + m.send(json!({ "jsonrpc": "2.0", "method": "notifications/cancelled", "params": { "requestId": 21, "reason": "the agent moved on" } })); + assert_eq!(m.call(22, "tasks/cancel", json!({ "taskId": dropped }))["result"]["status"], "cancelled"); + m.call(23, "ping", json!({})); + assert!(m.reply(21, Duration::from_secs(3)).is_none(), "a cancelled request was answered"); + // Every job recorded here is a task, the oldest first. + let list = m.call(24, "tasks/list", json!({})); + let seen: Vec<(String, String)> = list["result"]["tasks"].as_array().unwrap().iter().map(|t| (t["taskId"].as_str().unwrap().to_string(), t["status"].as_str().unwrap().to_string())).collect(); + assert_eq!(seen, [(quick.clone(), "completed".to_string()), (slow.clone(), "cancelled".to_string()), (dropped.clone(), "cancelled".to_string())], "{list}"); + // What makes no task: a tool that is not a run, a run not yet agreed to, a task that is not one. + assert_eq!(m.call(30, "tools/call", json!({ "name": "critique_qasm", "arguments": { "qasm": BELL }, "task": {} }))["error"]["code"], -32601); + let refused = m.call(31, "tools/call", json!({ "name": "run_on_provider", "arguments": { "provider": "paid", "qasm": BELL }, "task": {} })); + assert_eq!(refused["error"]["code"], -32602, "{refused}"); + assert!(refused["error"]["message"].as_str().unwrap().contains("confirm_cost: true"), "{refused}"); + assert_eq!(refused["error"]["data"]["plan"]["estimate"]["amount"], "0.25", "{refused}"); + assert_eq!(m.call(32, "tasks/list", json!({}))["result"]["tasks"].as_array().unwrap().len(), 3, "a refused run is not recorded"); + for bad in ["pj-0000000000000000", "../quantum-providers", ""] { + assert_eq!(m.call(33, "tasks/get", json!({ "taskId": bad }))["error"]["code"], -32602, "{bad:?}"); + } + assert_eq!(m.call(34, "tasks/list", json!({ "cursor": "2" }))["error"]["code"], -32602); + // A task outlives the session that made it. + let kept = task(&mut m, 40, "long"); + m.close(); + let mut m = Mcp::start(&h); + m.call(1, "initialize", json!({ "protocolVersion": "2025-11-25" })); + assert_eq!(m.call(2, "tasks/get", json!({ "taskId": kept }))["result"]["status"], "working"); + assert_eq!(m.call(3, "tasks/cancel", json!({ "taskId": kept }))["result"]["status"], "cancelled"); + m.close(); +}