fix(transport): stop keeping a session and GET stream at 2026-07-28

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"]

crates/rmcp/src/transport/streamable_http_client.rs

@@ -90,6 +90,51 @@ 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<Arc<str>>,
+    negotiated_version: &ProtocolVersion,
+) -> Option<Arc<str>> {
+    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
+}
+
+/// 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<Arc<str>>,
+    cleanup_session_id: Option<Arc<str>>,
+    negotiated_version: ProtocolVersion,
+    protocol_headers: HashMap<HeaderName, HeaderValue>,
+}
+
 fn cache_tools_from_response(
     cache: &mut HashMap<String, Arc<JsonObject>>,
     message: &mut ServerJsonRpcMessage,
@@ -950,24 +995,21 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
     /// 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,
         uri: Arc<str>,
         auth_header: Option<String>,
         custom_headers: HashMap<HeaderName, HeaderValue>,
         max_sse_event_size: usize,
-    ) -> Result<
-        (
-            Option<Arc<str>>,
-            ProtocolVersion,
-            HashMap<HeaderName, HeaderValue>,
-        ),
-        StreamableHttpError<C::Error>,
-    > {
+    ) -> Result<Reinitialized, StreamableHttpError<C::Error>> {
         let (init_msg, new_session_id_str) = client
             .post_message_with_max_sse_event_size(
                 uri.clone(),
@@ -981,10 +1023,13 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
             .expect_initialized::<C::Error>()
             .await?;
 
-        let new_session_id: Option<Arc<str>> = new_session_id_str.map(|s| Arc::from(s.as_str()));
+        let cleanup_session_id: Option<Arc<str>> =
+            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 {
@@ -1011,7 +1056,12 @@ impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
             .await?
             .expect_accepted_or_json::<C::Error>()?;
 
-        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,
+        })
     }
 }
 
@@ -1133,6 +1183,7 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
             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 {
@@ -1235,7 +1286,12 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                     ) => 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();
@@ -1254,10 +1310,12 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                         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(),
                         });
@@ -1517,6 +1575,7 @@ impl<C: StreamableHttpClient> Worker for StreamableHttpClientWorker<C> {
                                 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;

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<Value>) -> BoxStream<'static, Result<Sse, S
         .boxed()
 }
 
-/// Scripted server: initialize -> 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<Mutex<Option<mpsc::Receiver<Value>>>>,
+    notification_stream: Arc<Mutex<Option<mpsc::Receiver<Value>>>>,
     post_stream: Arc<Mutex<Option<mpsc::Receiver<Value>>>>,
     posted: mpsc::UnboundedSender<Value>,
 }
@@ -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<String>,
         _custom_headers: HashMap<HeaderName, HeaderValue>,
     ) -> Result<BoxStream<'static, Result<Sse, SseError>>, StreamableHttpError<Self::Error>> {
-        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<RoleClient, SamplingClient>,
     /// Every message the client POSTs to the scripted server.
     posted: mpsc::UnboundedReceiver<Value>,
-    /// Feeds the standalone GET stream.
-    get_tx: mpsc::Sender<Value>,
+    /// Feeds the unassociated stream (the SSE body of the initialized
+    /// notification POST).
+    unassociated_tx: mpsc::Sender<Value>,
     /// Feeds the SSE stream of the in-flight tools/list POST.
     post_tx: mpsc::Sender<Value>,
     /// 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,31 +242,46 @@ 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,
         sampled: sampled_rx,
     }
 }
 
-/// #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