rmcp - cargo Package Compare versions

Comparing version 2.1.0 to 2.2.0

tests/test_cancelled_response.rs

+//! A receiver SHOULD NOT send a response for a request it has already been told
+//! to cancel. This drives a real stdio server with raw JSON-RPC: the tool blocks
+//! until the request is cancelled, so its result is only produced *after* the
+//! cancellation — the service loop must drop it rather than write it to the wire.
+
+use std::{collections::BTreeSet, process::Stdio, time::Duration};
+
+use rmcp::{
+    ErrorData as McpError, RoleServer, ServerHandler, ServiceExt,
+    model::{CallToolRequestParams, CallToolResult, ContentBlock, ServerCapabilities, ServerInfo},
+    service::RequestContext,
+};
+use serde_json::{Value, json};
+use tokio::{
+    io::{AsyncBufReadExt, AsyncWrite, AsyncWriteExt, BufReader},
+    process::{Child, Command},
+};
+
+const HELPER_ENV: &str = "RMCP_CANCELLED_RESPONSE_HELPER";
+const READ_TIMEOUT: Duration = Duration::from_secs(10);
+
+#[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+async fn cancelled_request_receives_no_response() -> anyhow::Result<()> {
+    let mut child = spawn_helper();
+    let mut writer = child.stdin.take().expect("helper stdin");
+    let stdout = child.stdout.take().expect("helper stdout");
+    let mut reader = BufReader::new(stdout);
+
+    send_json(
+        &mut writer,
+        &json!({
+            "jsonrpc": "2.0",
+            "id": 1,
+            "method": "initialize",
+            "params": {
+                "protocolVersion": "2024-11-05",
+                "capabilities": {},
+                "clientInfo": { "name": "raw-test-client", "version": "0.0.0" }
+            }
+        }),
+    )
+    .await?;
+    collect_ids_until(&mut reader, 1, READ_TIMEOUT).await?;
+    send_json(
+        &mut writer,
+        &json!({ "jsonrpc": "2.0", "method": "notifications/initialized" }),
+    )
+    .await?;
+
+    // Start a request that blocks until cancelled, then cancel it. Its response is
+    // produced only after the cancellation arrives, so it must be suppressed.
+    send_json(
+        &mut writer,
+        &json!({
+            "jsonrpc": "2.0",
+            "id": 2,
+            "method": "tools/call",
+            "params": { "name": "wait-for-cancel", "arguments": {} }
+        }),
+    )
+    .await?;
+    send_json(
+        &mut writer,
+        &json!({
+            "jsonrpc": "2.0",
+            "method": "notifications/cancelled",
+            "params": { "requestId": 2 }
+        }),
+    )
+    .await?;
+    // A ping proves the server is alive past the cancellation, so the absence of
+    // an id=2 response is genuine suppression rather than a dead connection.
+    send_json(
+        &mut writer,
+        &json!({ "jsonrpc": "2.0", "id": 3, "method": "ping" }),
+    )
+    .await?;
+
+    let seen = collect_ids_until(&mut reader, 3, READ_TIMEOUT).await?;
+    assert!(seen.contains(&3));
+    assert!(!seen.contains(&2));
+
+    drop(writer);
+    wait_for_child(&mut child).await;
+    Ok(())
+}
+
+struct WaitForCancelServer;
+
+impl ServerHandler for WaitForCancelServer {
+    fn get_info(&self) -> ServerInfo {
+        ServerInfo::new(ServerCapabilities::builder().enable_tools().build())
+    }
+
+    async fn call_tool(
+        &self,
+        _request: CallToolRequestParams,
+        context: RequestContext<RoleServer>,
+    ) -> Result<CallToolResult, McpError> {
+        context.ct.cancelled().await;
+        Ok(CallToolResult::success(vec![ContentBlock::text(
+            "late response",
+        )]))
+    }
+}
+
+#[tokio::test]
+async fn cancelled_response_helper() -> anyhow::Result<()> {
+    if std::env::var(HELPER_ENV).as_deref() != Ok("1") {
+        return Ok(());
+    }
+    run_helper_server().await?;
+    Ok(())
+}
+
+#[cfg(feature = "local")]
+async fn run_helper_server() -> anyhow::Result<()> {
+    tokio::task::LocalSet::new()
+        .run_until(serve_helper_stdio())
+        .await
+}
+
+#[cfg(not(feature = "local"))]
+async fn run_helper_server() -> anyhow::Result<()> {
+    serve_helper_stdio().await
+}
+
+async fn serve_helper_stdio() -> anyhow::Result<()> {
+    let server = WaitForCancelServer.serve(rmcp::transport::stdio()).await?;
+    server.waiting().await?;
+    Ok(())
+}
+
+fn spawn_helper() -> Child {
+    let exe = std::env::current_exe().expect("current test exe");
+    Command::new(exe)
+        .arg("--exact")
+        .arg("cancelled_response_helper")
+        .arg("--quiet")
+        .arg("--nocapture")
+        .arg("--test-threads")
+        .arg("1")
+        .env(HELPER_ENV, "1")
+        .stdin(Stdio::piped())
+        .stdout(Stdio::piped())
+        .stderr(Stdio::null())
+        .kill_on_drop(true)
+        .spawn()
+        .expect("spawn helper")
+}
+
+async fn wait_for_child(child: &mut Child) {
+    let _ = tokio::time::timeout(Duration::from_secs(2), child.wait()).await;
+    if child.id().is_some() {
+        let _ = child.kill().await;
+    }
+}
+
+async fn send_json<W>(writer: &mut W, message: &Value) -> anyhow::Result<()>
+where
+    W: AsyncWrite + Unpin,
+{
+    let serialized = serde_json::to_string(message)?;
+    writer.write_all(serialized.as_bytes()).await?;
+    writer.write_all(b"\n").await?;
+    writer.flush().await?;
+    Ok(())
+}
+
+/// Read response lines, collecting every message id seen, until `stop_id` is seen
+/// (then a short grace read to catch any straggler) or the timeout elapses.
+async fn collect_ids_until<R>(
+    reader: &mut BufReader<R>,
+    stop_id: u64,
+    timeout: Duration,
+) -> anyhow::Result<BTreeSet<u64>>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    let mut seen = BTreeSet::new();
+    let mut deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        let remaining = deadline.saturating_duration_since(tokio::time::Instant::now());
+        if remaining.is_zero() {
+            break;
+        }
+        let mut line = String::new();
+        let Ok(read_result) = tokio::time::timeout(remaining, reader.read_line(&mut line)).await
+        else {
+            break;
+        };
+        if read_result? == 0 {
+            break;
+        }
+        let trimmed = line.trim();
+        if trimmed.is_empty() {
+            continue;
+        }
+        let Ok(value) = serde_json::from_str::<Value>(trimmed) else {
+            continue;
+        };
+        if let Some(id) = value.get("id").and_then(Value::as_u64) {
+            seen.insert(id);
+            if id == stop_id {
+                // Give any late (incorrectly-sent) response a brief window to arrive.
+                deadline = tokio::time::Instant::now() + Duration::from_millis(300);
+            }
+        }
+    }
+    Ok(seen)
+}

