Isolate the shared timeout timer; keep an event whose write fails

From the cold reviews of the post-merge hardening:

- The shared timer kept cancelled timeout tasks queued until they would
  have fired, and delivered timeout errors on its only thread, so a
  subscriber blocking in its error handler delayed every other
  session's timeouts. Cancelled tasks are now removed, and
  AcpSchedulers.withTimeout hands the TimeoutException to a separate
  daemon thread. Tested.
- An SSE event whose socket write threw was dropped; it now goes back to
  the front of the mailbox for the next subscriber. Tested.
- One connection whose keep-alive throws no longer stops keep-alives for
  all the others.
- StreamableHttpAcpServlet is package-private again: its constructor
  needs the transport's internal registry, so a public class name would
  have promised an embedding API that does not exist yet.

Author: Mark Pollack

acp-core/src/main/java/com/agentclientprotocol/sdk/spec/AcpAgentSession.java

@@ -55,7 +55,7 @@ public class AcpAgentSession implements AcpSession {
 	private final Duration requestTimeout;
 
 	/**
-	 * Per-session daemon scheduler for timeout operations. Disposed when session closes.
+	 * The JVM-wide daemon timer shared by every session (see AcpSchedulers); never disposed here.
 	 */
 	private final Scheduler timeoutScheduler;
 
@@ -405,7 +405,7 @@ public <T> Mono<T> sendRequest(String method, Object requestParams, TypeRef<T> t
 				this.pendingResponses.remove(requestId);
 				pendingResponseSink.error(error);
 			});
