fix(transport): retain terminal cause for startup failures (#1441)

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));

crates/rmcp/src/service/client.rs

@@ -147,6 +147,45 @@ impl ClientInitializeError {
     }
 }
 
+/// Auto's fallback can lose its send queue when the original startup request
+/// 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<T: Transport<RoleClient> + 'static>(
+    transport: &mut T,
+    error: ClientInitializeError,
+) -> ClientInitializeError {
+    let ClientInitializeError::TransportError {
+        error: original, ..
+    } = &error
+    else {
+        return error;
+    };
+    let is_closed = original
+        .error
+        .downcast_ref::<T::Error>()
+        .is_some_and(T::is_closed_error);
+    if !is_closed {
+        return error;
+    }
+    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.
+        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<T>(
     transport: &mut T,
@@ -155,10 +194,25 @@ async fn expect_next_message<T>(
 where
     T: Transport<RoleClient>,
 {
-    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, 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())),
+            }
+        }
+    }
 }
 
 /// Whether `expect_response` answers a server ping that arrives before the
@@ -888,6 +942,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),
@@ -900,8 +956,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);
+                    }
                 }
             }
         }
@@ -2656,6 +2716,136 @@ mod tests {
         client.cancel().await.unwrap();
     }
 
+    #[tokio::test]
+    async fn auto_fallback_send_recovers_retained_startup_cause() {
+        struct FallbackFailureTransport {
+            sends: Arc<std::sync::atomic::AtomicUsize>,
+            closed: bool,
+        }
+
+        impl Transport<RoleClient> for FallbackFailureTransport {
+            type Error = std::io::Error;
+
+            fn send(
+                &mut self,
+                _item: TxJsonRpcMessage<RoleClient>,
+            ) -> impl Future<Output = Result<(), Self::Error>> + 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<RxJsonRpcMessage<RoleClient>> {
+                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(())
+            }
+
+            fn supports_quit_status(&self) -> bool {
+                true
+            }
+
+            fn quit_status(&self) -> Option<DynamicTransportError> {
+                if !self.closed {
+                    return None;
+                }
+                Some(DynamicTransportError::from_parts(
+                    Self::name(),
+                    std::any::TypeId::of::<Self>(),
+                    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::<std::io::Error>()
+                    .expect("retained typed cause");
+                assert_eq!(cause.kind(), std::io::ErrorKind::ConnectionRefused);
+            }
+            other => panic!("expected retained transport cause, got {other:?}"),
+        }
+    }
+
+    #[tokio::test]
+    async fn fallback_error_does_not_close_transport_without_quit_status() {
+        struct NoStatusTransport {
+            closed: bool,
+        }
+
+        impl Transport<RoleClient> for NoStatusTransport {
+            type Error = std::io::Error;
+
+            fn send(
+                &mut self,
+                _item: TxJsonRpcMessage<RoleClient>,
+            ) -> impl Future<Output = Result<(), Self::Error>> + Send + 'static {
+                std::future::ready(Ok(()))
+            }
+
+            async fn receive(&mut self) -> Option<RxJsonRpcMessage<RoleClient>> {
+                None
+            }
+
+            async fn close(&mut self) -> Result<(), Self::Error> {
+                self.closed = true;
+                Ok(())
+            }
+        }
+
+        let mut transport = NoStatusTransport { closed: false };
+        let error = ClientInitializeError::transport::<NoStatusTransport>(
+            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::<NoStatusTransport, RoleClient>()
+                    .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};

crates/rmcp/src/transport.rs

@@ -147,6 +147,38 @@ where
 
     /// Close the transport
     fn close(&mut self) -> impl Future<Output = Result<(), Self::Error>> + 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).
+    ///
+    /// 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
+    /// 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<DynamicTransportError> {
+        None
+    }
 }
 
 pub trait IntoTransport<R, E, A>: Send + 'static