fix(server): deliver awaited final events to gracefully closed child queues

A blocking message/send closes the caller's child queue as soon as the agent
returns, while the final event can still be in the MainEventBus. Since
dabca6c, distributeToChildren skipped every closed child, so the final event
was lost and the EventConsumer waited out its 3-second grace period for it.
Deliver to a closed child that is still awaiting a final event; a gracefully
closed child still drains its queue.

Fixes #1226

Author: jeroenbellen

server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java

@@ -776,7 +776,7 @@ void distributeToChildren(EventQueueItem item) {
                 // a narrow residual race remains for graceful close but is harmless for
                 // non-final events (they'd land in the draining queue) and impossible for
                 // final events (submittedFinalEvents prevents close while one is in flight).
-                if (child.isClosed()) {
+                if (child.isClosed() && !child.isAwaitingFinalEvent()) {
                     LOGGER.debug("MainQueue[{}]: Skipping closed child queue for event {}",
                             taskId, item.getEvent().getClass().getSimpleName());
                     return;

server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java

@@ -1696,6 +1696,64 @@ public void cancel(RequestContext context, AgentEmitter emitter) {
         assertEquals(TaskState.TASK_STATE_FAILED, storedTask.status().state());
     }
 
+    /**
+     * A blocking call returns to the handler on the agent's first event, and the handler closes the
+     * caller's queue as soon as the agent returns. A final event the event bus distributes after that
+     * must still reach the queue, which was told to expect it; otherwise the consumer waits out its
+     * grace period for it and the call fails or returns late.
+     */
+    @Test
+    void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exception {
+        CountDownLatch agentReturned = new CountDownLatch(1);
+        holdEventsAfterWorkingUntilAgentReturned(agentReturned);
+        agentExecutorExecute = (context, emitter) -> {
+            emitter.startWork();
+            emitter.complete();
+            agentReturned.countDown();
+        };
+
+        EventKind result = requestHandler.onMessageSend(blockingParams("msg-blocking-final-event"), NULL_CONTEXT);
+
+        Task task = assertInstanceOf(Task.class, result);
+        assertEquals(TaskState.TASK_STATE_COMPLETED, task.status().state());
+    }
+
+    private void holdEventsAfterWorkingUntilAgentReturned(CountDownLatch agentReturned) {
+        mainEventBusProcessor.setCallback(new MainEventBusProcessorCallback() {
+            @Override
+            public void onEventProcessed(String taskId, Event event) {
+                if (event instanceof TaskStatusUpdateEvent statusUpdate
+                        && statusUpdate.status().state() == TaskState.TASK_STATE_WORKING) {
+                    try {
+                        // Hold the agent's next event in the bus until the handler has closed the queue
+                        assertTrue(agentReturned.await(5, TimeUnit.SECONDS));
+                        Thread.sleep(200);
+                    } catch (InterruptedException e) {
+                        Thread.currentThread().interrupt();
+                    }
+                }
+            }
+
+            @Override
+            public void onTaskFinalized(String taskId) {
+            }
+        });
+    }
+
+    private static MessageSendParams blockingParams(String messageId) {
+        return MessageSendParams.builder()
+            .message(Message.builder()
+                .messageId(messageId)
+                .role(Message.Role.ROLE_USER)
+                .parts(new TextPart("hello"))
+                .build())
+            .configuration(MessageSendConfiguration.builder()
+                .returnImmediately(false)
+                .acceptedOutputModes(List.of())
+                .build())
+            .build();
+    }
+
     @Test
     void interruptedTerminalEnqueueStillReportsAgentFailure() throws Exception {
         assertInterruptedTerminalEnqueueReportsError(