-		})).timeout(this.requestTimeout, timeoutScheduler).handle((jsonRpcResponse, deliveredResponseSink) -> {
+		})).transform(response -> AcpSchedulers.withTimeout(response, this.requestTimeout)).handle((jsonRpcResponse, deliveredResponseSink) -> {
 			if (jsonRpcResponse.error() != null) {
 				logger.error("Error handling request: {}", jsonRpcResponse.error());
 				deliveredResponseSink.error(new AcpError(jsonRpcResponse.error()));

acp-core/src/main/java/com/agentclientprotocol/sdk/spec/AcpClientSession.java

@@ -58,7 +58,7 @@ public class AcpClientSession implements AcpSession {
 	private final Duration requestTimeout;
 
 	/**
-	 * Per-session daemon scheduler for timeout operations. Disposed when session closes.
+	 * The JVM-wide daemon timer shared by every session (see AcpSchedulers); never disposed here.
 	 */
 	private final Scheduler timeoutScheduler;
 
@@ -418,7 +418,7 @@ public <T> Mono<T> sendRequest(String method, Object requestParams, TypeRef<T> t
 				this.pendingResponses.remove(requestId);
 				pendingResponseSink.error(error);
 			});
-		})).timeout(this.requestTimeout, timeoutScheduler).handle((jsonRpcResponse, deliveredResponseSink) -> {
+		})).transform(response -> AcpSchedulers.withTimeout(response, this.requestTimeout)).handle((jsonRpcResponse, deliveredResponseSink) -> {
 			if (jsonRpcResponse.error() != null) {
 				logger.error("Error handling request: {}", jsonRpcResponse.error());
 				deliveredResponseSink.error(new AcpError(jsonRpcResponse.error()));

acp-core/src/main/java/com/agentclientprotocol/sdk/util/AcpSchedulers.java

@@ -4,8 +4,12 @@
 
 package com.agentclientprotocol.sdk.util;
 
+import java.time.Duration;
 import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledThreadPoolExecutor;
+import java.util.concurrent.TimeoutException;
 
+import reactor.core.publisher.Mono;
 import reactor.core.scheduler.Scheduler;
 import reactor.core.scheduler.Schedulers;
 
@@ -26,11 +30,29 @@ public final class AcpSchedulers {
 
 	private static final class TimeoutHolder {
 
-		static final Scheduler SCHEDULER = Schedulers.fromExecutorService(Executors.newScheduledThreadPool(1, r -> {
-			Thread t = new Thread(r, "acp-timeout");
+		static final Scheduler SCHEDULER = Schedulers.fromExecutorService(timerExecutor(), "acp-timeout");
+
+		/**
+		 * Timeout errors are handed off the timer so a subscriber that blocks in its error
+		 * handler cannot delay every other session's timeouts.
+		 */
+		static final Scheduler DELIVERY = Schedulers.fromExecutorService(Executors.newCachedThreadPool(r -> {
+			Thread t = new Thread(r, "acp-timeout-delivery");
 			t.setDaemon(true);
 			return t;
-		}), "acp-timeout");
+		}), "acp-timeout-delivery");
+
+		private static ScheduledThreadPoolExecutor timerExecutor() {
+			ScheduledThreadPoolExecutor executor = new ScheduledThreadPoolExecutor(1, r -> {
+				Thread t = new Thread(r, "acp-timeout");
+				t.setDaemon(true);
+				return t;
+			});
+			// A request that completes cancels its timeout; drop the task instead of
+			// keeping it queued until it would have fired (up to the request timeout).
+			executor.setRemoveOnCancelPolicy(true);
+			return executor;
+		}
 
 	}
 
@@ -45,4 +67,16 @@ public static Scheduler timeouts() {
 		return TimeoutHolder.SCHEDULER;
 	}
 
+	/**
+	 * Applies a timeout on the shared timer and delivers the resulting
+	 * {@link TimeoutException} on a separate daemon thread.
+	 * @param mono the source
+	 * @param timeout how long to wait for its first signal
+	 * @return the source with the timeout applied
+	 */
+	public static <T> Mono<T> withTimeout(Mono<T> mono, Duration timeout) {
+		return mono.timeout(timeout, TimeoutHolder.SCHEDULER)
+			.onErrorResume(TimeoutException.class, e -> Mono.<T>error(e).subscribeOn(TimeoutHolder.DELIVERY));
+	}
+
 }

acp-core/src/test/java/com/agentclientprotocol/sdk/util/AcpSchedulersTest.java

@@ -0,0 +1,56 @@
+/*
+ * Copyright 2025-2026 the original author or authors.
+ */
+
+package com.agentclientprotocol.sdk.util;
+
+import java.time.Duration;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.junit.jupiter.api.Test;
+import reactor.core.publisher.Mono;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/**
+ * The shared timeout timer serves every session in the JVM, so one session's error handler
+ * must not be able to delay another session's timeout.
+ */
+class AcpSchedulersTest {
+
+	@Test
+	void aBlockingTimeoutHandlerDoesNotDelayAnotherSessionsTimeout() throws Exception {
+		CountDownLatch release = new CountDownLatch(1);
+		CountDownLatch firstBlocked = new CountDownLatch(1);
+		AcpSchedulers.withTimeout(Mono.never(), Duration.ofMillis(20)).subscribe(v -> {
+		}, error -> {
+			firstBlocked.countDown();
+			try {
+				release.await(5, TimeUnit.SECONDS); // a misbehaving subscriber
+			}
+			catch (InterruptedException ignored) {
+				Thread.currentThread().interrupt();
+			}
+		});
+		assertThat(firstBlocked.await(2, TimeUnit.SECONDS)).isTrue();
+
+		long start = System.nanoTime();
+		CountDownLatch secondTimedOut = new CountDownLatch(1);
+		AcpSchedulers.withTimeout(Mono.never(), Duration.ofMillis(50)).subscribe(v -> {
+		}, error -> {
+			if (error instanceof TimeoutException) {
+				secondTimedOut.countDown();
+			}
+		});
+		try {
+			assertThat(secondTimedOut.await(2, TimeUnit.SECONDS)).isTrue();
+			assertThat(Duration.ofNanos(System.nanoTime() - start)).isLessThan(Duration.ofMillis(1000));
+		}
+		finally {
+			release.countDown();
+		}
+	}
+
+}

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/SseOutboundStream.java

@@ -190,14 +190,24 @@ void drain() {
 					flushIfReady();
 					while (!detached && output.isReady()) {
 						byte[] bytes = comments.pollFirst();
+						String payload = null;
 						if (bytes == null) {
-							String payload = mailbox.pollFirst();
+							payload = mailbox.pollFirst();
 							if (payload == null) {
 								break;
 							}
 							bytes = ("data: " + payload + "\n\n").getBytes(StandardCharsets.UTF_8);
 						}
-						output.write(bytes);
+						try {
+							output.write(bytes);
+						}
+						catch (IOException | IllegalStateException e) {
+							if (payload != null) {
+								// The write failed, so the event did not go out: keep it first in line.
+								mailbox.addFirst(payload);
+							}
+							throw e;
+						}
 						flushPending = true;
 					}
 					flushIfReady();

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpAcpAgentTransport.java

@@ -247,7 +247,15 @@ private void startKeepAlive() {
 		// A comment every interval keeps proxies from cutting idle streams and surfaces
 		// dead subscribers (the write fails) without waiting for the next real event.
 		this.keepAliveTask = Flux.interval(interval, interval, scheduler)
-			.subscribe(tick -> connections.values().forEach(StreamableHttpConnection::keepAlive));
+			.subscribe(tick -> connections.values().forEach(connection -> {
+				try {
+					connection.keepAlive();
+				}
+				catch (RuntimeException e) {
+					// One broken connection must not stop keep-alives for every other one.
+					logger.debug("Keep-alive failed for connection {}: {}", connection.id(), e.getMessage());
+				}
+			}));
 	}
 
 	private void stopKeepAlive() {

acp-streamable-http-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/StreamableHttpAcpServlet.java

@@ -47,7 +47,7 @@
  *
  * @author Kaiser Dandangi
  */
-public class StreamableHttpAcpServlet extends HttpServlet {
+class StreamableHttpAcpServlet extends HttpServlet {
 
 	private static final Logger logger = LoggerFactory.getLogger(StreamableHttpAcpServlet.class);
 

acp-streamable-http-jetty/src/test/java/com/agentclientprotocol/sdk/agent/transport/SseOutboundStreamTest.java

@@ -139,6 +139,17 @@ void anEventPushedWhileTheSubscriberIsClosingIsKept() throws IOException {
 		assertThat(next.output.written()).isEqualTo(OPEN + "data: \"x\"\n\n");
 	}
 
+	@Test
+	void anEventWhoseWriteFailsIsKeptForTheNextSubscriber() throws IOException {
+		SseOutboundStream stream = new SseOutboundStream(8, 8);
+		Attached broken = attach(stream, true);
+		broken.output.failWrites = true;
+		stream.push("\"lost?\"");
+
+		Attached next = attach(stream, true);
+		assertThat(next.output.written()).isEqualTo(OPEN + "data: \"lost?\"\n\n");
+	}
+
 	private static Attached attach(SseOutboundStream stream, boolean ready) throws IOException {
 		FakeOutput output = new FakeOutput(ready);
 		AsyncContext asyncContext = mock(AsyncContext.class);
@@ -177,8 +188,13 @@ public void setWriteListener(WriteListener writeListener) {
 			this.listener = writeListener;
 		}
 
+		boolean failWrites;
+
 		@Override
-		public void write(int b) {
+		public void write(int b) throws IOException {
+			if (failWrites) {
+				throw new IOException("broken pipe");
+			}
 			bytes.write(b);
 		}