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..392af9765 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 { @@ -2667,6 +3247,525 @@ mod tests { } } + // ── 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); + } + /// 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();