TCK fixes

Author: kabir

server-common/src/main/java/io/a2a/server/events/EventConsumer.java

@@ -64,7 +64,10 @@ public Flow.Publisher<EventQueueItem> consumeAll() {
                         }
                         event = item.getEvent();
 
+                        // Defensive logging for error handling
                         if (event instanceof Throwable thr) {
+                            LOGGER.info("EventConsumer detected Throwable event: {} - triggering tube.fail()",
+                                    thr.getClass().getSimpleName());
                             tube.fail(thr);
                             return;
                         }

server-common/src/main/java/io/a2a/server/events/MainEventBusProcessor.java

@@ -166,6 +166,11 @@ private void processEvent(MainEventBusContext context) {
             }
 
             // Step 3: Then distribute to ChildQueues (clients see either event or error AFTER persistence attempt)
+            if (eventToDistribute == null) {
+                LOGGER.error("MainEventBusProcessor: eventToDistribute is NULL for task {} - this should never happen!", taskId);
+                eventToDistribute = new InternalError("Internal error: event processing failed");
+            }
+
             if (eventQueue instanceof EventQueue.MainQueue mainQueue) {
                 int childCount = mainQueue.getChildCount();
                 LOGGER.debug("MainEventBusProcessor: Distributing {} to {} children for task {}",

server-common/src/main/java/io/a2a/server/events/MainEventBusProcessorCallback.java

@@ -8,7 +8,6 @@
  * This interface is primarily intended for testing, allowing tests to synchronize
  * with the asynchronous MainEventBusProcessor. Production code should not rely on this.
  * </p>
- * <p>
  * Usage in tests:
  * <pre>
  * {@code
@@ -31,7 +30,6 @@
  * }
  * }
  * </pre>
- * </p>
  */
 public interface MainEventBusProcessorCallback {
 

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

@@ -90,6 +90,7 @@ public EventKind consumeAll(EventConsumer consumer) throws JSONRPCError {
     public EventTypeAndInterrupt consumeAndBreakOnInterrupt(EventConsumer consumer, boolean blocking) throws JSONRPCError {
         Flow.Publisher<EventQueueItem> allItems = consumer.consumeAll();
         AtomicReference<Message> message = new AtomicReference<>();
+        AtomicReference<Task> capturedTask = new AtomicReference<>();  // Capture Task events
         AtomicBoolean interrupted = new AtomicBoolean(false);
         AtomicReference<Throwable> errorRef = new AtomicReference<>();
         CompletableFuture<Void> completionFuture = new CompletableFuture<>();
@@ -123,6 +124,11 @@ public EventTypeAndInterrupt consumeAndBreakOnInterrupt(EventConsumer consumer,
                         return false;
                     }
 
+                    // Capture Task events (especially for new tasks where taskManager.getTask() would return null)
+                    if (event instanceof Task t) {
+                        capturedTask.set(t);
+                    }
+
                     // TaskStore update moved to MainEventBusProcessor
 
                     // Determine interrupt behavior
@@ -223,8 +229,17 @@ else if (blocking) {
             Utils.rethrow(error);
         }
 
+        // Return Message if captured, otherwise Task if captured, otherwise fetch from TaskStore
+        EventKind eventKind = message.get();
+        if (eventKind == null) {
+            eventKind = capturedTask.get();
+        }
+        if (eventKind == null) {
+            eventKind = taskManager.getTask();
+        }
+
         return new EventTypeAndInterrupt(
-                message.get() != null ? message.get() : taskManager.getTask(),
+                eventKind,
                 interrupted.get(),
                 consumptionCompletionFuture);
     }