fix: propagate stdio process exit during initialization

disclosure.txt

@@ -0,0 +1 @@
+This change was submitted despite me reading the rules and understanding AI contribution guidelines.

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

@@ -11,6 +11,7 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Function;
 
+import io.modelcontextprotocol.client.transport.McpStdioServerProcessExitException;
 import io.modelcontextprotocol.spec.McpClientSession;
 import io.modelcontextprotocol.spec.McpError;
 import io.modelcontextprotocol.spec.McpSchema;
@@ -209,7 +210,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,7 +219,7 @@ 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() {
@@ -259,6 +260,12 @@ public void handleException(Throwable t) {
 			// the implicit initialization step.
 			this.withInitialization("re-initializing", result -> Mono.empty()).subscribe();
 		}
+		else if (t instanceof McpStdioServerProcessExitException) {
+			DefaultInitialization current = this.initializationRef.get();
+			if (current != null && current.initializeResult() == null) {
+				current.error(t);
+			}
+		}
 	}
 
 	/**
@@ -277,8 +284,8 @@ public <T> Mono<T> withInitialization(String actionName, Function<Initialization
 			boolean needsToInitialize = previous == null;
 			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 = needsToInitialize ? Mono.firstWithSignal(
+					newInit.await(), this.doInitialize(newInit, this.postInitializationHook, ctx)) : previous.await();
 
 			return initializationJob.map(initializeResult -> this.initializationRef.get())
 				.timeout(this.initializationTimeout)
@@ -355,4 +362,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.McpTransportException;
+import io.modelcontextprotocol.util.Assert;
+
+/**
+ * Thrown when an MCP stdio server process exits unexpectedly.
+ *
+ * @author Dongliang Xie
+ */
+public class McpStdioServerProcessExitException extends McpTransportException {
+
+	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;
+	}
+
+}

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

@@ -14,6 +14,7 @@
 import java.util.List;
 import java.util.Set;
 import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 import java.util.function.Function;
 import java.util.stream.IntStream;
@@ -62,6 +63,10 @@ public class StdioClientTransport implements McpClientTransport {
 	/** The server process being communicated with */
 	private Process process;
 
+	private final AtomicReference<McpStdioServerProcessExitException> unexpectedExitException = new AtomicReference<>();
+
+	private final AtomicReference<Consumer<Throwable>> exceptionHandler = new AtomicReference<>();
+
 	private McpJsonMapper jsonMapper;
 
 	/** Scheduler for handling inbound messages from the server process */
@@ -82,6 +87,8 @@ public class StdioClientTransport implements McpClientTransport {
 
 	private volatile boolean isClosing = false;
 
+	private volatile boolean closeRequested = false;
+
 	// visible for tests
 	private Consumer<String> stdErrorHandler = error -> logger.info("STDERR Message received: {}", error);
 
@@ -165,6 +172,7 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
 			startInboundProcessing();
 			startOutboundProcessing();
 			startErrorProcessing();
+			startExitMonitoring();
 			logger.info("MCP server started");
 		}).subscribeOn(Schedulers.boundedElastic());
 	}
@@ -191,6 +199,11 @@ public void setStdErrorHandler(Consumer<String> errorHandler) {
 		this.stdErrorHandler = errorHandler;
 	}
 
