fix: Use thread pool executor in ResultAggregatorTest instead of direct executor

Problem:
- Tests used directExecutor (Runnable::run) which executes tasks synchronously
- When ResultAggregator calls CompletableFuture.runAsync(task, directExecutor),
  the task runs on the calling thread instead of a background thread
- In SSE streaming tests on REST transport, this blocks the Vert.x worker thread
- Client times out waiting for response headers (httpx.ReadTimeout)

Root Cause:
- directExecutor executes runAsync() tasks synchronously on calling thread
- This defeats the purpose of using runAsync() for non-blocking operation
- Real production code uses @internal Executor (thread pool) which works correctly

Solution:
- Use Executors.newCachedThreadPool() in tests instead of directExecutor
- Now runAsync() properly schedules work on background threads
- Vert.x worker threads don't block during consumer subscription

Evidence:
- test_sse_header_compliance and test_sse_event_format_compliance timed out
- Only on REST transport in CI (works locally, works on JSON-RPC/gRPC)
- Server logs show onMessageSendStream completing but client never receives response

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

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

Author: kabir

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

@@ -11,6 +11,7 @@
 
 import java.util.Collections;
 import java.util.concurrent.Executor;
+import java.util.concurrent.Executors;
 
 import io.a2a.server.events.EventConsumer;
 import io.a2a.server.events.EventQueue;
@@ -40,12 +41,14 @@ public class ResultAggregatorTest {
     private TaskManager mockTaskManager;
 
     private ResultAggregator aggregator;
-    private final Executor directExecutor = Runnable::run;
+    // Use a real thread pool executor instead of direct executor
+    // to avoid blocking the calling thread during async operations
+    private final Executor testExecutor = Executors.newCachedThreadPool();
 
     @BeforeEach
     void setUp() {
         MockitoAnnotations.openMocks(this);
-        aggregator = new ResultAggregator(mockTaskManager, null, directExecutor);
+        aggregator = new ResultAggregator(mockTaskManager, null, testExecutor);
     }
 
     // Helper methods for creating sample data
@@ -71,7 +74,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, directExecutor);
+        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, initialMessage, testExecutor);
 
         // Test that the message is properly stored by checking getCurrentResult
         assertEquals(initialMessage, aggregatorWithMessage.getCurrentResult());
@@ -82,7 +85,7 @@ void testConstructorWithMessage() {
     @Test
     void testGetCurrentResultWithMessageSet() {
         Message sampleMessage = createSampleMessage("hola", "msg1", Message.Role.USER);
-        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, directExecutor);
+        ResultAggregator aggregatorWithMessage = new ResultAggregator(mockTaskManager, sampleMessage, testExecutor);
 
         EventKind result = aggregatorWithMessage.getCurrentResult();
 
@@ -117,7 +120,7 @@ void testConstructorStoresTaskManagerCorrectly() {
 
     @Test
     void testConstructorWithNullMessage() {
-        ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, directExecutor);
+        ResultAggregator aggregatorWithNullMessage = new ResultAggregator(mockTaskManager, null, testExecutor);
         Task expectedTask = createSampleTask("null_msg_task", TaskState.WORKING, "ctx1");
         when(mockTaskManager.getTask()).thenReturn(expectedTask);
 
@@ -177,7 +180,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, directExecutor);
+        ResultAggregator messageAggregator = new ResultAggregator(mockTaskManager, message, testExecutor);
 
         // 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");