fix: serialize streamable HTTP sink emissions

Author: kamikaz1k

acp-core/src/main/java/com/agentclientprotocol/sdk/agent/transport/RemoteAcpConnection.java

@@ -154,6 +154,14 @@ private final class ConnectionTransport implements AcpAgentTransport {
 
 		private final Sinks.Many<JSONRPCMessage> inboundSink = Sinks.many().unicast().onBackpressureBuffer();
 
+		/*
+		 * Streamable HTTP can deliver multiple POST requests for one ACP connection on
+		 * different server threads. Reactor unicast sinks require serialized producers,
+		 * so all transport-adapter ingress is funneled through this monitor before
+		 * emission.
+		 */
+		private final Object inboundEmitMonitor = new Object();
+
 		private final Sinks.One<Void> terminationSink = Sinks.one();
 
 		private final AtomicBoolean transportStarted = new AtomicBoolean(false);
@@ -190,9 +198,11 @@ void acceptInbound(JSONRPCMessage message) {
 			if (transportClosing.get()) {
 				throw new AcpConnectionException("Remote ACP connection is closing");
 			}
-			Sinks.EmitResult result = inboundSink.tryEmitNext(message);
-			if (result.isFailure()) {
-				throw new AcpConnectionException("Failed to enqueue inbound message: " + result);
+			synchronized (inboundEmitMonitor) {
+				Sinks.EmitResult result = inboundSink.tryEmitNext(message);
+				if (result.isFailure()) {
+					throw new AcpConnectionException("Failed to enqueue inbound message: " + result);
+				}
 			}
 		}
 

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

@@ -146,6 +146,13 @@ private record HttpClientBundle(HttpClient httpClient, ExecutorService ownedExec
 
 	private final Sinks.Many<JSONRPCMessage> inboundSink;
 
+	/*
+	 * A streamable HTTP client may have one connection SSE reader and multiple session
+	 * SSE readers active at the same time. Reactor unicast sinks require serialized
+	 * producers, so every SSE reader emits through this monitor.
+	 */
+	private final Object inboundEmitMonitor = new Object();
+
 	private final AtomicBoolean connected = new AtomicBoolean(false);
 
 	private final AtomicBoolean initialized = new AtomicBoolean(false);
@@ -581,9 +588,11 @@ private Mono<Void> processInbound(RouteScope actualScope, JSONRPCMessage message
 
 	private Mono<Void> emitInbound(JSONRPCMessage message) {
 		return Mono.fromRunnable(() -> {
-			Sinks.EmitResult result = inboundSink.tryEmitNext(message);
-			if (result.isFailure()) {
-				throw new AcpConnectionException("Failed to enqueue inbound message: " + result);
+			synchronized (inboundEmitMonitor) {
+				Sinks.EmitResult result = inboundSink.tryEmitNext(message);
+				if (result.isFailure()) {
+					throw new AcpConnectionException("Failed to enqueue inbound message: " + result);
+				}
 			}
 		});
 	}