Reconnect a dropped SSE stream instead of failing the transport

The Streamable HTTP client ended the whole transport when its connection
stream closed, and when a session stream with a pending response
closed, even when the server had only detached it (backpressure, a
proxy in between restarting). The server keeps undelivered events in
the stream's mailbox, so the client now reopens the stream with a short
backoff and carries on. It still gives up when the server answers 404,
meaning the connection or session is gone, or after three reconnects in
a row that delivered no event, so a server that keeps closing cannot
cause an endless loop. The reopened stream replaces the old one only if
nothing else already did.

Author: Mark Pollack

CHANGELOG.md

@@ -39,6 +39,11 @@ building a second client on an already-connected transport now fails at construc
   GET first on `http://` endpoints, so against a server that speaks h2c (the SDK's own does) every
   request, streams included, runs on HTTP/2. Against one that does not, the probe settles on HTTP/1.1
   and later requests stop offering the upgrade, which some servers route to their WebSocket handler.
+- **The Streamable HTTP client reconnects a dropped SSE stream.** When the server or the network closes
+  the connection stream or a session stream, the client reopens it with a short backoff instead of
+  failing the transport; the server's mailbox delivers whatever was not yet written. It gives up, as
+  before, when the server answers 404 (the connection is gone) or after three reconnects in a row that
+  delivered nothing.
 - **Client sessions learn that their transport died.** `AcpClientTransport.awaitTermination()` (default:
   never) is implemented by the Streamable HTTP and WebSocket client transports; `AcpClientSession` fails
   pending requests at once with the cause, and every later request, instead of waiting out the request

acp-core/src/main/java/com/agentclientprotocol/sdk/client/transport/StreamableHttpAcpClientTransport.java

@@ -783,22 +783,73 @@ private void clearState() {
 		}
 	}
 
