Fix flaky WebSocket tests: use emitNext with busyLooping for concurrent emission safety

Replace tryEmitNext with emitNext/busyLooping in all transport
sendMessage methods to handle FAIL_NON_SERIALIZED race conditions
on Reactor Sinks.many().unicast() sinks.

Author: Mark Pollack

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

@@ -4,6 +4,7 @@
 
 package com.agentclientprotocol.sdk.agent.transport;
 
+import java.time.Duration;
 import java.io.BufferedReader;
 import java.io.IOException;
 import java.io.InputStream;
@@ -277,12 +278,9 @@ else if (isClosing.get()) {
 	@Override
 	public Mono<Void> sendMessage(JSONRPCMessage message) {
 		return Mono.zip(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> {
-			if (outboundSink.tryEmitNext(message).isSuccess()) {
-				return Mono.empty();
-			}
-			else {
-				return Mono.error(new RuntimeException("Failed to enqueue message"));
-			}
+			outboundSink.emitNext(message,
+					Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
+			return Mono.empty();
 		}));
 	}
 

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

@@ -260,12 +260,9 @@ private void handleIncomingErrors() {
 
 	@Override
 	public Mono<Void> sendMessage(JSONRPCMessage message) {
-		if (this.outboundSink.tryEmitNext(message).isSuccess()) {
-			return Mono.empty();
-		}
-		else {
-			return Mono.error(new RuntimeException("Failed to enqueue message"));
-		}
+		this.outboundSink.emitNext(message,
+				Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
+		return Mono.empty();
 	}
 
 	/**

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

@@ -199,12 +199,9 @@ private void startOutboundProcessing() {
 	@Override
 	public Mono<Void> sendMessage(JSONRPCMessage message) {
 		return connectionReady.asMono().then(Mono.defer(() -> {
-			if (outboundSink.tryEmitNext(message).isSuccess()) {
-				return Mono.empty();
-			}
-			else {
-				return Mono.error(new RuntimeException("Failed to enqueue message"));
-			}
+			outboundSink.emitNext(message,
+					Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
+			return Mono.empty();
 		}));
 	}
 

acp-websocket-jetty/src/main/java/com/agentclientprotocol/sdk/agent/transport/WebSocketAcpAgentTransport.java

@@ -224,12 +224,9 @@ private void startOutboundProcessing() {
 	@Override
 	public Mono<Void> sendMessage(JSONRPCMessage message) {
 		return connectionReady.asMono().then(Mono.defer(() -> {
-			if (outboundSink.tryEmitNext(message).isSuccess()) {
-				return Mono.empty();
-			}
-			else {
-				return Mono.error(new RuntimeException("Failed to enqueue message"));
-			}
+			outboundSink.emitNext(message,
+					Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
+			return Mono.empty();
 		}));
 	}