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..c7bd57a01 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/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>, + 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 +} + +/// 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, @@ -950,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, @@ -960,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(), @@ -981,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 { @@ -1011,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, + }) } } @@ -1133,6 +1183,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 { @@ -1235,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(); @@ -1254,10 +1310,12 @@ 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(), }); @@ -1517,6 +1575,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_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 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..80b06a78d --- /dev/null +++ b/crates/rmcp/tests/test_streamable_http_sessionless_version.rs @@ -0,0 +1,420 @@ +#![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. That holds for the replacement +//! handshake after an expired-session 404 as much as for the first one. + +use std::{ + collections::VecDeque, + 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"; +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); + +/// 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>>, + /// 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() + } + + fn get_requests(&self) -> Vec { + self.calls() + .into_iter() + .filter(|(http_method, ..)| http_method == "GET") + .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(|(http_method, jsonrpc_method, _)| { + jsonrpc_method != "initialize" && http_method != "DELETE" + }) + .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( + 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"); + } + + // 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(( + "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 (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) + .body(Body::from( + json!({ + "jsonrpc": "2.0", + "id": request["id"], + "result": { + "protocolVersion": version, + "capabilities": {"tools": {}}, + "serverInfo": {"name": "dual-era", "version": "1.0"} + } + }) + .to_string(), + )) + .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") +} + +/// 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)) + .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; + } + }); + (format!("http://{address}/mcp"), ct, server) +} + +async fn start_client(uri: String) -> rmcp::service::RunningService { + let transport = StreamableHttpClientTransport::from_config( + StreamableHttpClientTransportConfig::with_uri(uri), + ); + ClientInfo::default() + .serve_with_lifecycle(transport, ClientLifecycleMode::Initialize) + .await + .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; + client.cancel().await.expect("cancel client"); + ct.cancel(); + let _ = server.await; + 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; + + // 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:?}" + ); +} + +#[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(|(http_method, _, session)| session.is_some() && http_method != "DELETE") + .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() + ); + + // 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] +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() + ); +}