+	/** Reconnects in a row that delivered nothing before the transport gives up on a stream. */
+	private static final int MAX_BARREN_RECONNECTS = 3;
+
+	private static final Duration RECONNECT_BACKOFF = Duration.ofMillis(200);
+
+	/**
+	 * A stream closed that this client did not close: the server detached it (backpressure,
+	 * a restart of the proxy in between), or the network dropped it. The server keeps what
+	 * it has not delivered in the stream's mailbox, so reopening loses nothing: reconnect
+	 * with a short backoff. Give up, as before, when the server answers that the connection
+	 * is gone, or after {@value #MAX_BARREN_RECONNECTS} reconnects in a row that delivered no
+	 * event (a server that keeps closing must not cause an endless loop).
+	 */
 	private void handleUnexpectedSseClosure(SseStream stream, Throwable error) {
 		if (closing.get()) {
 			return;
 		}
-
 		RouteScope scope = stream.scope;
-		if (!scope.isSession()) {
-			terminateAfterSseFailure(error);
+		int barren = stream.delivered ? 0 : stream.barrenReconnects + 1;
+		if (barren > MAX_BARREN_RECONNECTS) {
+			giveUpOn(stream, error);
 			return;
 		}
+		logger.info("SSE stream closed unexpectedly; reconnecting: {}", scope);
+		Mono.defer(() -> openSseStream(scope))
+			.retryWhen(reactor.util.retry.Retry.backoff(2, RECONNECT_BACKOFF)
+				.scheduler(AcpSchedulers.timeouts())
+				.filter(e -> !isConnectionGone(e)))
+			.subscribe(reopened -> {
+				reopened.barrenReconnects = barren;
+				if (closing.get() || !replaceStream(stream, reopened)) {
+					reopened.close();
+					return;
+				}
+				reopened.start();
+				logger.info("SSE stream reconnected: {}", scope);
+			}, reconnectError -> giveUpOn(stream, reconnectError));
+	}
+
+	private boolean replaceStream(SseStream old, SseStream reopened) {
+		if (old.scope.isSession()) {
+			return sessionStreams.replace(old.scope.sessionId(), old, reopened);
+		}
+		synchronized (this) {
+			if (this.connectionStream != old) {
+				return false;
+			}
+			this.connectionStream = reopened;
+			return true;
+		}
+	}
+
+	/** 404 on reconnect: the server no longer knows this connection or session. */
+	private static boolean isConnectionGone(Throwable error) {
+		return error instanceof AcpConnectionException && String.valueOf(error.getMessage()).contains("got 404");
+	}
 
-		if (hasPendingResponseFor(scope)) {
+	/** The old behaviour: a dead connection stream, or a session stream owing a response, ends the transport. */
+	private void giveUpOn(SseStream stream, Throwable error) {
+		if (closing.get()) {
+			return;
+		}
+		RouteScope scope = stream.scope;
+		if (!scope.isSession() || hasPendingResponseFor(scope)) {
 			terminateAfterSseFailure(error);
 			return;
 		}
-
 		if (sessionStreams.remove(scope.sessionId(), stream)) {
 			sessionStreamOpenOperations.remove(scope.sessionId());
 			logger.info("Session SSE stream closed; it will be reopened before the next session request: {}", scope);
@@ -873,6 +924,12 @@ private class SseStream {
 
 		private Future<?> readerTask;
 
+		/** Whether this stream carried at least one event; resets the reconnect budget. */
+		private volatile boolean delivered;
+
+		/** Reconnects in a row, ending with this stream, that delivered nothing. */
+		private volatile int barrenReconnects;
+
 		SseStream(RouteScope scope, InputStream body) {
 			this.scope = scope;
 			this.body = body;
@@ -939,6 +996,7 @@ private void dispatchEvent(StringBuilder dataBuffer) {
 			}
 			try {
 				JSONRPCMessage message = AcpSchema.deserializeJsonRpcMessage(jsonMapper, dataBuffer.toString());
+				delivered = true;
 				processInbound(scope, message).subscribe(v -> {
 				}, error -> {
 					if (!closed.get() && !closing.get()) {

acp-core/src/test/java/com/agentclientprotocol/sdk/client/transport/StreamableHttpAcpClientTransportTest.java

@@ -828,9 +828,12 @@ void cleartextServerWithoutH2cGetsPlainHttp11Requests() throws Exception {
 			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_INITIALIZE, "init-1",
 					AcpTestFixtures.createInitializeRequest())).block();
 
-			assertThat(requests.get(0).method()).as("the probe").isEqualTo("GET");
-			assertThat(requests.get(0).headers().firstValue("Acp-Connection-Id")).isEmpty();
-			assertThat(requests.subList(1, requests.size()))
+			// The mocked connection stream ends at once, so the client keeps reconnecting while
+			// this runs; take a snapshot. Reconnects must be pinned too.
+			List<HttpRequest> seen = List.copyOf(requests);
+			assertThat(seen.get(0).method()).as("the probe").isEqualTo("GET");
+			assertThat(seen.get(0).headers().firstValue("Acp-Connection-Id")).isEmpty();
+			assertThat(seen.subList(1, seen.size()))
 				.as("every request after the probe is pinned to HTTP/1.1")
 				.isNotEmpty()
 				.allMatch(request -> request.version().equals(java.util.Optional.of(HttpClient.Version.HTTP_1_1)));
@@ -858,6 +861,63 @@ void initializeRequiresConnectionIdHeader() throws Exception {
 			.hasMessageContaining("Acp-Connection-Id");
 	}
 
+	/**
+	 * The server closed the connection stream once (backpressure, a proxy restart). The
+	 * client reconnects, and a response the server sends afterwards is delivered: its
+	 * mailbox kept it.
+	 */
+	@Test
+	void droppedConnectionStreamIsReconnectedAndLaterResponsesArrive() throws Exception {
+		HttpClient httpClient = mock(HttpClient.class);
+		AtomicInteger connectionGets = new AtomicInteger();
+		PipedInputStream secondBody = new PipedInputStream();
+		PipedOutputStream secondWriter = new PipedOutputStream(secondBody);
+		BlockingQueue<AcpSchema.JSONRPCMessage> inbound = new LinkedBlockingQueue<>();
+		when(httpClient.sendAsync(any(), any())).thenAnswer(invocation -> {
+			HttpRequest request = invocation.getArgument(0);
+			if ("POST".equals(request.method()) && request.headers().firstValue("Acp-Connection-Id").isEmpty()) {
+				String init = jsonMapper.writeValueAsString(AcpTestFixtures.createJsonRpcResponse("init-1",
+						AcpTestFixtures.createInitializeResponse()));
+				return CompletableFuture.completedFuture(response(200,
+						Map.of("Content-Type", "application/json", "Acp-Connection-Id", "conn-1"), init));
+			}
+			if ("GET".equals(request.method()) && request.headers().firstValue("Acp-Session-Id").isEmpty()) {
+				// First connection stream ends at once; the reconnect gets a live one.
+				InputStream body = connectionGets.incrementAndGet() == 1 ? emptyBody() : secondBody;
+				return CompletableFuture.completedFuture(response(200, Map.of("Content-Type", "text/event-stream"), body));
+			}
+			if ("GET".equals(request.method())) {
+				return CompletableFuture.completedFuture(
+						response(200, Map.of("Content-Type", "text/event-stream"), new PipedInputStream(1024)));
+			}
+			return CompletableFuture.completedFuture(response(202, Map.of(), null));
+		});
+		StreamableHttpAcpClientTransport transport = new StreamableHttpAcpClientTransport(
+				URI.create("https://localhost:8443/acp"), jsonMapper, httpClient);
+		try {
+			transport.connect(message -> message.doOnNext(inbound::add).then(Mono.empty())).block();
+			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_INITIALIZE, "init-1",
+					AcpTestFixtures.createInitializeRequest())).block();
+			awaitResponse(inbound, "init-1");
+
+			long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5);
+			while (connectionGets.get() < 2 && System.nanoTime() < deadline) {
+				Thread.sleep(20);
+			}
+			assertThat(connectionGets.get()).as("the connection stream was reopened").isGreaterThanOrEqualTo(2);
+
+			transport.sendMessage(AcpTestFixtures.createJsonRpcRequest(AcpSchema.METHOD_SESSION_NEW, "new-1",
+					AcpTestFixtures.createNewSessionRequest("/workspace"))).block();
+			writeSse(secondWriter, new AcpSchema.JSONRPCResponse(AcpSchema.JSONRPC_VERSION, "new-1",
+					new AcpSchema.NewSessionResponse("sess-1", null, null), null));
+			assertThat(awaitResponse(inbound, "new-1").error()).isNull();
+		}
+		finally {
+			secondWriter.close();
+			transport.close();
+		}
+	}
+
 	private void awaitSessionSseClosure() throws InterruptedException {
 		Thread.sleep(100);
 	}