Transaction-Science/open-standards
Watch
1
0
Fork
You've already forked open-standards
 0
Code Issues Pull requests Projects Releases 15 Packages Wiki Activity Actions

wai-quantum-cli: a job on a provider is an MCP task #609

Merged
dcharlot merged 1 commit from wai/quantum-cli-provider-tasks into main 2026-11-04 00:36:36 -05:00
Conversation 0 Commits 1 Files changed 8 +735 -55
8 changed files with 735 additions and 55 deletions
Download patch file Download diff file Expand all files Collapse all files

8
.github/workflows/wai-quantum-cli.yml vendored
View file

@ -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.

19
wai/CHANGELOG.md
View file

@ -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

1
wai/wai-quantum-cli/README.md
View file

@ -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.

3
wai/wai-quantum-cli/docs/local-mcp.json
View file

@ -49823,6 +49823,9 @@
}
}
},
"execution": {
"taskSupport": "optional"
},
"outputSchema": {
"$id": "wai-quantum://schemas/run_on_provider/1",
"type": "object",

201
wai/wai-quantum-cli/src/mcp.rs
View file

@ -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<Mutex<Box<dyn Write + Send>>>);
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<String>,
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<Value> {
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, &params, &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<Value, String>, 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<Value> {
use crate::providers::{Books, tasks};
let reply = |r: Result<Value, tasks::Refusal>| 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<Value, String>| {
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;
}

48
wai/wai-quantum-cli/src/providers.rs
View file

@ -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<Books, String> {
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<Value> {
"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<Result<Val
if !matches!(name, "providers_list" | "provider_estimate" | "run_on_provider" | "get_provider_job" | "cancel_provider_job" | "list_provider_jobs") {
return None;
}
let reg = match Registry::load_default() {
Ok(r) => 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, &reg, &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)]

328
wai/wai-quantum-cli/src/providers/tasks.rs Normal file
View file

@ -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<Value, Refusal> {
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<Value, Refusal> {
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::<Vec<_>>() })
}
/// `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<Value, Refusal> {
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<Option<Value>, 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<Waiting>, Condvar)>);
#[derive(Default)]
struct Waiting {
/// The requests waiting, by their JSON-RPC id.
requests: HashSet<String>,
/// 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<Option<Value>, 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<Value, String>) + 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<Result<Value, String>> {
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);
}
}

182
wai/wai-quantum-cli/tests/providers.rs
View file

