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 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)?; }