fix: Use @internal Executor instead of ForkJoinPool for event consumption

Problem:
- ResultAggregator.consumeAndBreakOnInterrupt() used CompletableFuture.runAsync()
  without executor parameter
- Defaulted to ForkJoinPool.commonPool (only 3 threads on CI)
- During concurrent request bursts (6+ simultaneous), ForkJoinPool saturates
- Some consumer subscriptions never scheduled, Vert.x threads block forever
- Agents accumulate in runningAgents map, never cleaned up

Root Cause:
- ForkJoinPool.commonPool designed for CPU-bound tasks
- Our queue polling loop blocks for I/O (queue.dequeueEvent(500ms))
- Blocking I/O saturates limited ForkJoinPool workers
- Subsequent tasks queued but never execute

Solution:
- Pass @internal Executor to ResultAggregator constructor
- Use executor in runAsync() call instead of default ForkJoinPool
- @internal executor configured with 15 threads for I/O-bound operations
- Can handle concurrent request bursts without saturation

Evidence:
- Run 2/3 logs: Agents accumulate during 6-request burst at 30s intervals
- Thread dumps: Vert.x workers blocked at completionFuture.join()
- ForkJoinPool stops after initial requests, subsequent tasks never poll queues
- Detailed investigation in claudedocs/TCK_TIMEOUT_INVESTIGATION.md

Related:
- Issue #248
- Commit 3e93dd2 (removed awaitQueuePollerStart workaround)

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

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

Author: kabir

server-common/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java

@@ -138,7 +138,7 @@ public Task onCancelTask(TaskIdParams params, ServerCallContext context) throws
                 taskStore,
                 null);
 
-        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null);
+        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor);
 
         EventQueue queue = queueManager.tap(task.getId());
         if (queue == null) {
@@ -180,7 +180,7 @@ public EventKind onMessageSend(MessageSendParams params, ServerCallContext conte
         LOGGER.debug("Request context taskId: {}", taskId);
 
         EventQueue queue = queueManager.createOrTap(taskId);
-        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null);
+        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor);
 
         boolean blocking = true; // Default to blocking behavior
         if (params.configuration() != null && Boolean.FALSE.equals(params.configuration().blocking())) {
@@ -239,7 +239,7 @@ public Flow.Publisher<StreamingEventKind> onMessageSendStream(
         AtomicReference<String> taskId = new AtomicReference<>(mss.requestContext.getTaskId());
         EventQueue queue = queueManager.createOrTap(taskId.get());
         LOGGER.debug("Created/tapped queue for task {}: {}", taskId.get(), queue);
-        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null);
+        ResultAggregator resultAggregator = new ResultAggregator(mss.taskManager, null, executor);
 
         EnhancedRunnable producerRunnable = registerAndExecuteAgentAsync(taskId.get(), mss.requestContext, queue);
 
@@ -422,7 +422,7 @@ public Flow.Publisher<StreamingEventKind> onResubscribeToTask(
         }
 
         TaskManager taskManager = new TaskManager(task.getId(), task.getContextId(), taskStore, null);
-        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null);
+        ResultAggregator resultAggregator = new ResultAggregator(taskManager, null, executor);
         EventQueue queue = queueManager.tap(task.getId());
 
         if (queue == null) {

server-common/src/main/java/io/a2a/server/tasks/ResultAggregator.java

@@ -6,6 +6,7 @@
 
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
@@ -23,11 +24,13 @@
 
 public class ResultAggregator {
     private final TaskManager taskManager;
+    private final Executor executor;
     private volatile Message message;
 
-    public ResultAggregator(TaskManager taskManager, Message message) {
+    public ResultAggregator(TaskManager taskManager, Message message, Executor executor) {
         this.taskManager = taskManager;
         this.message = message;
+        this.executor = executor;
     }
 
     public EventKind getCurrentResult() {
@@ -102,6 +105,8 @@ public EventTypeAndInterrupt consumeAndBreakOnInterrupt(EventConsumer consumer,
         // CRITICAL: The subscription itself must run on a background thread to avoid blocking
         // the Vert.x worker thread. EventConsumer.consumeAll() starts a polling loop that
         // blocks in dequeueEvent(), so we must subscribe from a background thread.
+        // Use the @Internal executor (not ForkJoinPool.commonPool) to avoid saturation
+        // during concurrent request bursts.
         CompletableFuture.runAsync(() -> {
             consumer(
                 createTubeConfig(),
@@ -195,7 +200,7 @@ else if (!blocking) {
                     }
                 }
             );
-        });
+        }, executor);
 
         // Wait for completion or interruption
         try {

server-common/src/test/java/io/a2a/server/tasks/ResultAggregatorTest.java

@@ -10,6 +10,7 @@
 import static org.mockito.Mockito.when;
 
 import java.util.Collections;
+import java.util.concurrent.Executor;
 
 import io.a2a.server.events.EventConsumer;
 import io.a2a.server.events.EventQueue;
@@ -39,11 +40,12 @@ public class ResultAggregatorTest {
     private TaskManager mockTaskManager;
 
     private ResultAggregator aggregator;
+    private final Executor directExecutor = Runnable::run;
 
     @BeforeEach
     void setUp() {
         MockitoAnnotations.openMocks(this);
-        aggregator = new ResultAggregator(mockTaskManager, null);
+        aggregator = new ResultAggregator(mockTaskManager, null, directExecutor);
     }
 
     // Helper methods for creating sample data
@@ -69,7 +71,7 @@ private Task createSampleTask(String taskId, TaskState statusState, String conte
     @Test
     void testConstructorWithMessage() {
         Message initialMessage = createSampleMessage("initial", "msg1", Message.Role.USER);
-        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage);
+        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage, directExecutor);
 
         // Test that the message is properly stored by checking getCurrentResult
         assertEquals(initialMessage, aggregatorWithMessage.getCurrentResult());
@@ -80,7 +82,7 @@ void testConstructorWithMessage() {
     @Test
     void testGetCurrentResultWithMessageSet() {
         Message sampleMessage = createSampleMessage("hola", "msg1", Message.Role.USER);
-        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage);
+        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, directExecutor);
 
         EventKind result = aggregatorWithMessage.getCurrentResult();
 
@@ -115,7 +117,7 @@ void testConstructorStoresTaskManagerCorrectly() {
 
     @Test
     void testConstructorWithNullMessage() {
-        ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null);
+        ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, directExecutor);
         Task expectedTask = createSampleTask("null_msg_task", TaskState.WORKING, "ctx1");
         when(mockTaskManager.getTask()).thenReturn(expectedTask);
 
@@ -175,7 +177,7 @@ void testMultipleGetCurrentResultCalls() {
     void testGetCurrentResultWithMessageTakesPrecedence() {
         // Test that when both message and task are available, message takes precedence
         Message message = createSampleMessage("priority message", "pri1", Message.Role.USER);
-        ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message);
+        ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message, directExecutor);
 
         // Even if we set up the task manager to return something, message should take precedence
         Task task = createSampleTask("should_not_be_returned", TaskState.WORKING, "ctx1");