From fa83d4742e5150a7c6e3c62dc3b9500392a0da2b Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 19 Oct 2026 07:31:05 -0400 Subject: [PATCH 1/2] jcp-gateway: a task's exchanges and an interim result's retry are decided as continuations of the call MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit MCP 2026-07-28 lets a gated call answer before its work is done, and both ways it does so went around the gate. A task handle (resultType "task"): the client fetches the outcome with tasks/get, answers the server with tasks/update, stops it with tasks/cancel. All three passed through as discovery. The outcome, which is the tool's actual output, reached the model unmetered and unsealed and was never admitted to the flow store, so a value read through a task could be sent on by a later call with nothing to fold it against. And any task id at all was relayed, whoever's it was. An interim result (resultType "input_required"): the client repeats the call with its answers in inputResponses. Only `arguments` were folded, so an answer carrying a value from an untrusted read reached the server past the flow ceiling. Now: - A task handle is bound to the grant of the call it came back from. Each tasks/* exchange naming it is decided against that grant at the moment it is made (revoked, expired or spent refuses it), metered, debited, and sealed over a payload that commits to the receipt it continues. Its reply is admitted as untrusted data under the join of everything the client has sent the task. The request sent upstream is rebuilt from the fields decided over, so nothing else the client adds reaches the server. - The tasks/get that brings the outcome closes the task: one zero-cost result frame (tool-result, data) sealed over the content address of a closing record that names the opening receipt, every exchange's receipt, the joules metered across them and the outcome's digest. The record is Partial: the server works between exchanges, in no span the gateway reads. - A tasks/* exchange naming a task this gateway did not open, or has closed, is refused, sealed, and never reaches the server. - A retry's answers (inputResponses, requestState) are part of its sealed payload and are flow-checked with the arguments. The payload commits to the receipt of the interim result it answers. - Handles the server minted (a task id, a request key, a requestState) are left out of the flow check only when they are an exact echo of what the gateway saw the server mint. A 36-byte UUID would otherwise taint every poll under a ceiling, since the scan window is 16 bytes. A handle the client made up is checked. - server/discover carries the advertisement, as initialize does. The advertisement lists the task methods and `continuations: 1`. - Open tasks and pending interim results are bounded (1024 each). SPEC §6.1 states the rule in protocol-neutral terms. §10 gains a row for it. Known limits, stated in the code: - Open tasks are kept in memory. After a restart, polls of earlier tasks are refused (fail closed). - Work the server does between exchanges is not metered. - Server-pushed notifications/tasks are outside the request/response relay. Tests: 12 new, against scripted upstreams. All 19 mutations of the new logic are killed. Co-Authored-By: Claude Opus 5.5 --- jcp/SPEC.md | 48 + .../crates/jcp-gateway/src/bin/jcp_gateway.rs | 7 +- jcp/jcp-rs/crates/jcp-gateway/src/lib.rs | 1181 ++++++++++++++++- 3 files changed, 1192 insertions(+), 44 deletions(-) diff --git a/jcp/SPEC.md b/jcp/SPEC.md index 7f6a93373..543a148de 100644 --- a/jcp/SPEC.md +++ b/jcp/SPEC.md @@ -1049,6 +1049,53 @@ before the field existed. exercise any capability from a `data` frame (§7), regardless of its content. Tool and agent outputs MUST be tagged `data`. +### 6.1 Continuations: work that answers later (REQUIRED where an enforcement point relays it) + +An endpoint may answer a request before its work is done. It may return a +**handle** to work that goes on, whose outcome the client fetches in later +exchanges that name the handle (MCP's tasks; an A2A task). Or it may return an +**interim result** asking the client for input, which the client answers by +repeating the request with its answers attached (MCP's `input_required`). Each +later exchange moves the first request's work and the first request's kind of +payload: the outcome is the action's output, and the answers go to the +endpoint as its arguments do. So an enforcement point that relays such +exchanges and treats the later ones as discovery has a door beside its gate. +Such an enforcement point: + +1. MUST decide each later exchange as a **continuation** of the first: against + the grant the first was decided on, at the moment the later exchange is + made, so a grant revoked, expired or spent since refuses it (§7), and the + outcome of work the grant no longer covers is not fetched under it. It MUST + seal each continuation's decision (§8) over a payload that commits to the id + of the receipt it continues (the first's, or the previous continuation's), + so that one piece of work's receipts form a chain within the stream. +2. MUST refuse, and seal the refusal of, a later exchange naming a handle it did + not see returned to a request it decided. A handle is a bearer string the + endpoint minted, and no grant covers it until the enforcement point has bound + it to one. A handle it has closed (item 4) or forgotten (a restart, a bound on + what it keeps) is refused the same way: fail closed. +3. MUST flow-check what the client authored in a later exchange (the answers) + as it does the first request's arguments (§4.10), and MUST admit what comes + back — status text and the outcome alike — as `data`, labelled as the output + of the first request's action and joined with the label of everything the + client has sent the work. An exact echo of a handle or state the endpoint + minted MAY be left out of the flow check, being the endpoint's own text; + anything else the client sends, a handle it made up included, MUST be + checked. What the enforcement point forwards SHOULD be built from the fields + it decided over, so that nothing it did not check reaches the endpoint. +4. MUST meter and debit each later exchange (§5), and, when the outcome is + final, MUST seal one **closing record**: a `tool-result` (or `agent-reply`) + frame tagged `data`, debiting nothing, whose payload is the content address + (§3) of a record naming the first receipt, every continuation's receipt, the + joules metered across them, and the outcome's digest. The endpoint works + between exchanges, inside no exchange's span, so the record MUST mark that + sum `Partial` (§5.2) unless the meter read the work between them. + +The reference gateway (`jcp-gateway`) does this for MCP's `tasks/get`, +`tasks/update` and `tasks/cancel`, and for a `tools/call`, `resources/read` or +`prompts/get` retried with `inputResponses`; it advertises it in its +`server/discover` and `initialize` responses as `continuations`. + ## 7. The decision (REQUIRED) Given a Frame, its [SignedGrant](#4-the-grant-required), the caller's DID, a @@ -1212,6 +1259,7 @@ jcp-conformance --bin gen-vectors`). | Runaway loops, no energy accounting | §5 enforced joule budgets, conserved across delegation | | Mutable / optional audit | §8 signed JCR-1 receipts | | Fragmentation across protocols | §6 one frame for tools and agents | +| Results fetched later, outside the gate (task handles, interim input requests) | §6.1 continuations, chained to the receipt they continue | ## 11. Wire binding: `jcp/1` over a byte stream (OPTIONAL) diff --git a/jcp/jcp-rs/crates/jcp-gateway/src/bin/jcp_gateway.rs b/jcp/jcp-rs/crates/jcp-gateway/src/bin/jcp_gateway.rs index f22c3f4c2..d0b52ef46 100644 --- a/jcp/jcp-rs/crates/jcp-gateway/src/bin/jcp_gateway.rs +++ b/jcp/jcp-rs/crates/jcp-gateway/src/bin/jcp_gateway.rs @@ -8,9 +8,10 @@ //! //! The gateway speaks newline-delimited JSON-RPC with its own client on //! stdin/stdout and with the spawned MCP server over the child's stdio. Every -//! `tools/call` is gated, metered, debited, and sealed; receipts stream to -//! `JCP_GATEWAY_RECEIPTS` (default `./jcp-gateway-receipts.cose`, length- -//! prefixed envelopes) and a one-line summary per call goes to stderr. +//! method in `JCP_ENFORCED_METHODS` is gated, metered, debited, and sealed (a +//! task's exchanges as continuations of the call that opened it); receipts +//! stream to `JCP_GATEWAY_RECEIPTS` (default `./jcp-gateway-receipts.cose`, +//! length-prefixed envelopes) and a one-line summary per call goes to stderr. //! //! That file is also read back at startup, because SPEC §8.1 is REQUIRED of //! an enforcement point: it rebuilds its budget state by replaying its own diff --git a/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs b/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs index 6aca29c49..5b4ef797e 100644 --- a/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs +++ b/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs @@ -1,11 +1,14 @@ //! The energy-native MCP gateway. //! //! An MCP client speaks to an MCP server through this gateway instead of -//! directly. Discovery traffic (`initialize`, the `*/list` methods, -//! notifications) passes through untouched. But every `tools/call`, +//! directly. Discovery traffic (`initialize`, `server/discover`, the `*/list` +//! methods, notifications) passes through untouched. But every `tools/call`, //! `resources/read` and `prompts/get` is a **JCP enforcement point** — the //! last two because each moves a payload the model then reads, and each used -//! to pass through unbudgeted, unmetered and unsealed: +//! to pass through unbudgeted, unmetered and unsealed. So is every exchange +//! that carries one of those calls on after its first reply (SPEC §6.1): the +//! `tasks/get`, `tasks/update` and `tasks/cancel` of a task the server opened +//! in answer to one, and the retry that answers an `input_required` result: //! //! 1. The tool must have a capability grant; the frame is content-addressed to //! `(server, tool)`, so a grant for `read_file` cannot authorize `shell`. @@ -55,6 +58,15 @@ //! 5. Every decision — allow and deny — seals into a **hash-chained receipt //! log** ([`jcp_receipt::ReceiptLog`]): a tamper-evident, non-repudiable //! record of what the agent did and what it cost. +//! 6. A call whose work goes on after its reply is followed through it +//! ([`Gateway::relay`] says how): each later exchange is decided against +//! the call's grant at the moment it is made, metered, debited and sealed +//! as a continuation of the receipt before it; what the client sends in +//! one is flow-checked like the call's arguments, and what comes back is +//! admitted as untrusted data like the call's result; and a task that +//! reaches its outcome is closed by one sealed record of all of it. An +//! exchange that names a task this gateway did not open under a grant is +//! refused and sealed. //! //! This is the chokepoint the AI-gateway field occupies (LiteLLM, Bifrost, //! Portkey, et al.) rebuilt on the family's two unclaimed axes: the unit of @@ -74,7 +86,7 @@ use jcp_flow::{FlowStore, Mode, Witness}; use jcp_grant::issue; use jcp_receipt::{FlowSeal, ReceiptLog, ResumeError}; use jcp_runtime::Runtime; -use jcp_schema::flow::{FlowCeiling, Readers}; +use jcp_schema::flow::{FlowCeiling, Label, Readers}; use jcp_schema::{Capability, Caveat, ContentId, Coverage, Did, Energy, FlowFacts, Grant, SignedGrant}; use sandbox_meter::{Measurement, PowerMeter, SpanEnd, TreeError, TreeGap, TreeMeter, TreeReading}; use sandbox_schema::DeviceClass; @@ -324,6 +336,93 @@ pub struct Gateway { /// The witnesses this session emitted, by id — the disclosure store an auditor /// fetches to re-derive a sealed verdict. witnesses: BTreeMap, + /// The tasks the server opened in answer to a gated call, by the id it + /// minted for each: the only ones a `tasks/*` exchange may name. + tasks: BTreeMap, + /// How many tasks have been opened, for evicting the oldest. + tasks_opened: u64, + /// Gated calls the server answered `input_required`, by action and + /// arguments: the receipt a retry carrying the answers continues, and + /// the handles the server minted for it. + awaiting_input: BTreeMap<(String, ContentId), Interim>, +} + +/// An interim result a gated call got back (`resultType: "input_required"`): +/// its receipt, which the retry answering it continues, and the handles the +/// server minted in it, which the retry hands back. +#[derive(Debug, Clone)] +struct Interim { + receipt_id: ContentId, + /// The `requestState` the server handed back, if any. + request_state: Option, + /// The keys of the requests in its `inputRequests`. + keys: BTreeSet, +} + +/// A task the server opened in answer to a gated call (MCP's tasks +/// extension: the reply was a handle, `resultType: "task"`, and the outcome +/// is fetched later with `tasks/get`). +#[derive(Debug, Clone)] +struct OpenTask { + /// The action the call was gated as: the grant every exchange of the + /// task is decided against. + action: String, + /// The receipt of the call that opened it. + opened_by: ContentId, + /// The receipt of each exchange since, in order. + legs: Vec, + /// The joules metered across the task: the opening call and every + /// exchange since. + metered_uj: u64, + /// The flow label of everything the client has sent into the task: what + /// it sends later carries on from the call's arguments, so the outcome + /// is admitted under the join of them all. + label: Label, + /// The keys of every request the server has put to the client in the + /// task (`inputRequests`), which a `tasks/update` hands back. + request_keys: BTreeSet, + /// When it was opened, counted in tasks: the oldest is evicted first. + seq: u64, +} + +impl OpenTask { + /// The receipt the next exchange continues: the last one sealed. + fn last(&self) -> ContentId { + self.legs.last().copied().unwrap_or(self.opened_by) + } +} + +/// The most tasks one gateway keeps open, and the most interim results it +/// keeps waiting for answers. A server that opens tasks it never finishes +/// would otherwise grow these without bound. Past it the oldest task is +/// dropped, and an exchange naming it is refused, as one naming an unknown +/// task is; and an interim result is dropped, whose retry is then decided +/// as a call with nothing to continue. +const MAX_OPEN: usize = 1024; + +/// The task methods (MCP's tasks extension) a client uses to follow a task +/// it was handed: each is a continuation of the call that opened it. +const TASK_METHODS: [&str; 3] = ["tasks/get", "tasks/update", "tasks/cancel"]; + +/// Why every task's closing record is `Partial`: the server works on a task +/// between the exchanges, inside no exchange's span, and no meter here reads +/// that work. +const BETWEEN_EXCHANGES: &str = "between-exchanges"; + +/// One gated exchange the gateway authorized, metered, debited, sealed, and +/// whose reply it admitted as a labelled source. +struct Leg { + /// The upstream's reply as it came back, before any `_meta` is added. + response: String, + metered: Metered, + budget_remaining_uj: u64, + /// The flow label of what the client sent. + label: Label, + /// The receipt sealed for it, and that receipt's id (its envelope's + /// content id, which the next entry's `prev` names): what a + /// continuation of this exchange commits to. + receipt: Vec, + receipt_id: ContentId, } /// The tree meter that reads a call to an upstream naming `pid`: the one @@ -381,6 +480,9 @@ impl Gateway { flow: FlowStore::new(4096), mode: Mode::Explicit, witnesses: BTreeMap::new(), + tasks: BTreeMap::new(), + tasks_opened: 0, + awaiting_input: BTreeMap::new(), } } @@ -561,8 +663,59 @@ impl Gateway { } /// Relay one client message to the upstream under policy. The single entry - /// point: classifies the message, gates `tools/call`, meters and debits, - /// seals the receipt, and returns what to send the client. + /// point: classifies the message, gates `tools/call`, `resources/read` and + /// `prompts/get`, meters and debits, seals the receipt, and returns what to + /// send the client. + /// + /// # Work that answers later (SPEC §6.1) + /// + /// MCP's 2026-07-28 revision lets a gated call's first reply be something + /// other than its result, and each such reply opened a door: + /// + /// - **A task handle** (`resultType: "task"`, the tasks extension): the + /// server works on, and the client fetches the outcome with `tasks/get`, + /// answers the server's requests with `tasks/update`, or stops it with + /// `tasks/cancel`. Those passed through as discovery. The outcome, which + /// is the tool's actual output, reached the model unmetered and + /// unsealed, and was never admitted to the flow store, so a value read + /// through a task could be sent on by a later call with nothing to fold + /// it against. And any task id at all was relayed, whoever's it was. + /// Now the handle is bound to the call's grant when it comes back, and + /// each `tasks/*` exchange naming it is a continuation of the call: + /// decided against that grant at the moment it is made (so a grant + /// revoked, expired or spent since stops the outcome being fetched + /// under it), sealed over a payload naming the method, the task, any + /// answers and the id of the receipt it continues, flow-checked over + /// what the client authored in it as a call's arguments are, metered + /// and debited, and its reply admitted as untrusted data under the + /// join of everything the client has sent the task. What goes + /// upstream is rebuilt from those fields, so nothing else the client + /// puts in the request reaches the server. The `tasks/get` that brings the outcome also closes the + /// task with one more record: the task's result frame (`tool-result`, + /// `data`) sealed at zero cost, whose payload is the content address + /// (§3) of the closing record returned in the reply's `_meta`, which + /// names the opening receipt, every exchange's receipt, the joules + /// metered across them, and the outcome's digest. It is `Partial` + /// always: the server works between the exchanges, in no span this + /// gateway reads. A `tasks/*` exchange naming a task this gateway did + /// not open, or has closed, is refused and sealed, and never reaches + /// the server. + /// - **An interim result** (`resultType: "input_required"`): the server + /// asks the client for input, and the client repeats the call with its + /// answers in `inputResponses`. Only `arguments` were folded, so the + /// answers went upstream unscanned, and an answer carrying a value + /// derived from an untrusted read reached the server past the flow + /// ceiling. Now the answers, and the server's `requestState`, are part + /// of the retry's payload, sealed with the arguments, and what the + /// client authored in them is folded with the arguments (`authored`: + /// a handle the server minted, handed back unchanged, is not); and the + /// retry's payload commits to the receipt of the interim result it + /// answers, so a call's rounds form one chain in the log. + /// + /// What a task or an interim result waits on is kept in memory, so a + /// restarted gateway knows none of them: an exchange naming a task opened + /// before the restart is refused (fail closed), and a retry after one is + /// decided as a call with nothing to continue. pub fn relay(&mut self, line: &str, up: &mut impl Upstream) -> std::io::Result { let msg: Value = match serde_json::from_str(line) { Ok(v) => v, @@ -573,6 +726,9 @@ impl Gateway { let id = msg.get("id"); let params = msg.get("params").cloned().unwrap_or(Value::Null); + if TASK_METHODS.contains(&method) { + return self.relay_task(up, method, id, ¶ms); + } let Some((action, args)) = gated_action(method, ¶ms) else { if id.is_none() || msg.get("method").is_none() { // A notification (no `id`), or the client's response to a @@ -582,10 +738,11 @@ impl Gateway { return Ok(Relayed::Notified); } let response = up.call(line)?; - // `initialize` is where a client learns what it is talking to, so - // it is the one discovery method this gateway does not relay - // verbatim: it adds its own advertisement and changes nothing else. - if method == "initialize" { + // `initialize`, and `server/discover` where MCP 2026-07-28 moved + // it, are where a client learns what it is talking to, so they + // are the discovery methods this gateway does not relay verbatim: + // it adds its own advertisement and changes nothing else. + if method == "initialize" || method == "server/discover" { return Ok(Relayed::PassedThrough(advertise_extension( &response, jcp_receipt::PROFILE_FLOW, @@ -602,14 +759,39 @@ impl Gateway { // ---- A gated method: the enforcement point. let tool = action.as_str(); + // A retry answering an interim result carries the answers beside the + // arguments. They go to the server, so they are payload; and the + // interim they answer is found by the action and the arguments alone. + let (args, answers) = split_answers(method, args, ¶ms); let args_bytes = serde_json::to_vec(&args).unwrap_or_default(); + let round = (action.clone(), ContentId::of_bytes(&args_bytes)); + let interim = answers.as_ref().and_then(|_| self.awaiting_input.get(&round).cloned()); + let continues = interim.as_ref().map(|i| i.receipt_id); + let (payload, scanned) = match answers { + None => (args_bytes.clone(), args_bytes), + Some(mut p) => { + let none = BTreeSet::new(); + let (keys, state) = match &interim { + Some(i) => (&i.keys, i.request_state.as_ref()), + None => (&none, None), + }; + let answers = authored(p.get("inputResponses"), p.get("requestState"), keys, state); + let scanned = serde_json::to_vec(&json!({"arguments": &args, "answers": answers})) + .unwrap_or_default(); + p.insert("arguments".into(), args); + if let Some(c) = continues { + p.insert("continues".into(), json!(c.to_hex())); + } + (serde_json::to_vec(&Value::Object(p)).unwrap_or_default(), scanned) + } + }; // A gated method sent as a notification (no `id`) cannot be answered: // its result and its receipt could reach no one, and no reply bounds // the span it would be metered over. So it is refused before anything // else and never reaches the server, and the attempt is sealed. if id.is_none() { - return Ok(self.deny_notification(tool, &args_bytes)); + return Ok(self.deny_notification(tool, &payload)); } // No grant for this tool ⇒ deny. We still SEAL the denial, using any @@ -625,21 +807,249 @@ impl Gateway { if self.tool_schema_changed(tool) { return Ok(self.deny_schema_changed(tool, id)); } + + let exchange = Exchange { + line, + id, + tool, + payload: &payload, + scanned: &scanned, + declared: declared_witness_ids(¶ms), + carried: &Label::bottom(), + }; + let leg = match self.enforce(up, &sg, exchange)? { + Ok(leg) => leg, + Err(denied) => return Ok(denied), + }; + + // The receipt for THIS call is the one just appended, so the caller + // receives evidence for its own request rather than having to trust + // that the gateway logged something somewhere. + let response = attach_receipt(&leg.response, &leg.receipt, &leg.metered, leg.budget_remaining_uj); + let mut note = serde_json::Map::new(); + if let Some(c) = continues { + note.insert("continues".into(), json!(c.to_hex())); + } + match reply_kind(&leg.response) { + Reply::Task(task_id) => { + note.insert( + "task".into(), + json!({"taskId": task_id, "opened_by": leg.receipt_id.to_hex()}), + ); + self.open_task(task_id, tool, &leg); + } + Reply::InputRequired => { + if self.awaiting_input.len() >= MAX_OPEN && !self.awaiting_input.contains_key(&round) { + self.awaiting_input.pop_first(); + } + let (request_state, keys) = minted(&leg.response); + self.awaiting_input.insert(round, Interim { receipt_id: leg.receipt_id, request_state, keys }); + note.insert("input_required".into(), json!({"continue_from": leg.receipt_id.to_hex()})); + } + Reply::Final => { + if continues.is_some() { + self.awaiting_input.remove(&round); + } + } + } + + Ok(Relayed::Allowed { + response: annotate(&response, note), + cost_uj: leg.metered.cost.uj(), + budget_remaining_uj: leg.budget_remaining_uj, + }) + } + + /// A `tasks/*` exchange: a continuation of the gated call that opened the + /// task it names ([`relay`](Self::relay) says what that means). + fn relay_task( + &mut self, + up: &mut impl Upstream, + method: &str, + id: Option<&Value>, + params: &Value, + ) -> std::io::Result { + let task_id = params.get("taskId").and_then(Value::as_str).unwrap_or("").to_string(); + // A task id is a bearer string the server minted. No grant covers one + // until this gateway has seen it come back from a call it gated, so + // any other is refused here, sealed, and never reaches the server. + let Some(task) = self.tasks.get(&task_id).cloned() else { + let key = format!("{method}:{task_id}"); + if id.is_none() { + return Ok(self.deny_notification(&key, b"")); + } + return Ok(self.deny_sealed(&key, id, "no task opened under a grant")); + }; + let tool = task.action.as_str(); + // What goes upstream is rebuilt from what is decided here: the task's + // id, the answers a `tasks/update` carries, and the request's `_meta` + // (where MCP 2026-07-28 puts the protocol version every request + // carries). Anything else the client put in the request stays here, + // so it cannot reach the server past the flow check. + let answers = if method == "tasks/update" { params.get("inputResponses") } else { None }; + let mut sent = serde_json::Map::new(); + sent.insert("taskId".into(), json!(task_id)); + if let Some(answers) = answers { + sent.insert("inputResponses".into(), answers.clone()); + } + if let Some(meta) = params.get("_meta") { + sent.insert("_meta".into(), meta.clone()); + } + let line = json!({"jsonrpc": "2.0", "id": id, "method": method, "params": sent}).to_string(); + // The payload binds the exchange to its task and to the receipt it + // continues; the answers ride in it, and what the client authored in + // them is flow-checked as a call's arguments are. + let mut payload = serde_json::Map::new(); + payload.insert("method".into(), json!(method)); + payload.insert("taskId".into(), json!(task_id)); + payload.insert("continues".into(), json!(task.last().to_hex())); + if let Some(answers) = answers { + payload.insert("inputResponses".into(), answers.clone()); + } + let payload = serde_json::to_vec(&Value::Object(payload)).unwrap_or_default(); + let scanned = + serde_json::to_vec(&authored(answers, None, &task.request_keys, None)).unwrap_or_default(); + if id.is_none() { + return Ok(self.deny_notification(tool, &payload)); + } + let Some(sg) = self.grants.get(tool).cloned() else { + return Ok(self.deny_unknown(tool, id)); + }; + if self.tool_schema_changed(tool) { + return Ok(self.deny_schema_changed(tool, id)); + } + + let exchange = Exchange { + line: &line, + id, + tool, + payload: &payload, + scanned: &scanned, + declared: declared_witness_ids(params), + carried: &task.label, + }; + let leg = match self.enforce(up, &sg, exchange)? { + Ok(leg) => leg, + Err(denied) => return Ok(denied), + }; + if let Some(t) = self.tasks.get_mut(&task_id) { + t.legs.push(leg.receipt_id); + t.metered_uj = t.metered_uj.saturating_add(leg.metered.cost.uj()); + t.label = t.label.join(&leg.label); + t.request_keys.extend(minted(&leg.response).1); + } + + let response = attach_receipt(&leg.response, &leg.receipt, &leg.metered, leg.budget_remaining_uj); + let mut task_note = json!({ + "taskId": task_id, + "opened_by": task.opened_by.to_hex(), + "continues": task.last().to_hex(), + }); + // Only `tasks/get` brings a task's state back; a `tasks/cancel` is + // acknowledged, and the task is cancelled when a poll says so. + if method == "tasks/get" { + if let Some((status, outcome)) = finished(&leg.response) { + if let Some((record, receipt)) = self.close_task(&task_id, &sg, &status, outcome.as_ref()) { + task_note["closed"] = json!({"record": record, "receipt_b64": b64(&receipt)}); + } + } + } + let mut note = serde_json::Map::new(); + note.insert("task".into(), task_note); + Ok(Relayed::Allowed { + response: annotate(&response, note), + cost_uj: leg.metered.cost.uj(), + budget_remaining_uj: leg.budget_remaining_uj, + }) + } + + /// Bind a task the server opened in answer to a gated call to that call's + /// action, so the exchanges that follow it are decided against its grant. + fn open_task(&mut self, task_id: String, tool: &str, leg: &Leg) { + if self.tasks.len() >= MAX_OPEN && !self.tasks.contains_key(&task_id) { + let oldest = self.tasks.iter().min_by_key(|(_, t)| t.seq).map(|(k, _)| k.clone()); + if let Some(oldest) = oldest { + self.tasks.remove(&oldest); + } + } + self.tasks_opened += 1; + self.tasks.insert( + task_id, + OpenTask { + action: tool.to_string(), + opened_by: leg.receipt_id, + legs: Vec::new(), + metered_uj: leg.metered.cost.uj(), + label: leg.label.clone(), + request_keys: minted(&leg.response).1, + seq: self.tasks_opened, + }, + ); + } + + /// Close a task whose outcome came back, and forget it: seal its result + /// frame, whose payload is the closing record's content address (§3), + /// and return the record with its receipt. + /// + /// Nothing runs in the close and nothing is debited, so it is sealed with + /// the accounting decision for zero joules — graded `Estimator`, because + /// no meter read that zero, as a refusal's is — and a replay of the + /// stream adds nothing for it. The joules the task cost are in its + /// exchanges' receipts; the record sums them, and is `Partial` because + /// the server's work between the exchanges is in none of them. + fn close_task( + &mut self, + task_id: &str, + sg: &SignedGrant, + status: &str, + outcome: Option<&Value>, + ) -> Option<(Value, Vec)> { + let task = self.tasks.remove(task_id)?; + let mut record = json!({ + "task": task_id, + "status": status, + "opened_by": task.opened_by.to_hex(), + "legs": task.legs.iter().map(|l| l.to_hex()).collect::>(), + "metered_uj": task.metered_uj, + "energy_coverage": Coverage::Partial.label(), + "energy_partial_because": [BETWEEN_EXCHANGES], + }); + if let Some(outcome) = outcome { + record["outcome"] = json!(json_digest(outcome).to_hex()); + } + let grant = &sg.grant; + let mut frame = mcp_result_frame(grant.id(), &self.server, &task.action, b""); + frame.payload_hash = json_digest(&record); + let decision = self.runtime.charge(grant, Energy::estimator(0)); + let receipt = self.log.append(&frame, grant, &decision); + Some((record, receipt)) + } + + /// Phases 0–4 of one gated exchange, whichever method carries it: fold + /// the payload's provenance, authorize against `sg`, run the line + /// upstream metered, debit and seal, and admit the reply as a labelled + /// source. `Err` is the sealed refusal to hand back. + fn enforce( + &mut self, + up: &mut impl Upstream, + sg: &SignedGrant, + ex: Exchange<'_>, + ) -> std::io::Result> { + let Exchange { line, id, tool, payload, scanned, declared, carried } = ex; let grant = sg.grant.clone(); - let frame = mcp_call_frame(grant.id(), &self.server, tool, &args_bytes); + let frame = mcp_call_frame(grant.id(), &self.server, tool, payload); // Phase 0: fold the argument's provenance. The scan finds any verbatim // carry from a tool result admitted earlier this session; the monitor's own // energy for this is metered and sealed separately, never as tool cost. // It is metered on this process's own threads (`scan_meter`): the // scan runs here, on this thread, and touches no other process. - let declared = declared_witness_ids(¶ms); let anchor = self.log.head(); let mode = self.mode; let (witness, eval) = { let flow = &mut self.flow; let ((_id, _label, witness), m) = - self.scan_meter.measure(|| flow.label_payload(&args_bytes, &declared, mode, anchor)); + self.scan_meter.measure(|| flow.label_payload(scanned, &declared, mode, anchor)); (witness, m) }; let facts = FlowFacts { @@ -659,14 +1069,14 @@ impl Gateway { // Phase 1: authorize with the flow facts — a tainted argument above the // ceiling is denied here, before any debit (zero joules). - let auth = self.runtime.authorize_flow(&frame, &sg, self.caller, self.now, &facts); + let auth = self.runtime.authorize_flow(&frame, sg, self.caller, self.now, &facts); if !auth.is_allow() { let reason = deny_reason_of(&auth); self.log.append_flow(&frame, &grant, &auth, &seal); - return Ok(Relayed::Denied { + return Ok(Err(Relayed::Denied { response: jsonrpc_error(id, -32001, &format!("JCP denied: {reason}")), reason, - }); + })); } // Phase 2: run the upstream call, metered twice: on the upstream @@ -710,43 +1120,39 @@ impl Gateway { let the_response = is_the_response(&response, id); let metered = grade_call(reading.as_ref(), self.upstream_self_contained, &measurement, the_response); let cost = metered.cost; - let cost_uj = cost.uj(); // Phase 3: debit the metered joules (and the tainted sub-budget, if the // call was tainted); seal the accounting decision as jcp/3 when tainted. // The receipt seals `cost`'s own provenance: an estimate never enters // the ledger, or the evidence, as a reading. let decision = self.runtime.charge_flow(&grant, cost, &facts); - self.log.append_flow(&frame, &grant, &decision, &seal); + let receipt = self.log.append_flow(&frame, &grant, &decision, &seal); + let receipt_id = self.log.head(); // Phase 4: the result becomes a labelled source. Its frame is Trust::Data // (an injected instruction in it can never exercise a capability), and it // is admitted so a later call carrying its bytes is folded against it — the - // whole point: derived taint that outlives this one call. + // whole point: derived taint that outlives this one call. A reply in a + // piece of work the client fed before (a task's) derives from all of it. let _result_frame = mcp_result_frame(grant.id(), &self.server, tool, response.as_bytes()); self.flow.admit_output( response.as_bytes(), mcp_resource(&self.server, tool), None, - &facts.label, + &facts.label.join(carried), Readers::Public, &[], self.log.head(), ); - // The receipt for THIS call is the one just appended, so the caller - // receives evidence for its own request rather than having to trust - // that the gateway logged something somewhere. - let response = match self.log.entries().last() { - Some(r) => attach_receipt(&response, r, &metered, decision.budget_remaining_uj), - None => response, - }; - - Ok(Relayed::Allowed { + Ok(Ok(Leg { response, - cost_uj, + metered, budget_remaining_uj: decision.budget_remaining_uj, - }) + label: facts.label, + receipt, + receipt_id, + })) } /// Record each listed tool's schema digest, and flag any that contradicts @@ -848,10 +1254,13 @@ impl Gateway { } /// The methods this gateway enforces, named on the wire so a client can see -/// what it is getting rather than infer it. -pub const JCP_ENFORCED_METHODS: [&str; 3] = ["tools/call", "resources/read", "prompts/get"]; +/// what it is getting rather than infer it. The task methods are enforced as +/// continuations of the gated call that opened the task (SPEC §6.1). +pub const JCP_ENFORCED_METHODS: [&str; 6] = + ["tools/call", "resources/read", "prompts/get", "tasks/get", "tasks/update", "tasks/cancel"]; -/// Announce, in an `initialize` response, that this endpoint enforces JCP. +/// Announce, in an `initialize` or `server/discover` response, that this +/// endpoint enforces JCP. /// /// # Why an advertisement is the missing half /// @@ -903,14 +1312,18 @@ fn advertise_extension(response: &str, profile: &str) -> String { // under an unchanged name. Advertised because a client has no // other way to learn that this endpoint does it. "tool_schema_pinned": 1, + // A task's exchanges and an interim result's retry are decided + // and sealed as continuations of the call (SPEC §6.1), and a + // task id this gateway did not see opened is refused. + "continuations": 1, "receipt_in": format!("result._meta.{JCP_META_NS}.receipt_b64"), }), ); v.to_string() } -/// What an endpoint advertised under [`JCP_META_NS`] in its `initialize` -/// response. +/// What an endpoint advertised under [`JCP_META_NS`] in its `initialize` or +/// `server/discover` response. #[derive(Debug, Clone, PartialEq, Eq)] pub struct JcpAdvertisement { /// The receipt profile the endpoint seals with. @@ -920,12 +1333,16 @@ pub struct JcpAdvertisement { /// Whether it refuses a call to a tool whose parameter schema changed /// under an unchanged name. pub tool_schema_pinned: bool, + /// Whether it decides a task's exchanges and an interim result's retry + /// as continuations of the call (SPEC §6.1). + pub continuations: bool, } /// Why a client refused an endpoint. #[derive(Debug, Clone, PartialEq, Eq)] pub enum RequireError { - /// The response was not a JSON-RPC `initialize` result. + /// The response was not a JSON-RPC `initialize` or `server/discover` + /// result. NotAnInitializeResult, /// The endpoint advertised no JCP extension: it may enforce nothing, and /// a client that proceeds cannot tell. @@ -939,7 +1356,9 @@ pub enum RequireError { impl std::fmt::Display for RequireError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { match self { - RequireError::NotAnInitializeResult => write!(f, "not an initialize result"), + RequireError::NotAnInitializeResult => { + write!(f, "not an initialize or server/discover result") + } RequireError::NotAdvertised(ns) => { write!(f, "endpoint advertises no `{ns}` extension") } @@ -950,7 +1369,8 @@ impl std::fmt::Display for RequireError { impl std::error::Error for RequireError {} -/// Read an `initialize` response and **refuse** an endpoint that does not +/// Read an `initialize` response, or a `server/discover` one (where MCP +/// 2026-07-28 moved discovery), and **refuse** an endpoint that does not /// advertise JCP enforcement over every method in `required`. /// /// This is the half that makes the advertisement worth making. A client calls @@ -984,6 +1404,7 @@ pub fn require_jcp( profile: ext.get("profile").and_then(Value::as_str).unwrap_or("").to_string(), enforced, tool_schema_pinned: ext.get("tool_schema_pinned").and_then(Value::as_u64) == Some(1), + continuations: ext.get("continuations").and_then(Value::as_u64) == Some(1), }) } @@ -1003,7 +1424,14 @@ struct ToolSchemaState { /// Canonical JSON via [`ContentId::of`], so key order in the server's response /// cannot change the digest while the schema is the same. fn schema_digest(schema: &Value) -> ContentId { - ContentId::of(schema).unwrap_or_else(|_| ContentId::of_bytes(schema.to_string().as_bytes())) + json_digest(schema) +} + +/// Content-address a JSON value: its canonical form's id ([`ContentId::of`], +/// SPEC §3), so key order cannot change it; and, for a value the canonical +/// form refuses (a float), the id of its JSON text as this gateway writes it. +fn json_digest(v: &Value) -> ContentId { + ContentId::of(v).unwrap_or_else(|_| ContentId::of_bytes(v.to_string().as_bytes())) } /// The tools a `tools/list` response advertised, with each one's schema digest. @@ -1251,6 +1679,158 @@ fn is_the_response(response: &str, id: Option<&Value>) -> bool { v.get("method").is_none() && (v.get("result").is_some() || v.get("error").is_some()) && v.get("id") == Some(id) } +/// One gated exchange to decide: the line to send upstream, its id, the +/// action it is gated as, the payload it is decided and sealed over, the +/// part of that payload the client authored, which is what is flow-checked +/// (for a call without answers, the whole payload), the witnesses the client +/// declared for it, and the label of what the client sent earlier in the +/// same piece of work (`⊥` for a call). +struct Exchange<'a> { + line: &'a str, + id: Option<&'a Value>, + tool: &'a str, + payload: &'a [u8], + scanned: &'a [u8], + declared: BTreeSet, + carried: &'a Label, +} + +/// What a gated call's reply is, by its `resultType` (MCP 2026-07-28). +#[derive(Debug, Clone, PartialEq, Eq)] +enum Reply { + /// Its result (`complete`), an error, or a result from before the field, + /// which a client reads as `complete`. + Final, + /// An interim result: the server asks the client for input, and the + /// client repeats the call with its answers. + InputRequired, + /// A task handle, with the id the server minted for the task. + Task(String), +} + +fn reply_kind(response: &str) -> Reply { + let Ok(v) = serde_json::from_str::(response) else { return Reply::Final }; + let Some(result) = v.get("result") else { return Reply::Final }; + match result.get("resultType").and_then(Value::as_str) { + Some("task") => match result.get("taskId").and_then(Value::as_str) { + Some(task_id) => Reply::Task(task_id.to_string()), + // A handle naming no task: nothing can follow it. + None => Reply::Final, + }, + Some("input_required") => Reply::InputRequired, + _ => Reply::Final, + } +} + +/// The status and outcome of a `tasks/get` reply whose task has finished: +/// `completed` with its `result`, `failed` with its `error`, and `cancelled` +/// with none. `None` for a task still going, and for a reply that is no task. +fn finished(response: &str) -> Option<(String, Option)> { + let v: Value = serde_json::from_str(response).ok()?; + let task = v.get("result")?; + let status = task.get("status")?.as_str()?; + let outcome = match status { + "completed" => task.get("result").cloned(), + "failed" => task.get("error").cloned(), + "cancelled" => None, + _ => return None, + }; + Some((status.to_string(), outcome)) +} + +/// A gated call's arguments, and apart from them the answers it carries to +/// an interim result: `inputResponses`, with the `requestState` the server +/// handed back, if any. A call that answers nothing has none. A +/// `resources/read` is gated over its whole `params`, so the answers are +/// taken out of that; a `tools/call` or `prompts/get` is gated over its +/// `arguments`, which are left exactly as sent. +fn split_answers(method: &str, mut args: Value, params: &Value) -> (Value, Option>) { + let Some(responses) = params.get("inputResponses") else { return (args, None) }; + let mut answers = serde_json::Map::new(); + answers.insert("inputResponses".into(), responses.clone()); + if let Some(state) = params.get("requestState") { + answers.insert("requestState".into(), state.clone()); + } + if method == "resources/read" { + if let Some(obj) = args.as_object_mut() { + obj.remove("inputResponses"); + obj.remove("requestState"); + } + } + (args, Some(answers)) +} + +/// The handles the server minted in a reply for the client to hand back: +/// its `requestState`, and the keys of the requests in its `inputRequests`. +fn minted(response: &str) -> (Option, BTreeSet) { + let Ok(v) = serde_json::from_str::(response) else { return (None, BTreeSet::new()) }; + let Some(result) = v.get("result") else { return (None, BTreeSet::new()) }; + let keys = result + .get("inputRequests") + .and_then(Value::as_object) + .map(|m| m.keys().cloned().collect()) + .unwrap_or_default(); + (result.get("requestState").cloned(), keys) +} + +/// What a client's answers put before the server that the server did not +/// mint: every answer, every key the server did not issue, and a +/// `requestState` other than the one it handed back. This is what answers +/// are flow-checked over. +/// +/// A request's key and a `requestState` are handles the server minted, which +/// the client hands back unchanged. Each came in a reply this gateway +/// admitted as untrusted, so folding an exact echo of one would taint every +/// answer that follows (a scan window is 16 bytes, and a UUID key is 36), +/// while telling the server nothing it had not sent. But a handle the client +/// makes up can carry anything, so only an exact echo of one this gateway +/// saw the server mint is left out. +fn authored( + responses: Option<&Value>, + state: Option<&Value>, + keys: &BTreeSet, + minted_state: Option<&Value>, +) -> Value { + let mut out = Vec::new(); + match responses { + Some(Value::Object(answers)) => { + for (key, answer) in answers { + if !keys.contains(key) { + out.push(json!(key)); + } + out.push(answer.clone()); + } + } + Some(other) => out.push(other.clone()), + None => {} + } + if let Some(state) = state { + if Some(state) != minted_state { + out.push(state.clone()); + } + } + Value::Array(out) +} + +/// Add `note`'s keys to the `_meta` block [`attach_receipt`] wrote. A +/// response with no such block is returned unchanged. +fn annotate(response: &str, note: serde_json::Map) -> String { + if note.is_empty() { + return response.to_string(); + } + let Ok(mut v) = serde_json::from_str::(response) else { + return response.to_string(); + }; + let block = v + .get_mut("result") + .and_then(|r| r.get_mut("_meta")) + .and_then(|m| m.get_mut(JCP_META_NS)) + .and_then(Value::as_object_mut); + let Some(block) = block else { return response.to_string() }; + block.extend(note); + v.to_string() +} + /// A gated call's metered cost, and where it came from. #[derive(Debug, Clone, PartialEq, Eq)] struct Metered { @@ -2668,6 +3248,525 @@ mod tests { } /// Decode a verified receipt envelope to its JSON payload (test helper). + // ── work that answers later: tasks and interim results (SPEC §6.1) ────── + + const MAIL: &str = "did:jcp:gateway:mail"; + /// What the inbox holds, read straight back by `read_inbox`. + const INBOX_SECRET: &str = "INBOX-SECRET-TOKEN-0123456789abcdef"; + /// What a `read_email` task brings back when it completes. + const TASK_SECRET: &str = "TASK-OUTCOME-PAYLOAD-zyxwvutsrqponm"; + /// The id the server mints for its one task: a UUID, as real servers mint, + /// far longer than a scan window. + const TASK_ID: &str = "786512e2-9e0d-44bd-8f29-789f320fe840"; + /// The key and the state the server mints for its interim results. + const ASK_KEY: &str = "0b6f8f3e-4c1d-4e7a-9a52-2f1c8d7e6a10"; + const ASK_STATE: &str = "state-5d0e1c2b-7f3a-4b9e-8c6d-1a2b3c4d5e6f"; + + /// A server that answers `read_email` with a task handle and brings the + /// outcome back through `tasks/get` (`working` on the first poll, + /// `completed` with [`TASK_SECRET`] on the next, or `cancelled` once it + /// has been cancelled; with `ask_in_task`, the first poll asks for input + /// instead). `ask` answers `input_required` until it is retried with + /// answers. `read_inbox` returns [`INBOX_SECRET`]. Every other call + /// succeeds. It keeps every line it is sent. + struct TaskServer { + polls: usize, + cancelled: bool, + ask_in_task: bool, + seen: Vec, + } + + impl TaskServer { + fn new() -> TaskServer { + TaskServer { polls: 0, cancelled: false, ask_in_task: false, seen: Vec::new() } + } + } + + impl Upstream for TaskServer { + fn call(&mut self, line: &str) -> std::io::Result { + self.seen.push(line.to_string()); + let v: Value = serde_json::from_str(line).unwrap_or(Value::Null); + let id = v.get("id").cloned().unwrap_or(Value::Null); + let method = v.get("method").and_then(Value::as_str).unwrap_or(""); + let params = v.get("params").cloned().unwrap_or(Value::Null); + let tool = params.get("name").and_then(Value::as_str).unwrap_or(""); + // A pinch of work so a hardware span is nonzero. + let mut x = 0u64; + for i in 0..200_000u64 { + x = x.wrapping_mul(6364136223846793005).wrapping_add(i); + } + std::hint::black_box(x); + let result = match (method, tool) { + ("tools/call", "read_email") => { + json!({"resultType": "task", "taskId": TASK_ID, "status": "working", "ttlMs": 60000}) + } + ("tools/call", "read_inbox") => json!({"resultType": "complete", "body": INBOX_SECRET}), + ("tools/call", "ask") if params.get("inputResponses").is_none() => json!({ + "resultType": "input_required", + "inputRequests": {ASK_KEY: {"method": "elicitation/create", "params": {"message": "which folder?"}}}, + "requestState": ASK_STATE, + }), + ("tasks/get", _) => { + self.polls += 1; + if self.cancelled { + json!({"resultType": "complete", "taskId": TASK_ID, "status": "cancelled"}) + } else if self.polls < 2 && self.ask_in_task { + json!({"resultType": "complete", "taskId": TASK_ID, "status": "input_required", + "inputRequests": {ASK_KEY: {"method": "elicitation/create", "params": {}}}}) + } else if self.polls < 2 { + json!({"resultType": "complete", "taskId": TASK_ID, "status": "working"}) + } else { + json!({"resultType": "complete", "taskId": TASK_ID, "status": "completed", + "result": {"content": [{"type": "text", "text": TASK_SECRET}]}}) + } + } + ("tasks/cancel", _) => { + self.cancelled = true; + json!({"resultType": "complete"}) + } + _ => json!({"resultType": "complete", "ok": true}), + }; + Ok(json!({"jsonrpc": "2.0", "id": id, "result": result}).to_string()) + } + } + + fn mail_gw() -> Gateway { + Gateway::new( + "mail", + &[ + ("read_email", 50_000_000), + ("read_inbox", 50_000_000), + ("send_email", 50_000_000), + ("ask", 50_000_000), + ], + SigningKey::from_bytes(&[7u8; 32]), + ) + } + + fn mail_pub() -> [u8; 32] { + SigningKey::from_bytes(&[7u8; 32]).verifying_key().to_bytes() + } + + fn tool_call(id: i64, tool: &str, arguments: Value) -> String { + json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", + "params": {"name": tool, "arguments": arguments}}) + .to_string() + } + + fn task_call(id: i64, method: &str, task_id: &str) -> String { + json!({"jsonrpc": "2.0", "id": id, "method": method, "params": {"taskId": task_id}}).to_string() + } + + /// The gateway's `_meta` block in a relayed response. + fn jcp_block(relayed: &Relayed) -> Value { + let Relayed::Allowed { response, .. } = relayed else { panic!("expected Allowed, got {relayed:?}") }; + let v: Value = serde_json::from_str(response).unwrap(); + v["result"]["_meta"][JCP_META_NS].clone() + } + + fn receipt_id(envelope: &[u8]) -> String { + ContentId::of_bytes(envelope).to_hex() + } + + /// The whole point of following a task: its outcome is the tool's output, + /// and it arrives through `tasks/get`, which used to pass through as + /// discovery — so it was never admitted to the flow store, and a value + /// read through a task could be sent on with nothing to fold it against. + #[test] + fn a_task_outcome_is_untrusted_data_and_cannot_be_sent_on() { + let mut g = mail_gw(); + g.set_flow_ceiling("send_email", principal_only()); + let mut up = TaskServer::new(); + + let opened = g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + assert_eq!(jcp_block(&opened)["task"]["taskId"], TASK_ID); + let working = g.relay(&task_call(2, "tasks/get", TASK_ID), &mut up).unwrap(); + assert!(jcp_block(&working)["task"].get("closed").is_none(), "a working task is not closed"); + let done = g.relay(&task_call(3, "tasks/get", TASK_ID), &mut up).unwrap(); + let Relayed::Allowed { response, .. } = &done else { panic!() }; + assert!(response.contains(TASK_SECRET), "the outcome reaches the client"); + + let exfil = tool_call(4, "send_email", json!({"body": format!("fwd {TASK_SECRET} now")})); + match g.relay(&exfil, &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow"), + other => panic!("a value read through a task must not be sent on: {other:?}"), + } + let clean = tool_call(5, "send_email", json!({"body": "meeting at noon"})); + assert!(matches!(g.relay(&clean, &mut up).unwrap(), Relayed::Allowed { .. })); + } + + /// A task's outcome derives from the call that opened it, so it carries + /// that call's provenance as well as its own channel's: an outcome a + /// ceiling would admit from the task's channel alone is refused when the + /// call that opened the task was fed from another. + #[test] + fn a_task_outcome_carries_the_provenance_of_the_call_that_opened_it() { + use jcp_schema::flow::Origin; + let mut g = mail_gw(); + let only_read_email = FlowCeiling::new(Label { + integrity: Integrity::Untrusted, + origins: Origins::one(Origin::channel(mcp_resource("mail", "read_email"))), + readers: Readers::Public, + }); + g.set_flow_ceiling("send_email", only_read_email); + let mut up = TaskServer::new(); + + // What came from read_email alone may be sent on. + g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + g.relay(&task_call(2, "tasks/get", TASK_ID), &mut up).unwrap(); + g.relay(&task_call(3, "tasks/get", TASK_ID), &mut up).unwrap(); + let ok = tool_call(4, "send_email", json!({"body": format!("fwd {TASK_SECRET} now")})); + assert!( + matches!(g.relay(&ok, &mut up).unwrap(), Relayed::Allowed { .. }), + "the ceiling admits read_email's channel" + ); + + // A task opened with an argument read from the inbox: its outcome + // carries the inbox's origin too, which the ceiling does not admit. + let mut g = mail_gw(); + g.set_flow_ceiling("send_email", FlowCeiling::new(Label { + integrity: Integrity::Untrusted, + origins: Origins::one(Origin::channel(mcp_resource("mail", "read_email"))), + readers: Readers::Public, + })); + let mut up = TaskServer::new(); + g.relay(&tool_call(1, "read_inbox", json!({})), &mut up).unwrap(); + g.relay(&tool_call(2, "read_email", json!({"query": INBOX_SECRET})), &mut up).unwrap(); + g.relay(&task_call(3, "tasks/get", TASK_ID), &mut up).unwrap(); + g.relay(&task_call(4, "tasks/get", TASK_ID), &mut up).unwrap(); + let exfil = tool_call(5, "send_email", json!({"body": format!("fwd {TASK_SECRET} now")})); + match g.relay(&exfil, &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow"), + other => panic!("the outcome carries the inbox's origin: {other:?}"), + } + } + + /// Each exchange of a task is decided, metered, debited and sealed, as a + /// continuation of the receipt before it, and the task closes with one + /// record naming them all. + #[test] + fn each_exchange_of_a_task_is_sealed_as_a_continuation_and_the_task_closes_once() { + let mut g = mail_gw(); + let mut up = TaskServer::new(); + + let opened = g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + let Relayed::Allowed { cost_uj: c0, budget_remaining_uj: r0, .. } = opened else { panic!() }; + let poll1 = g.relay(&task_call(2, "tasks/get", TASK_ID), &mut up).unwrap(); + let Relayed::Allowed { cost_uj: c1, budget_remaining_uj: r1, .. } = poll1 else { panic!() }; + let poll2 = g.relay(&task_call(3, "tasks/get", TASK_ID), &mut up).unwrap(); + let Relayed::Allowed { cost_uj: c2, budget_remaining_uj: r2, .. } = poll2 else { panic!() }; + + // Each poll is debited from the call's grant. + assert!(c1 > 0 && c2 > 0, "each poll is metered"); + assert_eq!(r1 + c1, r0, "the first poll is debited from the grant that opened the task"); + assert_eq!(r2 + c2, r1, "and so is the second"); + + // Open, two polls, and the close: four receipts, one verifying chain. + let receipts = g.receipts().to_vec(); + assert_eq!(receipts.len(), 4); + assert_eq!(jcp_receipt::verify_log(&receipts, MAIL, &mail_pub()), Ok(4)); + + // Each poll is a decision over read_email's own resource, and its + // `_meta` names the receipt it continues. + let (id0, id1, id2) = (receipt_id(&receipts[0]), receipt_id(&receipts[1]), receipt_id(&receipts[2])); + assert_eq!(jcp_block(&opened)["task"]["opened_by"], id0.as_str()); + for (poll, prev) in [(&poll1, &id0), (&poll2, &id1)] { + let task = &jcp_block(poll)["task"]; + assert_eq!(task["opened_by"], id0.as_str()); + assert_eq!(task["continues"], prev.as_str()); + } + for r in &receipts[1..3] { + let p = jcr1_payload(r, MAIL, &mail_pub()); + assert_eq!(p["action"], "read_email"); + assert_eq!(p["resource"], mcp_resource("mail", "read_email").to_hex().as_str()); + assert_eq!(p["frame_kind"], "tool-call"); + assert_eq!(p["verdict"], "allow"); + } + // The sealed payload of a poll commits to the receipt it continues. + let p1 = jcr1_payload(&receipts[1], MAIL, &mail_pub()); + let expected = serde_json::to_vec(&json!({"continues": &id0, "method": "tasks/get", "taskId": TASK_ID})).unwrap(); + assert_eq!(p1["payload_hash"], ContentId::of_bytes(&expected).to_hex().as_str()); + + // The close: the task's result frame, sealed at zero cost over the + // closing record's content address. + let closed = &jcp_block(&poll2)["task"]["closed"]; + let record = &closed["record"]; + assert_eq!(record["status"], "completed"); + assert_eq!(record["opened_by"], id0.as_str()); + assert_eq!(record["legs"], json!([&id1, &id2])); + assert_eq!(record["metered_uj"], c0 + c1 + c2); + assert_eq!(record["energy_coverage"], "Partial"); + assert_eq!(record["energy_partial_because"], json!(["between-exchanges"])); + let outcome = json!({"content": [{"type": "text", "text": TASK_SECRET}]}); + assert_eq!(record["outcome"], json_digest(&outcome).to_hex().as_str()); + assert_eq!(closed["receipt_b64"], b64(&receipts[3]).as_str()); + let close = jcr1_payload(&receipts[3], MAIL, &mail_pub()); + assert_eq!(close["frame_kind"], "tool-result"); + assert_eq!(close["trust"], "data"); + assert_eq!(close["verdict"], "allow"); + assert_eq!(close["cost_uj"], 0); + assert_eq!(close["action"], "read_email"); + assert_eq!(close["payload_hash"], ContentId::of(record).unwrap().to_hex().as_str()); + + // A replay of the stream re-applies the three debits and nothing for + // the close. + let gid = g.grants["read_email"].grant.id(); + let mut resumed = mail_gw(); + resumed.resume(receipts).unwrap(); + assert_eq!(resumed.runtime.spent(&gid), c0 + c1 + c2); + assert_eq!(g.runtime.spent(&gid), c0 + c1 + c2); + + // A task closes once: the next poll names a task this gateway has + // closed, and is refused before it reaches the server. + let before = up.seen.len(); + match g.relay(&task_call(4, "tasks/get", TASK_ID), &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "no task opened under a grant"), + other => panic!("a closed task is not polled again: {other:?}"), + } + assert_eq!(up.seen.len(), before); + } + + /// A task id is a bearer string the server minted. One this gateway never + /// saw come back from a call it gated is under no grant, so every task + /// method naming it is refused, sealed, and kept from the server. + #[test] + fn a_task_this_gateway_did_not_open_is_refused_and_sealed() { + let mut g = mail_gw(); + let mut up = TaskServer::new(); + for (i, method) in TASK_METHODS.iter().enumerate() { + match g.relay(&task_call(i as i64 + 1, method, TASK_ID), &mut up).unwrap() { + Relayed::Denied { reason, response } => { + assert_eq!(reason, "no task opened under a grant"); + assert!(response.contains("receipt_b64"), "the refusal carries its receipt"); + } + other => panic!("{method} on an unopened task must be refused: {other:?}"), + } + } + // As a notification, too. + let note = json!({"jsonrpc": "2.0", "method": "tasks/cancel", "params": {"taskId": TASK_ID}}).to_string(); + assert!(matches!(g.relay(¬e, &mut up).unwrap(), Relayed::Denied { .. })); + assert!(up.seen.is_empty(), "none reached the server"); + assert_eq!(g.receipts().len(), 4, "each refusal is sealed"); + for r in g.receipts() { + assert_eq!(jcp_receipt::extract(r, MAIL, &mail_pub()).unwrap().verdict, "deny"); + } + } + + /// A poll is decided at the moment it is made: once the grant the task + /// was opened under is revoked, its outcome is not fetched under it. + #[test] + fn a_poll_under_a_grant_revoked_since_the_task_opened_is_refused() { + let mut g = mail_gw(); + let mut up = TaskServer::new(); + g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + let grant = g.grants["read_email"].grant.clone(); + let revocation = jcp_grant::revoke(&g.issuer_sk, &grant, [9u8; 16]).unwrap(); + g.runtime.apply_revocation(&revocation, &grant).unwrap(); + + match g.relay(&task_call(2, "tasks/get", TASK_ID), &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "revoked"), + other => panic!("a revoked grant must not fetch the outcome: {other:?}"), + } + assert_eq!(up.seen.len(), 1, "only the opening call reached the server"); + let last = g.receipts().last().unwrap(); + let core = jcp_receipt::extract(last, MAIL, &mail_pub()).unwrap(); + assert_eq!(core.deny_reason.as_deref(), Some("revoked")); + } + + /// What the client sends into a task reaches the server, so it is + /// flow-checked against the grant's ceiling as a call's arguments are. + /// The handles the server minted, handed back unchanged, are not: a + /// `tasks/update` whose answers are clean goes through under a ceiling + /// the server's UUID task id and request keys would otherwise breach. + #[test] + fn answers_sent_into_a_task_are_flow_checked_and_echoed_handles_are_not() { + let mut g = mail_gw(); + g.set_flow_ceiling("read_email", principal_only()); + let mut up = TaskServer { ask_in_task: true, ..TaskServer::new() }; + + g.relay(&tool_call(1, "read_inbox", json!({})), &mut up).unwrap(); + g.relay(&tool_call(2, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + // The first poll asks for input; it is admitted as untrusted, keys and all. + let asked = g.relay(&task_call(3, "tasks/get", TASK_ID), &mut up).unwrap(); + assert!(matches!(asked, Relayed::Allowed { .. }), "the poll echoes only the task id: {asked:?}"); + + let update = |id: i64, answer: &str| { + json!({"jsonrpc": "2.0", "id": id, "method": "tasks/update", + "params": {"taskId": TASK_ID, "inputResponses": {ASK_KEY: {"action": "accept", "content": {"folder": answer}}}}}) + .to_string() + }; + // Clean first: a refused payload is admitted under its taint, and a + // later one sharing a scan window of its text would join it. + assert!( + matches!(g.relay(&update(4, "archive"), &mut up).unwrap(), Relayed::Allowed { .. }), + "a clean answer under the server's own key goes through" + ); + match g.relay(&update(5, &format!("send {INBOX_SECRET}")), &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow"), + other => panic!("an answer carrying the inbox must be refused: {other:?}"), + } + + // A key the client made up is the client's text, and is checked. + let forged = json!({"jsonrpc": "2.0", "id": 6, "method": "tasks/update", + "params": {"taskId": TASK_ID, "inputResponses": {INBOX_SECRET: {"action": "accept"}}}}) + .to_string(); + assert!(matches!(g.relay(&forged, &mut up).unwrap(), Relayed::Denied { .. })); + } + + /// What goes upstream for a task method is rebuilt from what was + /// decided: a field the client added beside the task id never reaches the + /// server. + #[test] + fn a_task_request_reaches_the_server_with_only_what_was_decided() { + let mut g = mail_gw(); + let mut up = TaskServer::new(); + g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + let smuggled = json!({"jsonrpc": "2.0", "id": 2, "method": "tasks/get", + "params": {"taskId": TASK_ID, "note": INBOX_SECRET, + "_meta": {"io.modelcontextprotocol/protocolVersion": "2026-07-28"}}}) + .to_string(); + assert!(matches!(g.relay(&smuggled, &mut up).unwrap(), Relayed::Allowed { .. })); + let sent: Value = serde_json::from_str(up.seen.last().unwrap()).unwrap(); + assert_eq!(sent["params"]["taskId"], TASK_ID); + assert!(sent["params"].get("note").is_none(), "{sent}"); + assert_eq!(sent["params"]["_meta"]["io.modelcontextprotocol/protocolVersion"], "2026-07-28"); + } + + /// A cancelled task closes on the poll that reports it, with no outcome. + #[test] + fn a_cancelled_task_closes_with_no_outcome() { + let mut g = mail_gw(); + let mut up = TaskServer::new(); + g.relay(&tool_call(1, "read_email", json!({"folder": "inbox"})), &mut up).unwrap(); + let ack = g.relay(&task_call(2, "tasks/cancel", TASK_ID), &mut up).unwrap(); + assert!(jcp_block(&ack)["task"].get("closed").is_none(), "a cancel is acknowledged, not final"); + let polled = g.relay(&task_call(3, "tasks/get", TASK_ID), &mut up).unwrap(); + let record = &jcp_block(&polled)["task"]["closed"]["record"]; + assert_eq!(record["status"], "cancelled"); + assert!(record.get("outcome").is_none()); + assert_eq!(record["legs"].as_array().unwrap().len(), 2); + assert_eq!(jcp_receipt::verify_log(g.receipts(), MAIL, &mail_pub()), Ok(4)); + } + + fn ask_retry(id: i64, answer: &str, state: &str) -> String { + json!({"jsonrpc": "2.0", "id": id, "method": "tools/call", + "params": {"name": "ask", "arguments": {"q": "x"}, + "inputResponses": {ASK_KEY: {"action": "accept", "content": {"folder": answer}}}, + "requestState": state}}) + .to_string() + } + + /// The answers in a retry go to the server beside the arguments, so they + /// are flow-checked with them. Only the arguments used to be. + #[test] + fn answers_in_a_retry_are_flow_checked_with_the_arguments() { + // Each case on a gateway of its own that has read the inbox and been + // asked for input: a retry ends the round, and a refused payload is + // admitted under its taint. + let asked = || { + let mut g = mail_gw(); + g.set_flow_ceiling("ask", principal_only()); + let mut up = TaskServer::new(); + g.relay(&tool_call(1, "read_inbox", json!({})), &mut up).unwrap(); + let interim = g.relay(&tool_call(2, "ask", json!({"q": "x"})), &mut up).unwrap(); + assert!(matches!(interim, Relayed::Allowed { .. })); + (g, up) + }; + + let (mut g, mut up) = asked(); + match g.relay(&ask_retry(3, &format!("send {INBOX_SECRET}"), ASK_STATE), &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow"), + other => panic!("an answer carrying the inbox must be refused: {other:?}"), + } + // A clean answer, handing back the server's own key and state, goes + // through although both are in a reply admitted as untrusted. + let (mut g, mut up) = asked(); + assert!(matches!(g.relay(&ask_retry(3, "archive", ASK_STATE), &mut up).unwrap(), Relayed::Allowed { .. })); + // A state the client made up is the client's text, and is checked. + let (mut g, mut up) = asked(); + match g.relay(&ask_retry(3, "archive", INBOX_SECRET), &mut up).unwrap() { + Relayed::Denied { reason, .. } => assert_eq!(reason, "tainted-flow"), + other => panic!("a made-up state carrying the inbox must be refused: {other:?}"), + } + } + + /// A retry answering an interim result is sealed as its continuation: the + /// retry's payload commits to the interim's receipt. + #[test] + fn a_retry_answering_an_interim_result_continues_its_receipt() { + let mut g = mail_gw(); + let mut up = TaskServer::new(); + let interim = g.relay(&tool_call(1, "ask", json!({"q": "x"})), &mut up).unwrap(); + let id0 = receipt_id(&g.receipts()[0]); + assert_eq!(jcp_block(&interim)["input_required"]["continue_from"], id0.as_str()); + + let retry = g.relay(&ask_retry(2, "archive", ASK_STATE), &mut up).unwrap(); + assert_eq!(jcp_block(&retry)["continues"], id0.as_str()); + let sealed = jcr1_payload(&g.receipts()[1], MAIL, &mail_pub()); + let payload = json!({ + "arguments": {"q": "x"}, + "continues": &id0, + "inputResponses": {ASK_KEY: {"action": "accept", "content": {"folder": "archive"}}}, + "requestState": ASK_STATE, + }); + assert_eq!( + sealed["payload_hash"], + ContentId::of_bytes(&serde_json::to_vec(&payload).unwrap()).to_hex().as_str(), + "the retry's sealed payload names the interim it answers" + ); + assert!(up.seen[1].contains("inputResponses"), "the retry reaches the server as sent"); + + // The round is over: another retry has nothing to continue. + let again = g.relay(&ask_retry(3, "archive", ASK_STATE), &mut up).unwrap(); + assert!(jcp_block(&again).get("continues").is_none()); + } + + /// A server that opens a task for every call and finishes none. + struct Minting(u64); + impl Upstream for Minting { + fn call(&mut self, line: &str) -> std::io::Result { + let v: Value = serde_json::from_str(line).unwrap_or(Value::Null); + let id = v.get("id").cloned().unwrap_or(Value::Null); + self.0 += 1; + let result = if v["method"] == "tools/call" { + json!({"resultType": "task", "taskId": format!("t-{}", self.0), "status": "working"}) + } else { + json!({"resultType": "complete", "taskId": v["params"]["taskId"], "status": "working"}) + }; + Ok(json!({"jsonrpc": "2.0", "id": id, "result": result}).to_string()) + } + } + + /// Open tasks are bounded: past [`MAX_OPEN`] the oldest is dropped, and an + /// exchange naming it is refused as one naming an unknown task is. + #[test] + fn a_server_that_never_finishes_its_tasks_cannot_grow_the_gateway_without_bound() { + let mut g = mail_gw(); + let mut up = Minting(0); + for i in 0..=MAX_OPEN { + g.relay(&tool_call(i as i64, "read_email", json!({"n": i})), &mut up).unwrap(); + } + assert_eq!(g.tasks.len(), MAX_OPEN); + let newest = format!("t-{}", MAX_OPEN + 1); + assert!(matches!(g.relay(&task_call(9_001, "tasks/get", "t-1"), &mut up).unwrap(), Relayed::Denied { .. })); + assert!(matches!(g.relay(&task_call(9_002, "tasks/get", "t-2"), &mut up).unwrap(), Relayed::Allowed { .. })); + assert!(matches!(g.relay(&task_call(9_003, "tasks/get", &newest), &mut up).unwrap(), Relayed::Allowed { .. })); + } + + /// MCP 2026-07-28 has no `initialize`: a client learns what it talks to + /// from `server/discover`, so that is where the advertisement must be. + #[test] + fn server_discover_carries_the_advertisement() { + let mut g = gw(); + let mut up = MockServer { calls: 0 }; + let line = json!({"jsonrpc": "2.0", "id": 1, "method": "server/discover", "params": {}}).to_string(); + let Relayed::PassedThrough(response) = g.relay(&line, &mut up).unwrap() else { panic!() }; + let adv = require_jcp(&response, &["tools/call", "tasks/get", "tasks/update", "tasks/cancel"]) + .expect("server/discover advertises JCP over the task methods"); + assert!(adv.continuations); + assert_eq!(adv.enforced, JCP_ENFORCED_METHODS); + } + fn jcr1_payload(envelope: &[u8], steward: &str, pubkey: &[u8; 32]) -> Value { let core = jcp_receipt::jcr1::verify_cose_sign1(envelope, steward, pubkey).unwrap(); jcp_schema::from_canonical_cbor(&core).unwrap() -- 2.54.0 (Apple Git-157) From 66f36fdeecb67f411f37b073b10618eea33fc646 Mon Sep 17 00:00:00 2001 From: Transaction Science Date: Mon, 19 Oct 2026 08:28:09 -0400 Subject: [PATCH 2/2] jcp-gateway: the receipt-decoding test helper keeps its doc comment The task tests were inserted between jcr1_payload's doc comment and the function, so the comment documented the first constant after it and clippy (-D warnings, macOS meter job) refused the empty line after it. Co-Authored-By: Claude Opus 5.5 --- jcp/jcp-rs/crates/jcp-gateway/src/lib.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs b/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs index 5b4ef797e..392af9765 100644 --- a/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs +++ b/jcp/jcp-rs/crates/jcp-gateway/src/lib.rs @@ -3247,7 +3247,6 @@ mod tests { } } - /// Decode a verified receipt envelope to its JSON payload (test helper). // ── work that answers later: tasks and interim results (SPEC §6.1) ────── const MAIL: &str = "did:jcp:gateway:mail"; @@ -3767,6 +3766,7 @@ mod tests { assert_eq!(adv.enforced, JCP_ENFORCED_METHODS); } + /// Decode a verified receipt envelope to its JSON payload (test helper). fn jcr1_payload(envelope: &[u8], steward: &str, pubkey: &[u8; 32]) -> Value { let core = jcp_receipt::jcr1::verify_cose_sign1(envelope, steward, pubkey).unwrap(); jcp_schema::from_canonical_cbor(&core).unwrap() -- 2.54.0 (Apple Git-157)