From 06b54aaf78fb00a50a9e42c1d859b308bda8d858 Mon Sep 17 00:00:00 2001 From: Serhii Lizunov <34252204+LizunovSergey@users.noreply.github.com> Date: Fri, 11 Sep 2026 00:02:24 +0200 Subject: [PATCH 1/4] fix(transport): stop keeping a session and GET stream at 2026-07-28 SEP-2567 removes sessions and the standalone GET endpoint at 2026-07-28, so an Mcp-Session-Id and a GET stream are both artifacts of a pre-2026-07-28 server shape. The modern startup path already gets this right: a server/discover bootstrap sets session_id to None unconditionally. The legacy-shaped paths did not, and all three spawn_common_stream call sites were guarded only on a session id being present, never on the negotiated version: - legacy startup, where a server can answer legacy initialize with both a session id and protocolVersion 2026-07-28; - the fallback initialize taken after server/discover fails; - session re-establishment after an expired-session 404, which respawned on the new id without reconsulting the version. Gating happens where a session id is adopted rather than at each spawn: session_id_for_version drops the id once the negotiated version has no sessions, which leaves all three call sites with nothing to open and stops the id being echoed on requests. PeerRequestAssociation Unassociated then becomes unreachable over streamable HTTP instead of merely rejected, and the receive-side enforcement stays as the defence for anything that still slips through. A volunteered id is dropped with a warning rather than being treated as fatal, and the id is still recorded for the shutdown DELETE, so a session the server really did create is torn down. --- crates/rmcp/Cargo.toml | 10 + .../src/transport/streamable_http_client.rs | 34 +++ ...est_streamable_http_sessionless_version.rs | 214 ++++++++++++++++++ 3 files changed, 258 insertions(+) create mode 100644 crates/rmcp/tests/test_streamable_http_sessionless_version.rs diff --git a/crates/rmcp/Cargo.toml b/crates/rmcp/Cargo.toml index ac51ebfcd..2ccdba34e 100644 --- a/crates/rmcp/Cargo.toml +++ b/crates/rmcp/Cargo.toml @@ -338,6 +338,16 @@ required-features = [ ] path = "tests/test_discover_http_client_startup.rs" +[[test]] +name = "test_streamable_http_sessionless_version" +required-features = [ + "client", + "reqwest", + "transport-streamable-http-client-reqwest", + "transport-streamable-http-server", +] +path = "tests/test_streamable_http_sessionless_version.rs" + [[test]] name = "test_streamable_http_standard_headers" required-features = ["server", "client", "transport-streamable-http-server", "reqwest"] diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 6f27abf90..276661580 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -90,6 +90,37 @@ fn request_version_headers( (version, headers) } +/// Decides whether a session id survives version negotiation. +/// +/// SEP-2567 removes sessions and the standalone GET endpoint at +/// [`ProtocolVersion::STANDARD_HEADERS`], so at that version an `Mcp-Session-Id` and a GET +/// stream are both artifacts of a pre-`2026-07-28` server shape. A legacy-shaped handshake +/// can still answer with a session id while negotiating that version; the id is dropped +/// rather than echoed, which also leaves every `spawn_common_stream` call site — all three +/// of which are guarded on a session id being present — with no stream to open. +/// +/// Dropping is deliberate rather than fatal: refusing to start would break clients against +/// servers that work today, and the receive-side enforcement added for SEP-2260 still +/// rejects anything that reaches the client over a stream it should not have. The caller +/// keeps the original id for the shutdown `DELETE`, so a session the server really did +/// create is still torn down. +fn session_id_for_version( + session_id: Option>, + negotiated_version: &ProtocolVersion, +) -> Option> { + if negotiated_version < &ProtocolVersion::STANDARD_HEADERS { + return session_id; + } + if session_id.is_some() { + tracing::warn!( + version = negotiated_version.as_str(), + "server returned an Mcp-Session-Id while negotiating a version that has no sessions; \ + the id will not be sent on requests and no standalone GET stream will be opened" + ); + } + None +} + fn cache_tools_from_response( cache: &mut HashMap>, message: &mut ServerJsonRpcMessage, @@ -1133,6 +1164,7 @@ impl Worker for StreamableHttpClientWorker { auth_header: config.auth_header.clone(), protocol_headers: protocol_headers.clone(), }); + session_id = session_id_for_version(session_id, &negotiated_version); context.send_to_handler(message).await?; if is_legacy_startup { @@ -1261,6 +1293,7 @@ impl Worker for StreamableHttpClientWorker { auth_header: config.auth_header.clone(), protocol_headers: protocol_headers.clone(), }); + session_id = session_id_for_version(session_id, &negotiated_version); // Do not send controls queued during recovery to the new session. context.advance_control_generation(); session_cancellation = CancellationToken::new(); @@ -1517,6 +1550,7 @@ impl Worker for StreamableHttpClientWorker { auth_header: config.auth_header.clone(), protocol_headers: protocol_headers.clone(), }); + session_id = session_id_for_version(session_id, &negotiated_version); context.send_to_handler(initialize_response).await?; awaiting_fallback_initialized = true; continue; diff --git a/crates/rmcp/tests/test_streamable_http_sessionless_version.rs b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs new file mode 100644 index 000000000..ce95df369 --- /dev/null +++ b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs @@ -0,0 +1,214 @@ +#![cfg(all( + not(feature = "local"), + feature = "client", + feature = "reqwest", + feature = "transport-streamable-http-server" +))] +//! SEP-2567 removes sessions and the standalone GET stream at 2026-07-28. A legacy-shaped +//! handshake can still answer with an `Mcp-Session-Id` while negotiating that version; the +//! client must not then echo the id or open the stream. + +use std::sync::{Arc, Mutex}; + +use axum::{ + Router, + body::{Body, Bytes}, + extract::State, + http::{HeaderMap, Response, StatusCode}, + routing::any, +}; +use rmcp::{ + ClientLifecycleMode, ClientServiceExt, + model::ClientInfo, + transport::{ + StreamableHttpClientTransport, streamable_http_client::StreamableHttpClientTransportConfig, + }, +}; +use serde_json::json; +use tokio_util::sync::CancellationToken; + +const SESSION_ID: &str = "session-the-server-should-not-have-issued"; + +/// One request the client made: HTTP method, JSON-RPC method, `Mcp-Session-Id` header. +type Call = (String, String, Option); + +#[derive(Clone, Default)] +struct Recorder { + seen: Arc>>, + negotiated_version: Arc>, +} + +impl Recorder { + fn calls(&self) -> Vec { + self.seen.lock().expect("recorder poisoned").clone() + } + + fn get_requests(&self) -> Vec { + self.calls() + .into_iter() + .filter(|(http_method, ..)| http_method == "GET") + .collect() + } + + fn session_headers_after_handshake(&self) -> Vec> { + self.calls() + .into_iter() + .filter(|(_, jsonrpc_method, _)| jsonrpc_method != "initialize") + .map(|(.., session)| session) + .collect() + } +} + +async fn handler( + State(state): State, + method: axum::http::Method, + headers: HeaderMap, + body: Bytes, +) -> Response { + let session = headers + .get("mcp-session-id") + .and_then(|value| value.to_str().ok()) + .map(str::to_owned); + + if method == axum::http::Method::GET { + state.seen.lock().expect("recorder poisoned").push(( + "GET".to_owned(), + "-".to_owned(), + session, + )); + // Hang so the client keeps the stream if it opens one; the test cancels it. + return Response::builder() + .status(StatusCode::METHOD_NOT_ALLOWED) + .body(Body::empty()) + .expect("build GET rejection"); + } + + let request: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON-RPC body"); + let jsonrpc_method = request["method"].as_str().unwrap_or("-").to_owned(); + state.seen.lock().expect("recorder poisoned").push(( + "POST".to_owned(), + jsonrpc_method.clone(), + session, + )); + + if jsonrpc_method == "server/discover" { + // Force the legacy initialize path. + return Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Body::empty()) + .expect("build discover rejection"); + } + + if jsonrpc_method == "initialize" { + let version = state + .negotiated_version + .lock() + .expect("recorder poisoned") + .clone(); + // The shape this issue is about: a session id alongside a negotiated version + // that, from 2026-07-28 on, has no sessions at all. + return Response::builder() + .status(StatusCode::OK) + .header("content-type", "application/json") + .header("mcp-session-id", SESSION_ID) + .body(Body::from( + json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { + "protocolVersion": version, + "capabilities": {}, + "serverInfo": {"name": "dual-era", "version": "1.0"} + } + }) + .to_string(), + )) + .expect("build initialize response"); + } + + Response::builder() + .status(StatusCode::ACCEPTED) + .body(Body::empty()) + .expect("build notification response") +} + +async fn connect_and_record(negotiated_version: &str) -> Recorder { + let recorder = Recorder::default(); + *recorder + .negotiated_version + .lock() + .expect("recorder poisoned") = negotiated_version.to_owned(); + + let ct = CancellationToken::new(); + let router = Router::new() + .route("/mcp", any(handler)) + .with_state(recorder.clone()); + let listener = tokio::net::TcpListener::bind("127.0.0.1:0") + .await + .expect("listener should bind"); + let address = listener.local_addr().expect("listener address"); + let server = tokio::spawn({ + let ct = ct.clone(); + async move { + let _ = axum::serve(listener, router) + .with_graceful_shutdown(async move { ct.cancelled_owned().await }) + .await; + } + }); + + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(format!("http://{address}/mcp")), + ); + let client = ClientInfo::default() + .serve_with_lifecycle(transport, ClientLifecycleMode::Initialize) + .await + .expect("client should start against a legacy handshake"); + + // Give a standalone GET stream, if one were opened, time to reach the server. + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + client.cancel().await.expect("cancel client"); + ct.cancel(); + let _ = server.await; + recorder +} + +#[tokio::test] +async fn modern_version_drops_the_session_and_opens_no_stream() { + let recorder = connect_and_record("2026-07-28").await; + + // No standalone GET stream: SEP-2567 removed the endpoint at this version. + assert_eq!( + recorder.get_requests(), + Vec::new(), + "client opened a standalone GET stream at a version that has none" + ); + // And the id the server volunteered is not echoed back on anything. + let sessions = recorder.session_headers_after_handshake(); + assert!( + !sessions.is_empty(), + "expected at least one post-handshake request to inspect" + ); + assert!( + sessions.iter().all(Option::is_none), + "client echoed Mcp-Session-Id at a version with no sessions: {sessions:?}" + ); +} + +#[tokio::test] +async fn legacy_version_keeps_the_session_and_opens_the_stream() { + let recorder = connect_and_record("2025-11-25").await; + + // The legacy shape is untouched: the stream is opened and carries the session id. + let gets = recorder.get_requests(); + assert_eq!( + gets.len(), + 1, + "expected exactly one standalone GET stream on a legacy session, got {gets:?}" + ); + assert_eq!(gets[0].2.as_deref(), Some(SESSION_ID)); + let sessions = recorder.session_headers_after_handshake(); + assert!( + sessions.iter().any(|s| s.as_deref() == Some(SESSION_ID)), + "legacy session id was not echoed after the handshake: {sessions:?}" + ); +} From ac58bf5b333b4f5d45cfd74bb76a1cbd292f7786 Mon Sep 17 00:00:00 2001 From: Serhii Lizunov <34252204+LizunovSergey@users.noreply.github.com> Date: Sat, 12 Sep 2026 09:51:34 +0200 Subject: [PATCH 2/4] test(sep-2260): drive enforcement from a stream that survives the fix The harness reached an unassociated stream via the standalone GET stream, which required the client to keep a session while negotiating 2026-07-28. That combination no longer exists, so the scenario was unreachable and the test timed out waiting for a rejection. Enforcement is strict only at >= 2026-07-28, so dropping to a legacy version would have disabled the very check under test. Use the SSE body of a post-startup notification POST instead: it carries no request id, so the stream is Unassociated at a strict protocol version, with an unrelated request still in flight. The scripted server also stops minting a session it is not entitled to at that version. --- .../tests/test_sep_2260_stream_enforcement.rs | 94 ++++++++++++------- 1 file changed, 61 insertions(+), 33 deletions(-) diff --git a/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs b/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs index 36595e05d..fb6973aaf 100644 --- a/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs +++ b/crates/rmcp/tests/test_sep_2260_stream_enforcement.rs @@ -1,14 +1,17 @@ //! 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. 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. +//! Scripted streamable HTTP "server" negotiating 2026-07-28, which is the +//! range where enforcement is strict (`enforce_peer_request_association` +//! only tightens at `>= V_2026_07_28`). +//! +//! The unassociated stream here is the SSE body the server returns for the +//! `notifications/initialized` POST. A POST carrying no request id gets +//! `InboundStreamOrigin::Unassociated`, so a server request arriving on it +//! is unassociated by construction — the same condition the standalone GET +//! stream used to provide, minus the session. Answering a notification POST +//! with a stream instead of `202 Accepted` is itself server misbehaviour, +//! which is the point: receive-side enforcement exists to protect the +//! client from non-conforming servers. #![cfg(all( feature = "client", feature = "transport-streamable-http-client", @@ -54,13 +57,14 @@ fn message_stream(rx: mpsc::Receiver) -> BoxStream<'static, Result JSON init result (2026-07-28 + session); +/// Scripted server: initialize -> JSON init result (2026-07-28, no session); /// first non-initialize request POST -> SSE stream fed by `post_stream`; +/// first notification POST -> SSE stream fed by `notification_stream`; /// everything else -> Accepted. Every message the client POSTs is forwarded /// to `posted`. #[derive(Clone)] struct ScriptedServer { - get_stream: Arc>>>, + notification_stream: Arc>>>, post_stream: Arc>>>, posted: mpsc::UnboundedSender, } @@ -86,10 +90,9 @@ impl StreamableHttpClient for ScriptedServer { rmcp::model::ServerResult::InitializeResult(info), serde_json::from_value(value["id"].clone()).expect("request id"), ); - return Ok(StreamableHttpPostResponse::Json( - response, - Some("scripted-session".into()), - )); + // No session id: 2026-07-28 removed protocol-level sessions + // (SEP-2567), so a conforming server mints none. + return Ok(StreamableHttpPostResponse::Json(response, None)); } if matches!(message, ClientJsonRpcMessage::Request(_)) { // Fail as a transport error rather than panicking: this code runs @@ -102,6 +105,16 @@ impl StreamableHttpClient for ScriptedServer { })?; return Ok(StreamableHttpPostResponse::Sse(message_stream(rx), None)); } + // A notification POST carries no request id, so the stream the client + // opens for it is `Unassociated`. `notifications/initialized` is + // excluded: startup requires `202 Accepted` or JSON there and treats a + // stream as fatal, so the harness uses a post-startup notification. + if matches!(message, ClientJsonRpcMessage::Notification(_)) + && value["method"] != "notifications/initialized" + && let Some(rx) = self.notification_stream.lock().await.take() + { + return Ok(StreamableHttpPostResponse::Sse(message_stream(rx), None)); + } Ok(StreamableHttpPostResponse::Accepted) } @@ -123,11 +136,10 @@ impl StreamableHttpClient for ScriptedServer { _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()), - } + // Unreached: with no session there is no standalone GET stream. Left + // silent rather than failing so an auto-reconnect after a scripted + // stream ends at teardown cannot surface as a spurious error. + Ok(futures::stream::pending().boxed()) } } @@ -184,8 +196,9 @@ 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 unassociated stream (the SSE body of the initialized + /// notification POST). + unassociated_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). @@ -197,12 +210,12 @@ struct Harness { /// Drive startup + one in-flight tools/list. async fn setup() -> Harness { - let (get_tx, get_rx) = mpsc::channel(8); + let (unassociated_tx, unassociated_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))), + notification_stream: Arc::new(Mutex::new(Some(unassociated_rx))), post_stream: Arc::new(Mutex::new(Some(post_rx))), posted: posted_tx, }; @@ -229,15 +242,29 @@ async fn setup() -> Harness { // 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 call = tokio::spawn({ + let peer = peer.clone(); + 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(); + // Open the unassociated stream. Any post-startup notification does: the + // POST carries no request id, so the SSE body the server returns for it + // is `InboundStreamOrigin::Unassociated`. + peer.notify_roots_list_changed() + .await + .expect("send roots/list_changed"); + assert_eq!( + next_posted(&mut posted_rx).await["method"], + "notifications/roots/list_changed" + ); + Harness { client, posted: posted_rx, - get_tx, + unassociated_tx, post_tx, call, tools_list_id, @@ -245,15 +272,16 @@ async fn setup() -> Harness { } } -/// #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.) +/// #1033 scenario 1: a restricted request on a stream unassociated with any +/// outbound request, while an unrelated outbound request is in flight, must +/// be rejected with -32602. (The coarse check from #1029 incorrectly accepted +/// this, because it only asked whether *any* request was in flight.) #[tokio::test] -async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_flight() +async fn restricted_request_on_unassociated_stream_rejected_while_unrelated_request_in_flight() -> anyhow::Result<()> { let mut h = setup().await; - h.get_tx.send(sampling_request(100)).await?; + h.unassociated_tx.send(sampling_request(100)).await?; let rejection = next_posted(&mut h.posted).await; assert_eq!( @@ -262,8 +290,8 @@ async fn restricted_request_on_get_stream_rejected_while_unrelated_request_in_fl ); assert_eq!( rejection["error"]["code"], -32602, - "SEP-2260: GET-stream request must be rejected even with an unrelated \ - request in flight, got {rejection}" + "SEP-2260: unassociated-stream request must be rejected even with an \ + unrelated request in flight, got {rejection}" ); h.post_tx From 7566a73661c8b114759a6f3cc3a977063d5a463b Mon Sep 17 00:00:00 2001 From: Serhii Lizunov <34252204+LizunovSergey@users.noreply.github.com> Date: Sat, 12 Sep 2026 22:08:03 +0200 Subject: [PATCH 3/4] fix(transport): drop the session before the recovery handshake completes The gate was applied to the id `perform_reinitialization` returned, but that function has already POSTed `notifications/initialized` on the new session by then. At 2026-07-28 that one request still echoed an Mcp-Session-Id. Move the gate inside `perform_reinitialization`, where the startup path already has it relative to its own `initialized` notification, and return the id as the server sent it alongside the gated one so the shutdown DELETE is unaffected. --- .../src/transport/streamable_http_client.rs | 55 +++-- ...est_streamable_http_sessionless_version.rs | 214 ++++++++++++++++-- 2 files changed, 233 insertions(+), 36 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 276661580..c7bd57a01 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -121,6 +121,20 @@ fn session_id_for_version( None } +/// The session established by [`StreamableHttpClientWorker::perform_reinitialization`]. +/// +/// The two ids are the same id at different stages of [`session_id_for_version`], and they +/// are deliberately not interchangeable: `cleanup_session_id` is what the server sent, kept +/// so the shutdown `DELETE` still tears down a session the server really created, while +/// `session_id` is what may be echoed on requests and used to open a standalone GET stream. +/// At [`ProtocolVersion::STANDARD_HEADERS`] the latter is always `None`. +struct Reinitialized { + session_id: Option>, + cleanup_session_id: Option>, + negotiated_version: ProtocolVersion, + protocol_headers: HashMap, +} + fn cache_tools_from_response( cache: &mut HashMap>, message: &mut ServerJsonRpcMessage, @@ -981,9 +995,13 @@ impl StreamableHttpClientWorker { /// future remains `Send` without requiring `C: Sync`). POSTs the saved /// initialize request without a session ID, extracts the new session ID and /// protocol version, sends `notifications/initialized`, and returns the new - /// `(session_id, protocol_headers)` pair. The init result message is **not** + /// session in a [`Reinitialized`]. The init result message is **not** /// forwarded to the handler because the handler already processed the original /// initialization. + /// + /// The handshake completes here, so [`session_id_for_version`] is applied here too: + /// the `initialized` notification is part of the new session and must not echo an id + /// the negotiated version has no sessions for. async fn perform_reinitialization( client: C, saved_init_request: ClientJsonRpcMessage, @@ -991,14 +1009,7 @@ impl StreamableHttpClientWorker { auth_header: Option, custom_headers: HashMap, max_sse_event_size: usize, - ) -> Result< - ( - Option>, - ProtocolVersion, - HashMap, - ), - StreamableHttpError, - > { + ) -> Result> { let (init_msg, new_session_id_str) = client .post_message_with_max_sse_event_size( uri.clone(), @@ -1012,10 +1023,13 @@ impl StreamableHttpClientWorker { .expect_initialized::() .await?; - let new_session_id: Option> = new_session_id_str.map(|s| Arc::from(s.as_str())); + let cleanup_session_id: Option> = + new_session_id_str.map(|s| Arc::from(s.as_str())); let (negotiated_version, new_protocol_headers) = negotiate_version_headers(&init_msg, custom_headers); + let new_session_id = + session_id_for_version(cleanup_session_id.clone(), &negotiated_version); let initialized_notification = ClientJsonRpcMessage::notification( ClientNotification::InitializedNotification(InitializedNotification { @@ -1042,7 +1056,12 @@ impl StreamableHttpClientWorker { .await? .expect_accepted_or_json::()?; - Ok((new_session_id, negotiated_version, new_protocol_headers)) + Ok(Reinitialized { + session_id: new_session_id, + cleanup_session_id, + negotiated_version, + protocol_headers: new_protocol_headers, + }) } } @@ -1267,7 +1286,12 @@ impl Worker for StreamableHttpClientWorker { ) => result.unwrap_or(Err(StreamableHttpError::SessionRecoveryTimeout)), }; match recovery { - Ok((new_session_id, new_version, new_headers)) => { + Ok(Reinitialized { + session_id: new_session_id, + cleanup_session_id, + negotiated_version: new_version, + protocol_headers: new_headers, + }) => { streams.abort_all(); while streams.join_next().await.is_some() {} request_stream_cancellations.clear(); @@ -1286,14 +1310,15 @@ impl Worker for StreamableHttpClientWorker { session_id = new_session_id; negotiated_version = new_version; protocol_headers = new_headers; - session_cleanup_info = session_id.as_ref().map(|sid| SessionCleanupInfo { + // Built from the id as sent, not the gated one, so a session the + // server really created is still torn down at shutdown. + session_cleanup_info = cleanup_session_id.map(|sid| SessionCleanupInfo { client: self.client.clone(), uri: config.uri.clone(), - session_id: sid.clone(), + session_id: sid, auth_header: config.auth_header.clone(), protocol_headers: protocol_headers.clone(), }); - session_id = session_id_for_version(session_id, &negotiated_version); // Do not send controls queued during recovery to the new session. context.advance_control_generation(); session_cancellation = CancellationToken::new(); diff --git a/crates/rmcp/tests/test_streamable_http_sessionless_version.rs b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs index ce95df369..d2c8bb204 100644 --- a/crates/rmcp/tests/test_streamable_http_sessionless_version.rs +++ b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs @@ -6,9 +6,13 @@ ))] //! SEP-2567 removes sessions and the standalone GET stream at 2026-07-28. A legacy-shaped //! handshake can still answer with an `Mcp-Session-Id` while negotiating that version; the -//! client must not then echo the id or open the stream. +//! client must not then echo the id or open the stream. That holds for the replacement +//! handshake after an expired-session 404 as much as for the first one. -use std::sync::{Arc, Mutex}; +use std::{ + collections::VecDeque, + sync::{Arc, Mutex}, +}; use axum::{ Router, @@ -28,17 +32,61 @@ use serde_json::json; use tokio_util::sync::CancellationToken; const SESSION_ID: &str = "session-the-server-should-not-have-issued"; +const REPLACEMENT_SESSION_ID: &str = "session-issued-by-the-replacement-handshake"; /// One request the client made: HTTP method, JSON-RPC method, `Mcp-Session-Id` header. type Call = (String, String, Option); -#[derive(Clone, Default)] +/// How the server answers one `initialize`: the id it volunteers and the version it settles on. +type Handshake = (&'static str, &'static str); + +#[derive(Clone)] struct Recorder { seen: Arc>>, - negotiated_version: Arc>, + /// Consumed one per `initialize`, in order; the last entry answers every later one. + handshakes: Arc>>, + /// While set, the next `tools/list` is refused with a 404 to expire the session. + expire_next_list: Arc>, } impl Recorder { + /// A server that answers every handshake the same way and never expires a session. + fn new(handshake: Handshake) -> Self { + Self { + seen: Arc::new(Mutex::new(Vec::new())), + handshakes: Arc::new(Mutex::new(VecDeque::from([handshake]))), + expire_next_list: Arc::new(Mutex::new(false)), + } + } + + /// A server that expires the first session, then answers the replacement handshake + /// differently from the first. + fn expiring(first: Handshake, replacement: Handshake) -> Self { + Self { + seen: Arc::new(Mutex::new(Vec::new())), + handshakes: Arc::new(Mutex::new(VecDeque::from([first, replacement]))), + expire_next_list: Arc::new(Mutex::new(true)), + } + } + + fn next_handshake(&self) -> Handshake { + let mut handshakes = self.handshakes.lock().expect("recorder poisoned"); + if handshakes.len() > 1 { + handshakes.pop_front().expect("checked non-empty") + } else { + *handshakes + .front() + .expect("at least one handshake is scripted") + } + } + + fn take_expiry(&self) -> bool { + std::mem::replace( + &mut self.expire_next_list.lock().expect("recorder poisoned"), + false, + ) + } + fn calls(&self) -> Vec { self.seen.lock().expect("recorder poisoned").clone() } @@ -57,6 +105,19 @@ impl Recorder { .map(|(.., session)| session) .collect() } + + /// Everything the client sent after the last `initialize`, i.e. on the session that + /// handshake established. Empty if it never re-initialized. + fn calls_on_replacement_session(&self) -> Vec { + let calls = self.calls(); + match calls + .iter() + .rposition(|(_, jsonrpc_method, _)| jsonrpc_method == "initialize") + { + Some(last_initialize) => calls[last_initialize + 1..].to_vec(), + None => Vec::new(), + } + } } async fn handler( @@ -100,24 +161,20 @@ async fn handler( } if jsonrpc_method == "initialize" { - let version = state - .negotiated_version - .lock() - .expect("recorder poisoned") - .clone(); + let (session_id, version) = state.next_handshake(); // The shape this issue is about: a session id alongside a negotiated version // that, from 2026-07-28 on, has no sessions at all. return Response::builder() .status(StatusCode::OK) .header("content-type", "application/json") - .header("mcp-session-id", SESSION_ID) + .header("mcp-session-id", session_id) .body(Body::from( json!({ "jsonrpc": "2.0", "id": request["id"], "result": { "protocolVersion": version, - "capabilities": {}, + "capabilities": {"tools": {}}, "serverInfo": {"name": "dual-era", "version": "1.0"} } }) @@ -126,19 +183,36 @@ async fn handler( .expect("build initialize response"); } + if jsonrpc_method == "tools/list" { + if state.take_expiry() { + // The expired-session 404 that sends the client through re-initialization. + return Response::builder() + .status(StatusCode::NOT_FOUND) + .body(Body::empty()) + .expect("build session-expired rejection"); + } + return Response::builder() + .status(StatusCode::OK) + .header("content-type", "application/json") + .body(Body::from( + json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": {"tools": []} + }) + .to_string(), + )) + .expect("build tools/list response"); + } + Response::builder() .status(StatusCode::ACCEPTED) .body(Body::empty()) .expect("build notification response") } -async fn connect_and_record(negotiated_version: &str) -> Recorder { - let recorder = Recorder::default(); - *recorder - .negotiated_version - .lock() - .expect("recorder poisoned") = negotiated_version.to_owned(); - +/// Serve `recorder` on a loopback port until the returned token is cancelled. +async fn serve(recorder: &Recorder) -> (String, CancellationToken, tokio::task::JoinHandle<()>) { let ct = CancellationToken::new(); let router = Router::new() .route("/mcp", any(handler)) @@ -155,14 +229,23 @@ async fn connect_and_record(negotiated_version: &str) -> Recorder { .await; } }); + (format!("http://{address}/mcp"), ct, server) +} +async fn start_client(uri: String) -> rmcp::service::RunningService { let transport = StreamableHttpClientTransport::from_config( - StreamableHttpClientTransportConfig::with_uri(format!("http://{address}/mcp")), + StreamableHttpClientTransportConfig::with_uri(uri), ); - let client = ClientInfo::default() + ClientInfo::default() .serve_with_lifecycle(transport, ClientLifecycleMode::Initialize) .await - .expect("client should start against a legacy handshake"); + .expect("client should start against a legacy handshake") +} + +async fn connect_and_record(negotiated_version: &'static str) -> Recorder { + let recorder = Recorder::new((SESSION_ID, negotiated_version)); + let (uri, ct, server) = serve(&recorder).await; + let client = start_client(uri).await; // Give a standalone GET stream, if one were opened, time to reach the server. tokio::time::sleep(std::time::Duration::from_millis(300)).await; @@ -172,6 +255,32 @@ async fn connect_and_record(negotiated_version: &str) -> Recorder { recorder } +/// Start on a legacy session, let the server expire it, and come back on a replacement +/// handshake that negotiates `replacement_version`. +async fn connect_through_recovery(replacement_version: &'static str) -> Recorder { + let recorder = Recorder::expiring( + (SESSION_ID, "2025-11-25"), + (REPLACEMENT_SESSION_ID, replacement_version), + ); + let (uri, ct, server) = serve(&recorder).await; + let client = start_client(uri).await; + + // The first tools/list is answered 404; the transport re-initializes and retries it. + client + .peer() + .list_tools(None) + .await + .expect("the retry after re-initialization should succeed"); + + // Give a standalone GET stream on the replacement session, if one were opened, time + // to reach the server. + tokio::time::sleep(std::time::Duration::from_millis(300)).await; + client.cancel().await.expect("cancel client"); + ct.cancel(); + let _ = server.await; + recorder +} + #[tokio::test] async fn modern_version_drops_the_session_and_opens_no_stream() { let recorder = connect_and_record("2026-07-28").await; @@ -212,3 +321,66 @@ async fn legacy_version_keeps_the_session_and_opens_the_stream() { "legacy session id was not echoed after the handshake: {sessions:?}" ); } + +#[tokio::test] +async fn replacement_handshake_at_a_modern_version_drops_the_session_too() { + let recorder = connect_through_recovery("2026-07-28").await; + + let replacement = recorder.calls_on_replacement_session(); + assert!( + !replacement.is_empty(), + "the session was never re-established: {:?}", + recorder.calls() + ); + + // The `initialized` notification completes the replacement handshake, so it is part of + // the new session and must not carry an id that version has no sessions for. + let initialized = replacement + .iter() + .find(|(_, jsonrpc_method, _)| jsonrpc_method == "notifications/initialized") + .expect("replacement handshake should send notifications/initialized"); + assert_eq!( + initialized.2, None, + "re-initialization echoed Mcp-Session-Id on the notification that completes it" + ); + + // Nor on anything after it, and no stream on the replacement session either. + let echoed: Vec<_> = replacement + .iter() + .filter(|(.., session)| session.is_some()) + .collect(); + assert!( + echoed.is_empty(), + "client echoed Mcp-Session-Id on the replacement session: {echoed:?}" + ); + assert_eq!( + recorder.get_requests().len(), + 1, + "expected only the legacy session's GET stream, got {:?}", + recorder.get_requests() + ); +} + +#[tokio::test] +async fn replacement_handshake_at_a_legacy_version_keeps_its_session() { + let recorder = connect_through_recovery("2025-11-25").await; + + let replacement = recorder.calls_on_replacement_session(); + assert!( + !replacement.is_empty(), + "the session was never re-established: {:?}", + recorder.calls() + ); + assert!( + replacement + .iter() + .any(|(.., session)| session.as_deref() == Some(REPLACEMENT_SESSION_ID)), + "replacement session id was not echoed on a legacy session: {replacement:?}" + ); + assert_eq!( + recorder.get_requests().len(), + 2, + "expected a GET stream on each legacy session, got {:?}", + recorder.get_requests() + ); +} From d36ce251b0914fbec309c61bb9244a8ab7afaa12 Mon Sep 17 00:00:00 2001 From: Serhii Lizunov <34252204+LizunovSergey@users.noreply.github.com> Date: Sat, 12 Sep 2026 22:11:17 +0200 Subject: [PATCH 4/4] test(sep-2567): pin the replacement handshake and its teardown Records the shutdown DELETE instead of parsing its empty body as JSON, so the teardown is assertable rather than a panic in the server task. --- ...est_streamable_http_sessionless_version.rs | 38 ++++++++++++++++++- 1 file changed, 36 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/tests/test_streamable_http_sessionless_version.rs b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs index d2c8bb204..80b06a78d 100644 --- a/crates/rmcp/tests/test_streamable_http_sessionless_version.rs +++ b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs @@ -98,10 +98,21 @@ impl Recorder { .collect() } + fn delete_requests(&self) -> Vec { + self.calls() + .into_iter() + .filter(|(http_method, ..)| http_method == "DELETE") + .collect() + } + + /// Session headers on everything after the handshake that is not the teardown DELETE, + /// which carries the id on purpose so the server tears the session down. fn session_headers_after_handshake(&self) -> Vec> { self.calls() .into_iter() - .filter(|(_, jsonrpc_method, _)| jsonrpc_method != "initialize") + .filter(|(http_method, jsonrpc_method, _)| { + jsonrpc_method != "initialize" && http_method != "DELETE" + }) .map(|(.., session)| session) .collect() } @@ -144,6 +155,19 @@ async fn handler( .expect("build GET rejection"); } + // The shutdown DELETE carries no body, so record it before anything parses one. + if method == axum::http::Method::DELETE { + state.seen.lock().expect("recorder poisoned").push(( + "DELETE".to_owned(), + "-".to_owned(), + session, + )); + return Response::builder() + .status(StatusCode::OK) + .body(Body::empty()) + .expect("build session teardown response"); + } + let request: serde_json::Value = serde_json::from_slice(&body).expect("valid JSON-RPC body"); let jsonrpc_method = request["method"].as_str().unwrap_or("-").to_owned(); state.seen.lock().expect("recorder poisoned").push(( @@ -347,7 +371,7 @@ async fn replacement_handshake_at_a_modern_version_drops_the_session_too() { // Nor on anything after it, and no stream on the replacement session either. let echoed: Vec<_> = replacement .iter() - .filter(|(.., session)| session.is_some()) + .filter(|(http_method, _, session)| session.is_some() && http_method != "DELETE") .collect(); assert!( echoed.is_empty(), @@ -359,6 +383,16 @@ async fn replacement_handshake_at_a_modern_version_drops_the_session_too() { "expected only the legacy session's GET stream, got {:?}", recorder.get_requests() ); + + // Dropping the id is a request-and-stream decision, not a licence to leak server + // state: the session the server really did create is still torn down at shutdown. + let deletes = recorder.delete_requests(); + assert!( + deletes + .iter() + .any(|(.., session)| session.as_deref() == Some(REPLACEMENT_SESSION_ID)), + "the replacement session was never deleted at shutdown: {deletes:?}" + ); } #[tokio::test]