From 5b8664f2979ad0745427606a95af555a06e7beb1 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:16:45 +0800 Subject: [PATCH 01/29] worker: retain terminal cause, StartupFailed variant, quit_status (#1441) --- crates/rmcp/src/transport/worker.rs | 126 ++++++++++++++++++++++++++-- 1 file changed, 117 insertions(+), 9 deletions(-) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index e32da7d7d..ecbdb04d3 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -27,6 +27,18 @@ pub enum WorkerQuitReason { error: E, context: Cow<'static, str>, }, + /// The transport's startup handshake failed with the typed error. + /// + /// Distinct from [`Fatal`](Self::Fatal): a startup failure is an ordinary, + /// often retryable lifecycle event (peer unreachable, TLS failure, an auth + /// challenge), not an internal fault. The real typed error is retained here + /// rather than replaced by a synthetic placeholder, so the service loop and + /// late observers can see the actual cause. + #[error("Startup failed {error}, when {context}")] + StartupFailed { + error: E, + context: Cow<'static, str>, + }, #[error("Transport cancelled")] Cancelled, #[error("Transport closed")] @@ -50,6 +62,18 @@ impl WorkerQuitReason { context: context.into(), } } + pub fn startup_failed(error: E, context: impl Into>) -> Self { + Self::StartupFailed { + error, + context: context.into(), + } + } + pub fn startup_failed_context(context: impl Into>) -> impl FnOnce(E) -> Self { + |e| Self::StartupFailed { + error: e, + context: context.into(), + } + } } pub trait Worker: Sized + Send + 'static { @@ -172,7 +196,11 @@ pub struct WorkerTransport { control_send_service: tokio::sync::mpsc::Sender>, request_cancellations: RequestCancellations, control_generation: Arc, - join_handle: Option>>>, + join_handle: Option>>>>, + /// Retained terminal status of the worker, written exactly once when its + /// task quits. Lets an observer whose own timeout has already fired still + /// learn why the worker died, rather than relying on a delivery window. + quit_status: tokio::sync::watch::Receiver>>>, _drop_guard: tokio_util::sync::DropGuard, ct: CancellationToken, } @@ -226,16 +254,26 @@ impl WorkerTransport { cancellation_token: transport_task_ct.clone(), }; + // Retained terminal status: written once when the worker task finishes, + // so the quit reason is not lost when no `Transport::close()` is pending + // at the moment the worker dies. The reason is shared via `Arc` so both + // the join handle (consumed by close()) and late observers can read it. + let (quit_status_tx, quit_status) = tokio::sync::watch::channel(None); + let join_handle = tokio::spawn(async move { - worker + let result = worker .run(context) .instrument(tracing::span!( Level::TRACE, "transport_worker", name = worker_name, )) - .await - .inspect_err(|e| match e { + .await; + match &result { + Ok(()) => { + tracing::debug!("worker quit"); + } + Err(e) => match e { WorkerQuitReason::Cancelled | WorkerQuitReason::TransportClosed | WorkerQuitReason::HandlerTerminated @@ -245,13 +283,30 @@ impl WorkerTransport { WorkerQuitReason::Join(e) => { tracing::error!("worker quit with join error: {:?}", e); } + WorkerQuitReason::StartupFailed { error, context } => { + // A startup failure is an ordinary lifecycle event + // (peer unreachable, TLS failure, auth challenge), not + // an internal fault — log at warn, not error. + tracing::warn!("worker startup failed: {error}, when {context}"); + } WorkerQuitReason::Fatal { error, context } => { tracing::error!("worker quit with fatal: {error}, when {context}"); } - }) - .inspect(|_| { - tracing::debug!("worker quit"); - }) + }, + } + // Retain the terminal reason for late observers; the same value is + // shared with the join handle so close() can report join errors. + match result { + Ok(()) => { + let _ = quit_status_tx.send(None); + Ok(()) + } + Err(e) => { + let shared = Arc::new(e); + let _ = quit_status_tx.send(Some(shared.clone())); + Err(shared) + } + } }); Self { rx: from_transport_rx, @@ -260,6 +315,7 @@ impl WorkerTransport { request_cancellations, control_generation, join_handle: Some(join_handle), + quit_status, ct: transport_task_ct.clone(), _drop_guard: transport_task_ct.drop_guard(), } @@ -402,12 +458,26 @@ impl Transport for WorkerTransport { async fn close(&mut self) -> Result<(), Self::Error> { if let Some(handle) = self.join_handle.take() { self.ct.cancel(); - let _quit_reason = handle.await.map_err(W::err_join)?; + // Await worker termination; the quit reason itself is retained in + // `quit_status` rather than moved out, so late observers can still + // read it after close() returns. + let _ = handle.await.map_err(W::err_join)?; Ok(()) } else { Ok(()) } } + + fn quit_status(&self) -> Option { + let guard = self.quit_status.borrow(); + let reason = guard.as_ref()?; + let message = format!("{reason}"); + Some(crate::transport::DynamicTransportError::from_parts( + std::any::type_name::(), + std::any::TypeId::of::(), + Box::new(std::io::Error::other(message)), + )) + } } #[cfg(all(test, feature = "client"))] @@ -538,4 +608,42 @@ mod tests { drop(current); assert!(pending.lock().unwrap().is_empty()); } + + /// A worker whose startup fails must retain the real typed error in its + /// terminal status, so an observer whose own timeout already fired (no + /// pending `close()`) can still learn the cause. Regression for #1441. + #[tokio::test] + async fn startup_failure_is_retained_in_quit_status() { + struct StartupFailWorker; + + impl Worker for StartupFailWorker { + type Error = io::Error; + type Role = RoleClient; + + fn err_closed() -> Self::Error { + io::Error::other("worker closed") + } + fn err_join(error: tokio::task::JoinError) -> Self::Error { + io::Error::other(error) + } + async fn run( + self, + _context: WorkerContext, + ) -> Result<(), WorkerQuitReason> { + Err(WorkerQuitReason::startup_failed( + io::Error::new(io::ErrorKind::ConnectionRefused, "peer unreachable"), + "startup request failed", + )) + } + } + + let mut transport = WorkerTransport::spawn(StartupFailWorker); + // No close() pending — the worker dies on its own; the quit reason must + // still be observable afterwards. + transport.close().await.unwrap(); + let status = transport.quit_status().expect("quit reason retained"); + let text = status.to_string(); + assert!(text.contains("Startup failed"), "got: {text}"); + assert!(text.contains("peer unreachable"), "got: {text}"); + } } From 8ac0f11b1ff2a27300b010cbd6570ea1f4c33c16 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:16:47 +0800 Subject: [PATCH 02/29] transport: add provided Transport::quit_status (#1441) --- crates/rmcp/src/transport.rs | 16 ++++++++++++++++ 1 file changed, 16 insertions(+) diff --git a/crates/rmcp/src/transport.rs b/crates/rmcp/src/transport.rs index a539927d1..8511f637b 100644 --- a/crates/rmcp/src/transport.rs +++ b/crates/rmcp/src/transport.rs @@ -147,6 +147,22 @@ where /// Close the transport fn close(&mut self) -> impl Future> + Send; + + /// Terminal status of the transport's background worker, if it has quit. + /// + /// `None` means either the transport has no background worker or the worker + /// is still running. Implementations backed by a [`Worker`] retain the quit + /// reason here so an observer whose own timeout already fired can still + /// learn why the worker died, instead of only seeing a channel closure. + /// + /// The cause is returned as an owned [`DynamicTransportError`] wrapping the + /// worker's typed error so it can cross the service API boundary without a + /// generic parameter. + /// + /// [`Worker`]: crate::transport::worker::Worker + fn quit_status(&self) -> Option { + None + } } pub trait IntoTransport: Send + 'static From 89ac7103f69ac186267b2e6d2b8ba0dc35f61b1e Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:16:50 +0800 Subject: [PATCH 03/29] service: surface QuitReason::Transport with terminal cause (#1441) --- crates/rmcp/src/service.rs | 18 ++++++++++++++++++ 1 file changed, 18 insertions(+) diff --git a/crates/rmcp/src/service.rs b/crates/rmcp/src/service.rs index a473f2d49..6b3f14d0a 100644 --- a/crates/rmcp/src/service.rs +++ b/crates/rmcp/src/service.rs @@ -1225,6 +1225,11 @@ pub enum QuitReason { Cancelled, Closed, JoinError(tokio::task::JoinError), + /// The transport's background worker terminated. Carries the worker's + /// terminal reason (including a startup failure's real typed error) so + /// `waiting()` / `close()` callers learn *why* the connection ended rather + /// than an undifferentiated `Closed`. + Transport(DynamicTransportError), } /// Request execution context @@ -1842,6 +1847,19 @@ where if let Err(e) = sink_close_result { tracing::error!(%e, "fail to close sink"); } + + // Now that the transport's worker has terminated, recover its terminal + // reason (a startup failure's typed error included) so the quit reason + // reports *why* the connection ended instead of an undifferentiated + // `Closed`. + let quit_reason = match quit_reason { + QuitReason::Closed => match transport.quit_status() { + Some(cause) => QuitReason::Transport(cause), + None => QuitReason::Closed, + }, + other => other, + }; + tracing::info!(?quit_reason, "serve finished"); quit_reason }.instrument(current_span)); From fde962cb873c9cd8256eebd0db885d45c02ea2f0 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:16:53 +0800 Subject: [PATCH 04/29] streamable-http: report startup failure with real typed error (#1441) --- .../src/transport/streamable_http_client.rs | 30 ++++++++++++++----- 1 file changed, 23 insertions(+), 7 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 7b1981ff0..7d7e3a561 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -1196,16 +1196,32 @@ impl Worker for StreamableHttpClientWorker { Ok(res) => { let _ = responder.send(Ok(())); res.expect_initialized::().await.map_err( - WorkerQuitReason::fatal_context("process initialize response"), + WorkerQuitReason::startup_failed_context("process initialize response"), )? } Err(err) => { - let msg = format!("{:?}", err); - let _ = responder.send(Err(err)); - return Err(WorkerQuitReason::fatal( - StreamableHttpError::TransportChannelClosed, - msg, - )); + // Report the startup handshake failure with its real typed + // error instead of laundering it into a synthetic + // `TransportChannelClosed`. The typed error is also delivered + // to the in-flight `send()`; when that receiver has already + // gone away (its own timeout fired), the real cause is retained + // in the quit reason for late observers and the service loop. + match responder.send(Err(err)) { + Ok(()) => { + // The parked caller received the typed error. + return Err(WorkerQuitReason::startup_failed( + StreamableHttpError::TransportChannelClosed, + "startup request failed", + )); + } + Err(Err(err)) => { + return Err(WorkerQuitReason::startup_failed( + err, + "startup request failed", + )); + } + Err(Ok(())) => unreachable!("only Err is sent in this branch"), + } } }; let mut uses_modern_http = !is_legacy_startup; From 919431b66cfc9ba9b242ebb5f58f2df5bebe6101 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:24:19 +0800 Subject: [PATCH 05/29] worker: report cause only for abnormal termination; regression test (#1441) --- crates/rmcp/src/transport/worker.rs | 20 ++++++++++++++++++-- 1 file changed, 18 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index ecbdb04d3..7927a85f7 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -471,7 +471,23 @@ impl Transport for WorkerTransport { fn quit_status(&self) -> Option { let guard = self.quit_status.borrow(); let reason = guard.as_ref()?; - let message = format!("{reason}"); + // Only report a cause for abnormal termination. Normal shutdown paths + // (cancel, clean transport close, handler termination, idle timeout) + // carry no diagnostic value and must not masquerade as a `Transport` + // quit reason on the service side. + let message = match reason.as_ref() { + WorkerQuitReason::StartupFailed { error, context } => { + format!("startup failed, when {context}: {error}") + } + WorkerQuitReason::Fatal { error, context } => { + format!("transport fatal, when {context}: {error}") + } + WorkerQuitReason::Join(e) => format!("worker join error: {e}"), + WorkerQuitReason::Cancelled + | WorkerQuitReason::TransportClosed + | WorkerQuitReason::HandlerTerminated + | WorkerQuitReason::IdleTimeout(_) => return None, + }; Some(crate::transport::DynamicTransportError::from_parts( std::any::type_name::(), std::any::TypeId::of::(), @@ -643,7 +659,7 @@ mod tests { transport.close().await.unwrap(); let status = transport.quit_status().expect("quit reason retained"); let text = status.to_string(); - assert!(text.contains("Startup failed"), "got: {text}"); + assert!(text.contains("startup failed"), "got: {text}"); assert!(text.contains("peer unreachable"), "got: {text}"); } } From ce07d77f971a7d004054c3763ae2cf06c5977922 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:24:23 +0800 Subject: [PATCH 06/29] streamable-http: keep real error text on ack-delivered path (#1441) --- .../src/transport/streamable_http_client.rs | 19 +++++++++++-------- 1 file changed, 11 insertions(+), 8 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 7d7e3a561..45c8d58d4 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -1200,18 +1200,21 @@ impl Worker for StreamableHttpClientWorker { )? } Err(err) => { - // Report the startup handshake failure with its real typed - // error instead of laundering it into a synthetic - // `TransportChannelClosed`. The typed error is also delivered - // to the in-flight `send()`; when that receiver has already - // gone away (its own timeout fired), the real cause is retained - // in the quit reason for late observers and the service loop. + // Report the startup handshake failure as `StartupFailed`, not + // a `Fatal` with a synthetic `TransportChannelClosed`. Capture + // the real error's text up front: the typed value can only live + // in one place (it is not `Clone`), so it goes to whichever + // receiver is still listening — the in-flight `send()` ack, or, + // if that receiver already went away (its own timeout fired), + // the quit reason for late observers and the service loop. + let err_text = format!("{err}"); match responder.send(Err(err)) { Ok(()) => { - // The parked caller received the typed error. + // The parked caller received the typed error; the quit + // reason still names the real cause in text form. return Err(WorkerQuitReason::startup_failed( StreamableHttpError::TransportChannelClosed, - "startup request failed", + format!("startup request failed: {err_text}"), )); } Err(Err(err)) => { From b2c484bcd9d3a5e8a2bd3746b148330d5f82f8e4 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:55:53 +0800 Subject: [PATCH 07/29] worker: typed RetainedQuitReason wrapper + downcast test (#1441) --- crates/rmcp/src/transport/worker.rs | 63 ++++++++++++++++++++++++----- 1 file changed, 52 insertions(+), 11 deletions(-) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index 7927a85f7..d2f2f1f33 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -190,6 +190,34 @@ impl WorkerSendRequest { } } +/// A retained worker terminal reason, shared via `Arc` so it can outlive the +/// single fire-and-forget ack delivery window. +/// +/// Boxed into [`DynamicTransportError`](crate::transport::DynamicTransportError) +/// by [`Transport::quit_status`](crate::transport::Transport::quit_status). +/// Callers can downcast to this type to recover the `Arc>` +/// (and thus the typed `E`), or walk [`Error::source`] to reach the inner typed +/// error directly — preserving source-only data such as an auth challenge. +#[derive(Debug)] +pub struct RetainedQuitReason(pub Arc>); + +impl std::fmt::Display for RetainedQuitReason { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + std::fmt::Display::fmt(self.0.as_ref(), f) + } +} + +impl std::error::Error for RetainedQuitReason { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self.0.as_ref() { + WorkerQuitReason::StartupFailed { error, .. } + | WorkerQuitReason::Fatal { error, .. } => Some(error), + WorkerQuitReason::Join(e) => Some(e), + _ => None, + } + } +} + pub struct WorkerTransport { rx: tokio::sync::mpsc::Receiver>, send_service: tokio::sync::mpsc::Sender>, @@ -475,23 +503,23 @@ impl Transport for WorkerTransport { // (cancel, clean transport close, handler termination, idle timeout) // carry no diagnostic value and must not masquerade as a `Transport` // quit reason on the service side. - let message = match reason.as_ref() { - WorkerQuitReason::StartupFailed { error, context } => { - format!("startup failed, when {context}: {error}") - } - WorkerQuitReason::Fatal { error, context } => { - format!("transport fatal, when {context}: {error}") - } - WorkerQuitReason::Join(e) => format!("worker join error: {e}"), + match reason.as_ref() { + WorkerQuitReason::StartupFailed { .. } + | WorkerQuitReason::Fatal { .. } + | WorkerQuitReason::Join(_) => {} WorkerQuitReason::Cancelled | WorkerQuitReason::TransportClosed | WorkerQuitReason::HandlerTerminated | WorkerQuitReason::IdleTimeout(_) => return None, - }; + } + // Retain the typed reason (not a stringified copy) so callers can still + // downcast to `WorkerQuitReason` and reach source-only data + // such as an auth challenge. The wrapper's `source()` exposes the inner + // typed error for source-chain walks like `is_authorization_required`. Some(crate::transport::DynamicTransportError::from_parts( std::any::type_name::(), std::any::TypeId::of::(), - Box::new(std::io::Error::other(message)), + Box::new(RetainedQuitReason(reason.clone())), )) } } @@ -659,7 +687,20 @@ mod tests { transport.close().await.unwrap(); let status = transport.quit_status().expect("quit reason retained"); let text = status.to_string(); - assert!(text.contains("startup failed"), "got: {text}"); + assert!(text.contains("Startup failed"), "got: {text}"); assert!(text.contains("peer unreachable"), "got: {text}"); + + // The retained reason is downcastable to the typed quit reason, not + // stringified — source-only data (here the io::Error kind) survives. + let retained = status + .error + .downcast::>() + .expect("typed reason downcast"); + match retained.0.as_ref() { + WorkerQuitReason::StartupFailed { error, .. } => { + assert_eq!(error.kind(), io::ErrorKind::ConnectionRefused); + } + other => panic!("expected StartupFailed, got {other:?}"), + } } } From 8c5ebbafdefc577e397584ed78c9288680102351 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:55:57 +0800 Subject: [PATCH 08/29] client: surface retained startup cause in pre-init expect_next_message (#1441) --- crates/rmcp/src/service/client.rs | 20 ++++++++++++++++---- 1 file changed, 16 insertions(+), 4 deletions(-) diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 21d68be89..5f46535d1 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -155,10 +155,22 @@ async fn expect_next_message( where T: Transport, { - transport - .receive() - .await - .ok_or_else(|| ClientInitializeError::ConnectionClosed(context.to_string())) + match transport.receive().await { + Some(m) => Ok(m), + None => { + // The input stream ended during the handshake. If the transport's + // worker already quit with a startup failure, surface that retained + // cause instead of a bare "connection closed" — `serve_inner` (which + // also reads quit_status) is not created until initialization + // succeeds, so this pre-init path is the only place a startup + // failure's cause can reach the caller. + let detail = match transport.quit_status() { + Some(cause) => format!("{context}: {cause}"), + None => context.to_string(), + }; + Err(ClientInitializeError::ConnectionClosed(detail)) + } + } } /// Whether `expect_response` answers a server ping that arrives before the From d944818fcc67e4578145ce184a8a3008435ce75d Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 09:56:01 +0800 Subject: [PATCH 09/29] streamable-http: startup failure retains real typed error (#1441) From 273216c04e8ff580b9d77921fcb5cac9303b207f Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 10:19:27 +0800 Subject: [PATCH 10/29] worker: publish quit_status before input channel closes; StartupFailed holds Arc (#1441) --- crates/rmcp/src/transport/worker.rs | 28 +++++++++++++++++++++------- 1 file changed, 21 insertions(+), 7 deletions(-) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index d2f2f1f33..8b865ccdf 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -34,9 +34,13 @@ pub enum WorkerQuitReason { /// challenge), not an internal fault. The real typed error is retained here /// rather than replaced by a synthetic placeholder, so the service loop and /// late observers can see the actual cause. + /// + /// The error is shared via `Arc` so it can be delivered to the in-flight + /// `send()` ack *and* retained here — a single typed value would be lost if + /// the ack's receiver is dropped (its own timeout) after a successful queue. #[error("Startup failed {error}, when {context}")] StartupFailed { - error: E, + error: Arc, context: Cow<'static, str>, }, #[error("Transport cancelled")] @@ -64,13 +68,13 @@ impl WorkerQuitReason { } pub fn startup_failed(error: E, context: impl Into>) -> Self { Self::StartupFailed { - error, + error: Arc::new(error), context: context.into(), } } pub fn startup_failed_context(context: impl Into>) -> impl FnOnce(E) -> Self { |e| Self::StartupFailed { - error: e, + error: Arc::new(e), context: context.into(), } } @@ -210,8 +214,8 @@ impl std::fmt::Display for RetainedQuitRe impl std::error::Error for RetainedQuitReason { fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { match self.0.as_ref() { - WorkerQuitReason::StartupFailed { error, .. } - | WorkerQuitReason::Fatal { error, .. } => Some(error), + WorkerQuitReason::StartupFailed { error, .. } => Some(error.as_ref()), + WorkerQuitReason::Fatal { error, .. } => Some(error), WorkerQuitReason::Join(e) => Some(e), _ => None, } @@ -289,6 +293,13 @@ impl WorkerTransport { let (quit_status_tx, quit_status) = tokio::sync::watch::channel(None); let join_handle = tokio::spawn(async move { + // Hold the input channel open until the terminal status is + // published: `run()` drops its `WorkerContext` (and thus + // `to_handler_tx`) on return, so without this a concurrently polled + // `receive()` could observe EOF and read `quit_status()` as `None` + // before the watch write lands. Keeping a clone alive until after + // the write orders EOF after the status is visible. + let keep_input_open = context.to_handler_tx.clone(); let result = worker .run(context) .instrument(tracing::span!( @@ -324,7 +335,7 @@ impl WorkerTransport { } // Retain the terminal reason for late observers; the same value is // shared with the join handle so close() can report join errors. - match result { + let quit = match result { Ok(()) => { let _ = quit_status_tx.send(None); Ok(()) @@ -334,7 +345,10 @@ impl WorkerTransport { let _ = quit_status_tx.send(Some(shared.clone())); Err(shared) } - } + }; + // Publish the status before the input channel closes. + drop(keep_input_open); + quit }); Self { rx: from_transport_rx, From 119a6f87189b9fcf42b5ed22c3a00c0afe6c67a1 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 10:19:31 +0800 Subject: [PATCH 11/29] client: pre-init returns TransportError with owned retained cause (#1441) --- crates/rmcp/src/service/client.rs | 23 +++++++++++++---------- 1 file changed, 13 insertions(+), 10 deletions(-) diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 5f46535d1..e7b55704a 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -159,16 +159,19 @@ where Some(m) => Ok(m), None => { // The input stream ended during the handshake. If the transport's - // worker already quit with a startup failure, surface that retained - // cause instead of a bare "connection closed" — `serve_inner` (which - // also reads quit_status) is not created until initialization - // succeeds, so this pre-init path is the only place a startup - // failure's cause can reach the caller. - let detail = match transport.quit_status() { - Some(cause) => format!("{context}: {cause}"), - None => context.to_string(), - }; - Err(ClientInitializeError::ConnectionClosed(detail)) + // worker already quit with a startup failure, return the retained + // cause as a transport error (not a stringified ConnectionClosed), + // so the caller can inspect its source and downcast it. + // `serve_inner` is only created after init succeeds, so this + // pre-init path is the only place a startup failure's cause reaches + // the caller. + match transport.quit_status() { + Some(cause) => Err(ClientInitializeError::TransportError { + error: cause, + context: context.to_string().into(), + }), + None => Err(ClientInitializeError::ConnectionClosed(context.to_string())), + } } } } From 3e22abf26aa753255c65a4b9ce597cecc7d4dd78 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 10:19:35 +0800 Subject: [PATCH 12/29] streamable-http: share typed startup error via Arc in ack + quit reason (#1441) --- .../src/transport/streamable_http_client.rs | 52 ++++++++++--------- 1 file changed, 28 insertions(+), 24 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 45c8d58d4..cc7847f1d 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -243,6 +243,16 @@ pub enum StreamableHttpError { Deserialize(#[from] serde_json::Error), #[error("Transport channel closed")] TransportChannelClosed, + /// A startup handshake failure whose typed error is shared with the + /// transport's retained quit reason. + /// + /// The startup error is not `Clone`, so it is shared via `Arc` and delivered + /// to the in-flight `send()` ack in this variant while the same value is + /// retained in the quit reason. Callers should prefer the typed error from + /// the quit reason; this variant only exists so a receiver that is dropped + /// mid-delivery does not destroy the only copy. + #[error("Startup handshake failed (typed error retained in quit reason)")] + StartupShared(#[source] Arc), #[error("Missing session id in HTTP response")] MissingSessionIdInResponse, #[cfg(feature = "auth")] @@ -1201,30 +1211,24 @@ impl Worker for StreamableHttpClientWorker { } Err(err) => { // Report the startup handshake failure as `StartupFailed`, not - // a `Fatal` with a synthetic `TransportChannelClosed`. Capture - // the real error's text up front: the typed value can only live - // in one place (it is not `Clone`), so it goes to whichever - // receiver is still listening — the in-flight `send()` ack, or, - // if that receiver already went away (its own timeout fired), - // the quit reason for late observers and the service loop. - let err_text = format!("{err}"); - match responder.send(Err(err)) { - Ok(()) => { - // The parked caller received the typed error; the quit - // reason still names the real cause in text form. - return Err(WorkerQuitReason::startup_failed( - StreamableHttpError::TransportChannelClosed, - format!("startup request failed: {err_text}"), - )); - } - Err(Err(err)) => { - return Err(WorkerQuitReason::startup_failed( - err, - "startup request failed", - )); - } - Err(Ok(())) => unreachable!("only Err is sent in this branch"), - } + // a `Fatal` with a synthetic `TransportChannelClosed`. The typed + // error is shared via `Arc`: the quit reason retains it AND the + // in-flight `send()` ack carries the same `Arc` (as + // `StartupShared`), so even if the ack's receiver is dropped + // after a successful queue (its own timeout or a cancellation + // winning), the typed error survives in the quit reason for late + // observers and the service loop. + let err = Arc::new(err); + let quit_reason = WorkerQuitReason::StartupFailed { + error: err.clone(), + context: "startup request failed".into(), + }; + // Coerce to a shared dyn-Error handle for the ack so the same + // typed error is reachable through the source chain (auth + // challenge included) without a self-referential Arc type. + let shared: Arc = err; + let _ = responder.send(Err(StreamableHttpError::StartupShared(shared))); + return Err(quit_reason); } }; let mut uses_modern_http = !is_legacy_startup; From 3a5d6ebedae6bfeb32d2a0cc2c14790acb1fe690 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:33:53 +0800 Subject: [PATCH 13/29] review round 3: preserve typed startup access and fallback cause --- crates/rmcp/tests/test_streamable_http_status_metadata.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/crates/rmcp/tests/test_streamable_http_status_metadata.rs b/crates/rmcp/tests/test_streamable_http_status_metadata.rs index e00cd5559..1a37418f8 100644 --- a/crates/rmcp/tests/test_streamable_http_status_metadata.rs +++ b/crates/rmcp/tests/test_streamable_http_status_metadata.rs @@ -110,7 +110,7 @@ fn assert_status_error(error: StreamableHttpError, status: u16) !rendered.contains("private-"), "retained private response/request state: {rendered}" ); - match error { + match error.startup_error() { StreamableHttpError::HttpStatus(metadata) => { assert_eq!(metadata.status, status); assert_eq!(metadata.retry_after.as_deref(), Some("17")); From 249a1d2be7d8d305e5e3f72f2de6ad7cd38ebcd5 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:35:04 +0800 Subject: [PATCH 14/29] review round 3: preserve typed startup access and fallback cause --- .../src/transport/streamable_http_client.rs | 93 +++++++++++++------ 1 file changed, 64 insertions(+), 29 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index cc7847f1d..90ca765fc 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -212,6 +212,40 @@ impl HttpStatusError { } } +/// Shared startup error delivered to a live sender while the same typed value +/// remains in the transport's quit reason. +#[derive(Debug)] +pub struct SharedStartupError { + error: Arc>, + // Keep an erased handle to the same allocation. Returning the typed field + // directly from `source()` would create a recursive `Error` trait bound. + source: Arc, +} + +impl SharedStartupError { + fn new(error: Arc>) -> Self { + let source: Arc = error.clone(); + Self { error, source } + } + + /// Return the original typed startup error. + pub fn error(&self) -> &StreamableHttpError { + self.error.as_ref() + } +} + +impl std::fmt::Display for SharedStartupError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + f.write_str("startup handshake failed") + } +} + +impl std::error::Error for SharedStartupError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + Some(self.source.as_ref()) + } +} + #[derive(Error, Debug)] #[non_exhaustive] pub enum StreamableHttpError { @@ -243,16 +277,11 @@ pub enum StreamableHttpError { Deserialize(#[from] serde_json::Error), #[error("Transport channel closed")] TransportChannelClosed, - /// A startup handshake failure whose typed error is shared with the - /// transport's retained quit reason. - /// - /// The startup error is not `Clone`, so it is shared via `Arc` and delivered - /// to the in-flight `send()` ack in this variant while the same value is - /// retained in the quit reason. Callers should prefer the typed error from - /// the quit reason; this variant only exists so a receiver that is dropped - /// mid-delivery does not destroy the only copy. - #[error("Startup handshake failed (typed error retained in quit reason)")] - StartupShared(#[source] Arc), + /// A startup handshake failure shared with the transport's retained quit + /// reason. Use [`startup_error`](Self::startup_error) to recover the original + /// typed error without changing how existing variants are inspected. + #[error("startup handshake failed")] + StartupShared(#[source] SharedStartupError), #[error("Missing session id in HTTP response")] MissingSessionIdInResponse, #[cfg(feature = "auth")] @@ -307,6 +336,20 @@ pub(crate) fn json_rpc_parse_error_message(err: &serde_json::Error, body: &[u8]) } impl StreamableHttpError { + /// Return the original startup error behind [`StartupShared`](Self::StartupShared), + /// or this error itself for every other variant. + /// + /// Startup errors are not generally `Clone`, so the live `send()` result and + /// the retained quit reason share one typed value through `Arc`. This + /// accessor preserves typed inspection of `HttpStatus`, `AuthRequired`, and + /// other variants without requiring callers to handle the sharing wrapper. + pub fn startup_error(&self) -> &Self { + match self { + Self::StartupShared(error) => error.error(), + _ => self, + } + } + /// The `WWW-Authenticate` challenge carried by this error, when the /// server answered 401 ([`AuthRequired`](Self::AuthRequired)) or 403 /// ([`InsufficientScope`](Self::InsufficientScope)). Feed it to @@ -314,7 +357,7 @@ impl StreamableHttpError { /// to authorize reactively. #[cfg(feature = "auth")] pub fn auth_challenge(&self) -> Option<&str> { - match self { + match self.startup_error() { Self::AuthRequired(error) => Some(&error.www_authenticate_header), Self::InsufficientScope(error) => Some(&error.www_authenticate_header), _ => None, @@ -1210,25 +1253,17 @@ impl Worker for StreamableHttpClientWorker { )? } Err(err) => { - // Report the startup handshake failure as `StartupFailed`, not - // a `Fatal` with a synthetic `TransportChannelClosed`. The typed - // error is shared via `Arc`: the quit reason retains it AND the - // in-flight `send()` ack carries the same `Arc` (as - // `StartupShared`), so even if the ack's receiver is dropped - // after a successful queue (its own timeout or a cancellation - // winning), the typed error survives in the quit reason for late - // observers and the service loop. - let err = Arc::new(err); - let quit_reason = WorkerQuitReason::StartupFailed { - error: err.clone(), + // Keep one typed error in both delivery paths. The live sender + // can inspect the original variant through `startup_error()`; + // dropping an unreceived ack cannot destroy the retained copy. + let error = Arc::new(err); + let _ = responder.send(Err(StreamableHttpError::StartupShared( + SharedStartupError::new(error.clone()), + ))); + return Err(WorkerQuitReason::StartupFailed { + error, context: "startup request failed".into(), - }; - // Coerce to a shared dyn-Error handle for the ack so the same - // typed error is reachable through the source chain (auth - // challenge included) without a self-referential Arc type. - let shared: Arc = err; - let _ = responder.send(Err(StreamableHttpError::StartupShared(shared))); - return Err(quit_reason); + }); } }; let mut uses_modern_http = !is_legacy_startup; From 145637d9da1821537a585a4b90e7a4ba6426cce3 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:36:49 +0800 Subject: [PATCH 15/29] review round 3: preserve typed startup access and fallback cause --- crates/rmcp/src/transport/worker.rs | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index 8b865ccdf..1931709c5 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -203,6 +203,7 @@ impl WorkerSendRequest { /// (and thus the typed `E`), or walk [`Error::source`] to reach the inner typed /// error directly — preserving source-only data such as an auth challenge. #[derive(Debug)] +#[non_exhaustive] pub struct RetainedQuitReason(pub Arc>); impl std::fmt::Display for RetainedQuitReason { @@ -222,13 +223,15 @@ impl std::error::Error for Retaine } } +type WorkerJoinHandle = tokio::task::JoinHandle>>>; + pub struct WorkerTransport { rx: tokio::sync::mpsc::Receiver>, send_service: tokio::sync::mpsc::Sender>, control_send_service: tokio::sync::mpsc::Sender>, request_cancellations: RequestCancellations, control_generation: Arc, - join_handle: Option>>>>, + join_handle: Option>, /// Retained terminal status of the worker, written exactly once when its /// task quits. Lets an observer whose own timeout has already fired still /// learn why the worker died, rather than relying on a delivery window. From 8a2d9c6ec3f89460d21533b84a2b066ad14960b2 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:37:05 +0800 Subject: [PATCH 16/29] review round 3: preserve typed startup access and fallback cause --- crates/rmcp/src/service/client.rs | 117 +++++++++++++++++++++++++++++- 1 file changed, 115 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index e7b55704a..96876d4af 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -147,6 +147,35 @@ impl ClientInitializeError { } } +/// Auto's fallback can lose its send queue when the original startup request +/// fails after the discovery timeout. Prefer the worker's retained terminal +/// cause over the resulting synthetic channel-closed error. Closing a failed +/// transport also waits for the worker to publish that status: its request +/// queue can close just before the terminal-status write. +async fn recover_fallback_error + 'static>( + transport: &mut T, + error: ClientInitializeError, +) -> ClientInitializeError { + if !matches!(error, ClientInitializeError::TransportError { .. }) { + return error; + } + if transport.quit_status().is_none() { + // Initialization has already failed; preserve its error rather than any + // cleanup error. WorkerTransport::close awaits the worker task, which + // orders this read after its terminal-status publication. + let _ = transport.close().await; + } + match (error, transport.quit_status()) { + (ClientInitializeError::TransportError { context, .. }, Some(cause)) => { + ClientInitializeError::TransportError { + error: cause, + context, + } + } + (error, _) => error, + } +} + /// Helper function to get the next message from the stream async fn expect_next_message( transport: &mut T, @@ -903,6 +932,8 @@ where legacy_startup(&service, &mut transport, &id_provider, &peer, legacy_info) .await { + let fallback_error = + recover_fallback_error(&mut transport, fallback_error).await; return Err(ClientInitializeError::LegacyFallbackFailed { discover: discover_error, fallback: Box::new(fallback_error), @@ -915,8 +946,12 @@ where if let Some(version) = legacy_version { legacy_info.protocol_version = version; } - legacy_startup(&service, &mut transport, &id_provider, &peer, legacy_info) - .await?; + if let Err(error) = + legacy_startup(&service, &mut transport, &id_provider, &peer, legacy_info) + .await + { + return Err(recover_fallback_error(&mut transport, error).await); + } } } } @@ -2671,6 +2706,84 @@ mod tests { client.cancel().await.unwrap(); } + #[tokio::test] + async fn auto_fallback_send_recovers_retained_startup_cause() { + struct FallbackFailureTransport { + sends: Arc, + closed: bool, + } + + impl Transport for FallbackFailureTransport { + type Error = std::io::Error; + + fn send( + &mut self, + _item: TxJsonRpcMessage, + ) -> impl Future> + Send + 'static { + let attempt = self.sends.fetch_add(1, std::sync::atomic::Ordering::SeqCst); + async move { + if attempt == 0 { + futures::future::pending().await + } + Err(std::io::Error::new( + std::io::ErrorKind::BrokenPipe, + "request queue closed", + )) + } + } + + async fn receive(&mut self) -> Option> { + futures::future::pending().await + } + + async fn close(&mut self) -> Result<(), Self::Error> { + self.closed = true; + Ok(()) + } + + fn quit_status(&self) -> Option { + if !self.closed { + return None; + } + Some(DynamicTransportError::from_parts( + Self::name(), + std::any::TypeId::of::(), + Box::new(std::io::Error::new( + std::io::ErrorKind::ConnectionRefused, + "peer unreachable", + )), + )) + } + } + + let error = serve_client_with_ct_inner( + (), + FallbackFailureTransport { + sends: Arc::default(), + closed: false, + }, + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_11_25), + }, + CancellationToken::new(), + Duration::from_millis(1), + ) + .await + .expect_err("fallback send must fail"); + match error { + ClientInitializeError::TransportError { error, context } => { + assert_eq!(context.as_ref(), "send initialize request"); + let cause = error + .error + .downcast::() + .expect("retained typed cause"); + assert_eq!(cause.kind(), std::io::ErrorKind::ConnectionRefused); + } + other => panic!("expected retained transport cause, got {other:?}"), + } + } + #[tokio::test] async fn auto_startup_falls_back_when_discover_is_ignored() { use crate::model::{InitializeResult, ServerCapabilities}; From d5431cf5c81c873e4f3c7266f8ac83f0758cc080 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:57:25 +0800 Subject: [PATCH 17/29] review round 4: retain startup display and gate status wait --- crates/rmcp/src/transport.rs | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/crates/rmcp/src/transport.rs b/crates/rmcp/src/transport.rs index 8511f637b..6721a5bc0 100644 --- a/crates/rmcp/src/transport.rs +++ b/crates/rmcp/src/transport.rs @@ -148,6 +148,16 @@ where /// Close the transport fn close(&mut self) -> impl Future> + Send; + /// Whether this transport retains background-worker terminal status through + /// [`quit_status`](Self::quit_status). + /// + /// The default is `false`. Worker-backed transports override this together + /// with `quit_status`, allowing initialization error recovery to wait for + /// status publication without closing unrelated transport implementations. + fn supports_quit_status(&self) -> bool { + false + } + /// Terminal status of the transport's background worker, if it has quit. /// /// `None` means either the transport has no background worker or the worker From 8ffe22f1694844204067594d6e1e83ba5f41ec57 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:57:28 +0800 Subject: [PATCH 18/29] review round 4: retain startup display and gate status wait --- crates/rmcp/src/transport/worker.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index 1931709c5..59982b5d8 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -513,6 +513,10 @@ impl Transport for WorkerTransport { } } + fn supports_quit_status(&self) -> bool { + true + } + fn quit_status(&self) -> Option { let guard = self.quit_status.borrow(); let reason = guard.as_ref()?; From 95e96d3e2476799761c941c77ed875fa336d4424 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:57:31 +0800 Subject: [PATCH 19/29] review round 4: retain startup display and gate status wait --- crates/rmcp/src/transport/streamable_http_client.rs | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 90ca765fc..4b3b63adc 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -236,7 +236,7 @@ impl SharedStartupError { impl std::fmt::Display for SharedStartupError { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { - f.write_str("startup handshake failed") + std::fmt::Display::fmt(self.error.as_ref(), f) } } @@ -280,7 +280,7 @@ pub enum StreamableHttpError { /// A startup handshake failure shared with the transport's retained quit /// reason. Use [`startup_error`](Self::startup_error) to recover the original /// typed error without changing how existing variants are inspected. - #[error("startup handshake failed")] + #[error("startup handshake failed: {0}")] StartupShared(#[source] SharedStartupError), #[error("Missing session id in HTTP response")] MissingSessionIdInResponse, From 73bc723b3bfe5656d821733120960a8e1f8a273f Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:57:36 +0800 Subject: [PATCH 20/29] review round 4: retain startup display and gate status wait --- crates/rmcp/src/service/client.rs | 50 ++++++++++++++++++++++++++++++- 1 file changed, 49 insertions(+), 1 deletion(-) diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index 96876d4af..eaedef766 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -159,7 +159,7 @@ async fn recover_fallback_error + 'static>( if !matches!(error, ClientInitializeError::TransportError { .. }) { return error; } - if transport.quit_status().is_none() { + if transport.supports_quit_status() && transport.quit_status().is_none() { // Initialization has already failed; preserve its error rather than any // cleanup error. WorkerTransport::close awaits the worker task, which // orders this read after its terminal-status publication. @@ -2741,6 +2741,10 @@ mod tests { Ok(()) } + fn supports_quit_status(&self) -> bool { + true + } + fn quit_status(&self) -> Option { if !self.closed { return None; @@ -2784,6 +2788,50 @@ mod tests { } } + #[tokio::test] + async fn fallback_error_does_not_close_transport_without_quit_status() { + struct NoStatusTransport { + closed: bool, + } + + impl Transport for NoStatusTransport { + type Error = std::io::Error; + + fn send( + &mut self, + _item: TxJsonRpcMessage, + ) -> impl Future> + Send + 'static { + std::future::ready(Ok(())) + } + + async fn receive(&mut self) -> Option> { + None + } + + async fn close(&mut self) -> Result<(), Self::Error> { + self.closed = true; + Ok(()) + } + } + + let mut transport = NoStatusTransport { closed: false }; + let error = ClientInitializeError::transport::( + std::io::Error::new(std::io::ErrorKind::BrokenPipe, "send failed"), + "send initialize request", + ); + let error = recover_fallback_error(&mut transport, error).await; + assert!(!transport.closed, "unsupported transport must not close"); + match error { + ClientInitializeError::TransportError { error, .. } => { + let cause = error + .downcast::() + .expect("original transport error"); + assert_eq!(cause.kind(), std::io::ErrorKind::BrokenPipe); + } + other => panic!("expected original transport error, got {other:?}"), + } + } + #[tokio::test] async fn auto_startup_falls_back_when_discover_is_ignored() { use crate::model::{InitializeResult, ServerCapabilities}; From 7cd491fc3bf2ea14874d92e23a10b5691bf27f57 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 11:57:38 +0800 Subject: [PATCH 21/29] review round 4: retain startup display and gate status wait --- crates/rmcp/tests/test_streamable_http_status_metadata.rs | 4 ++++ 1 file changed, 4 insertions(+) diff --git a/crates/rmcp/tests/test_streamable_http_status_metadata.rs b/crates/rmcp/tests/test_streamable_http_status_metadata.rs index 1a37418f8..81b791c4e 100644 --- a/crates/rmcp/tests/test_streamable_http_status_metadata.rs +++ b/crates/rmcp/tests/test_streamable_http_status_metadata.rs @@ -110,6 +110,10 @@ fn assert_status_error(error: StreamableHttpError, status: u16) !rendered.contains("private-"), "retained private response/request state: {rendered}" ); + assert!( + rendered.contains(format!("HTTP {status}").as_str()), + "underlying status missing from Display: {rendered}" + ); match error.startup_error() { StreamableHttpError::HttpStatus(metadata) => { assert_eq!(metadata.status, status); From 77c6d199d5a31c66339d03d27088a108a9181e00 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:23:16 +0800 Subject: [PATCH 22/29] review round 5: retain initialized notification errors --- .../src/transport/streamable_http_client.rs | 24 ++++++++++++------- 1 file changed, 16 insertions(+), 8 deletions(-) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 4b3b63adc..74263363f 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -1307,7 +1307,8 @@ impl Worker for StreamableHttpClientWorker { &tool_header_cache, &negotiated_version, ); - self.client + let initialized_result = self + .client .post_message_with_max_sse_event_size( config.uri.clone(), initialized_notification.message, @@ -1317,13 +1318,20 @@ impl Worker for StreamableHttpClientWorker { config.max_sse_event_size, ) .await - .map_err(WorkerQuitReason::fatal_context( - "send initialized notification", - ))? - .expect_accepted_or_json::() - .map_err(WorkerQuitReason::fatal_context( - "process initialized notification response", - ))?; + .and_then(|response| response.expect_accepted_or_json::()); + if let Err(error) = initialized_result { + // The initialized notification is still part of startup. Deliver + // the real typed error to its waiter and retain the same value + // for observers whose receiver has already been dropped. + let error = Arc::new(error); + let _ = initialized_notification.responder.send(Err( + StreamableHttpError::StartupShared(SharedStartupError::new(error.clone())), + )); + return Err(WorkerQuitReason::StartupFailed { + error, + context: "send initialized notification".into(), + }); + } let _ = initialized_notification.responder.send(Ok(())); } #[expect( From 9a8a93c45562264726b5c41889094609332da3e4 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:23:18 +0800 Subject: [PATCH 23/29] review round 5: retain initialized notification errors --- .../test_discover_http_client_startup.rs | 68 ++++++++++++++++++- 1 file changed, 66 insertions(+), 2 deletions(-) diff --git a/crates/rmcp/tests/test_discover_http_client_startup.rs b/crates/rmcp/tests/test_discover_http_client_startup.rs index 70aa219dc..f88976d7e 100644 --- a/crates/rmcp/tests/test_discover_http_client_startup.rs +++ b/crates/rmcp/tests/test_discover_http_client_startup.rs @@ -17,10 +17,10 @@ use axum::{ use rmcp::{ ClientLifecycleMode, ClientServiceExt, ServerHandler, model::{ClientConfig, DiscoverResult, ErrorCode, ErrorData, ProtocolVersion}, - service::{MaybeSendFuture, RequestContext, RoleServer}, + service::{ClientInitializeError, MaybeSendFuture, RequestContext, RoleClient, RoleServer}, transport::{ StreamableHttpClientTransport, - streamable_http_client::StreamableHttpClientTransportConfig, + streamable_http_client::{StreamableHttpClientTransportConfig, StreamableHttpError}, streamable_http_server::{ StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager, }, @@ -58,6 +58,7 @@ impl ServerHandler for LegacyHttpServer { #[derive(Clone, Default)] struct PlainTextLegacyHttpState { methods: Arc>>, + fail_initialized: bool, } async fn plain_text_legacy_http_handler( @@ -97,6 +98,14 @@ async fn plain_text_legacy_http_handler( .expect("build initialize response"); } + if method == "notifications/initialized" && state.fail_initialized { + return Response::builder() + .status(StatusCode::TOO_MANY_REQUESTS) + .header("retry-after", "17") + .body(Body::from("rate limited")) + .expect("build notification failure"); + } + Response::builder() .status(StatusCode::ACCEPTED) .body(Body::empty()) @@ -237,6 +246,61 @@ async fn auto_http_client_falls_back_after_plain_text_4xx_rejection() { server.await.expect("server task"); } +#[tokio::test] +async fn initialized_notification_failure_preserves_typed_http_status() { + let ct = CancellationToken::new(); + let state = PlainTextLegacyHttpState { + methods: Arc::default(), + fail_initialized: true, + }; + let methods = state.methods.clone(); + let router = Router::new() + .route("/mcp", post(plain_text_legacy_http_handler)) + .with_state(state); + 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 error = ClientConfig::default() + .serve_with_lifecycle(transport, ClientLifecycleMode::Initialize) + .await + .expect_err("initialized notification must fail"); + match error { + ClientInitializeError::TransportError { error, .. } => { + assert!(error.to_string().contains("HTTP 429")); + let error = error + .downcast::, RoleClient>() + .expect("typed streamable HTTP error"); + match error.startup_error() { + StreamableHttpError::HttpStatus(metadata) => { + assert_eq!(metadata.status, 429); + assert_eq!(metadata.retry_after.as_deref(), Some("17")); + } + other => panic!("expected HTTP status, got {other:?}"), + } + } + other => panic!("expected transport error, got {other:?}"), + } + assert_eq!( + methods.lock().await.as_slice(), + &["initialize", "notifications/initialized"] + ); + ct.cancel(); + server.await.expect("server task"); +} + #[derive(Clone, Default)] struct VersionNegotiatingHttpState { methods: Arc>>, From b6f1fc13fa0ae75aba0ab04e0e9e5f8a816b33dc Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:37:04 +0800 Subject: [PATCH 24/29] review round 6: preserve live errors during fallback recovery --- crates/rmcp/src/transport.rs | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/crates/rmcp/src/transport.rs b/crates/rmcp/src/transport.rs index 6721a5bc0..492c43579 100644 --- a/crates/rmcp/src/transport.rs +++ b/crates/rmcp/src/transport.rs @@ -148,6 +148,12 @@ where /// Close the transport fn close(&mut self) -> impl Future> + Send; + /// Whether this is a synthetic channel-closed error produced after a + /// background worker disappeared, rather than a real peer/transport error. + fn is_closed_error(_error: &Self::Error) -> bool { + false + } + /// Whether this transport retains background-worker terminal status through /// [`quit_status`](Self::quit_status). /// From c3f42c06696277a6f83d6968dccea4220afbf627 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:37:07 +0800 Subject: [PATCH 25/29] review round 6: preserve live errors during fallback recovery --- crates/rmcp/src/transport/worker.rs | 9 +++++++++ 1 file changed, 9 insertions(+) diff --git a/crates/rmcp/src/transport/worker.rs b/crates/rmcp/src/transport/worker.rs index 59982b5d8..9b6438cc3 100644 --- a/crates/rmcp/src/transport/worker.rs +++ b/crates/rmcp/src/transport/worker.rs @@ -84,6 +84,11 @@ pub trait Worker: Sized + Send + 'static { type Error: std::error::Error + Send + Sync + 'static; type Role: ServiceRole; fn err_closed() -> Self::Error; + /// Whether this is the synthetic error produced when the worker's channel + /// closes, rather than a real transport failure delivered by the worker. + fn is_closed_error(_error: &Self::Error) -> bool { + false + } fn err_join(e: tokio::task::JoinError) -> Self::Error; fn run( self, @@ -513,6 +518,10 @@ impl Transport for WorkerTransport { } } + fn is_closed_error(error: &Self::Error) -> bool { + W::is_closed_error(error) + } + fn supports_quit_status(&self) -> bool { true } From 081d2d3d0237a487245e939bb025fcb4f7f62dcc Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:37:10 +0800 Subject: [PATCH 26/29] review round 6: preserve live errors during fallback recovery --- crates/rmcp/src/transport/streamable_http_client.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/rmcp/src/transport/streamable_http_client.rs b/crates/rmcp/src/transport/streamable_http_client.rs index 74263363f..facd34f5b 100644 --- a/crates/rmcp/src/transport/streamable_http_client.rs +++ b/crates/rmcp/src/transport/streamable_http_client.rs @@ -1193,6 +1193,9 @@ impl Worker for StreamableHttpClientWorker { fn err_closed() -> Self::Error { StreamableHttpError::TransportChannelClosed } + fn is_closed_error(error: &Self::Error) -> bool { + matches!(error, StreamableHttpError::TransportChannelClosed) + } fn err_join(e: tokio::task::JoinError) -> Self::Error { StreamableHttpError::TokioJoinError(e) } From 6649e1c8f5435d8522825430e64349836b68fa18 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:37:13 +0800 Subject: [PATCH 27/29] review round 6: preserve live errors during fallback recovery --- .../rmcp/src/transport/streamable_http_server/session/local.rs | 3 +++ 1 file changed, 3 insertions(+) diff --git a/crates/rmcp/src/transport/streamable_http_server/session/local.rs b/crates/rmcp/src/transport/streamable_http_server/session/local.rs index a3e6d19f4..fda30e491 100644 --- a/crates/rmcp/src/transport/streamable_http_server/session/local.rs +++ b/crates/rmcp/src/transport/streamable_http_server/session/local.rs @@ -1044,6 +1044,9 @@ impl Worker for LocalSessionWorker { fn err_closed() -> Self::Error { LocalSessionWorkerError::TransportClosed } + fn is_closed_error(error: &Self::Error) -> bool { + matches!(error, LocalSessionWorkerError::TransportClosed) + } fn err_join(e: tokio::task::JoinError) -> Self::Error { LocalSessionWorkerError::TokioJoinError(e) } From 7b97e9375ca3c47f3bb6e4f492dde4041d99621c Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:37:17 +0800 Subject: [PATCH 28/29] review round 6: preserve live errors during fallback recovery --- crates/rmcp/src/service/client.rs | 24 +++++++++++++++++++----- 1 file changed, 19 insertions(+), 5 deletions(-) diff --git a/crates/rmcp/src/service/client.rs b/crates/rmcp/src/service/client.rs index eaedef766..9badab17d 100644 --- a/crates/rmcp/src/service/client.rs +++ b/crates/rmcp/src/service/client.rs @@ -148,15 +148,25 @@ impl ClientInitializeError { } /// Auto's fallback can lose its send queue when the original startup request -/// fails after the discovery timeout. Prefer the worker's retained terminal -/// cause over the resulting synthetic channel-closed error. Closing a failed -/// transport also waits for the worker to publish that status: its request -/// queue can close just before the terminal-status write. +/// fails after the discovery timeout. Replace only the synthetic channel-closed +/// error; a live typed error delivered by the worker is already more useful. +/// Closing a failed worker transport also waits for status publication: its +/// request queue can close just before the terminal-status write. async fn recover_fallback_error + 'static>( transport: &mut T, error: ClientInitializeError, ) -> ClientInitializeError { - if !matches!(error, ClientInitializeError::TransportError { .. }) { + let ClientInitializeError::TransportError { + error: original, .. + } = &error + else { + return error; + }; + let is_closed = original + .error + .downcast_ref::() + .is_some_and(T::is_closed_error); + if !is_closed { return error; } if transport.supports_quit_status() && transport.quit_status().is_none() { @@ -2736,6 +2746,10 @@ mod tests { futures::future::pending().await } + fn is_closed_error(error: &Self::Error) -> bool { + error.kind() == std::io::ErrorKind::BrokenPipe + } + async fn close(&mut self) -> Result<(), Self::Error> { self.closed = true; Ok(()) From 9133157932e857e26fa0e33af578a6a397811601 Mon Sep 17 00:00:00 2001 From: ip2a Date: Wed, 11 Nov 2026 12:37:20 +0800 Subject: [PATCH 29/29] review round 6: preserve live errors during fallback recovery --- .../test_discover_http_client_startup.rs | 27 +++++++++++++++---- 1 file changed, 22 insertions(+), 5 deletions(-) diff --git a/crates/rmcp/tests/test_discover_http_client_startup.rs b/crates/rmcp/tests/test_discover_http_client_startup.rs index f88976d7e..2280210df 100644 --- a/crates/rmcp/tests/test_discover_http_client_startup.rs +++ b/crates/rmcp/tests/test_discover_http_client_startup.rs @@ -246,8 +246,11 @@ async fn auto_http_client_falls_back_after_plain_text_4xx_rejection() { server.await.expect("server task"); } +#[rstest::rstest] +#[case::initialize(false)] +#[case::auto_fallback(true)] #[tokio::test] -async fn initialized_notification_failure_preserves_typed_http_status() { +async fn initialized_notification_failure_preserves_typed_http_status(#[case] auto_fallback: bool) { let ct = CancellationToken::new(); let state = PlainTextLegacyHttpState { methods: Arc::default(), @@ -273,10 +276,22 @@ async fn initialized_notification_failure_preserves_typed_http_status() { let transport = StreamableHttpClientTransport::from_config( StreamableHttpClientTransportConfig::with_uri(format!("http://{address}/mcp")), ); + let lifecycle = if auto_fallback { + ClientLifecycleMode::Auto { + preferred_versions: vec![ProtocolVersion::V_2026_07_28], + legacy_version: Some(ProtocolVersion::V_2025_11_25), + } + } else { + ClientLifecycleMode::Initialize + }; let error = ClientConfig::default() - .serve_with_lifecycle(transport, ClientLifecycleMode::Initialize) + .serve_with_lifecycle(transport, lifecycle) .await .expect_err("initialized notification must fail"); + let error = match error { + ClientInitializeError::LegacyFallbackFailed { fallback, .. } if auto_fallback => *fallback, + error => error, + }; match error { ClientInitializeError::TransportError { error, .. } => { assert!(error.to_string().contains("HTTP 429")); @@ -293,10 +308,12 @@ async fn initialized_notification_failure_preserves_typed_http_status() { } other => panic!("expected transport error, got {other:?}"), } - assert_eq!( - methods.lock().await.as_slice(), + let expected_methods: &[&str] = if auto_fallback { + &["server/discover", "initialize", "notifications/initialized"] + } else { &["initialize", "notifications/initialized"] - ); + }; + assert_eq!(methods.lock().await.as_slice(), expected_methods); ct.cancel(); server.await.expect("server task"); }