From e3011b0ecee7d7bc7c95baf8575c268bf32fe713 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Sun, 26 Jul 2026 23:56:04 +0200 Subject: [PATCH 01/11] refactor(service): express SEP-2260 receive-side association as an enum Replaces the has_pending_outbound_request bool on enforce_peer_request_association with PeerRequestAssociation, so a stream-separating transport can report per-request association (#1033). Behavior-preserving: the event loop still passes the coarse signal as Unknown. --- crates/rmcp/src/service.rs | 30 +++++++++- crates/rmcp/src/service/client.rs | 98 ++++++++++++++++++++++++++++--- 2 files changed, 116 insertions(+), 12 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index b1b2c2b7c..b9d5172b4 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -163,17 +163,39 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { /// SEP-2260 says clients receiving a server-to-client request with no /// associated outbound request should reject it with invalid params. An /// error return is sent back to the peer instead of dispatching to the - /// handler. + /// handler. The `association` argument is the transport-observed stream + /// association; see [`PeerRequestAssociation`]. #[doc(hidden)] fn enforce_peer_request_association( _peer_request: &Self::PeerReq, _peer_info: Option<&Self::PeerInfo>, - _has_pending_outbound_request: bool, + _association: PeerRequestAssociation, ) -> Result<(), McpError> { Ok(()) } } +/// How an inbound peer request relates to this side's in-flight outbound +/// requests, as observed by the transport (SEP-2260). +/// +/// SEP-2260 defines no wire field for request association; only a +/// stream-separating transport (streamable HTTP) can distinguish the first +/// two variants. Transports without stream separation (stdio, in-process) +/// yield [`Self::Unknown`], preserving the coarse in-flight check. +#[derive(Debug, Clone, PartialEq, Eq)] +#[expect(clippy::exhaustive_enums, reason = "intentionally exhaustive")] +pub enum PeerRequestAssociation { + /// Arrived on the response stream of an outbound request that is still + /// awaiting its response. + Associated, + /// Arrived on a stream tied to no in-flight outbound request (e.g. the + /// streamable HTTP standalone GET stream). + Unassociated, + /// The transport supplied no stream provenance; only the coarse + /// "is anything in flight" signal exists. + Unknown { has_pending_outbound_request: bool }, +} + pub(crate) fn uses_legacy_lifecycle( protocol_version: Option<&ProtocolVersion>, uses_discover_lifecycle: bool, @@ -1448,7 +1470,9 @@ where if let Err(error) = R::enforce_peer_request_association( &request, peer.peer_info().as_deref(), - !local_responder_pool.is_empty(), + PeerRequestAssociation::Unknown { + has_pending_outbound_request: !local_responder_pool.is_empty(), + }, ) { tracing::warn!(%id, message = %error.message, "rejected peer request"); // send directly: the sink proxy path would drop the diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 9fc1f641a..8832f9baf 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -238,14 +238,14 @@ impl ServiceRole for RoleClient { } } - // SEP-2260: with no outbound request in flight there is nothing the - // server request could be associated with, so reject it. With one in - // flight we cannot tell which request it belongs to (no wire field), so - // we accept — an under-approximation of the spec's SHOULD. + // SEP-2260: reject restricted server requests that the transport observed + // arriving with no in-flight outbound request association. Transports + // without stream separation (stdio) report `Unknown`, where the coarse + // in-flight check is the best available under-approximation of the SHOULD. fn enforce_peer_request_association( peer_request: &Self::PeerReq, peer_info: Option<&Self::PeerInfo>, - has_pending_outbound_request: bool, + association: PeerRequestAssociation, ) -> Result<(), ErrorData> { let restricted = matches!( peer_request, @@ -258,13 +258,24 @@ impl ServiceRole for RoleClient { } let strict = peer_info.is_some_and(|info| info.protocol_version >= ProtocolVersion::V_2026_07_28); - if strict && !has_pending_outbound_request { - return Err(ErrorData::invalid_params( + if !strict { + return Ok(()); + } + let associated = match association { + PeerRequestAssociation::Associated => true, + PeerRequestAssociation::Unassociated => false, + PeerRequestAssociation::Unknown { + has_pending_outbound_request, + } => has_pending_outbound_request, + }; + if associated { + Ok(()) + } else { + Err(ErrorData::invalid_params( "SEP-2260: server-to-client requests must be associated with an in-flight client request", None, - )); + )) } - Ok(()) } async fn invalidate_response_cache(peer: &Peer, notification: &Self::PeerNot) { @@ -2209,3 +2220,72 @@ mod tests { assert_eq!(peer.discover(meta).await.unwrap(), expected); } } + +#[cfg(test)] +mod sep2260_association_tests { + use super::*; + use crate::{ + model::{ + CreateMessageRequest, CreateMessageRequestParams, SamplingMessage, ServerCapabilities, + }, + service::PeerRequestAssociation, + }; + + fn sampling_request() -> ServerRequest { + ServerRequest::CreateMessageRequest(CreateMessageRequest::new( + CreateMessageRequestParams::new(vec![SamplingMessage::user_text("hi")], 16), + )) + } + + fn server_info(version: ProtocolVersion) -> ServerInfo { + let mut info = ServerInfo::new(ServerCapabilities::default()); + info.protocol_version = version; + info + } + + fn enforce(info: &ServerInfo, association: PeerRequestAssociation) -> Result<(), ErrorData> { + RoleClient::enforce_peer_request_association(&sampling_request(), Some(info), association) + } + + #[test] + fn strict_rejects_unassociated() { + let info = server_info(ProtocolVersion::V_2026_07_28); + let err = enforce(&info, PeerRequestAssociation::Unassociated).unwrap_err(); + assert_eq!(err.code, crate::model::ErrorCode::INVALID_PARAMS); + } + + #[test] + fn strict_accepts_associated() { + let info = server_info(ProtocolVersion::V_2026_07_28); + assert!(enforce(&info, PeerRequestAssociation::Associated).is_ok()); + } + + #[test] + fn strict_unknown_falls_back_to_coarse_check() { + let info = server_info(ProtocolVersion::V_2026_07_28); + assert!( + enforce( + &info, + PeerRequestAssociation::Unknown { + has_pending_outbound_request: true + } + ) + .is_ok() + ); + assert!( + enforce( + &info, + PeerRequestAssociation::Unknown { + has_pending_outbound_request: false + } + ) + .is_err() + ); + } + + #[test] + fn legacy_protocol_accepts_even_unassociated() { + let info = server_info(ProtocolVersion::V_2025_11_25); + assert!(enforce(&info, PeerRequestAssociation::Unassociated).is_ok()); + } +} From 2daa630509c976c3575829def71ba4dd6688f0a0 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 00:08:25 +0200 Subject: [PATCH 02/11] feat(transport): record inbound stream origin on streamable HTTP client (#1033) The worker attaches an InboundStreamOrigin extension to each inbound server request: Unassociated for the standalone GET stream, OutboundRequest(id) for a POST's SSE stream. Mirror of the OriginatingRequestId marker used by the server side. --- crates/rmcp/src/service.rs | 17 ++++ .../src/transport/streamable_http_client.rs | 90 +++++++++++++++++-- 2 files changed, 102 insertions(+), 5 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index b9d5172b4..6355bdd97 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -231,6 +231,23 @@ pub(crate) fn in_request_handler_scope() -> bool { #[expect(clippy::exhaustive_structs, reason = "intentionally exhaustive")] pub struct OriginatingRequestId(pub RequestId); +/// Marker in an inbound request's non-serialized [`Extensions`] recording +/// which HTTP response stream it arrived on (SEP-2260): the receive-side +/// mirror of [`OriginatingRequestId`]. Attached by the streamable HTTP +/// client transport; read by the service layer to enforce the client +/// receive-side association check. Never on the wire (SEP-2260 defines no +/// wire field), so transports that serialize messages cannot convey it and +/// the service layer falls back to the coarse in-flight check. +#[derive(Debug, Clone, PartialEq, Eq)] +#[expect(clippy::exhaustive_enums, reason = "intentionally exhaustive")] +pub enum InboundStreamOrigin { + /// The standalone GET stream, or a POST response stream not tied to an + /// outbound request. + Unassociated, + /// The SSE response stream of the POST that carried this outbound request. + OutboundRequest(RequestId), +} + pub type TxJsonRpcMessage = JsonRpcMessage<::Req, ::Resp, ::Not>; pub type RxJsonRpcMessage = JsonRpcMessage< diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index c43fb7dd5..a0db6fc84 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -19,10 +19,11 @@ use super::common::client_side_sse::{ use crate::{ RoleClient, model::{ - ClientJsonRpcMessage, ClientNotification, ClientRequest, ErrorData, GetMeta, + ClientJsonRpcMessage, ClientNotification, ClientRequest, ErrorData, GetExtensions, GetMeta, InitializedNotification, JsonObject, ProtocolVersion, RequestId, ServerJsonRpcMessage, ServerResult, }, + service::InboundStreamOrigin, transport::{ common::{client_side_sse::SseAutoReconnectStream, mcp_headers}, worker::{Worker, WorkerQuitReason, WorkerSendRequest, WorkerTransport}, @@ -635,6 +636,7 @@ impl StreamableHttpClientWorker { + Send + 'static, sse_worker_tx: tokio::sync::mpsc::Sender, + origin: InboundStreamOrigin, close_on_response: bool, ct: CancellationToken, ) -> Result<(), StreamableHttpError> { @@ -649,9 +651,14 @@ impl StreamableHttpClientWorker { break; } }; - let Some(message) = message.transpose()? else { + let Some(mut message) = message.transpose()? else { break; }; + // SEP-2260: record which stream this request arrived on so the + // service layer can enforce receive-side request association. + if let ServerJsonRpcMessage::Request(request) = &mut message { + request.request.extensions_mut().insert(origin.clone()); + } let is_response = matches!( message, ServerJsonRpcMessage::Response(_) | ServerJsonRpcMessage::Error(_) @@ -719,6 +726,7 @@ impl StreamableHttpClientWorker { Self::execute_sse_stream( sse_stream, sse_worker_tx, + InboundStreamOrigin::Unassociated, false, transport_task_ct.child_token(), ) @@ -1283,9 +1291,18 @@ impl Worker for StreamableHttpClientWorker { ); } let stream_tx = sse_worker_tx.clone(); + let origin = match &stream_request_id { + Some(id) => { + InboundStreamOrigin::OutboundRequest( + id.clone(), + ) + } + None => InboundStreamOrigin::Unassociated, + }; streams.spawn(async move { let result = Self::execute_sse_stream( - sse_stream, stream_tx, true, stream_ct, + sse_stream, stream_tx, origin, true, + stream_ct, ) .await; (stream_request_id, result) @@ -1342,9 +1359,13 @@ impl Worker for StreamableHttpClientWorker { .insert(request_id.clone(), stream_ct.clone()); } let stream_tx = sse_worker_tx.clone(); + let origin = match &stream_request_id { + Some(id) => InboundStreamOrigin::OutboundRequest(id.clone()), + None => InboundStreamOrigin::Unassociated, + }; streams.spawn(async move { let result = Self::execute_sse_stream( - sse_stream, stream_tx, true, stream_ct, + sse_stream, stream_tx, origin, true, stream_ct, ) .await; (stream_request_id, result) @@ -1769,7 +1790,66 @@ mod tests { use serde_json::json; use super::*; - use crate::model::{ListToolsResult, NumberOrString, ServerResult, Tool}; + use crate::{ + model::{ + GetExtensions, ListToolsResult, NumberOrString, ServerRequest, ServerResult, Tool, + }, + service::InboundStreamOrigin, + }; + + #[expect( + deprecated, + reason = "sampling (SEP-2577) is still a representative server request" + )] + fn sampling_request_message(id: i64) -> ServerJsonRpcMessage { + use crate::model::{CreateMessageRequest, CreateMessageRequestParams, SamplingMessage}; + ServerJsonRpcMessage::request( + ServerRequest::CreateMessageRequest(CreateMessageRequest::new( + CreateMessageRequestParams::new(vec![SamplingMessage::user_text("hi")], 16), + )), + NumberOrString::Number(id), + ) + } + + #[tokio::test] + async fn execute_sse_stream_marks_inbound_requests_with_origin() { + for origin in [ + InboundStreamOrigin::Unassociated, + InboundStreamOrigin::OutboundRequest(RequestId::Number(3)), + ] { + let response = ServerJsonRpcMessage::response( + ServerResult::ListToolsResult(ListToolsResult::default()), + NumberOrString::Number(1), + ); + let stream = futures::stream::iter([Ok(sampling_request_message(9)), Ok(response)]); + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + StreamableHttpClientWorker::::execute_sse_stream( + stream, + tx, + origin.clone(), + false, + CancellationToken::new(), + ) + .await + .expect("stream completes"); + + let ServerJsonRpcMessage::Request(request) = + rx.recv().await.expect("request forwarded") + else { + panic!("expected request first"); + }; + assert_eq!( + request.request.extensions().get::(), + Some(&origin), + "inbound requests must carry their stream origin" + ); + // Responses are correlated by JSON-RPC id; no marker needed or added. + assert!(matches!( + rx.recv().await.expect("response forwarded"), + ServerJsonRpcMessage::Response(_) + )); + } + } type ReconnectAttempt = (Option, Option); From fb86279b3f55685fbc049b0d3a88e13a8bf9d797 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 00:22:53 +0200 Subject: [PATCH 03/11] feat(service): enforce SEP-2260 client receive-side check per stream origin (#1033) The event loop maps InboundStreamOrigin plus the in-flight responder pool to PeerRequestAssociation: restricted requests arriving on the standalone GET stream are now rejected with -32602 even while unrelated outbound requests are in flight. --- crates/rmcp/src/service.rs | 95 ++++++++++++++++++++++++++++++++++++-- 1 file changed, 92 insertions(+), 3 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 6355bdd97..8c7138f31 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -204,6 +204,28 @@ pub(crate) fn uses_legacy_lifecycle( && protocol_version.is_none_or(|version| version < &ProtocolVersion::V_2026_07_28) } +/// Translate the transport-attached [`InboundStreamOrigin`] marker (if any) +/// and the in-flight outbound request pool into the association passed to +/// [`ServiceRole::enforce_peer_request_association`]. +pub(crate) fn peer_request_association( + request: &Req, + local_responder_pool: &std::collections::HashMap, +) -> PeerRequestAssociation { + match request.extensions().get::() { + None => PeerRequestAssociation::Unknown { + has_pending_outbound_request: !local_responder_pool.is_empty(), + }, + Some(InboundStreamOrigin::Unassociated) => PeerRequestAssociation::Unassociated, + Some(InboundStreamOrigin::OutboundRequest(id)) => { + if local_responder_pool.contains_key(id) { + PeerRequestAssociation::Associated + } else { + PeerRequestAssociation::Unassociated + } + } + } +} + tokio::task_local! { pub(crate) static ORIGINATING_REQUEST: RequestId; } @@ -1487,9 +1509,7 @@ where if let Err(error) = R::enforce_peer_request_association( &request, peer.peer_info().as_deref(), - PeerRequestAssociation::Unknown { - has_pending_outbound_request: !local_responder_pool.is_empty(), - }, + peer_request_association(&request, &local_responder_pool), ) { tracing::warn!(%id, message = %error.message, "rejected peer request"); // send directly: the sink proxy path would drop the @@ -1770,4 +1790,73 @@ mod sep2260_marker_tests { let request = send_and_capture(None).await; assert!(request.extensions().get::().is_none()); } + + #[test] + #[expect( + deprecated, + reason = "Sampling is deprecated by SEP-2577 but remains the canonical restricted request" + )] + fn peer_request_association_maps_stream_origin() { + use std::collections::HashMap; + + use crate::model::{ + CreateMessageRequest, CreateMessageRequestParams, SamplingMessage, ServerRequest, + }; + + fn sampling(origin: Option) -> ServerRequest { + let mut request = CreateMessageRequest::new(CreateMessageRequestParams::new( + vec![SamplingMessage::user_text("hi")], + 16, + )); + if let Some(origin) = origin { + request.extensions.insert(origin); + } + ServerRequest::CreateMessageRequest(request) + } + + let empty: HashMap = HashMap::new(); + let in_flight: HashMap = HashMap::from([(RequestId::Number(7), ())]); + + // No marker (stdio): coarse signal. + assert_eq!( + peer_request_association(&sampling(None), &in_flight), + PeerRequestAssociation::Unknown { + has_pending_outbound_request: true + } + ); + assert_eq!( + peer_request_association(&sampling(None), &empty), + PeerRequestAssociation::Unknown { + has_pending_outbound_request: false + } + ); + // Standalone GET stream: unassociated even with requests in flight. + assert_eq!( + peer_request_association( + &sampling(Some(InboundStreamOrigin::Unassociated)), + &in_flight + ), + PeerRequestAssociation::Unassociated + ); + // Originating POST stream of an in-flight request: associated. + assert_eq!( + peer_request_association( + &sampling(Some(InboundStreamOrigin::OutboundRequest( + RequestId::Number(7) + ))), + &in_flight + ), + PeerRequestAssociation::Associated + ); + // Stream of a request that is no longer in flight: unassociated. + assert_eq!( + peer_request_association( + &sampling(Some(InboundStreamOrigin::OutboundRequest( + RequestId::Number(8) + ))), + &in_flight + ), + PeerRequestAssociation::Unassociated + ); + } } From e2f1fdc88956dbc6cfea8531b0564beb20ce5cda Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 00:32:12 +0200 Subject: [PATCH 04/11] test: end-to-end SEP-2260 stream-based enforcement over streamable HTTP (#1033) --- .../tests/test_sep_2260_stream_enforcement.rs | 276 ++++++++++++++++++ 1 file changed, 276 insertions(+) create mode 100644 crates/rmcp/tests/test_sep_2260_stream_enforcement.rs diff --git a/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs b/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs new file mode 100644 index 000000000..4246b7d68 --- /dev/null +++ b/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs @@ -0,0 +1,276 @@ +//! SEP-2260 follow-up (#1033): stream-based receive-side enforcement. +//! +//! Scripted streamable HTTP "server": answers a legacy initialize with +//! protocol 2026-07-28 and a session id (spec-legal; rmcp's own server is +//! stateless at that version, but the client must be correct against any +//! server), so the client has BOTH a standalone GET stream and strict +//! SEP-2260 enforcement. +#![cfg(all( + feature = "client", + feature = "transport-streamable-http-client", + not(feature = "local") +))] +#![allow(deprecated)] + +use std::{collections::HashMap, sync::Arc, time::Duration}; + +use futures::{StreamExt, stream::BoxStream}; +use http::{HeaderName, HeaderValue}; +use rmcp::{ + ClientHandler, + model::{ + ClientInfo, ClientJsonRpcMessage, CreateMessageRequestParams, CreateMessageResult, + ProtocolVersion, SamplingMessage, ServerCapabilities, ServerInfo, ServerJsonRpcMessage, + }, + service::{ClientLifecycleMode, RequestContext, RoleClient, serve_client_with_lifecycle}, + transport::streamable_http_client::{ + StreamableHttpClient, StreamableHttpClientTransport, StreamableHttpClientTransportConfig, + StreamableHttpError, StreamableHttpPostResponse, + }, +}; +use serde_json::{Value, json}; +use sse_stream::{Error as SseError, Sse}; +use tokio::sync::{Mutex, mpsc}; + +fn to_sse(message: Value) -> Result { + Ok(Sse { + event: None, + data: Some(message.to_string()), + id: None, + retry: None, + }) +} + +fn message_stream(rx: mpsc::Receiver) -> BoxStream<'static, Result> { + tokio_stream::wrappers::ReceiverStream::new(rx) + .map(to_sse) + .boxed() +} + +/// Scripted server: initialize -> JSON init result (2026-07-28 + session); +/// first non-initialize request POST -> SSE stream fed by `post_stream`; +/// everything else -> Accepted. Every message the client POSTs is forwarded +/// to `posted`. +#[derive(Clone)] +struct ScriptedServer { + get_stream: Arc>>>, + post_stream: Arc>>>, + posted: mpsc::UnboundedSender, +} + +impl StreamableHttpClient for ScriptedServer { + type Error = std::io::Error; + + async fn post_message( + &self, + _uri: Arc, + message: ClientJsonRpcMessage, + _session_id: Option>, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result> { + let value = serde_json::to_value(&message).expect("serialize client message"); + self.posted.send(value.clone()).expect("test alive"); + if value["method"] == "initialize" { + let mut info = ServerInfo::new(ServerCapabilities::default()); + info.protocol_version = ProtocolVersion::V_2026_07_28; + let response = ServerJsonRpcMessage::response( + rmcp::model::ServerResult::InitializeResult(info), + serde_json::from_value(value["id"].clone()).expect("request id"), + ); + return Ok(StreamableHttpPostResponse::Json( + response, + Some("scripted-session".into()), + )); + } + if matches!(message, ClientJsonRpcMessage::Request(_)) { + let rx = self + .post_stream + .lock() + .await + .take() + .expect("exactly one non-initialize request POST in this test"); + return Ok(StreamableHttpPostResponse::Sse(message_stream(rx), None)); + } + Ok(StreamableHttpPostResponse::Accepted) + } + + async fn delete_session( + &self, + _uri: Arc, + _session_id: Arc, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result<(), StreamableHttpError> { + Ok(()) + } + + async fn get_stream( + &self, + _uri: Arc, + _session_id: Option>, + _last_event_id: Option, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result>, StreamableHttpError> { + match self.get_stream.lock().await.take() { + Some(rx) => Ok(message_stream(rx)), + // Reconnect after the scripted stream ends: stay silent. + None => Ok(futures::stream::pending().boxed()), + } + } +} + +#[derive(Clone)] +struct SamplingClient { + // Some(tx): forward sampling params; None: sampling must never reach the handler. + on_sampling: Option>, +} + +impl ClientHandler for SamplingClient { + async fn create_message( + &self, + params: CreateMessageRequestParams, + _context: RequestContext, + ) -> Result { + let Some(tx) = &self.on_sampling else { + panic!("sampling request must not reach the handler in this test"); + }; + tx.send(params).expect("test alive"); + Ok(CreateMessageResult::new( + SamplingMessage::assistant_text("pong"), + "test-model".to_string(), + )) + } + + fn get_info(&self) -> ClientInfo { + ClientInfo::default() + } +} + +fn sampling_request(id: u32) -> Value { + json!({ + "jsonrpc": "2.0", + "id": id, + "method": "sampling/createMessage", + "params": { + "messages": [{ "role": "user", "content": { "type": "text", "text": "hi" } }], + "maxTokens": 16 + } + }) +} + +fn tools_list_response(id: &Value) -> Value { + json!({ "jsonrpc": "2.0", "id": id, "result": { "tools": [] } }) +} + +async fn next_posted(posted: &mut mpsc::UnboundedReceiver) -> Value { + tokio::time::timeout(Duration::from_secs(5), posted.recv()) + .await + .expect("posted message within 5s") + .expect("channel open") +} + +/// Drive startup + one in-flight tools/list; return (client, posted rx, +/// get stream tx, post stream tx, in-flight call handle, tools/list id). +async fn setup( + handler: SamplingClient, +) -> ( + rmcp::service::RunningService, + mpsc::UnboundedReceiver, + mpsc::Sender, + mpsc::Sender, + tokio::task::JoinHandle>, + Value, +) { + let (get_tx, get_rx) = mpsc::channel(8); + let (post_tx, post_rx) = mpsc::channel(8); + let (posted_tx, mut posted_rx) = mpsc::unbounded_channel(); + let server = ScriptedServer { + get_stream: Arc::new(Mutex::new(Some(get_rx))), + post_stream: Arc::new(Mutex::new(Some(post_rx))), + posted: posted_tx, + }; + let transport = StreamableHttpClientTransport::with_client( + server, + StreamableHttpClientTransportConfig::with_uri("http://scripted/mcp"), + ); + let client = serve_client_with_lifecycle(handler, transport, ClientLifecycleMode::Initialize) + .await + .expect("initialize against scripted server"); + + // initialize + notifications/initialized already posted during startup. + assert_eq!(next_posted(&mut posted_rx).await["method"], "initialize"); + assert_eq!( + next_posted(&mut posted_rx).await["method"], + "notifications/initialized" + ); + + // Unrelated outbound request, kept in flight (response withheld). + let peer = client.peer().clone(); + let call = tokio::spawn(async move { peer.list_tools(None).await }); + let tools_list = next_posted(&mut posted_rx).await; + assert_eq!(tools_list["method"], "tools/list"); + let tools_list_id = tools_list["id"].clone(); + + (client, posted_rx, get_tx, post_tx, call, tools_list_id) +} + +/// #1033 scenario 1: a restricted request on the standalone GET stream while +/// an unrelated outbound request is in flight must be rejected with -32602. +/// (The coarse check from #1029 incorrectly accepted this.) +#[tokio::test] +async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_flight() +-> anyhow::Result<()> { + let (client, mut posted_rx, get_tx, post_tx, call, tools_list_id) = + setup(SamplingClient { on_sampling: None }).await; + + get_tx.send(sampling_request(100)).await?; + + let rejection = next_posted(&mut posted_rx).await; + assert_eq!( + rejection["id"], 100, + "reply to the sampling request: {rejection}" + ); + assert_eq!( + rejection["error"]["code"], -32602, + "SEP-2260: GET-stream request must be rejected even with an unrelated \ + request in flight, got {rejection}" + ); + + post_tx.send(tools_list_response(&tools_list_id)).await?; + call.await??; + client.cancel().await?; + Ok(()) +} + +/// Positive twin: the same restricted request arriving on the SSE stream of +/// the originating POST is dispatched to the handler and answered. +#[tokio::test] +async fn restricted_request_on_originating_post_stream_is_dispatched() -> anyhow::Result<()> { + let (sampled_tx, mut sampled_rx) = mpsc::unbounded_channel(); + let (client, mut posted_rx, _get_tx, post_tx, call, tools_list_id) = setup(SamplingClient { + on_sampling: Some(sampled_tx), + }) + .await; + + post_tx.send(sampling_request(200)).await?; + + let response = next_posted(&mut posted_rx).await; + assert_eq!( + response["id"], 200, + "reply to the sampling request: {response}" + ); + assert_eq!( + response["result"]["model"], "test-model", + "request on the originating POST stream must reach the handler, got {response}" + ); + tokio::time::timeout(Duration::from_secs(5), sampled_rx.recv()) + .await? + .expect("handler invoked"); + + post_tx.send(tools_list_response(&tools_list_id)).await?; + call.await??; + client.cancel().await?; + Ok(()) +} From 5f28f9e9d16ed751a643354c4c33e7babd8a5b61 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 00:46:31 +0200 Subject: [PATCH 05/11] docs: cross-link SEP-2260 stream markers --- crates/rmcp/src/service.rs | 2 ++ 1 file changed, 2 insertions(+) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 8c7138f31..c1f7c25ec 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -249,6 +249,8 @@ pub(crate) fn in_request_handler_scope() -> bool { /// outside a handler they return an `invalid_request` error. The association /// is task-local and does not cross `tokio::spawn`, so use the task manager /// for long-running work. +/// +/// The client receive-side mirror is [`InboundStreamOrigin`]. #[derive(Debug, Clone, PartialEq, Eq)] #[expect(clippy::exhaustive_structs, reason = "intentionally exhaustive")] pub struct OriginatingRequestId(pub RequestId); From f0b09c0d2661f5a1b1056ca7bdba4e193023bf04 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 16:04:57 +0200 Subject: [PATCH 06/11] test: origin marker survives SSE resumption per SEP-2260 (#1033) A POST SSE stream resumed via GET + Last-Event-ID (SEP-1699) reconnects beneath execute_sse_stream, so requests replayed after a resume keep their OutboundRequest origin. Pins the layering invariant: hoisting reconnection above the marker attach point would wrongly reject associated requests with -32602. --- .../src/transport/streamable_http_client.rs | 126 ++++++++++++++++++ 1 file changed, 126 insertions(+) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index a0db6fc84..45668186c 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -1948,6 +1948,132 @@ mod tests { ); } + #[derive(Clone, Default)] + struct ResumedRequestClient { + reconnects: Arc>>, + } + + impl StreamableHttpClient for ResumedRequestClient { + type Error = std::io::Error; + + async fn post_message( + &self, + _uri: Arc, + _message: ClientJsonRpcMessage, + _session_id: Option>, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result> { + Err(StreamableHttpError::UnexpectedServerResponse( + "unexpected POST".into(), + )) + } + + async fn delete_session( + &self, + _uri: Arc, + _session_id: Arc, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result<(), StreamableHttpError> { + Ok(()) + } + + async fn get_stream( + &self, + _uri: Arc, + session_id: Option>, + last_event_id: Option, + _auth_header: Option, + _custom_headers: HashMap, + ) -> Result> { + self.reconnects + .lock() + .expect("lock reconnects") + .push((session_id.map(|id| id.to_string()), last_event_id)); + let request = sampling_request_message(9); + let response = ServerJsonRpcMessage::response( + ServerResult::ListToolsResult(ListToolsResult::default()), + NumberOrString::Number(1), + ); + // Stay open after the response, like a live connection, so the + // post-response drain in `execute_sse_stream` doesn't trigger + // further reconnects. + Ok(futures::stream::iter([request, response].map(|message| { + Ok(Sse { + event: None, + data: Some(serde_json::to_string(&message).expect("serialize message")), + id: None, + retry: None, + }) + })) + .chain(futures::stream::pending()) + .boxed()) + } + } + + /// SEP-1699 resumes a broken POST SSE stream via GET + Last-Event-ID + /// beneath `execute_sse_stream`, so the SEP-2260 origin marker must span + /// resumes; if reconnection were hoisted above the marker attach point, + /// replayed associated requests would be wrongly rejected with -32602. + #[tokio::test] + async fn resumed_post_stream_requests_keep_outbound_origin() { + let initial = futures::stream::iter([Ok(Sse { + event: None, + data: None, + id: Some("e1".into()), + retry: Some(0), + })]) + .boxed(); + let client = ResumedRequestClient::default(); + let reconnects = client.reconnects.clone(); + let sse_stream = + StreamableHttpClientWorker::::response_sse_to_jsonrpc( + initial, + None, + client, + Arc::from("http://localhost/mcp"), + None, + HashMap::new(), + DEFAULT_MAX_SSE_EVENT_SIZE, + Arc::new(ExponentialBackoff { + max_times: Some(1), + base_duration: Duration::ZERO, + }), + ); + + let origin = InboundStreamOrigin::OutboundRequest(RequestId::Number(3)); + let (tx, mut rx) = tokio::sync::mpsc::channel(4); + StreamableHttpClientWorker::::execute_sse_stream( + sse_stream, + tx, + origin.clone(), + true, + CancellationToken::new(), + ) + .await + .expect("stream completes"); + + assert_eq!( + reconnects.lock().expect("lock reconnects").as_slice(), + &[(None, Some("e1".into()))], + "the request must arrive on the resumed connection" + ); + let ServerJsonRpcMessage::Request(request) = rx.recv().await.expect("request forwarded") + else { + panic!("expected request first"); + }; + assert_eq!( + request.request.extensions().get::(), + Some(&origin), + "origin marker must survive SSE resumption" + ); + assert!(matches!( + rx.recv().await.expect("response forwarded"), + ServerJsonRpcMessage::Response(_) + )); + } + fn tool(name: &'static str, annotation: serde_json::Value) -> Tool { let schema = json!({ "type": "object", From 1dd0c9a536255c8a4f1f85d6b3bc826b03076e9a Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 17:51:05 +0200 Subject: [PATCH 07/11] docs: trim SEP-2260 comments to spec rationale --- crates/rmcp/src/service.rs | 31 +++++++------------ crates/rmcp/src/service/client.rs | 7 ++--- .../src/transport/streamable_http_client.rs | 4 +-- 3 files changed, 16 insertions(+), 26 deletions(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index c1f7c25ec..0f47fa9a2 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -163,8 +163,7 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { /// SEP-2260 says clients receiving a server-to-client request with no /// associated outbound request should reject it with invalid params. An /// error return is sent back to the peer instead of dispatching to the - /// handler. The `association` argument is the transport-observed stream - /// association; see [`PeerRequestAssociation`]. + /// handler. #[doc(hidden)] fn enforce_peer_request_association( _peer_request: &Self::PeerReq, @@ -176,23 +175,21 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { } /// How an inbound peer request relates to this side's in-flight outbound -/// requests, as observed by the transport (SEP-2260). +/// requests (SEP-2260). /// -/// SEP-2260 defines no wire field for request association; only a -/// stream-separating transport (streamable HTTP) can distinguish the first -/// two variants. Transports without stream separation (stdio, in-process) -/// yield [`Self::Unknown`], preserving the coarse in-flight check. +/// SEP-2260 defines no wire field for association, so only stream-separating +/// transports (streamable HTTP) can observe it; other transports yield +/// [`Self::Unknown`]. #[derive(Debug, Clone, PartialEq, Eq)] #[expect(clippy::exhaustive_enums, reason = "intentionally exhaustive")] pub enum PeerRequestAssociation { - /// Arrived on the response stream of an outbound request that is still - /// awaiting its response. + /// Arrived on the response stream of an in-flight outbound request. Associated, /// Arrived on a stream tied to no in-flight outbound request (e.g. the /// streamable HTTP standalone GET stream). Unassociated, - /// The transport supplied no stream provenance; only the coarse - /// "is anything in flight" signal exists. + /// The transport cannot distinguish streams; only the coarse in-flight + /// signal is available. Unknown { has_pending_outbound_request: bool }, } @@ -204,9 +201,6 @@ pub(crate) fn uses_legacy_lifecycle( && protocol_version.is_none_or(|version| version < &ProtocolVersion::V_2026_07_28) } -/// Translate the transport-attached [`InboundStreamOrigin`] marker (if any) -/// and the in-flight outbound request pool into the association passed to -/// [`ServiceRole::enforce_peer_request_association`]. pub(crate) fn peer_request_association( request: &Req, local_responder_pool: &std::collections::HashMap, @@ -256,12 +250,9 @@ pub(crate) fn in_request_handler_scope() -> bool { pub struct OriginatingRequestId(pub RequestId); /// Marker in an inbound request's non-serialized [`Extensions`] recording -/// which HTTP response stream it arrived on (SEP-2260): the receive-side -/// mirror of [`OriginatingRequestId`]. Attached by the streamable HTTP -/// client transport; read by the service layer to enforce the client -/// receive-side association check. Never on the wire (SEP-2260 defines no -/// wire field), so transports that serialize messages cannot convey it and -/// the service layer falls back to the coarse in-flight check. +/// which HTTP response stream it arrived on: the receive-side mirror of +/// [`OriginatingRequestId`]. Never on the wire (SEP-2260 defines no wire +/// field); when absent, the coarse in-flight check applies. #[derive(Debug, Clone, PartialEq, Eq)] #[expect(clippy::exhaustive_enums, reason = "intentionally exhaustive")] pub enum InboundStreamOrigin { diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 8832f9baf..b829611fe 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -238,10 +238,9 @@ impl ServiceRole for RoleClient { } } - // SEP-2260: reject restricted server requests that the transport observed - // arriving with no in-flight outbound request association. Transports - // without stream separation (stdio) report `Unknown`, where the coarse - // in-flight check is the best available under-approximation of the SHOULD. + // SEP-2260: reject restricted server requests that arrived unassociated + // with any in-flight outbound request. Without stream separation + // (`Unknown`) the coarse in-flight check under-approximates the SHOULD. fn enforce_peer_request_association( peer_request: &Self::PeerReq, peer_info: Option<&Self::PeerInfo>, diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 45668186c..675c90ce3 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -654,8 +654,8 @@ impl StreamableHttpClientWorker { let Some(mut message) = message.transpose()? else { break; }; - // SEP-2260: record which stream this request arrived on so the - // service layer can enforce receive-side request association. + // SEP-2260: mark inbound requests with the stream they arrived on + // for the client receive-side association check. if let ServerJsonRpcMessage::Request(request) = &mut message { request.request.extensions_mut().insert(origin.clone()); } From 482d3e2888d0c1068268adfee313f3e9670da4f2 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Mon, 27 Jul 2026 17:56:46 +0200 Subject: [PATCH 08/11] docs: align SEP-2577 expect reason with sibling suppressions --- crates/rmcp/src/transport/streamable_http_client.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 675c90ce3..fe563ee3f 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -1799,7 +1799,7 @@ mod tests { #[expect( deprecated, - reason = "sampling (SEP-2577) is still a representative server request" + reason = "Sampling is deprecated by SEP-2577 but remains the canonical restricted request" )] fn sampling_request_message(id: i64) -> ServerJsonRpcMessage { use crate::model::{CreateMessageRequest, CreateMessageRequestParams, SamplingMessage}; From a3c5b65d4ca8f295fa1cfd25a3402c44b067dcd9 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 29 Jul 2026 02:41:58 +0200 Subject: [PATCH 09/11] test: harden SEP-2260 stream-enforcement e2e (#1033) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Detect handler invocation via a channel asserted empty instead of a panic in a spawned task (swallowed, cannot fail the test); surface scripted-server misuse as transport errors rather than panics in the transport task; bound the tail awaits with 5s timeouts. Correct the header comment: a 2026-07-28 server minting a session id is not spec-legal (SEP-2567 removes sessions and the GET endpoint) — the scripted server is deliberately non-conforming, which is the point of receive-side enforcement. --- .../tests/test_sep_2260_stream_enforcement.rs | 135 +++++++++++------- 1 file changed, 84 insertions(+), 51 deletions(-) diff --git a/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs b/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs index 4246b7d68..36595e05d 100644 --- a/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs +++ b/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs @@ -1,16 +1,23 @@ //! SEP-2260 follow-up (#1033): stream-based receive-side enforcement. //! //! Scripted streamable HTTP "server": answers a legacy initialize with -//! protocol 2026-07-28 and a session id (spec-legal; rmcp's own server is -//! stateless at that version, but the client must be correct against any -//! server), so the client has BOTH a standalone GET stream and strict +//! protocol 2026-07-28 AND a session id. That combination is NOT +//! spec-compliant: the 2026-07-28 revision removes protocol-level sessions +//! and the standalone GET stream (SEP-2567; transports spec: "do not mint +//! or echo session IDs"). Receive-side enforcement (#1033) exists precisely +//! to protect the client from non-conforming servers, and rmcp's client +//! tolerates the session id and opens the standalone GET stream — so this +//! is the reachable path where the client has BOTH a GET stream and strict //! SEP-2260 enforcement. #![cfg(all( feature = "client", feature = "transport-streamable-http-client", not(feature = "local") ))] -#![allow(deprecated)] +#![expect( + deprecated, + reason = "Sampling is deprecated by SEP-2577 but remains the canonical restricted request" +)] use std::{collections::HashMap, sync::Arc, time::Duration}; @@ -70,7 +77,8 @@ impl StreamableHttpClient for ScriptedServer { _custom_headers: HashMap, ) -> Result> { let value = serde_json::to_value(&message).expect("serialize client message"); - self.posted.send(value.clone()).expect("test alive"); + // Receiver drop is normal at test teardown; never panic in the transport task. + let _ = self.posted.send(value.clone()); if value["method"] == "initialize" { let mut info = ServerInfo::new(ServerCapabilities::default()); info.protocol_version = ProtocolVersion::V_2026_07_28; @@ -84,12 +92,14 @@ impl StreamableHttpClient for ScriptedServer { )); } if matches!(message, ClientJsonRpcMessage::Request(_)) { - let rx = self - .post_stream - .lock() - .await - .take() - .expect("exactly one non-initialize request POST in this test"); + // Fail as a transport error rather than panicking: this code runs + // in the transport task, where a panic is swallowed and shows up + // only as an opaque timeout in the test. + let rx = self.post_stream.lock().await.take().ok_or_else(|| { + StreamableHttpError::Client(std::io::Error::other( + "scripted server expects exactly one non-initialize request POST", + )) + })?; return Ok(StreamableHttpPostResponse::Sse(message_stream(rx), None)); } Ok(StreamableHttpPostResponse::Accepted) @@ -123,8 +133,10 @@ impl StreamableHttpClient for ScriptedServer { #[derive(Clone)] struct SamplingClient { - // Some(tx): forward sampling params; None: sampling must never reach the handler. - on_sampling: Option>, + // Every invocation is recorded here. Tests assert on the receiver (the + // negative test asserts it stays empty); a panic in this handler would + // run in a spawned task and be silently swallowed. + sampled: mpsc::UnboundedSender, } impl ClientHandler for SamplingClient { @@ -133,10 +145,7 @@ impl ClientHandler for SamplingClient { params: CreateMessageRequestParams, _context: RequestContext, ) -> Result { - let Some(tx) = &self.on_sampling else { - panic!("sampling request must not reach the handler in this test"); - }; - tx.send(params).expect("test alive"); + let _ = self.sampled.send(params); Ok(CreateMessageResult::new( SamplingMessage::assistant_text("pong"), "test-model".to_string(), @@ -171,21 +180,27 @@ async fn next_posted(posted: &mut mpsc::UnboundedReceiver) -> Value { .expect("channel open") } -/// Drive startup + one in-flight tools/list; return (client, posted rx, -/// get stream tx, post stream tx, in-flight call handle, tools/list id). -async fn setup( - handler: SamplingClient, -) -> ( - rmcp::service::RunningService, - mpsc::UnboundedReceiver, - mpsc::Sender, - mpsc::Sender, - tokio::task::JoinHandle>, - Value, -) { +struct Harness { + client: rmcp::service::RunningService, + /// Every message the client POSTs to the scripted server. + posted: mpsc::UnboundedReceiver, + /// Feeds the standalone GET stream. + get_tx: mpsc::Sender, + /// Feeds the SSE stream of the in-flight tools/list POST. + post_tx: mpsc::Sender, + /// In-flight tools/list call (response withheld until the test releases it). + call: tokio::task::JoinHandle>, + tools_list_id: Value, + /// Sampling params seen by the client handler. + sampled: mpsc::UnboundedReceiver, +} + +/// Drive startup + one in-flight tools/list. +async fn setup() -> Harness { let (get_tx, get_rx) = mpsc::channel(8); let (post_tx, post_rx) = mpsc::channel(8); let (posted_tx, mut posted_rx) = mpsc::unbounded_channel(); + let (sampled_tx, sampled_rx) = mpsc::unbounded_channel(); let server = ScriptedServer { get_stream: Arc::new(Mutex::new(Some(get_rx))), post_stream: Arc::new(Mutex::new(Some(post_rx))), @@ -195,9 +210,15 @@ async fn setup( server, StreamableHttpClientTransportConfig::with_uri("http://scripted/mcp"), ); - let client = serve_client_with_lifecycle(handler, transport, ClientLifecycleMode::Initialize) - .await - .expect("initialize against scripted server"); + let client = serve_client_with_lifecycle( + SamplingClient { + sampled: sampled_tx, + }, + transport, + ClientLifecycleMode::Initialize, + ) + .await + .expect("initialize against scripted server"); // initialize + notifications/initialized already posted during startup. assert_eq!(next_posted(&mut posted_rx).await["method"], "initialize"); @@ -213,7 +234,15 @@ async fn setup( assert_eq!(tools_list["method"], "tools/list"); let tools_list_id = tools_list["id"].clone(); - (client, posted_rx, get_tx, post_tx, call, tools_list_id) + Harness { + client, + posted: posted_rx, + get_tx, + post_tx, + call, + tools_list_id, + sampled: sampled_rx, + } } /// #1033 scenario 1: a restricted request on the standalone GET stream while @@ -222,12 +251,11 @@ async fn setup( #[tokio::test] async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_flight() -> anyhow::Result<()> { - let (client, mut posted_rx, get_tx, post_tx, call, tools_list_id) = - setup(SamplingClient { on_sampling: None }).await; + let mut h = setup().await; - get_tx.send(sampling_request(100)).await?; + h.get_tx.send(sampling_request(100)).await?; - let rejection = next_posted(&mut posted_rx).await; + let rejection = next_posted(&mut h.posted).await; assert_eq!( rejection["id"], 100, "reply to the sampling request: {rejection}" @@ -238,9 +266,16 @@ async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_fl request in flight, got {rejection}" ); - post_tx.send(tools_list_response(&tools_list_id)).await?; - call.await??; - client.cancel().await?; + h.post_tx + .send(tools_list_response(&h.tools_list_id)) + .await?; + tokio::time::timeout(Duration::from_secs(5), h.call).await???; + // Rejected means rejected: the handler must not ALSO have been invoked. + assert!( + h.sampled.try_recv().is_err(), + "handler must not see the rejected sampling request" + ); + tokio::time::timeout(Duration::from_secs(5), h.client.cancel()).await??; Ok(()) } @@ -248,15 +283,11 @@ async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_fl /// the originating POST is dispatched to the handler and answered. #[tokio::test] async fn restricted_request_on_originating_post_stream_is_dispatched() -> anyhow::Result<()> { - let (sampled_tx, mut sampled_rx) = mpsc::unbounded_channel(); - let (client, mut posted_rx, _get_tx, post_tx, call, tools_list_id) = setup(SamplingClient { - on_sampling: Some(sampled_tx), - }) - .await; + let mut h = setup().await; - post_tx.send(sampling_request(200)).await?; + h.post_tx.send(sampling_request(200)).await?; - let response = next_posted(&mut posted_rx).await; + let response = next_posted(&mut h.posted).await; assert_eq!( response["id"], 200, "reply to the sampling request: {response}" @@ -265,12 +296,14 @@ async fn restricted_request_on_originating_post_stream_is_dispatched() -> anyhow response["result"]["model"], "test-model", "request on the originating POST stream must reach the handler, got {response}" ); - tokio::time::timeout(Duration::from_secs(5), sampled_rx.recv()) + tokio::time::timeout(Duration::from_secs(5), h.sampled.recv()) .await? .expect("handler invoked"); - post_tx.send(tools_list_response(&tools_list_id)).await?; - call.await??; - client.cancel().await?; + h.post_tx + .send(tools_list_response(&h.tools_list_id)) + .await?; + tokio::time::timeout(Duration::from_secs(5), h.call).await???; + tokio::time::timeout(Duration::from_secs(5), h.client.cancel()).await??; Ok(()) } From ac61fd836c76938ce0902fa5e0e2703f56164c39 Mon Sep 17 00:00:00 2001 From: Camille Lawrence Date: Wed, 29 Jul 2026 03:06:59 +0200 Subject: [PATCH 10/11] test: adapt SEP-2260 association tests to ServerPeerInfo Rebase onto 3.0.0: RoleClient::PeerInfo is now ServerPeerInfo (#1065) and its constructor takes the protocol version directly. --- crates/rmcp/src/service/client.rs | 11 ++++++----- 1 file changed, 6 insertions(+), 5 deletions(-) diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index b829611fe..114b958ba 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -2236,13 +2236,14 @@ mod sep2260_association_tests { )) } - fn server_info(version: ProtocolVersion) -> ServerInfo { - let mut info = ServerInfo::new(ServerCapabilities::default()); - info.protocol_version = version; - info + fn server_info(version: ProtocolVersion) -> ServerPeerInfo { + ServerPeerInfo::new(version, ServerCapabilities::default()) } - fn enforce(info: &ServerInfo, association: PeerRequestAssociation) -> Result<(), ErrorData> { + fn enforce( + info: &ServerPeerInfo, + association: PeerRequestAssociation, + ) -> Result<(), ErrorData> { RoleClient::enforce_peer_request_association(&sampling_request(), Some(info), association) } From 8e6e3420fce77c68be26a8aebd0c0255df38efa2 Mon Sep 17 00:00:00 2001 From: camillelawrence Date: Wed, 29 Jul 2026 17:26:06 -0400 Subject: [PATCH 11/11] chore: add 'Copy' macro to new PeerRequestAssociation Co-authored-by: Dale Seo <5466341+DaleSeo@users.noreply.github.com> --- crates/rmcp/src/service.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index 0f47fa9a2..f4fa24c07 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -180,7 +180,7 @@ pub trait ServiceRole: std::fmt::Debug + Send + Sync + 'static + Copy + Clone { /// SEP-2260 defines no wire field for association, so only stream-separating /// transports (streamable HTTP) can observe it; other transports yield /// [`Self::Unknown`]. -#[derive(Debug, Clone, PartialEq, Eq)] +#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[expect(clippy::exhaustive_enums, reason = "intentionally exhaustive")] pub enum PeerRequestAssociation { /// Arrived on the response stream of an in-flight outbound request.