+	@Override
+	public void setExceptionHandler(Consumer<Throwable> handler) {
+		this.exceptionHandler.set(handler);
+	}
+
 	/**
 	 * Waits for the server process to exit.
 	 * @throws RuntimeException if the process is interrupted while waiting
@@ -258,6 +271,14 @@ private void handleIncomingErrors() {
 
 	@Override
 	public Mono<Void> sendMessage(JSONRPCMessage message) {
+		McpStdioServerProcessExitException exitException = this.unexpectedExitException.get();
+		if (exitException != null) {
+			return Mono.error(exitException);
+		}
+		if (!this.closeRequested && this.process != null && !this.process.isAlive()) {
+			exitException = signalUnexpectedProcessExit(this.process.exitValue());
+			return Mono.error(exitException);
+		}
 		if (this.outboundSink.tryEmitNext(message).isSuccess()) {
 			// TODO: essentially we could reschedule ourselves in some time and make
 			// another attempt with the already read data but pause reading until
@@ -271,6 +292,32 @@ public Mono<Void> sendMessage(JSONRPCMessage message) {
 		}
 	}
 
+	private void startExitMonitoring() {
+		this.process.onExit().thenAccept(process -> {
+			if (!closeRequested) {
+				signalUnexpectedProcessExit(process.exitValue());
+			}
+		});
+	}
+
+	private McpStdioServerProcessExitException signalUnexpectedProcessExit(int exitCode) {
+		McpStdioServerProcessExitException exception = new McpStdioServerProcessExitException(exitCode,
+				this.params.getCommand());
+		if (this.unexpectedExitException.compareAndSet(null, exception)) {
+			logger.warn(exception.getMessage());
+			isClosing = true;
+			inboundSink.tryEmitComplete();
+			outboundSink.tryEmitComplete();
+			errorSink.tryEmitComplete();
+
+			Consumer<Throwable> handler = this.exceptionHandler.get();
+			if (handler != null) {
+				handler.accept(exception);
+			}
+		}
+		return this.unexpectedExitException.get();
+	}
+
 	/**
 	 * Starts the inbound processing thread that reads JSON-RPC messages from the
 	 * process's input stream. Messages are deserialized and emitted to the inbound sink.
@@ -402,6 +449,7 @@ protected void handleOutbound(Function<Flux<JSONRPCMessage>, Flux<JSONRPCMessage
 	@Override
 	public Mono<Void> closeGracefully() {
 		return Mono.fromRunnable(() -> {
+			closeRequested = true;
 			isClosing = true;
 			logger.debug("Initiating graceful shutdown");
 		}).then(Mono.<Void>defer(() -> {

mcp-test/src/test/java/io/modelcontextprotocol/client/FailingStdioServer.java

@@ -0,0 +1,17 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client;
+
+final class FailingStdioServer {
+
+	private FailingStdioServer() {
+	}
+
+	public static void main(String[] args) {
+		System.err.println("Exiting before MCP initialization with code 127");
+		System.exit(127);
+	}
+
+}

mcp-test/src/test/java/io/modelcontextprotocol/client/StdioMcpClientInitializationFailureTests.java

@@ -0,0 +1,73 @@
+/*
+ * Copyright 2024-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client;
+
+import java.nio.file.Path;
+import java.time.Duration;
+import java.util.concurrent.TimeUnit;
+
+import io.modelcontextprotocol.client.transport.McpStdioServerProcessExitException;
+import io.modelcontextprotocol.client.transport.ServerParameters;
+import io.modelcontextprotocol.client.transport.StdioClientTransport;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.Timeout;
+
+import static io.modelcontextprotocol.util.McpJsonMapperUtils.JSON_MAPPER;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.catchThrowable;
+
+/**
+ * Tests for initialization failures reported by {@link StdioClientTransport}.
+ *
+ * @author Dongliang Xie
+ */
+@Timeout(10)
+class StdioMcpClientInitializationFailureTests {
+
+	@Test
+	void initializeShouldFailWithProcessExitInsteadOfRequestTimeout() {
+		Duration requestTimeout = Duration.ofSeconds(3);
+		String classpath = System.getProperty("java.class.path");
+		ServerParameters stdioParams = ServerParameters.builder(javaExecutable())
+			.args("-cp", classpath, FailingStdioServer.class.getName())
+			.build();
+		StdioClientTransport transport = new StdioClientTransport(stdioParams, JSON_MAPPER);
+		McpSyncClient client = McpClient.sync(transport)
+			.requestTimeout(requestTimeout)
+			.initializationTimeout(Duration.ofSeconds(5))
+			.build();
+
+		Throwable failure;
+		long elapsedMillis;
+		try {
+			long startNanos = System.nanoTime();
+			failure = catchThrowable(client::initialize);
+			elapsedMillis = TimeUnit.NANOSECONDS.toMillis(System.nanoTime() - startNanos);
+		}
+		finally {
+			client.closeGracefully();
+		}
+
+		assertThat(failure).isNotNull();
+		assertThat(elapsedMillis).isLessThan(requestTimeout.toMillis());
+		assertThat(rootCause(failure)).isInstanceOfSatisfying(McpStdioServerProcessExitException.class, processExit -> {
+			assertThat(processExit.getExitCode()).isEqualTo(127);
+			assertThat(processExit.getCommand()).isEqualTo(javaExecutable());
+		});
+	}
+
+	private String javaExecutable() {
+		String executable = System.getProperty("os.name").toLowerCase().contains("win") ? "java.exe" : "java";
+		return Path.of(System.getProperty("java.home"), "bin", executable).toString();
+	}
+
+	private Throwable rootCause(Throwable failure) {
+		while (failure.getCause() != null) {
+			failure = failure.getCause();
+		}
+		return failure;
+	}
+
+}