@ -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<std::process::ChildStdin>,
replies: std::sync::mpsc::Receiver<Value>,
early: Vec<Value>,
}
impl Mcp {
fn start(h: &Home) -> Mcp {
let mut child = h.cmd().args(["mcp", "--demo-key"]).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap();
let stdin = child.stdin.take();
let out = BufReader::new(child.stdout.take().unwrap());
let (tx, replies) = std::sync::mpsc::channel();
std::thread::spawn(move || {
for line in out.lines().map_while(Result::ok) {
if tx.send(serde_json::from_str::<Value>(&line).unwrap_or_else(|e| panic!("{e}: {line}"))).is_err() {
break;
}
}
});
Mcp { child, stdin, replies, early: Vec::new() }
}
fn send(&mut self, msg: Value) {
let stdin = self.stdin.as_mut().unwrap();
writeln!(stdin, "{msg}").unwrap();
stdin.flush().unwrap();
}
fn ask(&mut self, id: u64, method: &str, params: Value) {
self.send(json!({ "jsonrpc": "2.0", "id": id, "method": method, "params": params }));
}
/// The reply to `id`, if it comes within `within`; others are kept.
fn reply(&mut self, id: u64, within: std::time::Duration) -> Option<Value> {
if let Some(i) = self.early.iter().position(|r| r["id"] == id) {
return Some(self.early.remove(i));
}
let until = std::time::Instant::now() + within;
while let Some(left) = until.checked_duration_since(std::time::Instant::now()) {
match self.replies.recv_timeout(left) {
Ok(r) if r["id"] == id => return Some(r),
Ok(r) => self.early.push(r),
Err(_) => break,
}
}
None
}
fn call(&mut self, id: u64, method: &str, params: Value) -> Value {
self.ask(id, method, params);
self.reply(id, std::time::Duration::from_secs(120)).unwrap_or_else(|| panic!("no reply to {id} ({method})"))
}
/// Closes standard input, and waits for the server to end.
fn close(mut self) {
drop(self.stdin.take());
let _ = self.child.wait();
}
}
/// A test that fails part way leaves no server running.
impl Drop for Mcp {
fn drop(&mut self) {
drop(self.stdin.take());
let _ = self.child.kill();
let _ = self.child.wait();
}
}
/// A child process ended with its test, however the test ends.
struct Reaped(Child);
impl Drop for Reaped {
fn drop(&mut self) {
let _ = self.0.kill();
let _ = self.0.wait();
}
}
/// A job on a provider as an MCP task: made at once, its id the job's handle;
/// its result waited for while the server answers everything else; cancelled
/// by asking, which ends the wait; a result the client gave up on never
/// answered; and the task still there for the next session.
#[test]
fn an_agent_runs_on_a_provider_as_a_task_and_the_server_answers_while_it_waits() {
if Command::new("curl").arg("--version").output().is_err() {
println!("no curl on this host: the HTTP transport runs where curl is");
return;
}
use std::time::{Duration, Instant};
let h = Home::new("tasks");
let (web, url) = serve_http(&h, &["--polls", "2"]);
let _web = Reaped(web);
h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url], "");
let (long, long_url) = serve_http(&h, &["--polls", "1000000"]);
let _long = Reaped(long);
h.ok(&["providers", "add", "long", "--kind", "http", "--endpoint", &long_url], "");
add_process(&h, "paid", &["--price", "0.25", "--currency", "EUR", "--per", "job"], false);
let mut m = Mcp::start(&h);
// Tasks are declared to a client whose protocol has them.
let before = m.call(1, "initialize", json!({ "protocolVersion": "2025-06-18" }));
assert!(before["result"]["capabilities"].get("tasks").is_none(), "{before}");
let init = m.call(2, "initialize", json!({ "protocolVersion": "2025-11-25" }));
assert_eq!(init["result"]["capabilities"]["tasks"], json!({ "list": {}, "cancel": {}, "requests": { "tools": { "call": {} } } }), "{init}");
let tools = m.call(3, "tools/list", json!({}));
for t in tools["result"]["tools"].as_array().unwrap() {
let want = if t["name"] == "run_on_provider" { json!({ "taskSupport": "optional" }) } else { Value::Null };
assert_eq!(t["execution"], want, "{}", t["name"]);
}
let task = |m: &mut Mcp, id: u64, provider: &str| {
let made = m.call(id, "tools/call", json!({ "name": "run_on_provider", "arguments": { "provider": provider, "qasm": BELL, "shots": 100 }, "task": { "ttl": 60000 } }));
let t = made["result"]["task"].clone();
assert_eq!((t["status"].as_str(), t["ttl"].is_null()), (Some("working"), true), "{made}");
let id = t["taskId"].as_str().unwrap_or_else(|| panic!("{made}")).to_string();
assert!(id.starts_with("pj-"), "{id}");
id
};
// A short queue: the result is the job once it has ended, naming its task, and its receipt verifies.
let quick = task(&mut m, 4, "web");
m.ask(5, "tasks/result", json!({ "taskId": quick }));
let r = m.reply(5, Duration::from_secs(120)).expect("the result comes when the job ends");
let r = &r["result"];
assert_eq!((r["isError"].as_bool(), r["structuredContent"]["status"].as_str(), r["structuredContent"]["id"].as_str()), (Some(false), Some("done"), Some(quick.as_str())), "{r}");
assert_eq!(r["_meta"]["io.modelcontextprotocol/related-task"], json!({ "taskId": quick }));
let checked = m.call(6, "tools/call", json!({ "name": "verify_quantum_result", "arguments": { "job": r["structuredContent"] } }));
assert_eq!(checked["result"]["structuredContent"]["valid"], true, "{checked}");
assert_eq!(m.call(7, "tasks/get", json!({ "taskId": quick }))["result"]["status"], "completed");
let again = m.call(8, "tasks/cancel", json!({ "taskId": quick }));
assert_eq!(again["error"]["code"], -32602, "a task that has ended is not cancelled: {again}");
// A long queue: while its result waits, the server answers everything else.
let slow = task(&mut m, 10, "long");
m.ask(11, "tasks/result", json!({ "taskId": slow }));
let asked = Instant::now();
assert_eq!(m.call(12, "ping", json!({})), json!({ "jsonrpc": "2.0", "id": 12, "result": {} }));
assert!(asked.elapsed() < Duration::from_secs(20), "the waiting result held the server for {:?}", asked.elapsed());
assert_eq!(m.call(13, "tasks/get", json!({ "taskId": slow }))["result"]["status"], "working");
assert!(m.reply(11, Duration::from_millis(300)).is_none(), "the result came before its job ended");
// Cancelled by asking: the waiting result ends at once, with the cancelled
// job. By 7 s the wait asks every 8 s, so an answer within 3 s is the cancel's.
std::thread::sleep(Duration::from_secs(7).saturating_sub(asked.elapsed()));
let c = m.call(14, "tasks/cancel", json!({ "taskId": slow }));
assert_eq!(c["result"]["status"], "cancelled", "{c}");
let woke = Instant::now();
let r = m.reply(11, Duration::from_secs(30)).expect("a cancel ends the wait");
assert!(woke.elapsed() < Duration::from_secs(3), "the wait slept through the cancel: {:?}", woke.elapsed());
assert_eq!((r["result"]["isError"].as_bool(), r["result"]["structuredContent"]["status"].as_str()), (Some(false), Some("cancelled")), "{r}");
// A result the client gave up on is never answered.
let dropped = task(&mut m, 20, "long");
m.ask(21, "tasks/result", json!({ "taskId": dropped }));
m.send(json!({ "jsonrpc": "2.0", "method": "notifications/cancelled", "params": { "requestId": 21, "reason": "the agent moved on" } }));
assert_eq!(m.call(22, "tasks/cancel", json!({ "taskId": dropped }))["result"]["status"], "cancelled");
m.call(23, "ping", json!({}));
assert!(m.reply(21, Duration::from_secs(3)).is_none(), "a cancelled request was answered");
// Every job recorded here is a task, the oldest first.
let list = m.call(24, "tasks/list", json!({}));
let seen: Vec<(String, String)> = list["result"]["tasks"].as_array().unwrap().iter().map(|t| (t["taskId"].as_str().unwrap().to_string(), t["status"].as_str().unwrap().to_string())).collect();
assert_eq!(seen, [(quick.clone(), "completed".to_string()), (slow.clone(), "cancelled".to_string()), (dropped.clone(), "cancelled".to_string())], "{list}");
// What makes no task: a tool that is not a run, a run not yet agreed to, a task that is not one.
assert_eq!(m.call(30, "tools/call", json!({ "name": "critique_qasm", "arguments": { "qasm": BELL }, "task": {} }))["error"]["code"], -32601);
let refused = m.call(31, "tools/call", json!({ "name": "run_on_provider", "arguments": { "provider": "paid", "qasm": BELL }, "task": {} }));
assert_eq!(refused["error"]["code"], -32602, "{refused}");
assert!(refused["error"]["message"].as_str().unwrap().contains("confirm_cost: true"), "{refused}");
assert_eq!(refused["error"]["data"]["plan"]["estimate"]["amount"], "0.25", "{refused}");
assert_eq!(m.call(32, "tasks/list", json!({}))["result"]["tasks"].as_array().unwrap().len(), 3, "a refused run is not recorded");
for bad in ["pj-0000000000000000", "../quantum-providers", ""] {
assert_eq!(m.call(33, "tasks/get", json!({ "taskId": bad }))["error"]["code"], -32602, "{bad:?}");
}
assert_eq!(m.call(34, "tasks/list", json!({ "cursor": "2" }))["error"]["code"], -32602);
// A task outlives the session that made it.
let kept = task(&mut m, 40, "long");
m.close();
let mut m = Mcp::start(&h);
m.call(1, "initialize", json!({ "protocolVersion": "2025-11-25" }));
assert_eq!(m.call(2, "tasks/get", json!({ "taskId": kept }))["result"]["status"], "working");
assert_eq!(m.call(3, "tasks/cancel", json!({ "taskId": kept }))["result"]["status"], "cancelled");
m.close();
}