From ebd28a4100b9322d2333167fd32791f995f3415d Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 2 Nov 2026 13:09:24 -0500 Subject: [PATCH 1/2] wai-quantum-qpu and wai-quantum-cli: a job on a provider outlives what sent it A run on a person's provider waited in its process, and a job that had not finished in two hours was cancelled at the provider, losing its place in the queue. If the process stopped first, the job went on and its id was gone, so its counts could not be read. A device's queue can take days. A job is now written down before it is sent, with everything sealing it will need, and the provider's id as soon as the provider gives one. Nothing that cannot be recorded is sent. Any process can come back to the record, and its counts are read against the program that was sent, whatever the device's calibration has become. Running out of patience leaves the job where it is. A request that did not complete may have reached the provider, so it is never sent again: that could be a second bill. The command line gains run --no-wait and providers jobs, collect and cancel. The local MCP server's run_on_provider answers within a call, and gains get_provider_job, cancel_provider_job and list_provider_jobs. Co-Authored-By: Claude Opus 5.5 --- wai/CHANGELOG.md | 19 + wai/extensions/quantum-providers.md | 19 + wai/wai-quantum-cli/README.md | 14 +- wai/wai-quantum-cli/README.zh-CN.md | 14 +- wai/wai-quantum-cli/docs/llms-full.txt | 154 +++++- wai/wai-quantum-cli/docs/llms.txt | 3 + wai/wai-quantum-cli/docs/local-mcp.json | 362 ++++++++++++- .../schemas/local-tools.v1.json | 107 ++++ .../skills/wai-quantum/SKILL.md | 2 +- .../skills/wai-quantum/reference/tools.md | 3 + wai/wai-quantum-cli/src/mcp.rs | 18 +- wai/wai-quantum-cli/src/providers.rs | 238 ++++++++- wai/wai-quantum-cli/tests/providers.rs | 123 +++++ wai/wai-quantum-qpu/README.md | 2 + wai/wai-quantum-qpu/bring-your-own-quantum.md | 29 + wai/wai-quantum-qpu/src/job.rs | 499 +++++++++++++++++- wai/wai-quantum-qpu/src/ledger.rs | 314 +++++++++++ wai/wai-quantum-qpu/src/lib.rs | 1 + wai/wai-quantum-qpu/src/open_provider.rs | 39 ++ wai/wai-quantum-qpu/src/registry.rs | 2 +- 20 files changed, 1900 insertions(+), 62 deletions(-) create mode 100644 wai/wai-quantum-qpu/src/ledger.rs diff --git a/wai/CHANGELOG.md b/wai/CHANGELOG.md index a568a8fa1..661c87fea 100644 --- a/wai/CHANGELOG.md +++ b/wai/CHANGELOG.md @@ -44,6 +44,25 @@ wire format may change and such changes are recorded here. never reached: it does not call `extract`), and `elliptic-curve` 0.14, which leaves no `crypto-bigint` 0.5 in this workspace. A caller that passes its own `c2pa::Signer` builds against 0.91. +### wai-quantum-qpu and wai-quantum-cli: a job on a provider outlives what sent it + +A run on a person's provider waited in its process, and a job that had not finished in two hours was cancelled at the +provider, losing its place in the queue. If the process stopped first, the job went on and its id was gone, so its +counts could not be read. A device's queue can take days. + +- **wai-quantum-qpu:** `job::start` writes a job down (`ledger`) before it is sent, and the provider's id as soon as it + has one; nothing that cannot be recorded is sent. `job::check` and `job::follow` come back to it from any process, and + seal its counts with what was true when it was sent: the program, the provider, the estimate. Running out of patience + leaves the job where it is. `job::cancel` is someone asking. A request that did not complete may have reached the + provider, so it is never sent again. `job::run` keeps no record, and still cancels what its bound outlasts. +- **wai-quantum-cli:** `run --provider` records the job and says its handle, and `--no-wait` answers with it at once. + `providers jobs`, `providers collect` and `providers cancel` are new: `collect` exits 0 done, 1 ended otherwise, 2 still + at the provider, and 75 when it could not ask, which says nothing of the job. A run that outlasts its two hours now + exits 2, with the job left at the provider, where it was cancelled before. `run_on_provider` answers within a call + with the job running and its id, and `get_provider_job`, `cancel_provider_job` and `list_provider_jobs` are new tools. +- **`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. + ### 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/extensions/quantum-providers.md b/wai/extensions/quantum-providers.md index cd60cf481..671626d3b 100644 --- a/wai/extensions/quantum-providers.md +++ b/wai/extensions/quantum-providers.md @@ -229,6 +229,25 @@ A `result` asked twice of the same job MUST give the same counts. Field: `job`. The answer's `status` is `cancelled`, or `done` for a job that had already finished. A provider that cannot cancel answers `unsupported`. +### 5.7 A job outlasts its client + +A device's queue can outlast any process waiting on it, so a job is not tied to +the connection or the process that submitted it. A provider SHOULD answer +`status` and `result` for a job's id from any later connection, and SHOULD keep +a finished job's result for as long as it can, and answers `not_found` for one +it no longer holds. A provider that runs as a local process (§2.2) ends with its +client unless it keeps its jobs itself. + +A client: + +- SHOULD record a job before it submits it, and the job's id as soon as it has + one, with the program it sent, since a device's calibration can change before + the counts are read; +- MUST NOT cancel a job only because it stopped waiting for it, unless it keeps + no record to come back to; +- MUST NOT submit a job again when its `submit` did not complete, without the + person's confirmation: the first may have been taken, and may be billed. + ## 6. Counts and bit order An outcome is a string of `bits` characters, `0` or `1`, the highest bit diff --git a/wai/wai-quantum-cli/README.md b/wai/wai-quantum-cli/README.md index d77c8a4e7..f3c0ba301 100644 --- a/wai/wai-quantum-cli/README.md +++ b/wai/wai-quantum-cli/README.md @@ -258,6 +258,17 @@ wai-quantum run bell.qasm --provider lab --dry-run # the exact compile, its ch wai-quantum run bell.qasm --provider lab # free here; a run that is not free is refused (exit 3) without --yes ``` +A job is recorded before it is sent, in `provider-jobs` beside your registry, so it does not depend on what sent it. A device's queue can take days, and a run stops waiting after two hours with the job left at the provider: nothing cancels it but your asking. Come back to it by its handle, from any process, for as long as the provider keeps it: + +```sh norun=account +wai-quantum run bell.qasm --provider lab --no-wait # sends it, and answers with its handle +wai-quantum providers jobs # every job recorded here, and where it last stood +wai-quantum providers collect pj-0123456789abcdef # asks, and seals its result when it has one: exit 0 done, 1 ended otherwise, 2 still at the provider, 75 could not ask +wai-quantum providers cancel pj-0123456789abcdef # cancels it; one that finished first gives its result +``` + +A request that did not complete may have reached the provider, so it is never sent again: that could be a second bill. The record says so, and you look in the provider's own list. A job collected later is read against the program that was sent, whatever the device's calibration has become since. A provider run as a local process has to keep its jobs itself, or they end with it. + A credential is read from standard input (`--credential-stdin`) or named by an environment variable (`--credential-env VAR`), never given on a command line. `WAI_QUANTUM_CREDENTIALS=file` keeps it in a file only you can read instead of the keychain. A provider run's result is labelled `measured`, sealed with your key with the provider, the calibration snapshot, the compiled program's hash and the counts set beside the circuit's exact distribution, and checked by `verify`. `wai-quantum providers reference` serves the reference provider, the template for your own. ## Error correction @@ -417,7 +428,8 @@ wai-quantum verify sweep.json # the set's receipt, each item's, the r - `list_components` (every method, pass, noise channel and format, each with its identity) - `qasm_diagnostics` (the reader's codes and the job layer's: all of them, or one by `code`) - `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` +- `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 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/README.zh-CN.md b/wai/wai-quantum-cli/README.zh-CN.md index 4c63866aa..0212f4668 100644 --- a/wai/wai-quantum-cli/README.zh-CN.md +++ b/wai/wai-quantum-cli/README.zh-CN.md @@ -256,6 +256,17 @@ wai-quantum run bell.qasm --provider lab --dry-run # 确切的编译结果、 wai-quantum run bell.qasm --provider lab # 这里免费；不免费的运行在没有 --yes 时会被拒绝（退出状态 3） ``` +作业在发送之前就已记录在注册表旁边的 `provider-jobs` 中，因此它不依赖发送它的进程。设备的队列可能要等上几天；运行在等待两小时后就不再等待，而作业仍留在提供方处：只有你要求时才会取消它。只要提供方还保留着作业，就可以从任何进程凭句柄回来取它： + +```sh norun=account +wai-quantum run bell.qasm --provider lab --no-wait # 发送作业，并以句柄作答 +wai-quantum providers jobs # 这里记录的每个作业，以及它上次的状态 +wai-quantum providers collect pj-0123456789abcdef # 询问，并在有结果时封存：退出状态 0 完成，1 以其他方式结束，2 仍在提供方处，75 无法询问 +wai-quantum providers cancel pj-0123456789abcdef # 取消它；已先完成的作业则给出其结果 +``` + +未完成的请求可能已经到达提供方，所以绝不会再次发送：那可能造成第二笔费用。记录会说明这一点，由你去查看提供方自己的列表。稍后取回的作业，是对照当初发送的程序读取的，无论设备的校准此后变成了什么样。以本地进程运行的提供方必须自己保存作业，否则作业会随进程一起结束。 + 凭据从标准输入读取（`--credential-stdin`），或由环境变量指定（`--credential-env VAR`），绝不在命令行上给出。`WAI_QUANTUM_CREDENTIALS=file` 把它保存在只有你能读取的文件中，而不是钥匙串里。在提供方上运行的结果标为 `measured`，用你的密钥封存，并附有提供方、校准快照、编译后程序的哈希，以及与线路精确分布并列的计数，可由 `verify` 检查。`wai-quantum providers reference` 提供参考提供方，作为你自己的提供方的模板。 ## 纠错 @@ -413,7 +424,8 @@ wai-quantum verify sweep.json # 程序集的回执、每一项的回 - `list_components`（每个方法、编译遍、噪声信道和格式，各附其身份） - `qasm_diagnostics`（读取器的代码和作业层的代码：全部，或按 `code` 取其中一个） - `list_targets`、`show_target` 和 `run_on_target`（设备目标，见上文）；`run_quantum_circuit` 和 `plan_quantum_circuit` 也接受 `target` -- `providers_list`、`provider_estimate` 和 `run_on_provider`（你自己的硬件，见上文）：`run_on_provider` 会声明它可能花你的钱，对不免费的运行，没有 `confirm_cost: true` 时会拒绝 +- `providers_list`、`provider_estimate` 和 `run_on_provider`（你自己的硬件，见上文）：`run_on_provider` 会声明它可能花你的钱，对不免费的运行，没有 `confirm_cost: true` 时会拒绝。未在调用内完成的作业会以 running 状态连同其 id 返回 +- `get_provider_job`、`cancel_provider_job` 和 `list_provider_jobs`：同样的作业，可从任何会话访问 接受程序的工具也接受生成器，其输入的取值放在 `inputs` 中。运行在本机进行，并用当前用户的密钥签名；不会向任何地方发送任何内容。关于程序的每个回答都带有其判定（`check_equivalent` 的等级位于 `verification` 之下，因为它自身已有的 `verdict` 字段在先），每次拒绝都以数据形式带有其代码，每个工具都声明了带版本的输出模式（schema）。 diff --git a/wai/wai-quantum-cli/docs/llms-full.txt b/wai/wai-quantum-cli/docs/llms-full.txt index 606af7b1e..4ddbd24b1 100644 --- a/wai/wai-quantum-cli/docs/llms-full.txt +++ b/wai/wai-quantum-cli/docs/llms-full.txt @@ -588,7 +588,7 @@ Parameters: ### `run_on_provider`: Run on a provider (can spend the person's money) -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). +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. Parameters: - `provider` (string, required): The provider's name in the person's registry (providers_list). @@ -598,6 +598,24 @@ Parameters: - `seed` (integer): A seed, for a provider that is a seeded simulator; a device ignores it. - `confirm_cost` (boolean): true only when the person has agreed to this run's cost. A run that is not free is refused without it. +### `get_provider_job`: Ask after a job on a provider + +Where a job on one of the person's providers stands, by the id run_on_provider gave: still running, or finished, with its counts then sealed in a receipt, or ended, with why. Asking spends nothing and can be repeated; a finished job gives the same answer every time. It asks once and does not wait, so ask again later rather than in a loop: a device's queue can take hours or days. + +Parameters: +- `id` (string, required): The job's id, pj-… + +### `cancel_provider_job`: Cancel a job on a provider + +Cancels a job on one of the person's providers, by its id; only when the person asks, because it gives up the job's place in the queue. Waiting is never a reason to cancel: a job left alone can be collected later. A job that finished first gives its result. + +Parameters: +- `id` (string, required): The job's id, pj-… + +### `list_provider_jobs`: List the jobs on the person's providers + +Every job recorded on this machine: its id, its provider, its shots and where it last stood. It asks no provider, so a job shown running may have finished since; get_provider_job asks. + ## The job service's MCP tools: https://wai.transaction.science/quantum-jobs/mcp Streamable HTTP, no account and no key. @@ -1841,6 +1859,10 @@ Options: With --provider: confirm a run that is not free, or whose cost is unknown; without it such a run is refused (exit 3) + --no-wait + With --provider: send the job and answer with its handle, without waiting (`wai-quantum + providers collect`) + -h, --help Print help @@ -6882,6 +6904,10 @@ Commands: when every one is free, or with --yes free What anyone can use at no cost: the engine's methods here (with their limits), the free job service, and your providers that state free use + jobs Every job recorded here, and where it stands (asks no provider; `collect` does) + collect Ask after a job by its handle, and seal its result when it has one (exit 0 done, 1 + ended otherwise, 2 still at the provider, 75 could not ask: try again) + cancel Cancel a job at its provider. One that finished first gives its result reference Serve the reference provider, the template for your own: --stdio, or --http ADDR (`wai-quantum providers reference --help-args`) help Print this message or the help of the given subcommand(s) @@ -7132,6 +7158,132 @@ Options: Print version ``` +### `wai-quantum providers jobs` + +```text +Every job recorded here, and where it stands (asks no provider; `collect` does) + +Usage: wai-quantum providers jobs [OPTIONS] + +Options: + --input + A value for a generator's integer `input`, as n=5 (repeat for each input): the program is + read as its family's circuit at those values + + --json + + + --lang + The language of the words the command line writes itself, en or zh-CN (by default the one + LANG names); results, codes and the engine's own messages stay in English + + --from + The format of every program the command reads: qasm, qir (LLVM text) or jeff; by default + the file's extension says (.ll or .qir is QIR, .jeff is jeff), and OpenQASM otherwise + + -h, --help + Print help + + -V, --version + Print version +``` + +### `wai-quantum providers collect` + +```text +Ask after a job by its handle, and seal its result when it has one (exit 0 done, 1 ended otherwise, +2 still at the provider, 75 could not ask: try again) + +Usage: wai-quantum providers collect [OPTIONS] + +Arguments: + + The job's handle, pj-… + +Options: + --input + A value for a generator's integer `input`, as n=5 (repeat for each input): the program is + read as its family's circuit at those values + + --wait + Follow it, for up to two hours, rather than asking once + + --json + Print the job as the service returns it: request, plan, result and receipt + + --lang + The language of the words the command line writes itself, en or zh-CN (by default the one + LANG names); results, codes and the engine's own messages stay in English + + --from + The format of every program the command reads: qasm, qir (LLVM text) or jeff; by default + the file's extension says (.ll or .qir is QIR, .jeff is jeff), and OpenQASM otherwise + + --top + The most outcomes to list + + [default: 16] + + --key + Sign with the key in this file (64 hex digits) + + --demo-key + Sign with the published demo key: the receipt then shows a result is unaltered, not who + signed it + + -h, --help + Print help + + -V, --version + Print version +``` + +### `wai-quantum providers cancel` + +```text +Cancel a job at its provider. One that finished first gives its result + +Usage: wai-quantum providers cancel [OPTIONS] + +Arguments: + + The job's handle, pj-… + +Options: + --input + A value for a generator's integer `input`, as n=5 (repeat for each input): the program is + read as its family's circuit at those values + + --json + Print the job as the service returns it: request, plan, result and receipt + + --lang + The language of the words the command line writes itself, en or zh-CN (by default the one + LANG names); results, codes and the engine's own messages stay in English + + --top + The most outcomes to list + + [default: 16] + + --from + The format of every program the command reads: qasm, qir (LLVM text) or jeff; by default + the file's extension says (.ll or .qir is QIR, .jeff is jeff), and OpenQASM otherwise + + --key + Sign with the key in this file (64 hex digits) + + --demo-key + Sign with the published demo key: the receipt then shows a result is unaltered, not who + signed it + + -h, --help + Print help + + -V, --version + Print version +``` + ### `wai-quantum providers reference` ```text diff --git a/wai/wai-quantum-cli/docs/llms.txt b/wai/wai-quantum-cli/docs/llms.txt index fb8e4fc89..b336494b8 100644 --- a/wai/wai-quantum-cli/docs/llms.txt +++ b/wai/wai-quantum-cli/docs/llms.txt @@ -92,6 +92,9 @@ Runs on this machine, signed with the user's key; nothing leaves the machine. - `providers_list`: List the person's quantum providers - `provider_estimate`: Plan a run on a provider, and its cost - `run_on_provider`: Run on a provider (can spend the person's money) +- `get_provider_job`: Ask after a job on a provider +- `cancel_provider_job`: Cancel a job on a provider +- `list_provider_jobs`: List the jobs on the person's providers ## The job service's MCP server diff --git a/wai/wai-quantum-cli/docs/local-mcp.json b/wai/wai-quantum-cli/docs/local-mcp.json index 5ffc77539..3a4f60e27 100644 --- a/wai/wai-quantum-cli/docs/local-mcp.json +++ b/wai/wai-quantum-cli/docs/local-mcp.json @@ -49784,7 +49784,7 @@ { "name": "run_on_provider", "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).", + "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": { "type": "object", "required": [ @@ -57012,6 +57012,366 @@ } ] } + }, + { + "name": "get_provider_job", + "title": "Ask after a job on a provider", + "description": "Where a job on one of the person's providers stands, by the id run_on_provider gave: still running, or finished, with its counts then sealed in a receipt, or ended, with why. Asking spends nothing and can be repeated; a finished job gives the same answer every time. It asks once and does not wait, so ask again later rather than in a loop: a device's queue can take hours or days.", + "inputSchema": { + "type": "object", + "required": [ + "id" + ], + "properties": { + "id": { + "type": "string", + "description": "The job's id, pj-…" + } + } + }, + "annotations": { + "readOnlyHint": false, + "idempotentHint": true, + "openWorldHint": true + }, + "outputSchema": { + "$id": "wai-quantum://schemas/get_provider_job/1", + "type": "object", + "description": "A job on a provider, in the form of a job (job.v1.json), its id the record's handle: running, ended with its error, or done with its result and receipt, which verify_quantum_result checks.", + "anyOf": [ + { + "type": "object", + "description": "A job on a provider, in the form of a job (job.v1.json), its id the record's handle: running, ended with its error, or done with its result and receipt, which verify_quantum_result checks.", + "required": [ + "id", + "status", + "method", + "request" + ], + "properties": { + "id": { + "type": "string" + }, + "status": { + "enum": [ + "running", + "done", + "failed", + "cancelled" + ] + }, + "method": { + "const": "provider" + }, + "why": { + "type": "string" + }, + "request": { + "type": "object" + }, + "result": { + "type": "object" + }, + "error": { + "type": "string" + }, + "submitted_ms": { + "type": "integer" + }, + "finished_ms": { + "type": "integer" + } + } + }, + { + "type": "object", + "description": "A refusal as data: the message, every code in it (the last is the code), its outcome and meaning as the code's table gives them. A host that refuses for its own reasons gives the error alone.", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + }, + "message": { + "type": "string" + }, + "code": { + "type": [ + "string", + "null" + ] + }, + "codes": { + "type": "array", + "items": { + "type": "string" + } + }, + "outcome": { + "enum": [ + "refused", + "retry", + "failed", + "lowered", + "carried", + "preserved", + "warned", + "dropped", + "defaulted" + ] + }, + "meaning": { + "type": "string" + } + } + } + ] + } + }, + { + "name": "cancel_provider_job", + "title": "Cancel a job on a provider", + "description": "Cancels a job on one of the person's providers, by its id; only when the person asks, because it gives up the job's place in the queue. Waiting is never a reason to cancel: a job left alone can be collected later. A job that finished first gives its result.", + "inputSchema": { + "type": "object", + "required": [ + "id" + ], + "properties": { + "id": { + "type": "string", + "description": "The job's id, pj-…" + } + } + }, + "annotations": { + "readOnlyHint": false, + "destructiveHint": true, + "idempotentHint": true, + "openWorldHint": true + }, + "outputSchema": { + "$id": "wai-quantum://schemas/cancel_provider_job/1", + "type": "object", + "description": "A job on a provider, in the form of a job (job.v1.json), its id the record's handle: running, ended with its error, or done with its result and receipt, which verify_quantum_result checks.", + "anyOf": [ + { + "type": "object", + "description": "A job on a provider, in the form of a job (job.v1.json), its id the record's handle: running, ended with its error, or done with its result and receipt, which verify_quantum_result checks.", + "required": [ + "id", + "status", + "method", + "request" + ], + "properties": { + "id": { + "type": "string" + }, + "status": { + "enum": [ + "running", + "done", + "failed", + "cancelled" + ] + }, + "method": { + "const": "provider" + }, + "why": { + "type": "string" + }, + "request": { + "type": "object" + }, + "result": { + "type": "object" + }, + "error": { + "type": "string" + }, + "submitted_ms": { + "type": "integer" + }, + "finished_ms": { + "type": "integer" + } + } + }, + { + "type": "object", + "description": "A refusal as data: the message, every code in it (the last is the code), its outcome and meaning as the code's table gives them. A host that refuses for its own reasons gives the error alone.", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + }, + "message": { + "type": "string" + }, + "code": { + "type": [ + "string", + "null" + ] + }, + "codes": { + "type": "array", + "items": { + "type": "string" + } + }, + "outcome": { + "enum": [ + "refused", + "retry", + "failed", + "lowered", + "carried", + "preserved", + "warned", + "dropped", + "defaulted" + ] + }, + "meaning": { + "type": "string" + } + } + } + ] + } + }, + { + "name": "list_provider_jobs", + "title": "List the jobs on the person's providers", + "description": "Every job recorded on this machine: its id, its provider, its shots and where it last stood. It asks no provider, so a job shown running may have finished since; get_provider_job asks.", + "inputSchema": { + "type": "object", + "properties": {} + }, + "annotations": { + "readOnlyHint": true, + "idempotentHint": true, + "openWorldHint": false + }, + "outputSchema": { + "$id": "wai-quantum://schemas/list_provider_jobs/1", + "type": "object", + "description": "Every job recorded on this machine, as it last stood: no provider is asked.", + "anyOf": [ + { + "type": "object", + "description": "Every job recorded on this machine, as it last stood: no provider is asked.", + "required": [ + "ledger", + "jobs", + "unread" + ], + "properties": { + "ledger": { + "type": "string", + "description": "The directory the records are in." + }, + "jobs": { + "type": "array", + "items": { + "type": "object", + "required": [ + "id", + "provider", + "status", + "shots", + "submitted_ms" + ], + "properties": { + "id": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "status": { + "enum": [ + "running", + "done", + "failed", + "cancelled" + ] + }, + "shots": { + "type": "integer" + }, + "submitted_ms": { + "type": "integer" + }, + "finished_ms": { + "type": "integer" + }, + "error": { + "type": "string" + } + } + } + }, + "unread": { + "type": "array", + "items": { + "type": "string" + }, + "description": "Why any file among the records could not be read." + } + } + }, + { + "type": "object", + "description": "A refusal as data: the message, every code in it (the last is the code), its outcome and meaning as the code's table gives them. A host that refuses for its own reasons gives the error alone.", + "required": [ + "error" + ], + "properties": { + "error": { + "type": "string" + }, + "message": { + "type": "string" + }, + "code": { + "type": [ + "string", + "null" + ] + }, + "codes": { + "type": "array", + "items": { + "type": "string" + } + }, + "outcome": { + "enum": [ + "refused", + "retry", + "failed", + "lowered", + "carried", + "preserved", + "warned", + "dropped", + "defaulted" + ] + }, + "meaning": { + "type": "string" + } + } + } + ] + } } ] } diff --git a/wai/wai-quantum-cli/schemas/local-tools.v1.json b/wai/wai-quantum-cli/schemas/local-tools.v1.json index a4a34b377..63cb70dcc 100644 --- a/wai/wai-quantum-cli/schemas/local-tools.v1.json +++ b/wai/wai-quantum-cli/schemas/local-tools.v1.json @@ -297,6 +297,113 @@ } } }, + "provider_job": { + "type": "object", + "description": "A job on a provider, in the form of a job (job.v1.json), its id the record's handle: running, ended with its error, or done with its result and receipt, which verify_quantum_result checks.", + "required": [ + "id", + "status", + "method", + "request" + ], + "properties": { + "id": { + "type": "string" + }, + "status": { + "enum": [ + "running", + "done", + "failed", + "cancelled" + ] + }, + "method": { + "const": "provider" + }, + "why": { + "type": "string" + }, + "request": { + "type": "object" + }, + "result": { + "type": "object" + }, + "error": { + "type": "string" + }, + "submitted_ms": { + "type": "integer" + }, + "finished_ms": { + "type": "integer" + } + } + }, + "provider_jobs": { + "type": "object", + "description": "Every job recorded on this machine, as it last stood: no provider is asked.", + "required": [ + "ledger", + "jobs", + "unread" + ], + "properties": { + "ledger": { + "type": "string", + "description": "The directory the records are in." + }, + "jobs": { + "type": "array", + "items": { + "type": "object", + "required": [ + "id", + "provider", + "status", + "shots", + "submitted_ms" + ], + "properties": { + "id": { + "type": "string" + }, + "provider": { + "type": "string" + }, + "status": { + "enum": [ + "running", + "done", + "failed", + "cancelled" + ] + }, + "shots": { + "type": "integer" + }, + "submitted_ms": { + "type": "integer" + }, + "finished_ms": { + "type": "integer" + }, + "error": { + "type": "string" + } + } + } + }, + "unread": { + "type": "array", + "items": { + "type": "string" + }, + "description": "Why any file among the records could not be read." + } + } + }, "provider_estimate": { "type": "object", "description": "The dry run of a job on a provider: nothing submitted, nothing spent.", diff --git a/wai/wai-quantum-cli/skills/wai-quantum/SKILL.md b/wai/wai-quantum-cli/skills/wai-quantum/SKILL.md index b8d954bd3..e630d56c5 100644 --- a/wai/wai-quantum-cli/skills/wai-quantum/SKILL.md +++ b/wai/wai-quantum-cli/skills/wai-quantum/SKILL.md @@ -25,7 +25,7 @@ The `wai-quantum` command line reads, runs and checks OpenQASM 3 programs on thi - **Its resources:** the language, the diagnostics (the reader's codes and the job layer's), the methods, the verdict's levels and worked examples. **Its prompts:** writing and verifying a circuit for an intent, explaining a result and its receipt, and repairing a program from its diagnostics. The same text offline: `wai-quantum docs`. - **The list of tools, resources and prompts:** [reference/tools.md](reference/tools.md), generated from the servers' own lists. - **`wai-quantum setup-agent`** writes these instructions into a project's `AGENTS.md`, between `` and ``. -- **The person's own quantum providers:** `wai-quantum providers list` shows what they added; `wai-quantum run FILE --provider NAME --dry-run` gives the plan and the provider's cost before anything is sent. A run that is not free needs the person's yes (`--yes`, or `confirm_cost: true` on the MCP tool `run_on_provider`). Never add a provider or handle its credential yourself. +- **The person's own quantum providers:** `wai-quantum providers list` shows what they added; `wai-quantum run FILE --provider NAME --dry-run` gives the plan and the provider's cost before anything is sent. A run that is not free needs the person's yes (`--yes`, or `confirm_cost: true` on the MCP tool `run_on_provider`). A device's queue can take days: a job not finished in time comes back running with an id, and `wai-quantum providers collect ID` or the MCP tool `get_provider_job` collects it later. Never run it again to find out, and never cancel it because you stopped waiting. Never add a provider or handle its credential yourself. - **Too large for this machine:** `wai-quantum submit FILE` sends it to the job service at https://wai.transaction.science/quantum-jobs, whose MCP server is https://wai.transaction.science/quantum-jobs/mcp. It is free, with no account and no key. ## Rules diff --git a/wai/wai-quantum-cli/skills/wai-quantum/reference/tools.md b/wai/wai-quantum-cli/skills/wai-quantum/reference/tools.md index 14d4f2905..02c4a3468 100644 --- a/wai/wai-quantum-cli/skills/wai-quantum/reference/tools.md +++ b/wai/wai-quantum-cli/skills/wai-quantum/reference/tools.md @@ -55,6 +55,9 @@ Generated by `wai-quantum docs skill` from the servers' own lists; `wai-quantum - `providers_list`: List the person's quantum providers - `provider_estimate`: Plan a run on a provider, and its cost - `run_on_provider`: Run on a provider (can spend the person's money) +- `get_provider_job`: Ask after a job on a provider +- `cancel_provider_job`: Cancel a job on a provider +- `list_provider_jobs`: List the jobs on the person's providers ## The job service's MCP server: https://wai.transaction.science/quantum-jobs/mcp diff --git a/wai/wai-quantum-cli/src/mcp.rs b/wai/wai-quantum-cli/src/mcp.rs index c965c2649..a589a1023 100644 --- a/wai/wai-quantum-cli/src/mcp.rs +++ b/wai/wai-quantum-cli/src/mcp.rs @@ -54,6 +54,8 @@ pub(crate) fn schemas() -> &'static Registry { fn output_of(tool: &str) -> Option<&'static str> { Some(match tool { "run_quantum_circuit" | "run_on_provider" => "job.v1.json", + "get_provider_job" | "cancel_provider_job" => "local-tools.v1.json#/$defs/provider_job", + "list_provider_jobs" => "local-tools.v1.json#/$defs/provider_jobs", "plan_quantum_circuit" => "plan.v1.json", "run_on_target" => "job.v1.json", "repair_qasm" => "tools.v1.json#/$defs/repair", @@ -932,7 +934,7 @@ mod tests { #[allow(unused_mut)] let mut want = vec!["run_quantum_circuit", "plan_quantum_circuit", "lint_qasm", "repair_qasm", "critique_qasm", "check_equivalent", "convert_program", "improve_qasm", "classical_shortcuts", "compile_qasm", "estimate_resources", "draw_circuit", "debug_step", "debug_branches", "verify_quantum_result", "discover_evaluate", "discover_submit", "discover_archive", "novelty_qasm", "discover_novelty", "rewrite_qasm", "converge_cohort", "converge_history", "study_measure", "study_cohort", "taste_rank", "taste_next", "history_commit", "history_changes", "history_blame", "history_log", "list_methods", "list_fabric", "list_components", "qasm_diagnostics", "list_targets", "show_target", "run_on_target", "qec_code", "qec_decode", "qec_threshold", "pulse_schedule", "pulse_check", "gradient_qasm", "optimize_qasm", "vqe_ground_state", "qaoa_maxcut", "build_algorithm"]; #[cfg(feature = "providers")] - want.extend(["providers_list", "provider_estimate", "run_on_provider"]); + want.extend(["providers_list", "provider_estimate", "run_on_provider", "get_provider_job", "cancel_provider_job", "list_provider_jobs"]); assert_eq!(names, want); let run = handle( &json!({ "jsonrpc": "2.0", "id": 3, "method": "tools/call", "params": { "name": "run_quantum_circuit", "arguments": { "qasm": "OPENQASM 3.0; qubit[2] q; h q[0]; cx q[0], q[1];", "observables": ["Z0 Z1"] } } }), @@ -1531,13 +1533,25 @@ mod tests { let mut reg = Providers::load(dir.join("quantum-providers.json")).unwrap(); reg.add(Entry { name: "lab".into(), kind: Kind::Process, endpoint: None, command: vec!["unused".into()], settings: serde_json::Map::new(), credential: Credential::None }, false).unwrap(); let wait = wai_quantum_qpu::Wait { interval: std::time::Duration::ZERO, polls: 5 }; - let list = crate::providers::mcp_call_with("providers_list", &json!({ "free": true }), &key(), ®, &Stores::memory(), wait).unwrap().unwrap(); + let ledger = wai_quantum_qpu::ledger::Ledger::at(dir.join("provider-jobs")); + let list = crate::providers::mcp_call_with("providers_list", &json!({ "free": true }), &key(), ®, &Stores::memory(), &ledger, wait).unwrap().unwrap(); hold("providers_list", &list, &mut called); let r = std::sync::Arc::new(Reference::new(Config::default()).unwrap()); let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(Local(move |m: &Value| r.handle(m, None)))), None); let req = json!({ "qasm": bell, "shots": 50, "provider": "lab" }); hold("provider_estimate", &pjob::plan(&t, &req).unwrap().json, &mut called); hold("run_on_provider", &pjob::run(&t, &req, false, &key(), wait, |_| {}).unwrap(), &mut called); + // A job still at the provider, then finished, then one cancelled, each in the job's form. + let waiting = std::sync::Arc::new(Reference::new(Config { polls: 1, ..Config::default() }).unwrap()); + let w = std::sync::Arc::clone(&waiting); + let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(Local(move |m: &Value| w.handle(m, None)))), None); + let mut ticket = pjob::start(&t, &req, false, &ledger).unwrap(); + hold("run_on_provider", &ticket.job(), &mut called); + assert!(matches!(pjob::check(&t, &mut ticket, &key(), &ledger).unwrap(), pjob::Progress::Waiting(_))); + hold("get_provider_job", pjob::check(&t, &mut ticket, &key(), &ledger).unwrap().job(), &mut called); + let mut other = pjob::start(&t, &req, false, &ledger).unwrap(); + hold("cancel_provider_job", pjob::cancel(&t, &mut other, &key(), &ledger).unwrap().job(), &mut called); + hold("list_provider_jobs", &crate::providers::mcp_call_with("list_provider_jobs", &json!({}), &key(), ®, &Stores::memory(), &ledger, wait).unwrap().unwrap(), &mut called); } let _ = std::fs::remove_dir_all(&dir); assert!(problems.is_empty(), "{} answers do not follow their schemas:\n{}\n{}", problems.len(), problems.join("\n"), REGENERATE); diff --git a/wai/wai-quantum-cli/src/providers.rs b/wai/wai-quantum-cli/src/providers.rs index 39328eb1f..4a07066c7 100644 --- a/wai/wai-quantum-cli/src/providers.rs +++ b/wai/wai-quantum-cli/src/providers.rs @@ -16,12 +16,20 @@ //! wai-quantum providers free # what anyone can use at no cost //! wai-quantum providers reference --stdio # the reference provider, the template for your own //! wai-quantum run bell.qasm --provider lab --dry-run -//! wai-quantum run bell.qasm --provider lab [--yes] +//! wai-quantum run bell.qasm --provider lab [--yes] [--no-wait] +//! wai-quantum providers jobs # every job recorded here, and where it stands +//! wai-quantum providers collect pj-… # asks after one, and seals its result when it has one +//! wai-quantum providers cancel pj-… //! ``` //! -//! The same, for an agent, is three MCP tools ([`mcp_tools`]): -//! `providers_list`, `provider_estimate` and `run_on_provider`, the last of -//! which refuses a run that is not free without `confirm_cost: true`. +//! A job is recorded before it is sent ([`wai_quantum_qpu::ledger`]), so what +//! waits for it can stop, and a device's queue can outlast any wait: when the +//! wait ends the job is left where it is, and is collected later by its handle. +//! +//! 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`. use crate::{Job, KeyArgs, Output, die, job_request, load_key, print_job}; use clap::{Args, Subcommand}; @@ -32,7 +40,8 @@ use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use wai_quantum::ed25519_dalek::SigningKey; use wai_quantum_qpu::conformance::{self, Report}; -use wai_quantum_qpu::job::{self as pjob, JobError, Target}; +use wai_quantum_qpu::job::{self as pjob, JobError, Progress, Target}; +use wai_quantum_qpu::ledger::Ledger; use wai_quantum_qpu::registry::{self, Credential, Entry, Kind, Registry, Stores}; use wai_quantum_qpu::{Billing, Details, Secret, Wait}; @@ -40,9 +49,21 @@ use wai_quantum_qpu::{Billing, Details, Secret, Wait}; /// person did not confirm it. pub const UNCONFIRMED: i32 = 3; -/// How long a run waits for its job: every two seconds, for up to two hours. +/// The exit status of a job still at the provider. +pub const STILL_WAITING: i32 = 2; + +/// The exit status of a question that could not be asked: it says nothing of +/// the job, and asking again is safe. +pub const TRY_AGAIN: i32 = 75; + +/// How long a run on the command line waits for its job, every two seconds: +/// two hours. Then the job is left at the provider, to be collected. const WAIT: Wait = Wait { interval: Duration::from_secs(2), polls: 3600 }; +/// How long a tool call waits for its job before answering with its id, as the +/// job service's run does. +const CALL_WAIT: Wait = Wait { interval: Duration::from_secs(2), polls: 25 }; + /// `run`'s flags for a provider. #[derive(Args, Clone, Default)] pub struct RunArgs { @@ -55,6 +76,9 @@ pub struct RunArgs { /// With --provider: confirm a run that is not free, or whose cost is unknown; without it such a run is refused (exit 3) #[arg(long, requires = "provider")] pub yes: bool, + /// With --provider: send the job and answer with its handle, without waiting (`wai-quantum providers collect`) + #[arg(long, requires = "provider")] + pub no_wait: bool, } #[derive(Subcommand)] @@ -115,6 +139,32 @@ pub enum Cmd { #[arg(long, default_value_t = crate::DEFAULT_MEMORY_GB)] memory_gb: u64, }, + /// Every job recorded here, and where it stands (asks no provider; `collect` does) + Jobs { + #[arg(long)] + json: bool, + }, + /// Ask after a job by its handle, and seal its result when it has one (exit 0 done, 1 ended otherwise, 2 still at the provider, 75 could not ask: try again) + Collect { + /// The job's handle, pj-… + handle: String, + /// Follow it, for up to two hours, rather than asking once + #[arg(long)] + wait: bool, + #[command(flatten)] + out: Output, + #[command(flatten)] + key: KeyArgs, + }, + /// Cancel a job at its provider. One that finished first gives its result + Cancel { + /// The job's handle, pj-… + handle: String, + #[command(flatten)] + out: Output, + #[command(flatten)] + key: KeyArgs, + }, /// Serve the reference provider, the template for your own: --stdio, or --http ADDR (`wai-quantum providers reference --help-args`) Reference { #[arg(trailing_var_arg = true, allow_hyphen_values = true)] @@ -232,15 +282,9 @@ pub fn run(job: &Job, out: &Output, key: &KeyArgs, args: &RunArgs) { return; } let k = load_key(key); - match pjob::run(&t, &req, args.yes, &k.key, WAIT, sleeper()) { - Ok(body) => { - if out.json { - println!("{body}"); - } else { - print_job(&body, out.top); - print_measured(&body); - } - } + let ledger = Ledger::load_default().unwrap_or_else(|e| die(&e)); + let mut ticket = match pjob::start(&t, &req, args.yes, &ledger) { + Ok(ticket) => ticket, Err(JobError::Unconfirmed { plan, message }) => { if out.json { println!("{}", json!({ "refused": message, "plan": plan })); @@ -251,7 +295,75 @@ pub fn run(job: &Job, out: &Output, key: &KeyArgs, args: &RunArgs) { std::process::exit(UNCONFIRMED); } Err(e) => die(&e.to_string()), + }; + if args.no_wait { + show(&Progress::Waiting(ticket.job()), out); + return; } + eprintln!("note: recorded as {h}: if this stops, the job goes on at {name}, and `wai-quantum providers collect {h}` comes back to it", h = ticket.handle()); + let progress = pjob::follow(&t, &mut ticket, &k.key, &ledger, WAIT, sleeper()).unwrap_or_else(|e| die(&e.to_string())); + std::process::exit(show(&progress, out)); +} + +/// A job as the command line shows it, and the exit status that says where it +/// stands. +fn show(progress: &Progress, out: &Output) -> i32 { + let job = progress.job(); + if out.json { + println!("{job}"); + } + match progress { + Progress::Done(body) => { + if !out.json { + print_job(body, out.top); + print_measured(body); + } + 0 + } + Progress::Waiting(_) => { + let id = job["id"].as_str().unwrap_or_default(); + if !out.json { + println!("job {id}"); + println!("status {}", job["why"].as_str().unwrap_or("still at the provider")); + println!("collect wai-quantum providers collect {id}"); + } + STILL_WAITING + } + Progress::Ended(_) => { + eprintln!("wai-quantum: {}: {} ({})", job["id"].as_str().unwrap_or_default(), job["error"].as_str().unwrap_or("it ended with no result"), job["status"].as_str().unwrap_or_default()); + 1 + } + } +} + +/// The provider a recorded job went to: only a job still waiting needs one, so +/// a finished job reads back after its provider is gone. +fn target_of(reg: &Registry, stores: &Stores, provider: &str) -> Result { + registry::open(entry(reg, provider).map_err(|e| format!("{e}; the job is still recorded"))?, stores) +} + +/// `providers collect` and `providers cancel`: `ask` put to the job of +/// `handle`, or how it stands if nothing more can come of it. A question that +/// could not be asked exits here, as it says nothing of the job. +fn asked(handle: &str, key: &KeyArgs, ask: impl FnOnce(&Target, &mut wai_quantum_qpu::ledger::Ticket, &SigningKey, &Ledger) -> Result) -> Progress { + let ledger = Ledger::load_default().unwrap_or_else(|e| die(&e)); + let mut ticket = ledger.get(handle).unwrap_or_else(|e| die(&e)); + if !ticket.waiting() { + return pjob::progress_of(&ticket); + } + let (reg, stores) = load(); + let t = target_of(®, &stores, ticket.provider()).unwrap_or_else(|e| die(&e)); + let k = load_key(key); + ask(&t, &mut ticket, &k.key, &ledger).unwrap_or_else(|e| { + eprintln!("wai-quantum: could not ask after {handle}, and nothing about the job has changed: {e}"); + std::process::exit(TRY_AGAIN); + }) +} + +/// Every job recorded in `ledger`, as a list. +fn jobs_json(ledger: &Ledger) -> Value { + let (tickets, unread) = ledger.list(); + json!({ "ledger": ledger.dir().display().to_string(), "jobs": tickets.iter().map(|t| t.listing()).collect::>(), "unread": unread }) } /// What `providers describe` shows. @@ -315,6 +427,33 @@ fn free(reg: &Registry, stores: &Stores, memory_gb: u64) -> Value { pub fn main(cmd: Cmd) { match cmd { + Cmd::Jobs { json } => { + let v = jobs_json(&Ledger::load_default().unwrap_or_else(|e| die(&e))); + if json { + println!("{v}"); + return; + } + let jobs = v["jobs"].as_array().map_or(&[][..], Vec::as_slice); + if jobs.is_empty() { + println!("no jobs recorded yet ({})", v["ledger"].as_str().unwrap_or_default()); + } + for j in jobs { + let status = if j["status"] == "running" { "at the provider" } else { j["status"].as_str().unwrap_or_default() }; + println!("{} {:<16} {:>8} shots {status}", j["id"].as_str().unwrap_or_default(), j["provider"].as_str().unwrap_or_default(), j["shots"]); + } + for why in v["unread"].as_array().map_or(&[][..], Vec::as_slice) { + eprintln!("note: could not read {}", why.as_str().unwrap_or_default()); + } + } + Cmd::Collect { handle, wait, out, key } => { + let progress = asked(&handle, &key, |t, ticket, k, ledger| if wait { pjob::follow(t, ticket, k, ledger, WAIT, sleeper()) } else { pjob::check(t, ticket, k, ledger) }); + std::process::exit(show(&progress, &out)); + } + Cmd::Cancel { handle, out, key } => { + let progress = asked(&handle, &key, pjob::cancel); + let status = show(&progress, &out); + std::process::exit(if progress.job()["status"] == "cancelled" { 0 } else { status }); + } Cmd::Add { name, kind, endpoint, settings, credential_stdin, credential_env, replace, command } => { let kind = Kind::parse(&kind).unwrap_or_else(|| die(&format!("--kind {kind}: one of {}", Kind::ALL.map(Kind::name).join(", ")))); let (mut reg, stores) = load(); @@ -457,7 +596,7 @@ fn provider_schema(extra: Value) -> Value { s } -/// The three MCP tools. +/// The MCP tools. pub fn mcp_tools() -> Vec { vec![ json!({ @@ -475,13 +614,34 @@ pub fn mcp_tools() -> Vec { json!({ "name": "run_on_provider", "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).", + "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." } })), }), + json!({ + "name": "get_provider_job", + "title": "Ask after a job on a provider", + "description": "Where a job on one of the person's providers stands, by the id run_on_provider gave: still running, or finished, with its counts then sealed in a receipt, or ended, with why. Asking spends nothing and can be repeated; a finished job gives the same answer every time. It asks once and does not wait, so ask again later rather than in a loop: a device's queue can take hours or days.", + "inputSchema": { "type": "object", "required": ["id"], "properties": { "id": { "type": "string", "description": "The job's id, pj-…" } } }, + "annotations": { "readOnlyHint": false, "idempotentHint": true, "openWorldHint": true }, + }), + json!({ + "name": "cancel_provider_job", + "title": "Cancel a job on a provider", + "description": "Cancels a job on one of the person's providers, by its id; only when the person asks, because it gives up the job's place in the queue. Waiting is never a reason to cancel: a job left alone can be collected later. A job that finished first gives its result.", + "inputSchema": { "type": "object", "required": ["id"], "properties": { "id": { "type": "string", "description": "The job's id, pj-…" } } }, + "annotations": { "readOnlyHint": false, "destructiveHint": true, "idempotentHint": true, "openWorldHint": true }, + }), + json!({ + "name": "list_provider_jobs", + "title": "List the jobs on the person's providers", + "description": "Every job recorded on this machine: its id, its provider, its shots and where it last stood. It asks no provider, so a job shown running may have finished since; get_provider_job asks.", + "inputSchema": { "type": "object", "properties": {} }, + "annotations": { "readOnlyHint": true, "idempotentHint": true, "openWorldHint": false }, + }), ] } -/// Appends the three tools to a tools list. +/// Appends these tools to a tools list. pub fn add_tools(tools: &mut Value) { if let Some(a) = tools.as_array_mut() { a.extend(mcp_tools()); @@ -507,8 +667,8 @@ fn mcp_request(args: &Value) -> Result<(String, Value), String> { Ok((name, req)) } -/// One of the three tools, with `reg` and `stores`; `None` for another tool. -pub fn mcp_call_with(name: &str, args: &Value, key: &SigningKey, reg: &Registry, stores: &Stores, wait: Wait) -> Option> { +/// One of the tools, with `reg`, `stores` and `ledger`; `None` for another tool. +pub fn mcp_call_with(name: &str, args: &Value, key: &SigningKey, reg: &Registry, stores: &Stores, ledger: &Ledger, wait: Wait) -> Option> { Some(match name { "providers_list" => { let mut v = json!({ "registry": reg.path().display().to_string(), "providers": reg.entries().iter().map(Entry::listing).collect::>() }); @@ -524,20 +684,31 @@ pub fn mcp_call_with(name: &str, args: &Value, key: &SigningKey, reg: &Registry, return pjob::plan(&t, &req).map(|p| p.json).map_err(|e| e.to_string()); } let confirmed = args.get("confirm_cost") == Some(&Value::Bool(true)); - match pjob::run(&t, &req, confirmed, key, wait, std::thread::sleep) { - Ok(body) => Ok(body), + match pjob::start(&t, &req, confirmed, ledger).and_then(|mut ticket| pjob::follow(&t, &mut ticket, key, ledger, wait, std::thread::sleep)) { + Ok(progress) => Ok(progress.job().clone()), Err(JobError::Unconfirmed { plan, message }) => Err(format!("refused: {message}. Show the person this estimate; call again with confirm_cost: true only when they agree. The plan: {plan}")), Err(e) => Err(e.to_string()), } })(), + "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 mut ticket = ledger.get(handle)?; + if !ticket.waiting() { + return Ok(ticket.job()); + } + let t = target_of(reg, stores, ticket.provider())?; + let asked = if name == "cancel_provider_job" { pjob::cancel(&t, &mut ticket, key, ledger) } else { pjob::check(&t, &mut ticket, key, ledger) }; + asked.map(|progress| progress.job().clone()).map_err(|e| format!("could not ask after {handle}, and nothing about the job has changed; asking again is safe: {e}")) + })(), + "list_provider_jobs" => Ok(jobs_json(ledger)), _ => return None, }) } -/// One of the three tools, with the person's own registry; `None` for +/// One of the tools, with the person's own registry and records; `None` for /// another tool. pub fn mcp_call(name: &str, args: &Value, key: &SigningKey) -> Option> { - if !matches!(name, "providers_list" | "provider_estimate" | "run_on_provider") { + 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() { @@ -548,7 +719,11 @@ pub fn mcp_call(name: &str, args: &Value, key: &SigningKey) -> Option s, Err(e) => return Some(Err(e)), }; - mcp_call_with(name, args, key, ®, &stores, WAIT) + 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) } #[cfg(test)] @@ -583,7 +758,7 @@ mod tests { fn the_tools_say_what_they_cost_and_run_on_provider_says_it_spends_money() { let t = mcp_tools(); let names: Vec<&str> = t.iter().map(|x| x["name"].as_str().unwrap()).collect(); - assert_eq!(names, ["providers_list", "provider_estimate", "run_on_provider"]); + assert_eq!(names, ["providers_list", "provider_estimate", "run_on_provider", "get_provider_job", "cancel_provider_job", "list_provider_jobs"]); let run = t[2]["description"].as_str().unwrap(); assert!(run.contains("SPEND THE PERSON'S MONEY") && run.contains("confirm_cost"), "{run}"); assert!(t[2]["inputSchema"]["properties"]["confirm_cost"].is_object()); @@ -602,9 +777,16 @@ mod tests { reg.add(Entry { name: "lab".into(), kind: Kind::Process, endpoint: None, command: vec!["unused".into()], settings: Map::new(), credential: Credential::None }, false).unwrap(); let stores = Stores::memory(); let key = SigningKey::from_bytes(&[8; 32]); - let list = mcp_call_with("providers_list", &json!({}), &key, ®, &stores, Wait { interval: Duration::ZERO, polls: 5 }).unwrap().unwrap(); + let ledger = Ledger::at(dir.join("provider-jobs")); + let list = mcp_call_with("providers_list", &json!({}), &key, ®, &stores, &ledger, Wait { interval: Duration::ZERO, polls: 5 }).unwrap().unwrap(); assert_eq!(list["providers"][0]["name"], "lab"); - assert!(mcp_call_with("lint_qasm", &json!({}), &key, ®, &stores, WAIT).is_none()); + assert!(mcp_call_with("lint_qasm", &json!({}), &key, ®, &stores, &ledger, WAIT).is_none()); + // An id comes from an agent: it reaches no file but a record's. + for bad in ["../quantum-providers", "/etc/passwd", ""] { + let e = mcp_call_with("get_provider_job", &json!({ "id": bad }), &key, ®, &stores, &ledger, WAIT).unwrap().unwrap_err(); + assert!(e.contains("is not a job's handle"), "{e}"); + } + assert!(mcp_call_with("cancel_provider_job", &json!({}), &key, ®, &stores, &ledger, WAIT).unwrap().unwrap_err().contains("give \"id\"")); // The job layer as the tools call it, on a priced reference. let r = Arc::new(Reference::new(Config { price: Some(("0.5".into(), "EUR".into(), Per::Job)), ..Config::default() }).unwrap()); let t = Target::new("lab", Kind::Process, Box::new(wai_quantum_qpu::open_provider::OpenProvider::new(wai_quantum_qpu::open_provider::Local(move |m: &Value| r.handle(m, None)))), None); diff --git a/wai/wai-quantum-cli/tests/providers.rs b/wai/wai-quantum-cli/tests/providers.rs index caea9cadb..51672815e 100644 --- a/wai/wai-quantum-cli/tests/providers.rs +++ b/wai/wai-quantum-cli/tests/providers.rs @@ -278,3 +278,126 @@ fn an_agent_lists_estimates_is_refused_and_runs_only_with_confirm_cost() { drop(stdin); let _ = child.wait(); } + +fn jobs(h: &Home) -> Vec { + let v: Value = serde_json::from_str(&h.ok(&["providers", "jobs", "--json"], "")).unwrap(); + v["jobs"].as_array().unwrap().clone() +} + +/// What a job's record is for: the command that sent it can stop, the wait can +/// run out, and whoever comes back later collects the job by its handle. +#[test] +fn a_job_outlives_the_command_that_sent_it_and_is_collected_by_its_handle() { + if Command::new("curl").arg("--version").output().is_err() { + println!("no curl on this host: the HTTP transport runs where curl is"); + return; + } + let h = Home::new("collect"); + let prog = h.program(); + let hash = wai_quantum_qpu::reference::credential_hash(SECRET); + let (mut queue, url) = serve_http(&h, &["--polls", "3", "--credential-blake3", &hash]); + h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url, "--credential-stdin"], SECRET); + // Sent, and answered with its handle: no one is waiting. + let sent: Value = serde_json::from_str(&h.ok(&["run", &prog, "--provider", "web", "--shots", "200", "--no-wait", "--json"], "")).unwrap(); + assert_eq!((sent["status"].as_str(), sent["method"].as_str()), (Some("running"), Some("provider")), "{sent}"); + let id = sent["id"].as_str().unwrap().to_string(); + assert!(id.starts_with("pj-"), "{id}"); + let listed = jobs(&h); + assert_eq!((listed.len(), listed[0]["id"].as_str(), listed[0]["status"].as_str(), listed[0]["provider"].as_str()), (1, Some(id.as_str()), Some("running"), Some("web")), "{listed:?}"); + // Each question is its own process. Still at the provider is exit 2, and says how to come back. + for _ in 0..3 { + let out = h.run(&["providers", "collect", &id], ""); + assert_eq!(out.status.code(), Some(2), "{}", String::from_utf8_lossy(&out.stderr)); + assert!(String::from_utf8_lossy(&out.stdout).contains(&format!("wai-quantum providers collect {id}"))); + } + let body = h.ok(&["providers", "collect", &id, "--json"], ""); + let job: Value = serde_json::from_str(&body).unwrap(); + assert_eq!((job["status"].as_str(), job["result"]["job"]["polls"].as_u64()), (Some("done"), Some(4)), "{job}"); + let saved = h.dir.join("collected.json"); + std::fs::write(&saved, &body).unwrap(); + let check = h.ok(&["verify", saved.to_str().unwrap()], ""); + assert!(check.contains("signature VALID"), "{check}"); + // Done is kept: the same job, once the provider and its queue are gone. + let _ = queue.kill(); + let _ = queue.wait(); + h.ok(&["providers", "remove", "web"], ""); + assert_eq!(h.ok(&["providers", "collect", &id, "--json"], "").trim(), body.trim()); + // The command is killed while it waits, and the job goes on. + let (mut queue, url) = serve_http(&h, &["--polls", "1000000"]); + h.ok(&["providers", "add", "slow", "--kind", "http", "--endpoint", &url], ""); + let mut waiting = h.cmd().args(["run", &prog, "--provider", "slow"]).stdout(Stdio::null()).stderr(Stdio::null()).spawn().unwrap(); + let mut again = None; + for _ in 0..300 { + again = jobs(&h).iter().find(|j| j["provider"] == "slow" && j["status"] == "running").cloned(); + if again.is_some() { + break; + } + std::thread::sleep(std::time::Duration::from_millis(100)); + } + let _ = waiting.kill(); + let _ = waiting.wait(); + let again = again.expect("the job is recorded while the command waits"); + let id = again["id"].as_str().unwrap(); + let out = h.run(&["providers", "collect", id], ""); + assert_eq!(out.status.code(), Some(2), "{}", String::from_utf8_lossy(&out.stderr)); + // A provider that cannot be reached says nothing of the job: exit 75, and it is still recorded. + let _ = queue.kill(); + let _ = queue.wait(); + let out = h.run(&["providers", "collect", id], ""); + assert_eq!(out.status.code(), Some(75), "{}", String::from_utf8_lossy(&out.stderr)); + assert!(String::from_utf8_lossy(&out.stderr).contains("nothing about the job has changed")); + assert!(jobs(&h).iter().any(|j| j["id"] == id && j["status"] == "running")); + // A handle that is not one reaches nothing. + let out = h.run(&["providers", "collect", "../quantum-providers"], ""); + assert_eq!(out.status.code(), Some(1)); + assert!(String::from_utf8_lossy(&out.stderr).contains("is not a job's handle")); + // No credential is in any record, and none was printed. + for e in std::fs::read_dir(h.dir.join("config/wai/provider-jobs")).unwrap().flatten() { + assert!(!std::fs::read_to_string(e.path()).unwrap().contains(SECRET), "{}", e.path().display()); + } + assert!(!h.seen.borrow().contains(SECRET)); +} + +/// The same for an agent: a job it did not wait for, asked after and cancelled by its id. +#[test] +fn an_agent_comes_back_to_a_job_by_its_id_and_cancels_only_by_asking() { + if Command::new("curl").arg("--version").output().is_err() { + println!("no curl on this host: the HTTP transport runs where curl is"); + return; + } + let h = Home::new("agent-jobs"); + let prog = h.program(); + let (mut queue, url) = serve_http(&h, &["--polls", "1"]); + h.ok(&["providers", "add", "web", "--kind", "http", "--endpoint", &url], ""); + let (mut long, long_url) = serve_http(&h, &["--polls", "1000000"]); + h.ok(&["providers", "add", "long", "--kind", "http", "--endpoint", &long_url], ""); + let first: Value = serde_json::from_str(&h.ok(&["run", &prog, "--provider", "long", "--no-wait", "--json"], "")).unwrap(); + let id = first["id"].as_str().unwrap().to_string(); + let mut child = h.cmd().args(["mcp", "--demo-key"]).stdin(Stdio::piped()).stdout(Stdio::piped()).stderr(Stdio::null()).spawn().unwrap(); + let mut stdin = child.stdin.take().unwrap(); + let mut stdout = BufReader::new(child.stdout.take().unwrap()); + rpc(&mut stdin, &mut stdout, 1, "initialize", json!({ "protocolVersion": "2025-06-18" })); + let call = |stdin: &mut std::process::ChildStdin, stdout: &mut BufReader, n: u64, name: &str, args: Value| rpc(stdin, stdout, n, "tools/call", json!({ "name": name, "arguments": args }))["result"].clone(); + let list = call(&mut stdin, &mut stdout, 2, "list_provider_jobs", json!({})); + assert_eq!(list["structuredContent"]["jobs"][0]["id"].as_str(), Some(id.as_str()), "{list}"); + let asked = call(&mut stdin, &mut stdout, 3, "get_provider_job", json!({ "id": id })); + assert_eq!((asked["isError"].as_bool(), asked["structuredContent"]["status"].as_str()), (Some(false), Some("running")), "{asked}"); + let cancelled = call(&mut stdin, &mut stdout, 4, "cancel_provider_job", json!({ "id": id })); + assert_eq!(cancelled["structuredContent"]["status"], "cancelled", "{cancelled}"); + // A short queue finishes within the call, and its receipt verifies. + let ran = call(&mut stdin, &mut stdout, 5, "run_on_provider", json!({ "provider": "web", "qasm": BELL })); + assert_eq!(ran["structuredContent"]["status"], "done", "{ran}"); + let job = ran["structuredContent"].clone(); + let checked = call(&mut stdin, &mut stdout, 6, "verify_quantum_result", json!({ "job": job })); + assert_eq!(checked["structuredContent"]["valid"], true, "{checked}"); + let same = call(&mut stdin, &mut stdout, 7, "get_provider_job", json!({ "id": job["id"] })); + assert_eq!(same["structuredContent"], job); + let bad = call(&mut stdin, &mut stdout, 8, "get_provider_job", json!({ "id": "../../secret" })); + assert_eq!(bad["isError"], true, "{bad}"); + drop(stdin); + let _ = child.wait(); + for c in [&mut queue, &mut long] { + let _ = c.kill(); + let _ = c.wait(); + } +} diff --git a/wai/wai-quantum-qpu/README.md b/wai/wai-quantum-qpu/README.md index 97f108d9c..30baf56c6 100644 --- a/wai/wai-quantum-qpu/README.md +++ b/wai/wai-quantum-qpu/README.md @@ -8,6 +8,8 @@ For every circuit, a `QpuSampler`: 2. compiles the circuit to them exactly, with `wai_quantum::quantum_toolchain::native`, and checks the program against the circuit by simulation before anything is sent; 3. submits it, waits for the job, and reads the counts back onto the circuit's qubits: logical qubit `i` is measured into classical bit `i`, which is bit `i` of the counts' index, whatever physical qubit it ended on. +A queue can outlast any wait, so `job::start` records a job before it is sent (`ledger`), and `job::check`, `job::follow` and `job::cancel` come back to it by its record, from any process. Running out of patience leaves the job where it is. `job::run` keeps no record, so it cancels what does not finish within its bound. + `QpuSampler::member` certifies the device statistically against the exact distribution of every circuit of a corpus, as the fabric certifies any sampler. Its certificates carry the device's provenance: silicon for a processor, a simulator for a service's simulator. Placement ranks a simulator after silicon. ## Services diff --git a/wai/wai-quantum-qpu/bring-your-own-quantum.md b/wai/wai-quantum-qpu/bring-your-own-quantum.md index 3ba2701d6..92de9f9fd 100644 --- a/wai/wai-quantum-qpu/bring-your-own-quantum.md +++ b/wai/wai-quantum-qpu/bring-your-own-quantum.md @@ -121,6 +121,35 @@ switch says that such runs go through the desktop app or the command line. The hosted job service never routes a job to a provider and never takes a credential; a request that names a provider is refused there. +## When the queue is long + +A device's queue can take days, and nothing has to stay running for it. A job +is written down before it is sent, in `provider-jobs` beside your registry +(readable only by you, with no credential in it), and the provider's id beside +that as soon as the provider gives one. + +```sh norun=account +wai-quantum run bell.qasm --provider lab --no-wait # sends it, and answers with its handle +wai-quantum providers jobs # every job recorded here, and where it last stood +wai-quantum providers collect pj-0123456789abcdef # asks, and seals its result when it has one +wai-quantum providers cancel pj-0123456789abcdef +``` + +A plain `run` waits two hours and then stops waiting, with the job left at the +provider and its handle on the screen. Only your asking cancels a job. If the +command, the terminal or the machine stops, the job goes on, and `collect` +picks it up: exit 0 done, 1 ended otherwise, 2 still at the provider, 75 could +not ask, which says nothing of the job. An agent has the same in +`get_provider_job`, `cancel_provider_job` and `list_provider_jobs`, and the job +from `run_on_provider` is the same job. + +The counts are read against the program that was sent, and the receipt states +that program and what the provider said when it was sent, whatever its +calibration has become since. A request that did not complete may have reached +the provider, so it is never sent again, since that could be a second bill: the +record says so, and you look in the provider's own list. A provider you run as +a local process has to keep its jobs itself, or they end with it. + ## The protocol, and the reference as your template The [open provider protocol](../extensions/quantum-providers.md) is six diff --git a/wai/wai-quantum-qpu/src/job.rs b/wai/wai-quantum-qpu/src/job.rs index 47be916f5..b875c98de 100644 --- a/wai/wai-quantum-qpu/src/job.rs +++ b/wai/wai-quantum-qpu/src/job.rs @@ -20,17 +20,28 @@ //! snapshot, the compiled program's hash, the counts, the cost, and the //! counts against the circuit's exact distribution. //! +//! [`run`] waits within its bound and, keeping no record, cancels the job when +//! the bound passes, so that nothing goes on being spent with no one to +//! collect it. A device's queue can outlast any bound, so a job that should +//! survive is recorded instead ([`crate::ledger`]): [`start`] writes it down +//! before it is sent and gives its [`Ticket`]; [`check`] asks once and +//! [`follow`] asks within a bound, and neither cancels it when the answer is +//! still no; [`cancel`] is only ever someone asking. Any process can come back +//! to a ticket, and its result is the one [`run`] would have sealed. +//! //! A credential never appears in what these return: every message is passed //! through [`crate::Secret::redact`] first. use crate::open_provider::PROTOCOL; use crate::registry::Kind; -use crate::{Billing, Details, Device, Estimate, JobRecord, Provider, QpuSampler, Secret, Wait, basis_label}; +use crate::ledger::{Ledger, Ticket}; +use crate::{Billing, Details, Device, Estimate, JobInfo, JobRecord, JobStatus, Provider, QpuError, QpuSampler, Secret, Wait, basis_label}; use serde_json::{Value, json}; use std::time::Duration; use wai_quantum::ed25519_dalek::SigningKey; use wai_quantum::engine::targets::{import, layout}; use wai_quantum::engine::{catalog, provider, receipt}; +use wai_quantum::quantum_fabric::sampling::Counts; use wai_quantum::quantum_toolchain::native::{NativeGate, NativeProgram}; /// The free job service's address. @@ -259,44 +270,219 @@ pub fn plan(t: &Target, request: &Value) -> Result { plan_with(t, &sampler, request) } -fn energy_json(r: &JobRecord) -> Value { +fn energy_json(service_kwh: Option) -> Value { let mut v = json!({ "unmetered": "the work ran at the provider; this host's meter would read only the wait" }); - if let Some(k) = r.service_kwh { + if let Some(k) = service_kwh { v["service_kwh"] = json!(k); v["service_kwh_is"] = json!("the provider's own figure, not a measurement here"); } v } +/// The refusal of a run that is not free when no one has confirmed it. +fn refuse_unconfirmed(t: &Target, p: &Plan, confirmed: bool) -> Result<(), JobError> { + if p.free || confirmed { + return Ok(()); + } + let message = format!("{} on {}: {}", if p.estimate.as_ref().is_ok_and(|e| e.is_some()) { "this run is" } else { "this run's cost is not known to be free, and it is" }, t.name, p.confirmation()); + Err(JobError::Unconfirmed { plan: p.json.clone(), message }) +} + +/// What a job's receipt says of it that is known before it is sent, kept with +/// its record: what is true of the job when it was sent is not asked again +/// when its result is collected. +fn sealing_of(t: &Target, p: &Plan, confirmed: bool) -> Value { + let protocol = p.json["provider"]["protocol"].as_str().unwrap_or(PROTOCOL).to_string(); + let mut compiled = compiled_json(&p.program); + compiled["placement"] = p.placement.clone(); + json!({ + "provider": provider_json(t, &p.device, &p.details), + "calibration": p.details.calibration, + "compiled": compiled, + "free": p.free, + "estimate": estimate_json(&p.estimate), + "confirmed": confirmed && !p.free, + "why": format!("on {}, {} ({protocol}): compiled exactly to {} on its qubits and checked by simulation here before it was sent", t.name, p.device.name, basis_label(p.device.basis)), + }) +} + +/// The counts of job `id`, as a sealed job. +#[allow(clippy::too_many_arguments)] +fn sealed(t: &Target, request: &Value, read: &provider::Request, sealing: &Value, id: &str, polls: u32, counts: Counts, info: &JobInfo, key: &SigningKey) -> Result { + let charged = info.charged.as_ref().map(|c| json!({ "amount": c.amount, "currency": c.currency })); + let measured = provider::Measured { + counts, + provider: sealing["provider"].clone(), + calibration: info.calibration.clone().or_else(|| sealing["calibration"].as_str().map(str::to_string)), + compiled: sealing["compiled"].clone(), + job: json!({ "id": t.redact(id), "polls": polls }), + cost: json!({ "free": sealing["free"], "estimate": sealing["estimate"], "charged": charged, "confirmed": sealing["confirmed"] }), + energy: energy_json(info.service_kwh), + }; + let result = provider::result(read, measured).map_err(|e| t.fail(e))?; + Ok(provider::seal(request, result, key, sealing["why"].as_str().unwrap_or_default())) +} + /// Runs `request` on `t`: refused, with the plan, when it is not free and /// `confirmed` is false; otherwise submitted, waited for within `wait` -/// (sleeping with `sleep`), and sealed with `key`. +/// (sleeping with `sleep`), and sealed with `key`. A job that does not finish +/// within `wait` is cancelled, where the provider can cancel one: this keeps +/// no record to come back to. [`start`] and [`follow`] keep one. pub fn run(t: &Target, request: &Value, confirmed: bool, key: &SigningKey, wait: Wait, sleep: impl Fn(Duration) + 'static) -> Result { let sampler = QpuSampler::new(&*t.provider).map_err(|e| t.fail(e))?.with_wait(wait, sleep); let p = plan_with(t, &sampler, request)?; - if !p.free && !confirmed { - let message = format!("{} on {}: {}", if p.estimate.as_ref().is_ok_and(|e| e.is_some()) { "this run is" } else { "this run's cost is not known to be free, and it is" }, t.name, p.confirmation()); - return Err(JobError::Unconfirmed { plan: p.json, message }); + refuse_unconfirmed(t, &p, confirmed)?; + let sealing = sealing_of(t, &p, confirmed); + let (counts, record): (_, JobRecord) = sampler.run_compiled(&p.read.circuit, &p.program, p.read.shots, p.read.seed).map_err(|e| t.fail(e))?; + let info = JobInfo { service_kwh: record.service_kwh, calibration: record.calibration, charged: record.charged }; + sealed(t, request, &p.read, &sealing, &record.id, record.polls, counts, &info, key) +} + +/// What came of asking after a recorded job. Each holds the job in the form +/// every host gives one, its `id` the ticket's handle. +#[derive(Clone, Debug, PartialEq)] +pub enum Progress { + /// It finished, and its counts are sealed. + Done(Value), + /// It is still at the provider, queued or running. + Waiting(Value), + /// It ended with no result: refused, failed, cancelled, or not known to + /// have been sent. + Ended(Value), +} + +impl Progress { + /// The job. + pub fn job(&self) -> &Value { + match self { + Progress::Done(j) | Progress::Waiting(j) | Progress::Ended(j) => j, + } } - let (counts, record) = sampler.run_compiled(&p.read.circuit, &p.program, p.read.shots, p.read.seed).map_err(|e| t.fail(e))?; - let charged = record.charged.as_ref().map(|c| json!({ "amount": c.amount, "currency": c.currency })); - let measured = provider::Measured { - counts, - provider: provider_json(t, &p.device, &p.details), - calibration: record.calibration.clone().or_else(|| p.details.calibration.clone()), - compiled: { - let mut c = compiled_json(&p.program); - c["placement"] = p.placement.clone(); - c - }, - job: json!({ "id": t.redact(&record.id), "polls": record.polls }), - cost: json!({ "free": p.free, "estimate": estimate_json(&p.estimate), "charged": charged, "confirmed": confirmed && !p.free }), - energy: energy_json(&record), - }; - let result = provider::result(&p.read, measured).map_err(|e| t.fail(e))?; - let protocol = p.json["provider"]["protocol"].as_str().unwrap_or(PROTOCOL).to_string(); - let why = format!("on {}, {} ({protocol}): compiled exactly to {} on its qubits and checked by simulation here before it was sent", t.name, p.device.name, basis_label(p.device.basis)); - Ok(provider::seal(request, result, key, &why)) +} + +/// A recorded job's state, with no one asked. +pub fn progress_of(ticket: &Ticket) -> Progress { + match ticket.state() { + "done" => Progress::Done(ticket.job()), + "submitted" => Progress::Waiting(ticket.job()), + _ => Progress::Ended(ticket.job()), + } +} + +/// Plans `request` on `t` and sends it, having first written it down: refused, +/// with the plan, when it is not free and `confirmed` is false; refused too, +/// with nothing sent, when it cannot be written down. The ticket comes back as +/// soon as the provider has taken the job. +/// +/// The record's state says what is known if this stops part way. A request +/// that did not complete may have reached the provider, so it is `unknown`, +/// and is not sent again: that could be a second bill. +pub fn start(t: &Target, request: &Value, confirmed: bool, ledger: &Ledger) -> Result { + let sampler = QpuSampler::new(&*t.provider).map_err(|e| t.fail(e))?; + let p = plan_with(t, &sampler, request)?; + refuse_unconfirmed(t, &p, confirmed)?; + let mut ticket = Ticket::recording(&t.name, request, &p.program, p.read.shots, p.read.seed, sealing_of(t, &p, confirmed)); + ledger.put(&ticket).map_err(|e| JobError::Failed(format!("nothing was sent: the job could not be recorded first ({e})")))?; + match t.provider.submit_seeded(&p.program, p.read.shots, p.read.seed) { + Ok(id) => { + ticket.submitted(&id); + ledger.put(&ticket).map_err(|e| JobError::Failed(t.redact(&format!("{} took the job as {id}, and its id could not be recorded ({e}): keep that id", t.name))))?; + Ok(ticket) + } + Err(e) => { + let why = t.redact(&e.to_string()); + ticket.ended(if matches!(e, QpuError::Transport(_)) { "unknown" } else { "failed" }, &why); + let _ = ledger.put(&ticket); + Err(JobError::Failed(why)) + } + } +} + +/// Asks after a recorded job once, and seals its result if it has finished. +/// The answer is `Ok` for anything the job did, and an error only when asking +/// failed, which says nothing of the job: its record stays as it was, and +/// asking again is always safe. A job with a result already gives that result, +/// asking no one. +pub fn check(t: &Target, ticket: &mut Ticket, key: &SigningKey, ledger: &Ledger) -> Result { + let progress = ask(t, ticket, key, ledger)?; + if matches!(progress, Progress::Waiting(_)) { + let _ = ledger.put(ticket); // Only the count of checks: nothing depends on it. + } + Ok(progress) +} + +/// [`check`], writing the record only when something more than the count of +/// checks has changed: [`follow`] asks often, and writes the count when it stops. +fn ask(t: &Target, ticket: &mut Ticket, key: &SigningKey, ledger: &Ledger) -> Result { + if !ticket.waiting() { + return Ok(progress_of(ticket)); + } + let id = ticket.provider_id().ok_or_else(|| JobError::Failed(format!("the record of {} has no provider's id", ticket.handle())))?.to_string(); + ticket.polled(); + match t.provider.status(&id) { + Ok(JobStatus::Pending) => return Ok(progress_of(ticket)), + Ok(JobStatus::Done) => { + let read = provider::read(ticket.request()).map_err(|e| t.fail(e))?; + let program = ticket.program().map_err(|e| t.fail(e))?; + let (counts, info) = t.provider.counts(&id, &program, ticket.shots()).map_err(|e| t.fail(e))?; + let body = sealed(t, ticket.request(), &read, ticket.sealing(), &id, ticket.polls(), counts, &info, key)?; + ticket.done(body); + } + Ok(JobStatus::Ended { status, reason }) => { + let state = if status == "cancelled" { "cancelled" } else { "failed" }; + ticket.ended(state, &t.redact(&format!("{status} at {}: {reason}", t.name))); + } + Err(QpuError::Provider { code, message }) if code == "not_found" => { + ticket.ended("failed", &t.redact(&format!("{} no longer knows this job: {message}", t.name))); + } + Err(e) => return Err(t.fail(e)), + } + ledger.put(ticket).map_err(|e| JobError::Failed(format!("the job's state could not be recorded, and it is safe to ask again: {e}")))?; + Ok(progress_of(ticket)) +} + +/// Asks after a recorded job within `wait`, sleeping with `sleep`. When `wait` +/// passes, the job is left where it is, and the answer is that it is still +/// waiting: running out of patience is no reason to cancel it. A failure to +/// ask is tried again within `wait`, and named in the answer if it is the last +/// thing that happened. +pub fn follow(t: &Target, ticket: &mut Ticket, key: &SigningKey, ledger: &Ledger, wait: Wait, sleep: impl Fn(Duration)) -> Result { + let mut asked = 0u32; + loop { + let (last, why) = match ask(t, ticket, key, ledger) { + Ok(Progress::Waiting(job)) => (job, None), + Err(e) => (ticket.job(), Some(e.to_string())), + Ok(ended) => return Ok(ended), + }; + asked += 1; + if asked > wait.polls { + let _ = ledger.put(ticket); // Only the count of checks: nothing depends on it. + let mut job = last; + if let Some(why) = why { + job["why"] = json!(format!("{} The last check failed: {why}", job["why"].as_str().unwrap_or_default())); + } + return Ok(Progress::Waiting(job)); + } + sleep(wait.interval); + } +} + +/// Cancels a recorded job, because someone asked. A job that finished first +/// gives its result: it has been paid for. +pub fn cancel(t: &Target, ticket: &mut Ticket, key: &SigningKey, ledger: &Ledger) -> Result { + if !ticket.waiting() { + return Ok(progress_of(ticket)); + } + let id = ticket.provider_id().ok_or_else(|| JobError::Failed(format!("the record of {} has no provider's id", ticket.handle())))?.to_string(); + t.provider.cancel(&id).map_err(|e| t.fail(e))?; + if let Ok(Progress::Done(_)) = ask(t, ticket, key, ledger) { + return Ok(progress_of(ticket)); + } + if ticket.waiting() { + ticket.ended("cancelled", &format!("cancelled at {}", t.name)); + ledger.put(ticket).map_err(|e| JobError::Failed(format!("the job was cancelled, and that could not be recorded: {e}")))?; + } + Ok(progress_of(ticket)) } /// What anyone can use at no cost: the engine's own methods on this machine @@ -339,6 +525,7 @@ mod tests { use super::*; use crate::open_provider::{Local, OpenProvider}; use crate::reference::{Config, Per, Reference}; + use wai_quantum::quantum_toolchain::native::Basis; use std::sync::Arc; fn target(config: Config) -> Target { @@ -352,6 +539,266 @@ mod tests { const BELL: &str = "OPENQASM 3.0; qubit[2] q; bit[2] c; h q[0]; cx q[0], q[1]; c = measure q;"; + fn on(r: &Arc) -> Target { + let r = Arc::clone(r); + Target::new("lab", Kind::Process, Box::new(OpenProvider::new(Local(move |m: &Value| r.handle(m, None)))), None) + } + + /// A ledger of its own, gone when the test is. + struct Records(std::path::PathBuf); + + impl Records { + fn new(tag: &str) -> Records { + let d = std::env::temp_dir().join(format!("wai-qpu-job-{tag}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&d); + Records(d) + } + + fn ledger(&self) -> Ledger { + Ledger::at(&self.0) + } + } + + impl Drop for Records { + fn drop(&mut self) { + let _ = std::fs::remove_dir_all(&self.0); + } + } + + type Asked = Arc>>; + type Unreachable = Arc>>; + + /// The reference, counting what is asked of it, with a failure to reach it + /// on the operations named. + struct Watched { + r: Arc, + ops: Asked, + unreachable: Unreachable, + } + + impl crate::open_provider::Channel for Watched { + fn exchange(&self, m: &Value) -> Result { + let op = m["op"].as_str().unwrap_or_default().to_string(); + self.ops.lock().unwrap().push(op.clone()); + if self.unreachable.lock().unwrap().iter().any(|u| *u == op) { + return Err(QpuError::Transport("the connection was reset".into())); + } + Ok(self.r.handle(m, None)) + } + + fn label(&self) -> String { + "watched".into() + } + } + + fn watched(r: &Arc) -> (Target, Asked, Unreachable) { + let ops = Arc::new(std::sync::Mutex::new(Vec::new())); + let unreachable = Arc::new(std::sync::Mutex::new(Vec::new())); + let ch = Watched { r: Arc::clone(r), ops: Arc::clone(&ops), unreachable: Arc::clone(&unreachable) }; + (Target::new("lab", Kind::Process, Box::new(OpenProvider::new(ch)), None), ops, unreachable) + } + + fn count(ops: &std::sync::Mutex>, op: &str) -> usize { + ops.lock().unwrap().iter().filter(|o| *o == op).count() + } + + #[test] + fn a_recorded_job_outlives_what_sent_it_and_seals_as_a_run_would() { + let records = Records::new("outlives"); + let key = SigningKey::from_bytes(&[4; 32]); + let req = json!({ "qasm": BELL, "shots": 500, "seed": 7, "provider": "lab" }); + let r = Arc::new(Reference::new(Config { polls: 3, ..Config::default() }).unwrap()); + let handle = { + let t = on(&r); + let ticket = start(&t, &req, false, &records.ledger()).unwrap(); + assert_eq!(ticket.state(), "submitted"); + let job = ticket.job(); + assert_eq!((job["status"].as_str(), job["method"].as_str(), job["id"].as_str()), (Some("running"), Some("provider"), Some(ticket.handle())), "{job}"); + ticket.handle().to_string() + }; + // Another process, from the record alone. + let (t, ops, _) = watched(&r); + let mut back = records.ledger().get(&handle).unwrap(); + for _ in 0..3 { + assert!(matches!(check(&t, &mut back, &key, &records.ledger()).unwrap(), Progress::Waiting(_))); + } + let Progress::Done(body) = check(&t, &mut back, &key, &records.ledger()).unwrap() else { panic!("not done") }; + assert!(wai_quantum::engine::job::verify(&body).unwrap().0, "{body}"); + assert_eq!(body["id"].as_str(), Some(handle.as_str())); + assert_eq!(body["result"]["job"]["polls"], 4); + // What it sealed is what a run seals: nothing was asked again but the + // job's state and its counts. + assert_eq!((count(&ops, "status"), count(&ops, "result"), count(&ops, "describe"), count(&ops, "estimate"), count(&ops, "submit")), (4, 1, 0, 0, 0)); + let direct = run(&on(&Arc::new(Reference::new(Config::default()).unwrap())), &req, false, &key, no_wait(), |_| {}).unwrap(); + for k in ["counts", "compiled", "provider", "calibration", "cost", "energy"] { + assert_eq!(body["result"][k], direct["result"][k], "{k}"); + } + assert_eq!(body["why"], direct["why"]); + // Asking again gives the same job, and asks no one. + let mut again = records.ledger().get(&handle).unwrap(); + let Progress::Done(same) = check(&t, &mut again, &SigningKey::from_bytes(&[9; 32]), &records.ledger()).unwrap() else { panic!("not done") }; + assert_eq!(same, body); + assert_eq!(count(&ops, "status"), 4); + } + + #[test] + fn running_out_of_patience_does_not_cancel_the_job() { + let records = Records::new("patience"); + let key = SigningKey::from_bytes(&[4; 32]); + let req = json!({ "qasm": BELL, "shots": 100, "provider": "lab" }); + let r = Arc::new(Reference::new(Config { polls: 10, ..Config::default() }).unwrap()); + let (t, ops, _) = watched(&r); + let mut ticket = start(&t, &req, false, &records.ledger()).unwrap(); + let waited = follow(&t, &mut ticket, &key, &records.ledger(), Wait { interval: Duration::ZERO, polls: 2 }, |_| {}).unwrap(); + assert!(matches!(waited, Progress::Waiting(_)), "{waited:?}"); + assert_eq!((count(&ops, "cancel"), count(&ops, "status")), (0, 3)); + assert_eq!(records.ledger().get(ticket.handle()).unwrap().polls(), 3, "the checks are counted across processes"); + let Progress::Done(body) = follow(&t, &mut ticket, &key, &records.ledger(), Wait { interval: Duration::ZERO, polls: 20 }, |_| {}).unwrap() else { panic!("not done") }; + assert!(wai_quantum::engine::job::verify(&body).unwrap().0); + // A run keeps no record, so it does cancel: nothing goes on being spent. + let (t, ops, _) = watched(&Arc::new(Reference::new(Config { polls: 10, ..Config::default() }).unwrap())); + let e = run(&t, &req, false, &key, Wait { interval: Duration::ZERO, polls: 2 }, |_| {}).unwrap_err(); + assert!(e.to_string().contains("did not finish in time"), "{e}"); + assert_eq!(count(&ops, "cancel"), 1); + } + + #[test] + fn nothing_is_sent_that_could_not_be_recorded_first() { + let records = Records::new("unrecordable"); + std::fs::create_dir_all(&records.0).unwrap(); + let file = records.0.join("a-file"); + std::fs::write(&file, "not a directory").unwrap(); + let (t, ops, _) = watched(&Arc::new(Reference::new(Config::default()).unwrap())); + let e = start(&t, &json!({ "qasm": BELL, "provider": "lab" }), false, &Ledger::at(file.join("records"))).unwrap_err(); + assert!(e.to_string().contains("nothing was sent"), "{e}"); + assert_eq!(count(&ops, "submit"), 0); + } + + #[test] + fn a_job_that_may_have_been_sent_is_never_sent_again() { + let records = Records::new("unknown"); + let key = SigningKey::from_bytes(&[4; 32]); + let req = json!({ "qasm": BELL, "provider": "lab" }); + let (t, ops, unreachable) = watched(&Arc::new(Reference::new(Config::default()).unwrap())); + unreachable.lock().unwrap().push("submit"); + assert!(start(&t, &req, false, &records.ledger()).unwrap_err().to_string().contains("reset")); + unreachable.lock().unwrap().clear(); + let (tickets, unread) = records.ledger().list(); + assert!(unread.is_empty(), "{unread:?}"); + let mut ticket = tickets.into_iter().next().unwrap(); + assert_eq!(ticket.state(), "unknown"); + let job = ticket.job(); + assert_eq!(job["status"], "failed"); + assert!(job["error"].as_str().unwrap().contains("not sent again"), "{job}"); + assert!(matches!(check(&t, &mut ticket, &key, &records.ledger()).unwrap(), Progress::Ended(_))); + assert!(matches!(cancel(&t, &mut ticket, &key, &records.ledger()).unwrap(), Progress::Ended(_))); + assert_eq!((count(&ops, "submit"), count(&ops, "status"), count(&ops, "cancel")), (1, 0, 0)); + // A refusal is known not to have run. + let refusing = Local(|m: &Value| if m["op"] == "submit" { json!({ "ok": false, "error": { "code": "refused", "message": "not today" } }) } else { Reference::new(Config::default()).unwrap().handle(m, None) }); + let t = Target::new("lab", Kind::Process, Box::new(OpenProvider::new(refusing)), None); + assert!(start(&t, &req, false, &records.ledger()).is_err()); + assert!(records.ledger().list().0.iter().any(|t| t.state() == "failed" && !t.job()["error"].as_str().unwrap().contains("not sent again"))); + } + + #[test] + fn a_record_left_mid_send_says_so_and_asks_no_one() { + let records = Records::new("midsend"); + let (t, ops, _) = watched(&Arc::new(Reference::new(Config::default()).unwrap())); + let req = json!({ "qasm": BELL, "provider": "lab" }); + let p = plan(&t, &req).unwrap(); + let mut ticket = Ticket::recording("lab", &req, &p.program, 100, None, json!({})); + records.ledger().put(&ticket).unwrap(); + assert_eq!(ticket.state(), "submitting"); + assert!(matches!(check(&t, &mut ticket, &SigningKey::from_bytes(&[4; 32]), &records.ledger()).unwrap(), Progress::Ended(_))); + let job = ticket.job(); + assert!(job["error"].as_str().unwrap().contains("not known to have been sent") && job["error"].as_str().unwrap().contains("not sent again"), "{job}"); + assert_eq!((count(&ops, "submit"), count(&ops, "status")), (0, 0)); + } + + #[test] + fn a_failed_check_says_nothing_of_the_job_and_can_be_tried_again() { + let records = Records::new("flaky"); + let key = SigningKey::from_bytes(&[4; 32]); + let r = Arc::new(Reference::new(Config { polls: 1, ..Config::default() }).unwrap()); + let (t, ops, unreachable) = watched(&r); + let mut ticket = start(&t, &json!({ "qasm": BELL, "provider": "lab" }), false, &records.ledger()).unwrap(); + unreachable.lock().unwrap().push("status"); + assert!(check(&t, &mut ticket, &key, &records.ledger()).unwrap_err().to_string().contains("reset")); + assert_eq!(records.ledger().get(ticket.handle()).unwrap().state(), "submitted"); + // Within a bound it is tried again, and named if it is the last thing that happened. + let Progress::Waiting(job) = follow(&t, &mut ticket, &key, &records.ledger(), Wait { interval: Duration::ZERO, polls: 1 }, |_| {}).unwrap() else { panic!("not waiting") }; + assert!(job["why"].as_str().unwrap().contains("The last check failed"), "{job}"); + assert_eq!(count(&ops, "cancel"), 0); + unreachable.lock().unwrap().clear(); + // A failure to read the counts leaves the job to be collected too. + unreachable.lock().unwrap().push("result"); + assert!(check(&t, &mut ticket, &key, &records.ledger()).is_ok()); + assert!(check(&t, &mut ticket, &key, &records.ledger()).is_err()); + assert_eq!(records.ledger().get(ticket.handle()).unwrap().state(), "submitted"); + unreachable.lock().unwrap().clear(); + assert!(matches!(check(&t, &mut ticket, &key, &records.ledger()).unwrap(), Progress::Done(_))); + } + + #[test] + fn a_job_the_provider_has_forgotten_ends_with_why() { + let records = Records::new("forgotten"); + let r = Arc::new(Reference::new(Config { polls: 5, ..Config::default() }).unwrap()); + let mut ticket = start(&on(&r), &json!({ "qasm": BELL, "provider": "lab" }), false, &records.ledger()).unwrap(); + let elsewhere = on(&Arc::new(Reference::new(Config::default()).unwrap())); + let Progress::Ended(job) = check(&elsewhere, &mut ticket, &SigningKey::from_bytes(&[4; 32]), &records.ledger()).unwrap() else { panic!("not ended") }; + assert_eq!(job["status"], "failed"); + assert!(job["error"].as_str().unwrap().contains("no longer knows"), "{job}"); + assert_eq!(records.ledger().get(ticket.handle()).unwrap().state(), "failed"); + } + + #[test] + fn a_job_is_cancelled_only_by_asking_and_one_that_finished_first_gives_its_result() { + let records = Records::new("cancel"); + let key = SigningKey::from_bytes(&[4; 32]); + let req = json!({ "qasm": BELL, "provider": "lab" }); + let r = Arc::new(Reference::new(Config { polls: 50, ..Config::default() }).unwrap()); + let t = on(&r); + let mut ticket = start(&t, &req, false, &records.ledger()).unwrap(); + let Progress::Ended(job) = cancel(&t, &mut ticket, &key, &records.ledger()).unwrap() else { panic!("not ended") }; + assert_eq!(job["status"], "cancelled"); + assert_eq!(records.ledger().get(ticket.handle()).unwrap().state(), "cancelled"); + // Finished before anyone asked to cancel it. + let t = on(&Arc::new(Reference::new(Config::default()).unwrap())); + let mut ticket = start(&t, &req, false, &records.ledger()).unwrap(); + let Progress::Done(body) = cancel(&t, &mut ticket, &key, &records.ledger()).unwrap() else { panic!("not done") }; + assert!(wai_quantum::engine::job::verify(&body).unwrap().0); + } + + #[test] + fn a_recorded_run_is_refused_until_confirmed_and_records_nothing_until_then() { + let records = Records::new("priced"); + let key = SigningKey::from_bytes(&[4; 32]); + let t = on(&Arc::new(Reference::new(Config { price: Some(("0.0003".into(), "USD".into(), Per::Shot)), ..Config::default() }).unwrap())); + let req = json!({ "qasm": BELL, "shots": 1000, "provider": "lab" }); + assert!(matches!(start(&t, &req, false, &records.ledger()), Err(JobError::Unconfirmed { .. }))); + assert!(records.ledger().list().0.is_empty()); + let mut ticket = start(&t, &req, true, &records.ledger()).unwrap(); + let Progress::Done(body) = check(&t, &mut ticket, &key, &records.ledger()).unwrap() else { panic!("not done") }; + let cost = &body["result"]["cost"]; + assert_eq!((cost["free"].as_bool(), cost["confirmed"].as_bool(), cost["charged"]["amount"].as_str()), (Some(false), Some(true), Some("0.3")), "{cost}"); + } + + #[test] + fn a_program_reads_back_from_its_record_exactly_on_every_basis() { + let qft = "OPENQASM 3.0; qubit[3] q; bit[3] c; h q[0]; t q[1]; cx q[0], q[2]; rz(pi/8) q[2]; cx q[1], q[2]; c = measure q;"; + for basis in [Basis::RzSxXCz, Basis::PrxCz, Basis::GpiZz] { + let t = on(&Arc::new(Reference::new(Config { basis, ..Config::default() }).unwrap())); + let p = plan(&t, &json!({ "qasm": qft, "provider": "lab" })).unwrap().program; + let record = crate::open_provider::program_record(&p); + assert_eq!(crate::open_provider::program_from_record(&record).unwrap(), p, "{basis:?}"); + // A record that is not the program that was sent is not read as if it were. + let mut changed = record.clone(); + changed["qasm3"] = json!(format!("{}// another program\n", p.qasm3())); + let e = crate::open_provider::program_from_record(&changed).unwrap_err(); + assert!(e.contains("does not read back"), "{basis:?}: {e}"); + } + } + #[test] fn a_free_run_needs_no_confirmation_and_its_receipt_verifies() { let t = target(Config::default()); diff --git a/wai/wai-quantum-qpu/src/ledger.rs b/wai/wai-quantum-qpu/src/ledger.rs new file mode 100644 index 000000000..8f64aa597 --- /dev/null +++ b/wai/wai-quantum-qpu/src/ledger.rs @@ -0,0 +1,314 @@ +//! The record of a person's jobs on their providers: one file for each job, +//! in the configuration directory, readable by its owner only. +//! +//! A job on a device can wait in a queue for days. What waits for it should +//! hold nothing while it does, and lose nothing when it stops: so a job is +//! written down before it is sent ([`crate::job::start`]), with everything +//! sealing its result will need, and the provider's id beside that as soon as +//! the provider gives it. Whoever comes back later, in another process and +//! after another calibration, reads the record and collects the job, with the +//! program that was sent rather than one compiled again. +//! +//! A record is in one of these states: +//! +//! | state | what it means | +//! |---|---| +//! | `submitting` | written, and not yet known to have been sent. If it stays so, whatever was sending it stopped | +//! | `unknown` | the request to the provider did not complete, so the job may or may not be there | +//! | `submitted` | the provider took it, and gave its id | +//! | `done` | its counts were read and sealed; the sealed job is in the record | +//! | `failed` | the provider refused it or ended it | +//! | `cancelled` | it was cancelled | +//! +//! `submitting` and `unknown` are never sent again by this code: a second +//! submission could be a second bill. They say so, and the person looks at the +//! provider's own list. +//! +//! A record holds the request, the program and the provider's id. It never +//! holds a credential. + +use crate::open_provider; +use crate::registry::{config_dir, write_private}; +use serde_json::{Value, json}; +use std::path::{Path, PathBuf}; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{SystemTime, UNIX_EPOCH}; +use wai_quantum::quantum_toolchain::native::NativeProgram; + +/// A record's format. +pub const FORMAT: &str = "wai.quantum.provider-job/1"; + +const DIR_NAME: &str = "provider-jobs"; + +fn now_ms() -> u64 { + SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_millis() as u64) +} + +/// A handle no other record has: `pj-` and sixteen hexadecimal digits. +fn new_handle() -> String { + static NEXT: AtomicU64 = AtomicU64::new(0); + let nanos = SystemTime::now().duration_since(UNIX_EPOCH).map_or(0, |d| d.as_nanos()); + let seen = format!("{nanos}:{}:{}", std::process::id(), NEXT.fetch_add(1, Ordering::Relaxed)); + let h = blake3::hash(seen.as_bytes()); + let hex: String = h.as_bytes()[..8].iter().map(|b| format!("{b:02x}")).collect(); + format!("pj-{hex}") +} + +/// Whether `handle` is one this code gave: a handle comes back from an agent +/// or a command line and names a file, so nothing else may reach the path. +fn is_handle(handle: &str) -> bool { + handle.strip_prefix("pj-").is_some_and(|h| h.len() == 16 && h.bytes().all(|b| b.is_ascii_digit() || (b'a'..=b'f').contains(&b))) +} + +/// Where the records are kept. +pub struct Ledger { + dir: PathBuf, +} + +impl Ledger { + /// `WAI_QUANTUM_PROVIDER_JOBS`, or `provider-jobs` in the configuration + /// directory's `wai/`. + pub fn default_dir() -> Option { + std::env::var_os("WAI_QUANTUM_PROVIDER_JOBS").filter(|v| !v.is_empty()).map(PathBuf::from).or_else(|| config_dir().map(|d| d.join("wai").join(DIR_NAME))) + } + + /// The person's own records. + pub fn load_default() -> Result { + Ledger::default_dir().map(Ledger::at).ok_or_else(|| "no configuration directory: set XDG_CONFIG_HOME or HOME, or WAI_QUANTUM_PROVIDER_JOBS".to_string()) + } + + /// The records in `dir`. + pub fn at(dir: impl Into) -> Ledger { + Ledger { dir: dir.into() } + } + + pub fn dir(&self) -> &Path { + &self.dir + } + + /// Writes `ticket`, whole or not at all, on the disk before this returns. + pub fn put(&self, ticket: &Ticket) -> Result<(), String> { + let text = serde_json::to_string_pretty(&ticket.value).map_err(|e| e.to_string())? + "\n"; + let path = self.dir.join(format!("{}.json", ticket.handle())); + write_private(&path, text.as_bytes()).map_err(|e| format!("{}: {e}", path.display())) + } + + /// The record of `handle`. + pub fn get(&self, handle: &str) -> Result { + if !is_handle(handle) { + return Err(format!("{handle:?} is not a job's handle: they read pj- and sixteen hexadecimal digits (`wai-quantum providers jobs` lists them)")); + } + let path = self.dir.join(format!("{handle}.json")); + let text = std::fs::read_to_string(&path).map_err(|e| if e.kind() == std::io::ErrorKind::NotFound { format!("no job {handle} is recorded here ({})", self.dir.display()) } else { format!("{}: {e}", path.display()) })?; + Ticket::read(&text).map_err(|e| format!("{}: {e}", path.display())) + } + + /// Every record, the oldest first, with why any could not be read. + pub fn list(&self) -> (Vec, Vec) { + let (mut tickets, mut unread) = (Vec::new(), Vec::new()); + let Ok(dir) = std::fs::read_dir(&self.dir) else { return (tickets, unread) }; + for e in dir.flatten() { + let path = e.path(); + let handle = path.file_stem().and_then(|s| s.to_str()).unwrap_or_default(); + if path.extension().and_then(|x| x.to_str()) != Some("json") || !is_handle(handle) { + continue; + } + match self.get(handle) { + Ok(t) => tickets.push(t), + Err(why) => unread.push(why), + } + } + tickets.sort_by_key(|t| (t.submitted_ms(), t.handle().to_string())); + (tickets, unread) + } +} + +/// One job's record. +#[derive(Clone, Debug, PartialEq)] +pub struct Ticket { + value: Value, +} + +impl Ticket { + /// A job not yet sent: `sealing` is what its receipt will say of it. + pub(crate) fn recording(provider: &str, request: &Value, program: &NativeProgram, shots: u64, seed: Option, sealing: Value) -> Ticket { + Ticket { + value: json!({ + "format": FORMAT, + "handle": new_handle(), + "provider": provider, + "state": "submitting", + "submitted_ms": now_ms(), + "request": request, + "program": open_provider::program_record(program), + "shots": shots, + "seed": seed, + "sealing": sealing, + "polls": 0, + }), + } + } + + fn read(text: &str) -> Result { + let value: Value = serde_json::from_str(text).map_err(|e| format!("not a job's record: {e}"))?; + if value["format"] != FORMAT { + return Err(format!("not a job's record: its format is not {FORMAT}")); + } + for k in ["handle", "provider", "state"] { + if !value[k].is_string() { + return Err(format!("not a job's record: it has no {k}")); + } + } + Ok(Ticket { value }) + } + + pub fn handle(&self) -> &str { + self.value["handle"].as_str().unwrap_or_default() + } + + /// The person's name for the provider. + pub fn provider(&self) -> &str { + self.value["provider"].as_str().unwrap_or_default() + } + + /// The state: see the module's table. + pub fn state(&self) -> &str { + self.value["state"].as_str().unwrap_or_default() + } + + pub fn submitted_ms(&self) -> u64 { + self.value["submitted_ms"].as_u64().unwrap_or(0) + } + + pub fn request(&self) -> &Value { + &self.value["request"] + } + + pub(crate) fn sealing(&self) -> &Value { + &self.value["sealing"] + } + + /// The provider's id for the job, once it has given one. + pub(crate) fn provider_id(&self) -> Option<&str> { + self.value["id"].as_str() + } + + pub(crate) fn shots(&self) -> u64 { + self.value["shots"].as_u64().unwrap_or(0) + } + + pub(crate) fn polls(&self) -> u32 { + self.value["polls"].as_u64().unwrap_or(0) as u32 + } + + pub(crate) fn polled(&mut self) { + self.value["polls"] = json!(self.polls() + 1); + } + + /// The program that was sent, exactly: the one to read counts against, + /// whatever the device has become since. + pub(crate) fn program(&self) -> Result { + open_provider::program_from_record(&self.value["program"]) + } + + pub(crate) fn submitted(&mut self, id: &str) { + self.value["state"] = json!("submitted"); + self.value["id"] = json!(id); + } + + /// The job ended without a result: `state` is `failed`, `cancelled` or + /// `unknown`, and `why` has had any credential taken out. + pub(crate) fn ended(&mut self, state: &str, why: &str) { + self.value["state"] = json!(state); + self.value["error"] = json!(why); + self.value["finished_ms"] = json!(now_ms()); + } + + pub(crate) fn done(&mut self, sealed: Value) { + self.value["state"] = json!("done"); + self.value["sealed"] = sealed; + self.value["finished_ms"] = json!(now_ms()); + } + + /// The job in a list: what it is and where it stands, without its program, + /// its result or the provider's id. + pub fn listing(&self) -> Value { + let job = self.job(); + let mut v = json!({ "id": self.handle(), "provider": self.provider(), "status": job["status"], "shots": self.value["shots"], "submitted_ms": self.value["submitted_ms"] }); + for k in ["finished_ms", "error"] { + if !job[k].is_null() { + v[k] = job[k].clone(); + } + } + v + } + + /// Whether more can come of this job. + pub fn waiting(&self) -> bool { + self.state() == "submitted" + } + + /// The job in the form every host gives a job (`job.v1.json`), its `id` + /// the handle: finished with its result and receipt, still running, or + /// ended with why. + pub fn job(&self) -> Value { + let times = |mut j: Value| { + j["id"] = json!(self.handle()); + j["submitted_ms"] = self.value["submitted_ms"].clone(); + if !self.value["finished_ms"].is_null() { + j["finished_ms"] = self.value["finished_ms"].clone(); + } + j + }; + let ended = |status: &str, error: String| times(json!({ "status": status, "method": "provider", "request": self.request(), "error": error })); + match self.state() { + "done" => times(self.value["sealed"].clone()), + "submitted" => times(json!({ + "status": "running", + "method": "provider", + "why": format!("at {}, queued or running: nothing is held here while it waits. Ask again later with this id, for as long as the provider keeps it", self.provider()), + "request": self.request(), + })), + "cancelled" => ended("cancelled", self.value["error"].as_str().unwrap_or("cancelled").to_string()), + "failed" => ended("failed", self.value["error"].as_str().unwrap_or("it failed").to_string()), + "unknown" => ended("failed", format!("{}. It may have reached {}, and it was not sent again: look in the provider's own list", self.value["error"].as_str().unwrap_or("the request did not complete"), self.provider())), + _ => ended("failed", format!("it was recorded and not known to have been sent: whatever was sending it stopped, unless that is happening now. It may have reached {}, and it was not sent again: look in the provider's own list", self.provider())), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + fn dir(tag: &str) -> PathBuf { + let d = std::env::temp_dir().join(format!("wai-qpu-ledger-{tag}-{}", std::process::id())); + let _ = std::fs::remove_dir_all(&d); + d + } + + #[test] + fn a_handle_cannot_reach_outside_the_records() { + let d = dir("handles"); + let l = Ledger::at(&d); + for bad in ["../../etc/passwd", "pj-", "pj-0123456789abcde", "pj-0123456789abcdeg", "pj-0123456789ABCDEF", "pj-0123456789abcdef/../x", "", "0123456789abcdef"] { + assert!(l.get(bad).is_err_and(|e| e.contains("is not a job's handle")), "{bad:?}"); + } + assert!(l.get("pj-0123456789abcdef").is_err_and(|e| e.contains("no job"))); + let (a, b) = (new_handle(), new_handle()); + assert!(is_handle(&a) && a != b, "{a} {b}"); + } + + #[test] + fn a_file_that_is_not_a_record_is_said_to_be_unread_and_stops_nothing() { + let d = dir("unread"); + std::fs::create_dir_all(&d).unwrap(); + std::fs::write(d.join("pj-0123456789abcdef.json"), "{ not json").unwrap(); + std::fs::write(d.join("notes.txt"), "kept beside them").unwrap(); + let (tickets, unread) = Ledger::at(&d).list(); + assert!(tickets.is_empty()); + assert_eq!(unread.len(), 1, "{unread:?}"); + assert!(Ledger::at(d.join("never-made")).list().0.is_empty()); + let _ = std::fs::remove_dir_all(&d); + } +} diff --git a/wai/wai-quantum-qpu/src/lib.rs b/wai/wai-quantum-qpu/src/lib.rs index 3983be2ab..d3b415545 100644 --- a/wai/wai-quantum-qpu/src/lib.rs +++ b/wai/wai-quantum-qpu/src/lib.rs @@ -53,6 +53,7 @@ pub mod conformance; pub mod job; +pub mod ledger; pub mod sampler_v2; pub mod native_turns; pub mod open_provider; diff --git a/wai/wai-quantum-qpu/src/open_provider.rs b/wai/wai-quantum-qpu/src/open_provider.rs index dc58b35ba..2098fa1b2 100644 --- a/wai/wai-quantum-qpu/src/open_provider.rs +++ b/wai/wai-quantum-qpu/src/open_provider.rs @@ -451,6 +451,45 @@ pub fn program_json(p: &NativeProgram) -> Value { json!({ "format": NATIVE_FORMAT, "basis": basis_label(p.basis), "bits": p.n_qubits, "gates": gates, "measure": measure, "qasm3": p.qasm3() }) } +/// A compiled program as a job's record keeps it (`crate::ledger`): the wire +/// form, and what a provider is not sent but reading its counts back may need. +pub(crate) fn program_record(p: &NativeProgram) -> Value { + let mut v = program_json(p); + v["layout"] = json!(p.layout); + v["swaps"] = json!(p.swaps); + v["residual"] = p.residual.iter().map(|&(q, a)| json!({ "qubit": q, "turns": turns_json(a) })).collect(); + v +} + +/// The program of a record, and only if it reads back as the program that was +/// recorded: its OpenQASM 3 text is what its receipt's hash is of. +pub(crate) fn program_from_record(v: &Value) -> Result { + let w = program_from_json(v)?; + let qubits = |k: &str| -> Result, String> { + v[k].as_array().ok_or(format!("a recorded program's {k} is a list"))?.iter().map(|q| q.as_u64().and_then(|q| u16::try_from(q).ok()).ok_or(format!("a recorded program's {k} holds qubits"))).collect() + }; + let mut residual = Vec::new(); + for r in v["residual"].as_array().ok_or("a recorded program's residual is a list")? { + let q = r["qubit"].as_u64().and_then(|q| u16::try_from(q).ok()).ok_or("a residual names its qubit")?; + let (m, k) = (r["turns"][0].as_u64(), r["turns"][1].as_u64().and_then(|k| u8::try_from(k).ok())); + let a = m.zip(k).and_then(|(m, k)| Dyadic::new(m, k)).ok_or("a residual's turns are [m, k]")?; + residual.push((q, a)); + } + let p = NativeProgram { + basis: w.basis, + n_qubits: w.bits, + layout: qubits("layout")?, + final_layout: w.measure, + gates: w.gates, + residual, + swaps: v["swaps"].as_u64().ok_or("a recorded program counts its swaps")? as usize, + }; + if v["qasm3"].as_str() != Some(p.qasm3().as_str()) { + return Err("the recorded program does not read back as the program that was sent".into()); + } + Ok(p) +} + /// A program as a provider reads it from a message. #[derive(Clone, Debug, PartialEq, Eq)] pub struct WireProgram { diff --git a/wai/wai-quantum-qpu/src/registry.rs b/wai/wai-quantum-qpu/src/registry.rs index 7c704c03b..b2828e15f 100644 --- a/wai/wai-quantum-qpu/src/registry.rs +++ b/wai/wai-quantum-qpu/src/registry.rs @@ -242,7 +242,7 @@ pub fn config_dir() -> Option { /// Writes `text` to `path` readable by its owner only, through a file beside /// it renamed into place. -fn write_private(path: &Path, text: &[u8]) -> std::io::Result<()> { +pub(crate) fn write_private(path: &Path, text: &[u8]) -> std::io::Result<()> { if let Some(dir) = path.parent() { std::fs::create_dir_all(dir)?; } -- 2.54.0 (Apple Git-157) From e1857efb53ee5586040d8f1fe3e3d853e55b49c9 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 2 Nov 2026 19:07:10 -0500 Subject: [PATCH 2/2] wai-quantum-qpu.yml: the library has 46 tests, and the record's promises are named The library's pinned count goes from 34 to 46, for the twelve tests the provider job record adds. Five of them are named so CI fails if any stops running: a recorded job outlives what sent it, running out of patience does not cancel it, nothing is sent that was not recorded, a job that may have been sent is never sent again, and a handle reaches no file but a record's. Co-Authored-By: Claude Opus 5.5 --- .github/workflows/wai-quantum-qpu.yml | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/.github/workflows/wai-quantum-qpu.yml b/.github/workflows/wai-quantum-qpu.yml index 7afbdd509..06f581c58 100644 --- a/.github/workflows/wai-quantum-qpu.yml +++ b/.github/workflows/wai-quantum-qpu.yml @@ -79,15 +79,18 @@ jobs: # tests never touch one but the macOS leg's own. export WAI_QUANTUM_CREDENTIALS=file cargo test --locked 2>&1 | tee "$RUNNER_TEMP/test.log" - # The library (34), the clients against their contracts (8), and the + # The library (46), the clients against their contracts (8), and the # providers end to end (12), each binary's count pinned. - for n in 34 8 12; do + for n in 46 8 12; do grep -qE "^test result: ok\. $n passed; 0 failed" "$RUNNER_TEMP/test.log" || { echo "::error::a test binary did not pass all $n of its tests"; exit 1; } done for t in curl_carries_a_request_to_a_server_and_back every_operation_over_http every_operation_over_a_local_process \ the_reference_conforms_over_both_transports the_reference_gives_its_pinned_counts_on_every_platform \ a_run_that_is_not_free_is_refused_until_confirmed_and_carries_its_cost \ - a_credential_reaches_the_provider_and_nothing_a_run_returns; do + a_credential_reaches_the_provider_and_nothing_a_run_returns \ + job::tests::a_recorded_job_outlives_what_sent_it_and_seals_as_a_run_would job::tests::running_out_of_patience_does_not_cancel_the_job \ + job::tests::nothing_is_sent_that_could_not_be_recorded_first job::tests::a_job_that_may_have_been_sent_is_never_sent_again \ + ledger::tests::a_handle_cannot_reach_outside_the_records; do grep -q "^test $t \.\.\. ok$" "$RUNNER_TEMP/test.log" || { echo "::error::$t did not run and pass"; exit 1; } done - name: Clippy -- 2.54.0 (Apple Git-157)