Bound and keep alive the Streamable HTTP agent transport

StreamableHttpAcpAgentTransportOptions makes every network-facing
allocation finite: POST bodies are capped (16 MiB by default, 413
beyond), the WebSocket send queue and the number of provisional
session/load streams per connection are bounded, and the SSE mailbox
and per-subscriber queue limits are configurable. A failed session/load
discards its provisional state. Attached SSE streams receive a
keep-alive comment every 15 s so proxies do not cut idle connections
and dead subscribers are found without waiting for the next event. A
new GET on a stream takes it over from a subscriber the server may not
yet know is dead, instead of fanning out duplicates; the Rust and
TypeScript servers answer 409, which is harder on a reconnecting client.

From the cold reviews of PR #7.

Author: Mark Pollack

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

@@ -56,6 +56,10 @@
 import org.eclipse.jetty.websocket.server.WebSocketUpgradeHandler;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import reactor.core.Disposable;
+import reactor.core.publisher.Flux;
+import reactor.core.scheduler.Scheduler;
+import reactor.core.scheduler.Schedulers;
 import reactor.core.publisher.Mono;
 import reactor.core.publisher.Sinks;
 
@@ -94,9 +98,7 @@ public class StreamableHttpAcpAgentTransport {
 
 	private static final String CONTENT_TYPE_EVENT_STREAM = "text/event-stream";
 
-	private static final int MAX_REPLAY_EVENTS = 1024;
-
-	private static final int MAX_PENDING_SSE_EVENTS = 1024;
+	private static final byte[] SSE_KEEP_ALIVE_COMMENT = ": keep-alive\n\n".getBytes(StandardCharsets.UTF_8);
 
 	// Commits the SSE response when there are no replay events, without emitting an ACP message.
 	private static final byte[] SSE_OPEN_COMMENT = ": connected\n\n".getBytes(StandardCharsets.UTF_8);
@@ -164,6 +166,12 @@ private record ResolvedInboundRoute(JSONRPCMessage message, RouteScope requestSc
 
 	private final AcpAgentFactory agentFactory;
 
+	private final StreamableHttpAcpAgentTransportOptions options;
+
+	private volatile Scheduler keepAliveScheduler;
+
+	private volatile Disposable keepAliveTask;
+
 	private final ConcurrentMap<String, ConnectionState> connections = new ConcurrentHashMap<>();
 
 	private final ConcurrentMap<String, WebSocketConnectionState> webSocketConnections = new ConcurrentHashMap<>();
@@ -197,14 +205,29 @@ public StreamableHttpAcpAgentTransport(int port, AcpJsonMapper jsonMapper, AcpAg
 	 */
 	public StreamableHttpAcpAgentTransport(int port, String path, AcpJsonMapper jsonMapper,
 			AcpAgentFactory agentFactory) {
+		this(port, path, jsonMapper, agentFactory, StreamableHttpAcpAgentTransportOptions.defaults());
+	}
+
+	/**
+	 * Creates a new Streamable HTTP listener with explicit limits.
+	 * @param port port to listen on
+	 * @param path endpoint path
+	 * @param jsonMapper JSON mapper used for serialization
+	 * @param agentFactory factory used to create one agent runtime per connection
+	 * @param options bounds and timings
+	 */
+	public StreamableHttpAcpAgentTransport(int port, String path, AcpJsonMapper jsonMapper,
+			AcpAgentFactory agentFactory, StreamableHttpAcpAgentTransportOptions options) {
 		Assert.isTrue(port > 0, "Port must be positive");
 		Assert.hasText(path, "Path must not be empty");
 		Assert.notNull(jsonMapper, "The JsonMapper can not be null");
 		Assert.notNull(agentFactory, "The agentFactory can not be null");
+		Assert.notNull(options, "The options can not be null");
 		this.configuredPort = port;
 		this.path = path;
 		this.jsonMapper = jsonMapper;
 		this.agentFactory = agentFactory;
+		this.options = options;
 	}
 
 	/**
@@ -252,6 +275,7 @@ public Mono<Void> start() {
 			jettyServer.start();
 			this.server = jettyServer;
 			this.connector = jettyConnector;
+			startKeepAlive();
 			logger.info("Streamable HTTP agent listener started on port {} at path {}", getPort(), path);
 			return null;
 		}).then();
@@ -289,7 +313,32 @@ public Mono<Void> closeGracefully() {
 		});
 	}
 
+	private void startKeepAlive() {
+		Duration interval = options.keepAliveInterval();
+		if (interval.isZero()) {
+			return;
+		}
+		Scheduler scheduler = Schedulers.newSingle("acp-streamable-http-keepalive", true);
+		this.keepAliveScheduler = scheduler;
+		// 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(ConnectionState::keepAlive));
+	}
+
+	private void stopKeepAlive() {
+		Disposable task = this.keepAliveTask;
+		if (task != null) {
+			task.dispose();
+		}
+		Scheduler scheduler = this.keepAliveScheduler;
+		if (scheduler != null) {
+			scheduler.dispose();
+		}
+	}
+
 	private void stopServer() {
+		stopKeepAlive();
 		Server currentServer = this.server;
 		if (currentServer != null) {
 			try {
@@ -339,7 +388,20 @@ protected void doPost(HttpServletRequest request, HttpServletResponse response)
 				return;
 			}
 
-			String body = new String(request.getInputStream().readAllBytes(), StandardCharsets.UTF_8);
+			long declaredLength = request.getContentLengthLong();
+			if (declaredLength > options.maxPostBodyBytes()) {
+				writeText(response, HttpServletResponse.SC_REQUEST_ENTITY_TOO_LARGE,
+						"POST body exceeds " + options.maxPostBodyBytes() + " bytes");
+				return;
+			}
+			byte[] bodyBytes = request.getInputStream().readNBytes((int) Math.min(Integer.MAX_VALUE,
+					options.maxPostBodyBytes() + 1));
+			if (bodyBytes.length > options.maxPostBodyBytes()) {
+				writeText(response, HttpServletResponse.SC_REQUEST_ENTITY_TOO_LARGE,
+						"POST body exceeds " + options.maxPostBodyBytes() + " bytes");
+				return;
+			}
+			String body = new String(bodyBytes, StandardCharsets.UTF_8);
 			if (body.stripLeading().startsWith("[")) {
 				writeText(response, HttpServletResponse.SC_NOT_IMPLEMENTED, "JSON-RPC batches are not supported");
 				return;
@@ -693,8 +755,13 @@ private RouteScope resolveAgentOutboundScope(JSONRPCMessage message) {
 					String sessionId = extractSessionIdFromNewSessionResponse(response);
 					markSessionKnown(sessionId);
 				}
-				if (route.kind() == RequestKind.SESSION_LOAD && response.error() == null) {
-					markSessionKnown(route.requestScope().sessionId());
+				if (route.kind() == RequestKind.SESSION_LOAD) {
+					if (response.error() == null) {
+						markSessionKnown(route.requestScope().sessionId());
+					}
+					else {
+						discardProvisionalSession(route.requestScope().sessionId());
+					}
 				}
 				return route.responseScope();
 			}
@@ -816,7 +883,7 @@ private void prepareSessionForInbound(String sessionId, ClientRequestRoute route
 			SessionState current = sessions.get(sessionId);
 			if (route != null && route.kind() == RequestKind.SESSION_LOAD) {
 				if (current == null) {
-					sessions.putIfAbsent(sessionId, SessionState.PENDING_LOAD);
+					addProvisionalSession(sessionId);
 					sessionStream(sessionId);
 				}
 				return;
@@ -849,11 +916,37 @@ private OutboundStream openSessionStream(String sessionId) {
 				 * but its resume flow also asks clients to open a session stream before
 				 * sending session/load. Keep a provisional stream so practical resume can work.
 				 */
-				sessions.putIfAbsent(sessionId, SessionState.PENDING_LOAD);
+				addProvisionalSession(sessionId);
 			}
 			return sessionStream(sessionId);
 		}
 
+		/** Provisional sessions are bounded: a client cannot grow state with arbitrary ids. */
+		private void addProvisionalSession(String sessionId) {
+			long provisional = sessions.values().stream().filter(state -> state == SessionState.PENDING_LOAD).count();
+			if (provisional >= options.maxProvisionalSessions()
+					&& sessions.get(sessionId) != SessionState.PENDING_LOAD) {
+				throw new UnknownSessionException("Too many provisional sessions on connection " + id
+						+ " (limit " + options.maxProvisionalSessions() + ")");
+			}
+			sessions.putIfAbsent(sessionId, SessionState.PENDING_LOAD);
+		}
+
+		/** A failed session/load leaves no provisional state behind. */
+		private void discardProvisionalSession(String sessionId) {
+			if (sessions.remove(sessionId, SessionState.PENDING_LOAD)) {
+				OutboundStream stream = sessionStreams.remove(sessionId);
+				if (stream != null) {
+					stream.close();
+				}
+			}
+		}
+
+		void keepAlive() {
+			connectionStream.keepAlive();
+			sessionStreams.values().forEach(OutboundStream::keepAlive);
+		}
+
 		private OutboundStream sessionStream(String sessionId) {
 			return sessionStreams.computeIfAbsent(sessionId, ignored -> new OutboundStream());
 		}
@@ -909,9 +1002,9 @@ synchronized void push(String payload) {
 				return;
 			}
 			if (subscribers.isEmpty()) {
-				if (replay.size() == MAX_REPLAY_EVENTS) {
+				if (replay.size() == options.mailboxCapacity()) {
 					throw new AcpConnectionException(
-							"Outbound SSE replay buffer exceeded " + MAX_REPLAY_EVENTS + " events");
+							"Outbound SSE replay buffer exceeded " + options.mailboxCapacity() + " events");
 				}
 				replay.addLast(payload);
 				return;
@@ -926,6 +1019,14 @@ synchronized void subscribe(AsyncContext asyncContext, HttpServletResponse respo
 				asyncContext.complete();
 				return;
 			}
+			// One subscriber per stream: a new GET takes the stream over from a previous
+			// one that the server may not yet know is dead (proxy drop, client restart).
+			// Rust and TypeScript answer 409 instead; taking over is friendlier to a
+			// reconnecting client and never duplicates events.
+			for (SseSubscriber previous : new ArrayList<>(subscribers)) {
+				logger.debug("New SSE subscriber replaces the attached one");
+				previous.close();
+			}
 			SseSubscriber subscriber = new SseSubscriber(this, asyncContext, response);
 			subscribers.add(subscriber);
 			subscriber.start();
@@ -940,6 +1041,12 @@ void remove(SseSubscriber subscriber) {
 			subscribers.remove(subscriber);
 		}
 
+		void keepAlive() {
+			if (!closed.get()) {
+				subscribers.forEach(SseSubscriber::sendKeepAlive);
+			}
+		}
+
 		synchronized void close() {
 			if (closed.compareAndSet(false, true)) {
 				subscribers.forEach(SseSubscriber::close);
@@ -980,15 +1087,24 @@ synchronized void send(String payload) {
 			if (closed.get()) {
 				return;
 			}
-			if (pendingEvents.size() == MAX_PENDING_SSE_EVENTS) {
-				logger.warn("Closing backpressured SSE subscriber after {} pending events", MAX_PENDING_SSE_EVENTS);
+			if (pendingEvents.size() == options.maxPendingSseEvents()) {
+				logger.warn("Closing backpressured SSE subscriber after {} pending events",
+						options.maxPendingSseEvents());
 				close();
 				return;
 			}
 			pendingEvents.addLast(("data: " + payload + "\n\n").getBytes(StandardCharsets.UTF_8));
 			drain();
 		}
 
+		synchronized void sendKeepAlive() {
+			if (closed.get() || !pendingEvents.isEmpty()) {
+				return;
+			}
+			pendingEvents.addLast(SSE_KEEP_ALIVE_COMMENT);
+			drain();
+		}
+
 		synchronized void drain() {
 			try {
 				flushIfReady();
@@ -1165,6 +1281,10 @@ void send(String payload) {
 					if (closed.get()) {
 						throw new AcpConnectionException("Streamable ACP WebSocket connection is closed");
 					}
+					if (queue.size() >= options.maxWebSocketPendingFrames()) {
+						throw new AcpConnectionException("WebSocket send queue exceeded "
+								+ options.maxWebSocketPendingFrames() + " pending frames");
+					}
 					queue.addLast(payload);
 					shouldDrain = !sendInProgress;
 					if (shouldDrain) {

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

@@ -0,0 +1,117 @@
+/*
+ * Copyright 2025-2025 the original author or authors.
+ */
+
+package com.agentclientprotocol.sdk.agent.transport;
+
+import java.time.Duration;
+
+import com.agentclientprotocol.sdk.util.Assert;
+
+/**
+ * Limits and timings for {@link StreamableHttpAcpAgentTransport}. Every network-facing
+ * allocation is bounded by one of these so a slow or hostile client cannot grow memory
+ * without bound.
+ *
+ * @param maxPostBodyBytes largest accepted POST body; larger requests get 413
+ * @param mailboxCapacity events retained per outbound stream while no subscriber is
+ * attached; overflow closes the connection
+ * @param maxPendingSseEvents events queued for one attached SSE subscriber before it is
+ * closed as backpressured
+ * @param maxWebSocketPendingFrames frames queued for one WebSocket connection before it is
+ * closed as backpressured
+ * @param maxProvisionalSessions session-scoped streams a connection may open before the
+ * session is known (the {@code session/load} pre-open case)
+ * @param keepAliveInterval interval between SSE keep-alive comments on attached streams;
+ * {@link Duration#ZERO} disables them
+ * @author Mark Pollack
+ */
+public record StreamableHttpAcpAgentTransportOptions(long maxPostBodyBytes, int mailboxCapacity,
+		int maxPendingSseEvents, int maxWebSocketPendingFrames, int maxProvisionalSessions,
+		Duration keepAliveInterval) {
+
+	private static final long DEFAULT_MAX_POST_BODY_BYTES = 16L * 1024 * 1024;
+
+	private static final int DEFAULT_MAILBOX_CAPACITY = 1024;
+
+	private static final int DEFAULT_MAX_PENDING_SSE_EVENTS = 1024;
+
+	private static final int DEFAULT_MAX_WEBSOCKET_PENDING_FRAMES = 1024;
+
+	private static final int DEFAULT_MAX_PROVISIONAL_SESSIONS = 64;
+
+	private static final Duration DEFAULT_KEEP_ALIVE_INTERVAL = Duration.ofSeconds(15);
+
+	public StreamableHttpAcpAgentTransportOptions {
+		Assert.isTrue(maxPostBodyBytes > 0, "maxPostBodyBytes must be positive");
+		Assert.isTrue(mailboxCapacity > 0, "mailboxCapacity must be positive");
+		Assert.isTrue(maxPendingSseEvents > 0, "maxPendingSseEvents must be positive");
+		Assert.isTrue(maxWebSocketPendingFrames > 0, "maxWebSocketPendingFrames must be positive");
+		Assert.isTrue(maxProvisionalSessions > 0, "maxProvisionalSessions must be positive");
+		Assert.notNull(keepAliveInterval, "keepAliveInterval must not be null");
+		Assert.isTrue(!keepAliveInterval.isNegative(), "keepAliveInterval must not be negative");
+	}
+
+	public static StreamableHttpAcpAgentTransportOptions defaults() {
+		return builder().build();
+	}
+
+	public static Builder builder() {
+		return new Builder();
+	}
+
+	public static final class Builder {
+
+		private long maxPostBodyBytes = DEFAULT_MAX_POST_BODY_BYTES;
+
+		private int mailboxCapacity = DEFAULT_MAILBOX_CAPACITY;
+
+		private int maxPendingSseEvents = DEFAULT_MAX_PENDING_SSE_EVENTS;
+
+		private int maxWebSocketPendingFrames = DEFAULT_MAX_WEBSOCKET_PENDING_FRAMES;
+
+		private int maxProvisionalSessions = DEFAULT_MAX_PROVISIONAL_SESSIONS;
+
+		private Duration keepAliveInterval = DEFAULT_KEEP_ALIVE_INTERVAL;
+
+		private Builder() {
+		}
+
+		public Builder maxPostBodyBytes(long maxPostBodyBytes) {
+			this.maxPostBodyBytes = maxPostBodyBytes;
+			return this;
+		}
+
+		public Builder mailboxCapacity(int mailboxCapacity) {
+			this.mailboxCapacity = mailboxCapacity;
+			return this;
+		}
+
+		public Builder maxPendingSseEvents(int maxPendingSseEvents) {
+			this.maxPendingSseEvents = maxPendingSseEvents;
+			return this;
+		}
+
+		public Builder maxWebSocketPendingFrames(int maxWebSocketPendingFrames) {
+			this.maxWebSocketPendingFrames = maxWebSocketPendingFrames;
+			return this;
+		}
+
+		public Builder maxProvisionalSessions(int maxProvisionalSessions) {
+			this.maxProvisionalSessions = maxProvisionalSessions;
+			return this;
+		}
+
+		public Builder keepAliveInterval(Duration keepAliveInterval) {
+			this.keepAliveInterval = keepAliveInterval;
+			return this;
+		}
+
+		public StreamableHttpAcpAgentTransportOptions build() {
+			return new StreamableHttpAcpAgentTransportOptions(maxPostBodyBytes, mailboxCapacity, maxPendingSseEvents,
+					maxWebSocketPendingFrames, maxProvisionalSessions, keepAliveInterval);
+		}
+
+	}
+
+}