From 58620e4a1d4facd47aa10ddb557a5fe9bf858214 Mon Sep 17 00:00:00 2001 From: Jeroen Bellen <1204135+jeroenbellen@users.noreply.github.com> Date: Thu, 15 Oct 2026 19:08:08 +0200 Subject: [PATCH 1/2] 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 --- .../sdk/server/events/EventQueue.java | 2 +- .../DefaultRequestHandlerTest.java | 58 +++++++++++++++++++ 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java index c018bee5a..039822c1c 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java +++ b/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; diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 95ffd0f7c..0b05a83cc 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/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( From f715ea055d01fdcd16b6c02884b8395427b0fedb Mon Sep 17 00:00:00 2001 From: Jeroen Bellen <1204135+jeroenbellen@users.noreply.github.com> Date: Wed, 21 Oct 2026 13:48:42 +0200 Subject: [PATCH 2/2] fix(server): correct the child-queue comment and assert the held event on the main thread --- .../a2aproject/sdk/server/events/EventQueue.java | 14 +++++++------- .../requesthandlers/DefaultRequestHandlerTest.java | 9 ++++++--- 2 files changed, 13 insertions(+), 10 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java index 039822c1c..c35d35cab 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/EventQueue.java @@ -769,13 +769,13 @@ void distributeToChildren(EventQueueItem item) { taskId, item.getEvent().getClass().getSimpleName(), childCount); } children.forEach(child -> { - // Skip children that have already been closed. CopyOnWriteArrayList gives - // snapshot semantics, so a child removed by closeIfNotAwaitingFinalEvent() - // may still appear in this iteration. Checking isClosed() here narrows the - // TOCTOU window to the time between the check and internalEnqueueItem(); - // 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). + // Skip children that have already been closed, unless a child is still awaiting + // a final event. CopyOnWriteArrayList gives snapshot semantics, so a child removed + // by closeIfNotAwaitingFinalEvent() may still appear in this iteration; for a + // non-final event that is harmless (it lands in the draining queue). A child that + // was closed gracefully while awaiting a final event must still receive it: a + // blocking call closes the caller's queue before the event bus may have + // distributed the final event. if (child.isClosed() && !child.isAwaitingFinalEvent()) { LOGGER.debug("MainQueue[{}]: Skipping closed child queue for event {}", taskId, item.getEvent().getClass().getSimpleName()); diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 0b05a83cc..bbec1197b 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -23,6 +23,7 @@ import java.util.concurrent.Future; import java.util.concurrent.TimeUnit; import java.util.concurrent.TimeoutException; +import java.util.concurrent.atomic.AtomicBoolean; import java.util.concurrent.atomic.AtomicReference; import org.a2aproject.sdk.server.ServerCallContext; @@ -1705,7 +1706,8 @@ public void cancel(RequestContext context, AgentEmitter emitter) { @Test void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exception { CountDownLatch agentReturned = new CountDownLatch(1); - holdEventsAfterWorkingUntilAgentReturned(agentReturned); + AtomicBoolean agentReturnedInTime = new AtomicBoolean(); + holdEventsAfterWorkingUntilAgentReturned(agentReturned, agentReturnedInTime); agentExecutorExecute = (context, emitter) -> { emitter.startWork(); emitter.complete(); @@ -1714,11 +1716,12 @@ void blockingSendStillReturnsCompletedTaskWhenFinalEventArrivesLate() throws Exc EventKind result = requestHandler.onMessageSend(blockingParams("msg-blocking-final-event"), NULL_CONTEXT); + assertTrue(agentReturnedInTime.get(), "the agent did not return while its next event was held"); Task task = assertInstanceOf(Task.class, result); assertEquals(TaskState.TASK_STATE_COMPLETED, task.status().state()); } - private void holdEventsAfterWorkingUntilAgentReturned(CountDownLatch agentReturned) { + private void holdEventsAfterWorkingUntilAgentReturned(CountDownLatch agentReturned, AtomicBoolean agentReturnedInTime) { mainEventBusProcessor.setCallback(new MainEventBusProcessorCallback() { @Override public void onEventProcessed(String taskId, Event event) { @@ -1726,7 +1729,7 @@ public void onEventProcessed(String taskId, Event event) { && 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)); + agentReturnedInTime.set(agentReturned.await(5, TimeUnit.SECONDS)); Thread.sleep(200); } catch (InterruptedException e) { Thread.currentThread().interrupt();