.cargo_vcs_info.json

 {
   "git": {
-    "sha1": "8e44af499bfa8d54f18fb475eb86a601a5a2e432"
+    "sha1": "519577601db3823616dbd7c4eb84ed569d8e17d4"
   },
   "path_in_vcs": "crates/rmcp"
 }

Cargo.toml

@@ -15,3 +15,3 @@ # THIS FILE IS AUTOMATICALLY GENERATED BY CARGO
 name = "rmcp"
-version = "2.1.0"
+version = "2.2.0"
 build = "build.rs"
@@ -171,2 +171,6 @@ autolib = false
 [[test]]
+name = "test_cancelled_response"
+path = "tests/test_cancelled_response.rs"
+
+[[test]]
 name = "test_client_credentials"
@@ -621,3 +625,3 @@ path = "tests/test_client_credentials.rs"
 [dependencies.rmcp-macros]
-version = "2.1.0"
+version = "2.2.0"
 optional = true
@@ -624,0 +628,0 @@

CHANGELOG.md

@@ -10,2 +10,15 @@ # Changelog
 
+## [2.2.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v2.1.0...rmcp-v2.2.0) - 2026-07-08
+
+### Added
+
+- reject auth servers lacking S256 PKCE support ([#955](https://github.com/modelcontextprotocol/rust-sdk/pull/955))
+
+### Fixed
+
+- pass client conformance suite ([#960](https://github.com/modelcontextprotocol/rust-sdk/pull/960))
+- don't respond to cancelled requests ([#957](https://github.com/modelcontextprotocol/rust-sdk/pull/957))
+- fail orphaned streamable HTTP responses on reinit ([#914](https://github.com/modelcontextprotocol/rust-sdk/pull/914))
+- address 2025-11-25 conformance audit findings ([#951](https://github.com/modelcontextprotocol/rust-sdk/pull/951))
+
 ## [2.1.0](https://github.com/modelcontextprotocol/rust-sdk/compare/rmcp-v2.0.0...rmcp-v2.1.0) - 2026-07-02
@@ -12,0 +25,0 @@

src/model/content.rs

@@ -300,3 +300,3 @@ //! Content types that flow between agents, tools, prompts, and LLMs.
                 uri: uri.into(),
-                mime_type: Some("text".to_string()),
+                mime_type: Some("text/plain".to_string()),
                 text: content.into(),
@@ -303,0 +303,0 @@ meta: None,

src/model/resource.rs

@@ -196,3 +196,3 @@ use serde::{Deserialize, Serialize};
             uri: uri.into(),
-            mime_type: Some("text".into()),
+            mime_type: Some("text/plain".into()),
             text: text.into(),
@@ -199,0 +199,0 @@ meta: None,

src/service.rs

@@ -1106,5 +1106,7 @@ use futures::FutureExt;
                     } {
-                        if let Some(ct) = local_ct_pool.remove(id) {
-                            ct.cancel();
-                        }
+                        let Some(ct) = local_ct_pool.remove(id) else {
+                            tracing::debug!(%id, "dropping response for cancelled request");
+                            continue;
+                        };
+                        ct.cancel();
                         let send = transport.send(m);
@@ -1111,0 +1113,0 @@ let current_span = tracing::Span::current();

src/task_manager.rs

@@ -52,2 +52,3 @@ use std::{any::Any, collections::HashMap, pin::Pin};
 
+    /// Time-to-live in milliseconds, matching `TaskMetadata.ttl` from the MCP spec.
     pub fn with_ttl(mut self, ttl: u64) -> Self {
@@ -79,3 +80,7 @@ self.ttl = Some(ttl);
 // ===== Operation Processor =====
-pub const DEFAULT_TASK_TIMEOUT_SECS: u64 = 300; // 5 minutes
+#[deprecated(note = "use DEFAULT_TASK_TIMEOUT_MS; ttl values are milliseconds per the MCP spec")]
+pub const DEFAULT_TASK_TIMEOUT_SECS: u64 = 300;
+/// Default execution timeout (5 minutes), in milliseconds, applied when a
+/// descriptor does not specify a `ttl`.
+pub const DEFAULT_TASK_TIMEOUT_MS: u64 = 300_000;
 /// Operation processor that coordinates extractors and handlers
@@ -170,3 +175,3 @@ pub struct OperationProcessor {
         let task_id = descriptor.operation_id.clone();
-        let timeout_secs = descriptor.ttl.or(Some(DEFAULT_TASK_TIMEOUT_SECS));
+        let timeout_ms = descriptor.ttl.or(Some(DEFAULT_TASK_TIMEOUT_MS));
         let sender = self.task_result_sender.clone();
@@ -176,4 +181,4 @@ let descriptor_for_result = descriptor.clone();
         let timed_future = async move {
-            if let Some(secs) = timeout_secs {
-                match timeout(Duration::from_secs(secs), future).await {
+            if let Some(ms) = timeout_ms {
+                match timeout(Duration::from_millis(ms), future).await {
                     Ok(result) => result,
@@ -198,3 +203,3 @@ Err(_) => Err(Error::TaskError("Operation timed out".to_string())),
             started_at: std::time::Instant::now(),
-            timeout: timeout_secs,
+            timeout: timeout_ms,
             descriptor,
@@ -221,3 +226,3 @@ };
             if let Some(timeout_duration) = task.timeout {
-                if now.duration_since(task.started_at).as_secs() > timeout_duration {
+                if now.duration_since(task.started_at).as_millis() > u128::from(timeout_duration) {
                     task.task_handle.abort();
@@ -224,0 +229,0 @@ timed_out_tasks.push(task_id.clone());

src/transport/common/reqwest/streamable_http_client.rs

@@ -87,3 +87,3 @@ use std::{borrow::Cow, collections::HashMap, sync::Arc};
         }
-        let event_stream = SseStream::from_byte_stream(response.bytes_stream()).boxed();
+        let event_stream = SseStream::from_bytes_stream(response.bytes_stream()).boxed();
         Ok(event_stream)
@@ -227,3 +227,3 @@ }
             Some(ct) if ct.as_bytes().starts_with(EVENT_STREAM_MIME_TYPE.as_bytes()) => {
-                let event_stream = SseStream::from_byte_stream(response.bytes_stream()).boxed();
+                let event_stream = SseStream::from_bytes_stream(response.bytes_stream()).boxed();
                 Ok(StreamableHttpPostResponse::Sse(event_stream, session_id))
@@ -230,0 +230,0 @@ }

src/transport/streamable_http_client.rs

@@ -1,2 +0,7 @@
-use std::{borrow::Cow, collections::HashMap, sync::Arc, time::Duration};
+use std::{
+    borrow::Cow,
+    collections::{HashMap, HashSet},
+    sync::Arc,
+    time::Duration,
+};
 
@@ -15,4 +20,4 @@ use futures::{Stream, StreamExt, future::BoxFuture, stream::BoxStream};
     model::{
-        ClientJsonRpcMessage, ClientNotification, InitializedNotification, ServerJsonRpcMessage,
-        ServerResult,
+        ClientJsonRpcMessage, ClientNotification, ErrorData, InitializedNotification, RequestId,
+        ServerJsonRpcMessage, ServerResult,
     },
@@ -302,2 +307,75 @@ transport::{
 impl<C: StreamableHttpClient> StreamableHttpClientWorker<C> {
+    fn client_request_id(message: &ClientJsonRpcMessage) -> Option<RequestId> {
+        match message {
+            ClientJsonRpcMessage::Request(request) => Some(request.id.clone()),
+            _ => None,
+        }
+    }
+
+    fn server_response_id(message: &ServerJsonRpcMessage) -> Option<&RequestId> {
+        match message {
+            ServerJsonRpcMessage::Response(response) => Some(&response.id),
+            ServerJsonRpcMessage::Error(error) => error.id.as_ref(),
+            _ => None,
+        }
+    }
+
+    fn mark_stream_response_pending(
+        pending_stream_response_ids: &mut HashSet<RequestId>,
+        request_id: Option<RequestId>,
+    ) {
+        if let Some(request_id) = request_id {
+            pending_stream_response_ids.insert(request_id);
+        }
+    }
+
+    fn clear_stream_response_pending(
+        pending_stream_response_ids: &mut HashSet<RequestId>,
+        message: &ServerJsonRpcMessage,
+    ) {
+        if let Some(id) = Self::server_response_id(message) {
+            pending_stream_response_ids.remove(id);
+        }
+    }
+
+    async fn drain_queued_stream_messages(
+        sse_worker_rx: &mut tokio::sync::mpsc::Receiver<ServerJsonRpcMessage>,
+        context: &mut super::worker::WorkerContext<Self>,
+        pending_stream_response_ids: &mut HashSet<RequestId>,
+    ) -> Result<(), WorkerQuitReason<StreamableHttpError<C::Error>>> {
+        loop {
+            match sse_worker_rx.try_recv() {
+                Ok(message) => {
+                    Self::clear_stream_response_pending(pending_stream_response_ids, &message);
+                    context.send_to_handler(message).await?;
+                }
+                Err(tokio::sync::mpsc::error::TryRecvError::Empty) => return Ok(()),
+                Err(tokio::sync::mpsc::error::TryRecvError::Disconnected) => return Ok(()),
+            }
+        }
+    }
+
+    async fn fail_pending_stream_responses(
+        context: &mut super::worker::WorkerContext<Self>,
+        pending_stream_response_ids: &mut HashSet<RequestId>,
+    ) -> Result<(), WorkerQuitReason<StreamableHttpError<C::Error>>> {
+        if pending_stream_response_ids.is_empty() {
+            return Ok(());
+        }
+
+        let pending_ids = std::mem::take(pending_stream_response_ids);
+        for id in pending_ids {
+            context
+                .send_to_handler(ServerJsonRpcMessage::error(
+                    ErrorData::internal_error(
+                        "streamable HTTP session was re-initialized before the response arrived",
+                        None,
+                    ),
+                    Some(id),
+                ))
+                .await?;
+        }
+        Ok(())
+    }
+
     /// Convert a raw SSE stream into a JSON-RPC message stream without
@@ -334,2 +412,61 @@ /// reconnection logic.
 
+    /// Convert an SSE stream into JSON-RPC messages with reconnect semantics.
+    ///
+    /// This is used for request-scoped SSE responses as well as the standalone
+    /// GET stream. A request-scoped stream can close before its response arrives,
+    /// and SEP-1699 requires the client to honor `retry` and resume with
+    /// `Last-Event-ID` in that case.
+    fn reconnecting_sse_to_jsonrpc(
+        stream: BoxedSseStream,
+        client: C,
+        session_id: Arc<str>,
+        uri: Arc<str>,
+        auth_header: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+        retry_config: Arc<dyn SseRetryPolicy>,
+    ) -> impl Stream<Item = Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>> + Send + 'static
+    {
+        SseAutoReconnectStream::new(
+            stream,
+            StreamableHttpClientReconnect {
+                client,
+                session_id,
+                uri,
+                auth_header,
+                custom_headers,
+            },
+            retry_config,
+        )
+    }
+
+    /// Convert a POST response SSE stream into JSON-RPC messages.
+    ///
+    /// Stateful sessions can resume via GET when the response stream closes
+    /// before the server sends the matching JSON-RPC response. Stateless
+    /// transports do not have enough state to resume, so they keep the raw
+    /// SSE-to-JSON-RPC mapping.
+    fn response_sse_to_jsonrpc(
+        stream: BoxedSseStream,
+        session_id: Option<Arc<str>>,
+        client: C,
+        uri: Arc<str>,
+        auth_header: Option<String>,
+        custom_headers: HashMap<HeaderName, HeaderValue>,
+        retry_config: Arc<dyn SseRetryPolicy>,
+    ) -> BoxStream<'static, Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>> {
+        match session_id {
+            Some(session_id) => Self::reconnecting_sse_to_jsonrpc(
+                stream,
+                client,
+                session_id,
+                uri,
+                auth_header,
+                custom_headers,
+                retry_config,
+            )
+            .boxed(),
+            None => Self::raw_sse_to_jsonrpc(stream).boxed(),
+        }
+    }
+
     async fn execute_sse_stream(
@@ -563,2 +700,3 @@ sse_stream: impl Stream<Item = Result<ServerJsonRpcMessage, StreamableHttpError<C::Error>>>
         let mut streams = tokio::task::JoinSet::new();
+        let mut pending_stream_response_ids = HashSet::new();
         if let Some(session_id) = &session_id {
@@ -653,2 +791,3 @@ let client = self.client.clone();
                     let WorkerSendRequest { message, responder } = send_request;
+                    let request_id = Self::client_request_id(&message);
                     // Pass a clone to the first attempt so `message` is retained for a
@@ -687,6 +826,23 @@ // potential re-init retry. `post_message` takes ownership and the
                                     Ok((new_session_id, new_protocol_headers)) => {
-                                        // Old streams hold the stale session ID; abort them
-                                        // so the new standalone SSE stream takes over.
+                                        // Old streams hold the stale session ID. Stop them first
+                                        // so no late stale-session messages can arrive after the
+                                        // pending requests below are completed.
                                         streams.abort_all();
+                                        while streams.join_next().await.is_some() {}
 
+                                        // Forward any already queued response messages and fail
+                                        // the remaining accepted requests so callers do not wait
+                                        // forever for responses that can no longer arrive.
+                                        Self::drain_queued_stream_messages(
+                                            &mut sse_worker_rx,
+                                            &mut context,
+                                            &mut pending_stream_response_ids,
+                                        )
+                                        .await?;
+                                        Self::fail_pending_stream_responses(
+                                            &mut context,
+                                            &mut pending_stream_response_ids,
+                                        )
+                                        .await?;
+
                                         session_id = new_session_id;
@@ -774,2 +930,6 @@ protocol_headers = new_protocol_headers;
                                             Ok(StreamableHttpPostResponse::Accepted) => {
+                                                Self::mark_stream_response_pending(
+                                                    &mut pending_stream_response_ids,
+                                                    request_id,
+                                                );
                                                 tracing::trace!(
@@ -785,4 +945,17 @@ "client message accepted after re-init"
                                             Ok(StreamableHttpPostResponse::Sse(stream, ..)) => {
+                                                Self::mark_stream_response_pending(
+                                                    &mut pending_stream_response_ids,
+                                                    request_id,
+                                                );
+                                                let sse_stream = Self::response_sse_to_jsonrpc(
+                                                    stream,
+                                                    session_id.clone(),
+                                                    self.client.clone(),
+                                                    config.uri.clone(),
+                                                    config.auth_header.clone(),
+                                                    protocol_headers.clone(),
+                                                    self.config.retry_config.clone(),
+                                                );
                                                 streams.spawn(Self::execute_sse_stream(
-                                                    Self::raw_sse_to_jsonrpc(stream),
+                                                    sse_stream,
                                                     sse_worker_tx.clone(),
@@ -803,2 +976,6 @@ true,
                         Ok(StreamableHttpPostResponse::Accepted) => {
+                            Self::mark_stream_response_pending(
+                                &mut pending_stream_response_ids,
+                                request_id,
+                            );
                             tracing::trace!("client message accepted");
@@ -812,4 +989,17 @@ Ok(())
                         Ok(StreamableHttpPostResponse::Sse(stream, ..)) => {
+                            Self::mark_stream_response_pending(
+                                &mut pending_stream_response_ids,
+                                request_id,
+                            );
+                            let sse_stream = Self::response_sse_to_jsonrpc(
+                                stream,
+                                session_id.clone(),
+                                self.client.clone(),
+                                config.uri.clone(),
+                                config.auth_header.clone(),
+                                protocol_headers.clone(),
+                                self.config.retry_config.clone(),
+                            );
                             streams.spawn(Self::execute_sse_stream(
-                                Self::raw_sse_to_jsonrpc(stream),
+                                sse_stream,
                                 sse_worker_tx.clone(),
@@ -826,2 +1016,6 @@ true,
                 Event::ServerMessage(json_rpc_message) => {
+                    Self::clear_stream_response_pending(
+                        &mut pending_stream_response_ids,
+                        &json_rpc_message,
+                    );
                     // send the message to the handler
@@ -828,0 +1022,0 @@ if let Err(e) = context.send_to_handler(json_rpc_message).await {

tests/test_streamable_http_stale_session.rs

@@ -8,7 +8,16 @@ #![cfg(all(
 
-use std::{collections::HashMap, sync::Arc};
+use std::{
+    collections::{HashMap, VecDeque},
+    sync::Arc,
+};
 
+use futures::stream;
+use http::{HeaderName, HeaderValue};
 use rmcp::{
     ServiceError, ServiceExt,
-    model::{ClientJsonRpcMessage, ClientRequest, PingRequest, RequestId},
+    model::{
+        CallToolRequestParams, ClientInfo, ClientJsonRpcMessage, ClientRequest, ErrorCode,
+        ErrorData, InitializeResult, PingRequest, RequestId, ServerCapabilities,
+        ServerJsonRpcMessage, ServerResult,
+    },
     transport::{
@@ -18,2 +27,3 @@ StreamableHttpClientTransport,
             StreamableHttpClient, StreamableHttpClientTransportConfig, StreamableHttpError,
+            StreamableHttpPostResponse,
         },
@@ -30,3 +40,212 @@ streamable_http_server::{
 
+#[derive(Debug, thiserror::Error)]
+#[error("mock streamable http client error")]
+struct MockClientError;
+
+#[derive(Clone)]
+struct ReinitDropsAcceptedResponseClient {
+    state: Arc<tokio::sync::Mutex<MockState>>,
+    stale_stream_cancelled: CancellationToken,
+    initial_request_accepted: Arc<tokio::sync::Semaphore>,
+    final_retry_accepted: Arc<tokio::sync::Semaphore>,
+}
+
+struct MockState {
+    session_counter: usize,
+    posts: VecDeque<MockPost>,
+}
+
+enum MockPost {
+    Initialize,
+    Initialized,
+    Accepted,
+    SessionExpired,
+}
+
+impl ReinitDropsAcceptedResponseClient {
+    fn new() -> Self {
+        Self {
+            state: Arc::new(tokio::sync::Mutex::new(MockState {
+                session_counter: 0,
+                posts: VecDeque::from([
+                    MockPost::Initialize,
+                    MockPost::Initialized,
+                    MockPost::Accepted,
+                    MockPost::SessionExpired,
+                    MockPost::Initialize,
+                    MockPost::Initialized,
+                    MockPost::Accepted,
+                ]),
+            })),
+            stale_stream_cancelled: CancellationToken::new(),
+            initial_request_accepted: Arc::new(tokio::sync::Semaphore::new(0)),
+            final_retry_accepted: Arc::new(tokio::sync::Semaphore::new(0)),
+        }
+    }
+}
+
+impl StreamableHttpClient for ReinitDropsAcceptedResponseClient {
+    type Error = MockClientError;
+
+    async fn post_message(
+        &self,
+        _uri: Arc<str>,
+        message: ClientJsonRpcMessage,
+        _session_id: Option<Arc<str>>,
+        _auth_header: Option<String>,
+        _custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<StreamableHttpPostResponse, StreamableHttpError<Self::Error>> {
+        let mut state = self.state.lock().await;
+        match state
+            .posts
+            .pop_front()
+            .expect("unexpected mock post_message call")
+        {
+            MockPost::Initialize => {
+                state.session_counter += 1;
+                let id = match message {
+                    ClientJsonRpcMessage::Request(request) => request.id,
+                    other => panic!("expected initialize request, got {other:?}"),
+                };
+                Ok(StreamableHttpPostResponse::Json(
+                    ServerJsonRpcMessage::response(
+                        ServerResult::InitializeResult(InitializeResult::new(
+                            ServerCapabilities::builder().enable_tools().build(),
+                        )),
+                        id,
+                    ),
+                    Some(format!("session-{}", state.session_counter)),
+                ))
+            }
+            MockPost::Initialized => {
+                assert!(
+                    matches!(message, ClientJsonRpcMessage::Notification(_)),
+                    "expected initialized notification, got {message:?}"
+                );
+                Ok(StreamableHttpPostResponse::Accepted)
+            }
+            MockPost::Accepted => {
+                if state.posts.is_empty() {
+                    self.final_retry_accepted.add_permits(1);
+                } else {
+                    self.initial_request_accepted.add_permits(1);
+                }
+                Ok(StreamableHttpPostResponse::Accepted)
+            }
+            MockPost::SessionExpired => Err(StreamableHttpError::SessionExpired),
+        }
+    }
+
+    async fn delete_session(
+        &self,
+        _uri: Arc<str>,
+        _session_id: Arc<str>,
+        _auth_header: Option<String>,
+        _custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<(), StreamableHttpError<Self::Error>> {
+        Ok(())
+    }
+
+    async fn get_stream(
+        &self,
+        _uri: Arc<str>,
+        session_id: Arc<str>,
+        _last_event_id: Option<String>,
+        _auth_header: Option<String>,
+        _custom_headers: HashMap<HeaderName, HeaderValue>,
+    ) -> Result<
+        futures::stream::BoxStream<'static, Result<sse_stream::Sse, sse_stream::Error>>,
+        StreamableHttpError<Self::Error>,
+    > {
+        if session_id.as_ref() == "session-1" {
+            let cancel = self.stale_stream_cancelled.clone();
+            Ok(Box::pin(stream::once(async move {
+                cancel.cancelled_owned().await;
+                Ok(sse_stream::Sse {
+                    event: None,
+                    data: Some(
+                        serde_json::to_string(&ServerJsonRpcMessage::error(
+                            ErrorData::new(
+                                ErrorCode::INTERNAL_ERROR,
+                                "stale stream should not deliver after re-init",
+                                None,
+                            ),
+                            Some(RequestId::Number(2)),
+                        ))
+                        .expect("serialize stale error"),
+                    ),
+                    id: None,
+                    retry: None,
+                })
+            })))
+        } else {
+            Ok(Box::pin(stream::pending()))
+        }
+    }
+}
+
 #[tokio::test]
+async fn test_reinitialization_completes_accepted_sse_request_instead_of_hanging()
+-> anyhow::Result<()> {
+    let mock_client = ReinitDropsAcceptedResponseClient::new();
+    let initial_request_accepted = mock_client.initial_request_accepted.clone();
+    let final_retry_accepted = mock_client.final_retry_accepted.clone();
+    let transport = StreamableHttpClientTransport::with_client(
+        mock_client,
+        StreamableHttpClientTransportConfig::with_uri("mock://mcp"),
+    );
+    let mut client = ClientInfo::default().serve(transport).await?;
+
+    let peer = client.peer().clone();
+    let pending_call = tokio::spawn(async move {
+        peer.call_tool(CallToolRequestParams::new("slow_tool"))
+            .await
+    });
+
+    let _initial_permit = tokio::time::timeout(
+        std::time::Duration::from_secs(1),
+        initial_request_accepted.acquire(),
+    )
+    .await
+    .expect("initial accepted request should be observed")
+    .expect("initial accepted request semaphore should stay open");
+
+    let reinit_trigger = {
+        let peer = client.peer().clone();
+        tokio::spawn(async move { peer.list_tools(None).await })
+    };
+
+    let _retry_permit = tokio::time::timeout(
+        std::time::Duration::from_secs(1),
+        final_retry_accepted.acquire(),
+    )
+    .await
+    .expect("re-initialization retry should be accepted")
+    .expect("re-initialization retry semaphore should stay open");
+
+    let err = tokio::time::timeout(std::time::Duration::from_millis(100), pending_call)
+        .await
+        .expect("accepted SSE-backed request should complete instead of hanging")?
+        .expect_err(
+            "accepted request should fail after re-initialization drops its response stream",
+        );
+
+    match err {
+        ServiceError::McpError(error) => {
+            assert_eq!(error.code, ErrorCode::INTERNAL_ERROR);
+            assert!(
+                error.message.contains("session"),
+                "expected session-related error, got: {error}"
+            );
+        }
+        other => panic!("expected McpError for orphaned request, got: {other:?}"),
+    }
+
+    reinit_trigger.abort();
+    let _ = client.close().await;
+
+    Ok(())
+}
+
+#[tokio::test]
 async fn test_stale_session_id_returns_status_aware_error() -> anyhow::Result<()> {
@@ -33,0 +252,0 @@ let ct = CancellationToken::new();

tests/test_task.rs

@@ -83,2 +83,34 @@ use std::{any::Any, time::Duration};
 
+#[tokio::test]
+async fn ttl_is_interpreted_as_milliseconds() {
+    let mut processor = OperationProcessor::new();
+    let descriptor = OperationDescriptor::new("slow", "dummy").with_ttl(50);
+    let future = Box::pin(async {
+        tokio::time::sleep(Duration::from_millis(500)).await;
+        Ok(Box::new(DummyTransport {
+            id: "slow".to_string(),
+            value: 0,
+        }) as Box<dyn OperationResultTransport>)
+    });
+
+    processor
+        .submit_operation(OperationMessage::new(descriptor, future))
+        .expect("submit operation");
+
+    tokio::time::sleep(Duration::from_millis(200)).await;
+    let results = processor.peek_completed();
+    assert_eq!(
+        results.len(),
+        1,
+        "50ms ttl should have timed out the operation well within 200ms"
+    );
+    match &results[0].result {
+        Err(err) => assert!(
+            err.to_string().contains("timed out"),
+            "unexpected error: {err}"
+        ),
+        Ok(_) => panic!("expected the operation to time out, but it completed"),
+    }
+}
+
 #[test]
@@ -85,0 +117,0 @@ fn task_status_notification_param_preserves_meta() {

No alert changes

No metric changes