fix: propagate stdio process exit during initialization

mcp-core/src/main/java/io/modelcontextprotocol/client/LifecycleInitializer.java

@@ -15,6 +15,7 @@
 import io.modelcontextprotocol.spec.McpError;
 import io.modelcontextprotocol.spec.McpSchema;
 import io.modelcontextprotocol.spec.McpTransportSessionNotFoundException;
+import io.modelcontextprotocol.spec.McpTransportTerminatedException;
 import io.modelcontextprotocol.util.Assert;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
@@ -93,6 +94,9 @@ class LifecycleInitializer {
 
 	private final AtomicReference<DefaultInitialization> initializationRef = new AtomicReference<>();
 
+	/** Permanent transport failure that prevents any further initialization attempt. */
+	private final AtomicReference<Throwable> terminalFailure = new AtomicReference<>();
+
 	/**
 	 * The max timeout to await for the client-server connection to be initialized.
 	 */
@@ -209,7 +213,7 @@ private Mono<McpSchema.InitializeResult> await() {
 
 		private void complete(McpSchema.InitializeResult initializeResult) {
 			// inform all the subscribers waiting for the initialization
-			this.initSink.emitValue(initializeResult, Sinks.EmitFailureHandler.FAIL_FAST);
+			this.initSink.tryEmitValue(initializeResult);
 		}
 
 		private void cacheResult(McpSchema.InitializeResult initializeResult) {
@@ -218,24 +222,38 @@ private void cacheResult(McpSchema.InitializeResult initializeResult) {
 		}
 
 		private void error(Throwable t) {
-			this.initSink.emitError(t, Sinks.EmitFailureHandler.FAIL_FAST);
+			this.initSink.tryEmitError(t);
 		}
 
 		private void close() {
 			this.mcpSession().close();
 		}
 
+		private void terminate(Throwable cause) {
+			// Initialization has a single shared outcome. Publish the terminal failure
+			// even when the session has not been installed yet so that both the owner
+			// and all concurrent joiners observe it immediately.
+			this.initSink.tryEmitError(cause);
+			McpClientSession mcpClientSession = this.mcpSession();
+			if (mcpClientSession != null) {
+				mcpClientSession.terminate(cause);
+			}
+		}
+
 		private Mono<Void> closeGracefully() {
 			return this.mcpSession().closeGracefully();
 		}
 
 	}
 
 	public boolean isInitialized() {
-		return this.currentInitializationResult() != null;
+		return this.terminalFailure.get() == null && this.currentInitializationResult() != null;
 	}
 
 	public McpSchema.InitializeResult currentInitializationResult() {
+		if (this.terminalFailure.get() != null) {
+			return null;
+		}
 		DefaultInitialization current = this.initializationRef.get();
 		McpSchema.InitializeResult initializeResult = current != null ? current.result.get() : null;
 		return initializeResult;
@@ -250,7 +268,15 @@ public McpSchema.InitializeResult currentInitializationResult() {
 	 * @param t The exception to handle
 	 */
 	public void handleException(Throwable t) {
-		if (t instanceof McpTransportSessionNotFoundException) {
+		if (t instanceof McpTransportTerminatedException) {
+			if (this.terminalFailure.compareAndSet(null, t)) {
+				DefaultInitialization current = this.initializationRef.get();
+				if (current != null) {
+					current.terminate(t);
+				}
+			}
+		}
+		else if (t instanceof McpTransportSessionNotFoundException && this.terminalFailure.get() == null) {
 			DefaultInitialization previous = this.initializationRef.getAndSet(null);
 			if (previous != null) {
 				previous.close();
@@ -271,14 +297,39 @@ public void handleException(Throwable t) {
 	 */
 	public <T> Mono<T> withInitialization(String actionName, Function<Initialization, Mono<T>> operation) {
 		return Mono.deferContextual(ctx -> {
+			Throwable terminal = this.terminalFailure.get();
+			if (terminal != null) {
+				return Mono.error(terminal);
+			}
+
 			DefaultInitialization newInit = new DefaultInitialization();
 			DefaultInitialization previous = this.initializationRef.compareAndExchange(null, newInit);
 
 			boolean needsToInitialize = previous == null;
+			DefaultInitialization activeInitialization = needsToInitialize ? newInit : previous;
+			// Complete the handoff with handleException(): if termination won before
+			// registration, this check publishes it; if registration won first, the
+			// exception handler observes the active initialization.
+			Throwable terminalAfterRegistration = this.terminalFailure.get();
+			if (terminalAfterRegistration != null) {
+				activeInitialization.terminate(terminalAfterRegistration);
+			}
 			logger.debug(needsToInitialize ? "Initialization process started" : "Joining previous initialization");
 
-			Mono<McpSchema.InitializeResult> initializationJob = needsToInitialize
-					? this.doInitialize(newInit, this.postInitializationHook, ctx) : previous.await();
+			Mono<McpSchema.InitializeResult> initializationJob;
+			if (needsToInitialize) {
+				// The work branch only publishes into the shared sink. Keeping it from
+				// winning directly makes the owner and all joiners consume the same
+				// first terminal signal.
+				Mono<McpSchema.InitializeResult> initializationWork = this
+					.doInitialize(newInit, this.postInitializationHook, ctx)
+					.onErrorComplete()
+					.then(Mono.never());
+				initializationJob = Mono.firstWithSignal(activeInitialization.await(), initializationWork);
+			}
+			else {
+				initializationJob = activeInitialization.await();
+			}
 
 			return initializationJob.map(initializeResult -> this.initializationRef.get())
 				.timeout(this.initializationTimeout)
@@ -295,42 +346,46 @@ public <T> Mono<T> withInitialization(String actionName, Function<Initialization
 	private Mono<McpSchema.InitializeResult> doInitialize(DefaultInitialization initialization,
 			Function<Initialization, Mono<Void>> postInitOperation, ContextView ctx) {
 
-		initialization.setMcpClientSession(this.sessionSupplier.apply(ctx));
-
-		McpClientSession mcpClientSession = initialization.mcpSession();
-
-		String latestVersion = this.protocolVersions.get(this.protocolVersions.size() - 1);
-
-		McpSchema.InitializeRequest initializeRequest = McpSchema.InitializeRequest
-			.builder(latestVersion, this.clientCapabilities, this.clientInfo)
-			.build();
-
-		Mono<McpSchema.InitializeResult> result = mcpClientSession.sendRequest(McpSchema.METHOD_INITIALIZE,
-				initializeRequest, McpAsyncClient.INITIALIZE_RESULT_TYPE_REF);
-
-		return result.flatMap(initializeResult -> {
-			logger.info("Server response with Protocol: {}, Capabilities: {}, Info: {} and Instructions {}",
-					initializeResult.protocolVersion(), initializeResult.capabilities(), initializeResult.serverInfo(),
-					initializeResult.instructions());
+		return Mono.defer(() -> {
+			initialization.setMcpClientSession(this.sessionSupplier.apply(ctx));
 
-			if (!this.protocolVersions.contains(initializeResult.protocolVersion())) {
-				return Mono.error(McpError.builder(-32602)
-					.message("Unsupported protocol version")
-					.data("Unsupported protocol version from the server: " + initializeResult.protocolVersion())
-					.build());
+			McpClientSession mcpClientSession = initialization.mcpSession();
+			Throwable terminal = this.terminalFailure.get();
+			if (terminal != null) {
+				mcpClientSession.terminate(terminal);
+				return Mono.error(terminal);
 			}
 
-			return mcpClientSession.sendNotification(McpSchema.METHOD_NOTIFICATION_INITIALIZED, null)
-				.contextWrite(
-						c -> c.put(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION, initializeResult.protocolVersion()))
-				.thenReturn(initializeResult);
-		}).flatMap(initializeResult -> {
-			initialization.cacheResult(initializeResult);
-			return postInitOperation.apply(initialization).thenReturn(initializeResult);
-		}).doOnNext(initialization::complete).onErrorResume(ex -> {
-			initialization.error(ex);
-			return Mono.error(ex);
-		});
+			String latestVersion = this.protocolVersions.get(this.protocolVersions.size() - 1);
+
+			McpSchema.InitializeRequest initializeRequest = McpSchema.InitializeRequest
+				.builder(latestVersion, this.clientCapabilities, this.clientInfo)
+				.build();
+
+			Mono<McpSchema.InitializeResult> result = mcpClientSession.sendRequest(McpSchema.METHOD_INITIALIZE,
+					initializeRequest, McpAsyncClient.INITIALIZE_RESULT_TYPE_REF);
+
+			return result.flatMap(initializeResult -> {
+				logger.info("Server response with Protocol: {}, Capabilities: {}, Info: {} and Instructions {}",
+						initializeResult.protocolVersion(), initializeResult.capabilities(),
+						initializeResult.serverInfo(), initializeResult.instructions());
+
+				if (!this.protocolVersions.contains(initializeResult.protocolVersion())) {
+					return Mono.error(McpError.builder(-32602)
+						.message("Unsupported protocol version")
+						.data("Unsupported protocol version from the server: " + initializeResult.protocolVersion())
+						.build());
+				}
+
+				return mcpClientSession.sendNotification(McpSchema.METHOD_NOTIFICATION_INITIALIZED, null)
+					.contextWrite(
+							c -> c.put(McpAsyncClient.NEGOTIATED_PROTOCOL_VERSION, initializeResult.protocolVersion()))
+					.thenReturn(initializeResult);
+			}).flatMap(initializeResult -> {
+				initialization.cacheResult(initializeResult);
+				return postInitOperation.apply(initialization).thenReturn(initializeResult);
+			});
+		}).doOnNext(initialization::complete).doOnError(initialization::error);
 	}
 
 	/**
@@ -355,4 +410,4 @@ public Mono<?> closeGracefully() {
 		});
 	}
 
-}
+}

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/McpStdioServerProcessExitException.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+import io.modelcontextprotocol.spec.McpTransportTerminatedException;
+import io.modelcontextprotocol.util.Assert;
+
+/**
+ * Thrown when an MCP stdio server process exits unexpectedly.
+ *
+ * @author Dongliang Xie
+ */
+public class McpStdioServerProcessExitException extends McpTransportTerminatedException {
+
+	private static final long serialVersionUID = 1L;
+
+	private final int exitCode;
+
+	private final String command;
+
+	public McpStdioServerProcessExitException(int exitCode, String command) {
+		super(message(exitCode, command));
+		this.exitCode = exitCode;
+		this.command = command;
+	}
+
+	public int getExitCode() {
+		return this.exitCode;
+	}
+
+	public String getCommand() {
+		return this.command;
+	}
+
+	private static String message(int exitCode, String command) {
+		Assert.hasText(command, "The command can not be empty");
+		return "MCP server process exited unexpectedly with code " + exitCode + " for command: " + command;
+	}
+
+}