fix: Use @internal executor for streaming subscriptions to prevent ForkJoinPool saturation

Root cause: Transport handlers used CompletableFuture.runAsync() without executor
parameter, defaulting to ForkJoinPool.commonPool (3 threads on CI). During concurrent
test execution, pool saturation prevented streaming subscriptions from starting,
causing SSE tests to timeout waiting for HTTP headers.

Changes:
- RestHandler: Add @internal Executor injection, use in runAsync()
- GrpcHandler: Add abstract getExecutor(), use in convertToStreamResponse()
- QuarkusGrpcHandler: Implement getExecutor() with @internal Executor
- JSONRPCHandler: Add @internal Executor injection, use in convertToSendStreamingMessageResponse()
- All test classes: Update handler instantiations to pass internalExecutor

Fixes SSE streaming test timeouts (test_sse_header_compliance, test_sse_event_format_compliance)

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude <[email protected]>

Author: kabir

reference/grpc/src/main/java/io/a2a/server/grpc/quarkus/QuarkusGrpcHandler.java

@@ -3,8 +3,11 @@
 import jakarta.enterprise.inject.Instance;
 import jakarta.inject.Inject;
 
+import java.util.concurrent.Executor;
+
 import io.a2a.server.PublicAgentCard;
 import io.a2a.server.requesthandlers.RequestHandler;
+import io.a2a.server.util.async.Internal;
 import io.a2a.spec.AgentCard;
 import io.a2a.transport.grpc.handler.CallContextFactory;
 import io.a2a.transport.grpc.handler.GrpcHandler;
@@ -20,14 +23,17 @@ public class QuarkusGrpcHandler extends GrpcHandler {
     private final AgentCard agentCard;
     private final RequestHandler requestHandler;
     private final Instance<CallContextFactory> callContextFactoryInstance;
+    private final Executor executor;
 
     @Inject
     public QuarkusGrpcHandler(@PublicAgentCard AgentCard agentCard,
                               RequestHandler requestHandler,
-                              Instance<CallContextFactory> callContextFactoryInstance) {
+                              Instance<CallContextFactory> callContextFactoryInstance,
+                              @Internal Executor executor) {
         this.agentCard = agentCard;
         this.requestHandler = requestHandler;
         this.callContextFactoryInstance = callContextFactoryInstance;
+        this.executor = executor;
     }
 
     @Override
@@ -44,4 +50,9 @@ protected AgentCard getAgentCard() {
     protected CallContextFactory getCallContextFactory() {
         return callContextFactoryInstance.isUnsatisfied() ? null : callContextFactoryInstance.get();
     }
+
+    @Override
+    protected Executor getExecutor() {
+        return executor;
+    }
 }

transport/grpc/src/main/java/io/a2a/transport/grpc/handler/GrpcHandler.java

@@ -11,6 +11,7 @@
 import java.util.Map;
 import java.util.Set;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 
 import io.grpc.Context;
 import java.util.concurrent.Flow;
@@ -294,7 +295,7 @@ public void onComplete() {
                     responseObserver.onCompleted();
                 }
             });
-        });
+        }, getExecutor());
     }
 
     @Override
@@ -504,6 +505,8 @@ public static void setStreamingSubscribedRunnable(Runnable runnable) {
 
     protected abstract CallContextFactory getCallContextFactory();
 
+    protected abstract Executor getExecutor();
+
     /**
      * Attempts to extract the X-A2A-Extensions header from the current gRPC context.
      * This will only work if a server interceptor has been configured to capture

transport/grpc/src/test/java/io/a2a/transport/grpc/handler/GrpcHandlerTest.java

@@ -7,6 +7,7 @@
 
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Flow;
 
 import io.a2a.server.AgentCardValidator;
@@ -44,6 +45,7 @@
 import io.a2a.spec.TaskNotFoundError;
 import io.a2a.spec.TaskPushNotificationConfig;
 import io.a2a.spec.TaskResubscriptionRequest;
+import io.a2a.server.util.async.Internal;
 import mutiny.zero.ZeroPublisher;
 
 @ApplicationScoped
@@ -52,23 +54,26 @@ public class JSONRPCHandler {
     private AgentCard agentCard;
     private Instance<AgentCard> extendedAgentCard;
     private RequestHandler requestHandler;
+    private final Executor executor;
 
     protected JSONRPCHandler() {
+        this.executor = null;
     }
 
     @Inject
     public JSONRPCHandler(@PublicAgentCard AgentCard agentCard, @ExtendedAgentCard Instance<AgentCard> extendedAgentCard,
-                          RequestHandler requestHandler) {
+                          RequestHandler requestHandler, @Internal Executor executor) {
         this.agentCard = agentCard;
         this.extendedAgentCard = extendedAgentCard;
         this.requestHandler = requestHandler;
-        
+        this.executor = executor;
+
         // Validate transport configuration
         AgentCardValidator.validateTransportConfiguration(agentCard);
     }
 
-    public JSONRPCHandler(@PublicAgentCard AgentCard agentCard, RequestHandler requestHandler) {
-        this(agentCard, null, requestHandler);
+    public JSONRPCHandler(@PublicAgentCard AgentCard agentCard, RequestHandler requestHandler, Executor executor) {
+        this(agentCard, null, requestHandler, executor);
     }
 
     public SendMessageResponse onMessageSend(SendMessageRequest request, ServerCallContext context) {
@@ -278,7 +283,7 @@ public void onComplete() {
                             tube.complete();
                         }
                     });
-                });
+                }, executor);
             });
     }
 }