From b285de43753eb815391ec2613001a0b1e2d47769 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Fri, 25 Sep 2026 15:31:12 +0100 Subject: [PATCH 1/7] fix(server): recover failed terminal enqueues This fixes #1183 --- .../DefaultRequestHandler.java | 15 +++++++- .../sdk/server/tasks/AgentEmitter.java | 17 ++++++++- .../DefaultRequestHandlerTest.java | 27 +++++++++++++ .../tasks/AgentEmitterConcurrencyTest.java | 38 +++++++++++++++++++ 4 files changed, 93 insertions(+), 4 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 4388b0443..967afb87f 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -1266,11 +1266,11 @@ public void run() { // Log unexpected runtime exceptions at ERROR level // These indicate bugs in agent implementation LOGGER.error("Agent execution threw unexpected RuntimeException for task {}", taskId, e); - emitter.fail(new org.a2aproject.sdk.spec.InternalError("Agent execution failed: " + e.getMessage())); + reportAgentExecutionFailure(emitter, "Agent execution failed: " + e.getMessage()); } catch (Exception e) { // Log other exceptions at ERROR level LOGGER.error("Agent execution threw unexpected Exception for task {}", taskId, e); - emitter.fail(new org.a2aproject.sdk.spec.InternalError("Agent execution failed: " + e.getMessage())); + reportAgentExecutionFailure(emitter, "Agent execution failed: " + e.getMessage()); } LOGGER.debug("Agent execution completed for task {}", taskId); // The consumer (running on the Vert.x worker thread) handles queue lifecycle. @@ -1310,6 +1310,17 @@ public void run() { return runnable; } + private void reportAgentExecutionFailure(AgentEmitter emitter, String message) { + boolean wasInterrupted = Thread.interrupted(); + try { + emitter.fail(new InternalError(message)); + } finally { + if (wasInterrupted) { + Thread.currentThread().interrupt(); + } + } + } + private CompletableFuture cleanupProducer(@Nullable CompletableFuture agentFuture, @Nullable CompletableFuture consumptionFuture, String taskId, EventQueue queue, boolean isStreaming) { LOGGER.debug("Starting cleanup for task {} (streaming={})", taskId, isStreaming); logThreadStats("CLEANUP START"); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java index 19d458fd6..b9ca2409e 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java @@ -140,7 +140,20 @@ public void updateStatus(TaskState taskState, @Nullable Message message) { .contextId(contextId) .status(new TaskStatus(taskState, message, null)) .build(); - eventQueue.enqueueEvent(event); + if (isFinal) { + enqueueTerminalEvent(event); + } else { + eventQueue.enqueueEvent(event); + } + } + + private void enqueueTerminalEvent(Event event) { + try { + eventQueue.enqueueEvent(event); + } catch (RuntimeException | Error e) { + terminalStateReached.compareAndSet(true, false); + throw e; + } } /** @@ -284,7 +297,7 @@ public void fail(A2AError error) { throw new IllegalStateException("Cannot update task status - terminal state already reached"); } - eventQueue.enqueueEvent(error); + enqueueTerminalEvent(error); // Status transition happens automatically in MainEventBusProcessor // The error event is terminal and will trigger FAILED state transition } 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 1e729bd20..2d928c7e4 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 @@ -46,6 +46,7 @@ import org.a2aproject.sdk.spec.Event; import org.a2aproject.sdk.spec.EventKind; import org.a2aproject.sdk.spec.GetTaskPushNotificationConfigParams; +import org.a2aproject.sdk.spec.InternalError; import org.a2aproject.sdk.spec.InvalidParamsError; import org.a2aproject.sdk.spec.ListTasksParams; import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams; @@ -1692,6 +1693,32 @@ public void cancel(RequestContext context, AgentEmitter emitter) { assertEquals(TaskState.TASK_STATE_FAILED, storedTask.status().state()); } + @Test + void interruptedTerminalEnqueueStillReportsAgentFailure() throws Exception { + CountDownLatch executorRan = new CountDownLatch(1); + agentExecutorExecute = (context, emitter) -> { + executorRan.countDown(); + Thread.currentThread().interrupt(); + emitter.complete(); + }; + + MessageSendParams params = MessageSendParams.builder() + .message(Message.builder() + .messageId("msg-interrupted-terminal-enqueue") + .role(Message.Role.ROLE_USER) + .parts(new TextPart("hello")) + .build()) + .configuration(DEFAULT_CONFIG) + .build(); + + InternalError failure = assertThrows(InternalError.class, + () -> requestHandler.onMessageSend(params, NULL_CONTEXT)); + + assertTrue(executorRan.await(5, TimeUnit.SECONDS), "Executor should have run"); + assertEquals("Agent execution failed: Unable to acquire the semaphore to enqueue the event", + failure.getMessage()); + } + private DefaultRequestHandler buildHandlerWithRouter(AgentExecutorRouter router) { return DefaultRequestHandler.builder() .agentExecutor(executor) diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterConcurrencyTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterConcurrencyTest.java index 92745c465..832501b7e 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterConcurrencyTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/tasks/AgentEmitterConcurrencyTest.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.server.agentexecution.RequestContext; import org.a2aproject.sdk.server.events.EventQueue; +import org.a2aproject.sdk.spec.InternalError; import org.a2aproject.sdk.spec.UnsupportedOperationError; import org.junit.jupiter.api.Test; @@ -219,6 +220,43 @@ public void testFailWithErrorThenFailWithMessage() { exception.getMessage()); } + @Test + public void testFailedTerminalStatusEnqueueAllowsErrorFallback() { + RequestContext context = mock(RequestContext.class); + when(context.getTaskId()).thenReturn("test-task-123"); + when(context.getContextId()).thenReturn("test-context-456"); + + EventQueue eventQueue = mock(EventQueue.class); + RuntimeException enqueueFailure = new RuntimeException("queue failure"); + doThrow(enqueueFailure).doNothing().when(eventQueue).enqueueEvent(any()); + AgentEmitter emitter = new AgentEmitter(context, eventQueue); + + assertSame(enqueueFailure, assertThrows(RuntimeException.class, emitter::complete)); + + emitter.fail(new InternalError("fallback")); + + verify(eventQueue, times(2)).enqueueEvent(any()); + } + + @Test + public void testFailedErrorEnqueueAllowsTerminalStatusRetry() { + RequestContext context = mock(RequestContext.class); + when(context.getTaskId()).thenReturn("test-task-123"); + when(context.getContextId()).thenReturn("test-context-456"); + + EventQueue eventQueue = mock(EventQueue.class); + RuntimeException enqueueFailure = new RuntimeException("queue failure"); + doThrow(enqueueFailure).doNothing().when(eventQueue).enqueueEvent(any()); + AgentEmitter emitter = new AgentEmitter(context, eventQueue); + + assertSame(enqueueFailure, assertThrows(RuntimeException.class, + () -> emitter.fail(new UnsupportedOperationError()))); + + emitter.complete(); + + verify(eventQueue, times(2)).enqueueEvent(any()); + } + @Test public void testNonTerminalThenTerminalState() throws InterruptedException { // Setup From b1b9a11b5c4da29666fa2a541516ba45120e7db4 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Fri, 25 Sep 2026 15:37:44 +0100 Subject: [PATCH 2/7] test(server): cover terminal enqueue backpressure This fixes #1183 --- .../DefaultRequestHandlerTest.java | 119 ++++++++++++++++-- 1 file changed, 112 insertions(+), 7 deletions(-) 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 2d928c7e4..f99d20e9d 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 @@ -35,12 +35,14 @@ import org.a2aproject.sdk.server.events.InMemoryQueueManager; import org.a2aproject.sdk.server.events.MainEventBus; import org.a2aproject.sdk.server.events.MainEventBusProcessor; +import org.a2aproject.sdk.server.events.MainEventBusProcessorCallback; import org.a2aproject.sdk.server.tasks.AgentEmitter; import org.a2aproject.sdk.server.tasks.InMemoryPushNotificationConfigStore; import org.a2aproject.sdk.server.tasks.InMemoryTaskStore; import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore; import org.a2aproject.sdk.server.tasks.PushNotificationSender; import org.a2aproject.sdk.server.tasks.TaskStore; +import org.a2aproject.sdk.server.tasks.TaskStateProvider; import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.CancelTaskParams; import org.a2aproject.sdk.spec.Event; @@ -1695,10 +1697,69 @@ public void cancel(RequestContext context, AgentEmitter emitter) { @Test void interruptedTerminalEnqueueStillReportsAgentFailure() throws Exception { + EventQueueUtil.stop(mainEventBusProcessor); + queueManager = new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) { + @Override + public EventQueue.EventQueueBuilder createBaseEventQueueBuilder(String taskId) { + return super.createBaseEventQueueBuilder(taskId).queueSize(1); + } + }; + mainEventBusProcessor = new MainEventBusProcessor(mainEventBus, taskStore, + NOOP_PUSHNOTIFICATION_SENDER, queueManager); + EventQueueUtil.start(mainEventBusProcessor); + + CountDownLatch workingEventProcessing = new CountDownLatch(1); + CountDownLatch releaseWorkingEvent = new CountDownLatch(1); + CountDownLatch fallbackErrorProcessed = new CountDownLatch(1); + mainEventBusProcessor.setCallback(new MainEventBusProcessorCallback() { + @Override + public void onEventProcessed(String taskId, Event event) { + if (event instanceof TaskStatusUpdateEvent statusUpdate + && statusUpdate.status().state() == TaskState.TASK_STATE_WORKING) { + workingEventProcessing.countDown(); + try { + releaseWorkingEvent.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + } + } else if (event instanceof InternalError) { + fallbackErrorProcessed.countDown(); + } + } + + @Override + public void onTaskFinalized(String taskId) { + } + }); + + CountDownLatch agentFinished = new CountDownLatch(1); + AtomicReference agentThread = new AtomicReference<>(); + AtomicReference interruptStatusAfterAgent = new AtomicReference<>(false); + Executor recordingExecutor = command -> internalExecutor.execute(() -> { + try { + command.run(); + } finally { + interruptStatusAfterAgent.set(Thread.currentThread().isInterrupted()); + agentFinished.countDown(); + } + }); + requestHandler = DefaultRequestHandler.builder() + .agentExecutor(executor) + .taskStore(taskStore) + .queueManager(queueManager) + .pushConfigStore(pushConfigStore) + .pushNotificationsEnabled(true) + .mainEventBusProcessor(mainEventBusProcessor) + .executor(recordingExecutor) + .eventConsumerExecutor(internalExecutor) + .authorizationRequired(false) + .build(); + CountDownLatch executorRan = new CountDownLatch(1); agentExecutorExecute = (context, emitter) -> { + agentThread.set(Thread.currentThread()); executorRan.countDown(); - Thread.currentThread().interrupt(); + emitter.startWork(); emitter.complete(); }; @@ -1708,15 +1769,59 @@ void interruptedTerminalEnqueueStillReportsAgentFailure() throws Exception { .role(Message.Role.ROLE_USER) .parts(new TextPart("hello")) .build()) - .configuration(DEFAULT_CONFIG) + .configuration(MessageSendConfiguration.builder() + .returnImmediately(false) + .acceptedOutputModes(List.of()) + .build()) .build(); - InternalError failure = assertThrows(InternalError.class, - () -> requestHandler.onMessageSend(params, NULL_CONTEXT)); + ExecutorService controller = Executors.newSingleThreadExecutor(); + try { + Future interruptWasRestored = controller.submit(() -> { + assertTrue(workingEventProcessing.await(5, TimeUnit.SECONDS), + "Processor should hold the queue capacity after the WORKING event"); + assertTrue(executorRan.await(5, TimeUnit.SECONDS), "Executor should have run"); + Thread worker = agentThread.get(); + assertNotNull(worker); + assertTrue(awaitSemaphoreAcquire(worker, false), + "Terminal enqueue should wait while the queue is full"); + + worker.interrupt(); + + assertTrue(awaitSemaphoreAcquire(worker, false), + "Fallback enqueue should wait with the interrupt temporarily cleared"); + releaseWorkingEvent.countDown(); + assertTrue(agentFinished.await(5, TimeUnit.SECONDS), "Agent failure reporting should finish"); + return interruptStatusAfterAgent.get(); + }); - assertTrue(executorRan.await(5, TimeUnit.SECONDS), "Executor should have run"); - assertEquals("Agent execution failed: Unable to acquire the semaphore to enqueue the event", - failure.getMessage()); + assertThrows(InternalError.class, + () -> requestHandler.onMessageSend(params, NULL_CONTEXT)); + + assertTrue(fallbackErrorProcessed.await(5, TimeUnit.SECONDS), + "The fallback InternalError should be processed by the event bus"); + assertTrue(interruptWasRestored.get(5, TimeUnit.SECONDS), + "The agent worker's interrupt status should be restored after reporting the error"); + } finally { + releaseWorkingEvent.countDown(); + controller.shutdownNow(); + } + } + + private static boolean awaitSemaphoreAcquire(Thread thread, boolean interrupted) throws InterruptedException { + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + do { + if (thread.getState() == Thread.State.WAITING && thread.isInterrupted() == interrupted) { + for (StackTraceElement frame : thread.getStackTrace()) { + if (frame.getClassName().equals("java.util.concurrent.Semaphore") + && frame.getMethodName().equals("acquire")) { + return true; + } + } + } + TimeUnit.MILLISECONDS.sleep(10); + } while (System.nanoTime() < deadline); + return false; } private DefaultRequestHandler buildHandlerWithRouter(AgentExecutorRouter router) { From fa4131338bf325a22c4ce5d8b1390c56eee59896 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Fri, 25 Sep 2026 15:43:58 +0100 Subject: [PATCH 3/7] fix(server): preserve interrupts for protocol errors This fixes #1183 --- .../DefaultRequestHandler.java | 12 +-- .../DefaultRequestHandlerTest.java | 75 +++++++++++++++---- 2 files changed, 69 insertions(+), 18 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 967afb87f..1bb545c2a 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -1261,16 +1261,18 @@ public void run() { // These are expected business errors but should be tracked LOGGER.warn("Agent execution threw A2AError for task {}: {} - {}", taskId, e.getClass().getSimpleName(), e.getMessage(), e); - emitter.fail(e); + enqueueErrorPreservingInterrupt(emitter, e); } catch (RuntimeException e) { // Log unexpected runtime exceptions at ERROR level // These indicate bugs in agent implementation LOGGER.error("Agent execution threw unexpected RuntimeException for task {}", taskId, e); - reportAgentExecutionFailure(emitter, "Agent execution failed: " + e.getMessage()); + enqueueErrorPreservingInterrupt(emitter, + new InternalError("Agent execution failed: " + e.getMessage())); } catch (Exception e) { // Log other exceptions at ERROR level LOGGER.error("Agent execution threw unexpected Exception for task {}", taskId, e); - reportAgentExecutionFailure(emitter, "Agent execution failed: " + e.getMessage()); + enqueueErrorPreservingInterrupt(emitter, + new InternalError("Agent execution failed: " + e.getMessage())); } LOGGER.debug("Agent execution completed for task {}", taskId); // The consumer (running on the Vert.x worker thread) handles queue lifecycle. @@ -1310,10 +1312,10 @@ public void run() { return runnable; } - private void reportAgentExecutionFailure(AgentEmitter emitter, String message) { + private void enqueueErrorPreservingInterrupt(AgentEmitter emitter, A2AError error) { boolean wasInterrupted = Thread.interrupted(); try { - emitter.fail(new InternalError(message)); + emitter.fail(error); } finally { if (wasInterrupted) { Thread.currentThread().interrupt(); 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 f99d20e9d..66d0d303b 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 @@ -1697,6 +1697,35 @@ public void cancel(RequestContext context, AgentEmitter emitter) { @Test void interruptedTerminalEnqueueStillReportsAgentFailure() throws Exception { + assertInterruptedTerminalEnqueueReportsError( + (context, emitter) -> emitter.complete(), InternalError.class, null, true, false); + } + + @Test + void interruptedProtocolErrorEnqueueStillReportsAgentFailure() throws Exception { + CountDownLatch agentReadyToBeInterrupted = new CountDownLatch(1); + CountDownLatch agentWait = new CountDownLatch(1); + try { + assertInterruptedTerminalEnqueueReportsError((context, emitter) -> { + agentReadyToBeInterrupted.countDown(); + try { + agentWait.await(); + } catch (InterruptedException e) { + Thread.currentThread().interrupt(); + throw new UnsupportedOperationError(); + } + }, UnsupportedOperationError.class, agentReadyToBeInterrupted, false, true); + } finally { + agentWait.countDown(); + } + } + + private void assertInterruptedTerminalEnqueueReportsError( + AgentExecutorMethod agentAction, + Class expectedErrorClass, + CountDownLatch agentReadyToBeInterrupted, + boolean interruptBlockedTerminalEnqueue, + boolean returnImmediately) throws Exception { EventQueueUtil.stop(mainEventBusProcessor); queueManager = new InMemoryQueueManager((TaskStateProvider) taskStore, mainEventBus) { @Override @@ -1710,7 +1739,8 @@ public EventQueue.EventQueueBuilder createBaseEventQueueBuilder(String taskId) { CountDownLatch workingEventProcessing = new CountDownLatch(1); CountDownLatch releaseWorkingEvent = new CountDownLatch(1); - CountDownLatch fallbackErrorProcessed = new CountDownLatch(1); + CountDownLatch reportedErrorProcessed = new CountDownLatch(1); + AtomicReference reportedError = new AtomicReference<>(); mainEventBusProcessor.setCallback(new MainEventBusProcessorCallback() { @Override public void onEventProcessed(String taskId, Event event) { @@ -1722,8 +1752,9 @@ public void onEventProcessed(String taskId, Event event) { } catch (InterruptedException e) { Thread.currentThread().interrupt(); } - } else if (event instanceof InternalError) { - fallbackErrorProcessed.countDown(); + } else if (event instanceof A2AError error && expectedErrorClass.isInstance(error)) { + reportedError.set(error); + reportedErrorProcessed.countDown(); } } @@ -1760,7 +1791,7 @@ public void onTaskFinalized(String taskId) { agentThread.set(Thread.currentThread()); executorRan.countDown(); emitter.startWork(); - emitter.complete(); + agentAction.invoke(context, emitter); }; MessageSendParams params = MessageSendParams.builder() @@ -1770,7 +1801,7 @@ public void onTaskFinalized(String taskId) { .parts(new TextPart("hello")) .build()) .configuration(MessageSendConfiguration.builder() - .returnImmediately(false) + .returnImmediately(returnImmediately) .acceptedOutputModes(List.of()) .build()) .build(); @@ -1783,23 +1814,41 @@ public void onTaskFinalized(String taskId) { assertTrue(executorRan.await(5, TimeUnit.SECONDS), "Executor should have run"); Thread worker = agentThread.get(); assertNotNull(worker); - assertTrue(awaitSemaphoreAcquire(worker, false), - "Terminal enqueue should wait while the queue is full"); + if (interruptBlockedTerminalEnqueue) { + assertTrue(awaitSemaphoreAcquire(worker, false), + "Terminal enqueue should wait while the queue is full"); + } else { + assertNotNull(agentReadyToBeInterrupted); + assertTrue(agentReadyToBeInterrupted.await(5, TimeUnit.SECONDS), + "Agent should be ready to throw the protocol error"); + } worker.interrupt(); assertTrue(awaitSemaphoreAcquire(worker, false), - "Fallback enqueue should wait with the interrupt temporarily cleared"); + "Error enqueue should wait with the interrupt temporarily cleared"); releaseWorkingEvent.countDown(); - assertTrue(agentFinished.await(5, TimeUnit.SECONDS), "Agent failure reporting should finish"); + assertTrue(agentFinished.await(5, TimeUnit.SECONDS), "Agent error reporting should finish"); return interruptStatusAfterAgent.get(); }); - assertThrows(InternalError.class, - () -> requestHandler.onMessageSend(params, NULL_CONTEXT)); + String taskId = ""; + if (returnImmediately) { + Task task = assertInstanceOf(Task.class, requestHandler.onMessageSend(params, NULL_CONTEXT)); + taskId = task.id(); + } else { + assertThrows(expectedErrorClass, + () -> requestHandler.onMessageSend(params, NULL_CONTEXT)); + } - assertTrue(fallbackErrorProcessed.await(5, TimeUnit.SECONDS), - "The fallback InternalError should be processed by the event bus"); + assertTrue(reportedErrorProcessed.await(5, TimeUnit.SECONDS), + "The agent error should be processed by the event bus"); + assertInstanceOf(expectedErrorClass, reportedError.get()); + if (returnImmediately) { + Task storedTask = taskStore.get(taskId); + assertNotNull(storedTask); + assertEquals(TaskState.TASK_STATE_FAILED, storedTask.status().state()); + } assertTrue(interruptWasRestored.get(5, TimeUnit.SECONDS), "The agent worker's interrupt status should be restored after reporting the error"); } finally { From 066ad470fd7a29a2dc8e17d536cbd5961be224b3 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Mon, 28 Sep 2026 12:45:45 +0100 Subject: [PATCH 4/7] Review fixes --- .../sdk/server/events/EventQueue.java | 28 ++++++--- .../DefaultRequestHandler.java | 62 +++++++++++++------ .../sdk/server/tasks/AgentEmitter.java | 39 +++++++----- .../DefaultRequestHandlerTest.java | 26 ++++---- 4 files changed, 98 insertions(+), 57 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 3c24a103d..92b93a741 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 @@ -11,6 +11,7 @@ import java.util.concurrent.atomic.AtomicBoolean; import org.a2aproject.sdk.server.tasks.TaskStateProvider; +import org.a2aproject.sdk.spec.A2AError; import org.a2aproject.sdk.spec.Event; import org.a2aproject.sdk.spec.Task; import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent; @@ -510,14 +511,7 @@ public void enqueueItem(EventQueueItem item) { // Validate event taskId matches queue taskId validateEventIds(event); - // Check if this is a final event BEFORE submitting to MainEventBus - // If it is, notify all children to expect it (so they wait for MainEventBusProcessor) - if (isFinalEvent(event)) { - LOGGER.debug("Final event detected, notifying {} children to expect it", children.size()); - for (ChildQueue child : children) { - child.expectFinalEvent(); - } - } + boolean finalEvent = isFinalEvent(event); // Acquire semaphore for backpressure try { @@ -527,6 +521,15 @@ public void enqueueItem(EventQueueItem item) { throw new RuntimeException("Unable to acquire the semaphore to enqueue the event", e); } + // Notify children only after acquiring capacity. An interrupted acquire must not + // leave them waiting for an event that was never submitted. + if (finalEvent) { + LOGGER.debug("Final event detected, notifying {} children to expect it", children.size()); + for (ChildQueue child : children) { + child.expectFinalEvent(); + } + } + LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this); // Submit to MainEventBus for centralized persistence + distribution @@ -541,6 +544,11 @@ public void enqueueItem(EventQueueItem item) { // Release the permit here to avoid leaking it and eventually blocking // all event processing for this task. semaphore.release(); + if (finalEvent) { + for (ChildQueue child : children) { + child.clearAwaitingFinalEvent(); + } + } throw e; } } @@ -595,6 +603,8 @@ private boolean isFinalEvent(Event event) { && task.status().state().isFinal(); } else if (event instanceof TaskStatusUpdateEvent statusUpdate) { return statusUpdate.isFinal(); + } else if (event instanceof A2AError) { + return true; } return false; } @@ -830,6 +840,8 @@ private boolean isFinalEvent(Event event) { && task.status().state().isFinal(); } else if (event instanceof TaskStatusUpdateEvent statusUpdate) { return statusUpdate.isFinal(); + } else if (event instanceof A2AError) { + return true; } return false; } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 1bb545c2a..f34788c49 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -1255,24 +1255,29 @@ public void run() { LOGGER.debug("Agent execution starting for task {}", taskId); AgentEmitter emitter = new AgentEmitter(requestContext, queue); try { - resolvedExecutor.execute(requestContext, emitter); - } catch (A2AError e) { - // Log A2A errors at WARN level with full stack trace - // These are expected business errors but should be tracked - LOGGER.warn("Agent execution threw A2AError for task {}: {} - {}", - taskId, e.getClass().getSimpleName(), e.getMessage(), e); - enqueueErrorPreservingInterrupt(emitter, e); - } catch (RuntimeException e) { - // Log unexpected runtime exceptions at ERROR level - // These indicate bugs in agent implementation - LOGGER.error("Agent execution threw unexpected RuntimeException for task {}", taskId, e); - enqueueErrorPreservingInterrupt(emitter, - new InternalError("Agent execution failed: " + e.getMessage())); - } catch (Exception e) { - // Log other exceptions at ERROR level - LOGGER.error("Agent execution threw unexpected Exception for task {}", taskId, e); - enqueueErrorPreservingInterrupt(emitter, - new InternalError("Agent execution failed: " + e.getMessage())); + try { + resolvedExecutor.execute(requestContext, emitter); + } catch (A2AError e) { + // Log A2A errors at WARN level with full stack trace + // These are expected business errors but should be tracked + LOGGER.warn("Agent execution threw A2AError for task {}: {} - {}", + taskId, e.getClass().getSimpleName(), e.getMessage(), e); + enqueueErrorPreservingInterrupt(emitter, e); + } catch (RuntimeException e) { + // Log unexpected runtime exceptions at ERROR level + // These indicate bugs in agent implementation + LOGGER.error("Agent execution threw unexpected RuntimeException for task {}", taskId, e); + enqueueErrorPreservingInterrupt(emitter, + new InternalError("Agent execution failed: " + e.getMessage())); + } catch (Exception e) { + // Log other exceptions at ERROR level + LOGGER.error("Agent execution threw unexpected Exception for task {}", taskId, e); + enqueueErrorPreservingInterrupt(emitter, + new InternalError("Agent execution failed: " + e.getMessage())); + } + } finally { + // Executor threads are reused. Do not return an agent's interrupt to the pool. + Thread.interrupted(); } LOGGER.debug("Agent execution completed for task {}", taskId); // The consumer (running on the Vert.x worker thread) handles queue lifecycle. @@ -1315,7 +1320,17 @@ public void run() { private void enqueueErrorPreservingInterrupt(AgentEmitter emitter, A2AError error) { boolean wasInterrupted = Thread.interrupted(); try { - emitter.fail(error); + while (true) { + try { + emitter.tryFail(error); + return; + } catch (RuntimeException e) { + if (!causedByInterruptedEnqueue(e)) { + throw e; + } + wasInterrupted |= Thread.interrupted(); + } + } } finally { if (wasInterrupted) { Thread.currentThread().interrupt(); @@ -1323,6 +1338,15 @@ private void enqueueErrorPreservingInterrupt(AgentEmitter emitter, A2AError erro } } + private boolean causedByInterruptedEnqueue(Throwable error) { + for (Throwable cause = error; cause != null; cause = cause.getCause()) { + if (cause instanceof InterruptedException) { + return true; + } + } + return false; + } + private CompletableFuture cleanupProducer(@Nullable CompletableFuture agentFuture, @Nullable CompletableFuture consumptionFuture, String taskId, EventQueue queue, boolean isStreaming) { LOGGER.debug("Starting cleanup for task {} (streaming={})", taskId, isStreaming); logThreadStats("CLEANUP START"); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java index b9ca2409e..36e61e897 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java @@ -3,7 +3,6 @@ import java.util.List; import java.util.Map; import java.util.UUID; -import java.util.concurrent.atomic.AtomicBoolean; import org.a2aproject.sdk.server.agentexecution.RequestContext; import org.slf4j.Logger; @@ -101,7 +100,7 @@ public class AgentEmitter { private final EventQueue eventQueue; private final String taskId; private final String contextId; - private final AtomicBoolean terminalStateReached = new AtomicBoolean(false); + private boolean terminalStateReached; /** * Creates a new AgentEmitter for the given request context and event queue. @@ -121,18 +120,16 @@ public AgentEmitter(RequestContext context, EventQueue eventQueue) { * @param taskState the new task state * @param message optional message to include with the status update */ - public void updateStatus(TaskState taskState, @Nullable Message message) { + public synchronized void updateStatus(TaskState taskState, @Nullable Message message) { boolean isFinal = taskState.isFinal(); // Check terminal state first (fail fast) - if (terminalStateReached.get()) { + if (terminalStateReached) { throw new IllegalStateException("Cannot update task status - terminal state already reached"); } - // For final states, atomically set the flag + // For final states, record the terminal transition while holding this emitter's lock. if (isFinal) { - if (!terminalStateReached.compareAndSet(false, true)) { - throw new IllegalStateException("Cannot update task status - terminal state already reached"); - } + terminalStateReached = true; } TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() @@ -150,8 +147,8 @@ public void updateStatus(TaskState taskState, @Nullable Message message) { private void enqueueTerminalEvent(Event event) { try { eventQueue.enqueueEvent(event); - } catch (RuntimeException | Error e) { - terminalStateReached.compareAndSet(true, false); + } catch (RuntimeException e) { + terminalStateReached = false; throw e; } } @@ -291,17 +288,29 @@ public void fail(@Nullable Message message) { * @since 1.0.0 */ public void fail(A2AError error) { - // Set terminal state flag BEFORE enqueueing error - // This prevents race conditions where agent calls fail(error) then complete() - if (!terminalStateReached.compareAndSet(false, true)) { + if (!tryFail(error)) { throw new IllegalStateException("Cannot update task status - terminal state already reached"); } - - enqueueTerminalEvent(error); // Status transition happens automatically in MainEventBusProcessor // The error event is terminal and will trigger FAILED state transition } + /** + * Attempts to fail the task unless a terminal state has already been claimed. + * + * @param error the A2A error to enqueue + * @return {@code true} if the error was enqueued, or {@code false} if a terminal state was already reached + */ + public synchronized boolean tryFail(A2AError error) { + if (terminalStateReached) { + return false; + } + terminalStateReached = true; + enqueueTerminalEvent(error); + // Status transition happens automatically in MainEventBusProcessor + return true; + } + /** * Marks the task as SUBMITTED. */ 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 66d0d303b..73759b331 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 @@ -1765,12 +1765,13 @@ public void onTaskFinalized(String taskId) { CountDownLatch agentFinished = new CountDownLatch(1); AtomicReference agentThread = new AtomicReference<>(); - AtomicReference interruptStatusAfterAgent = new AtomicReference<>(false); + AtomicReference interruptStatusAtPoolReturn = new AtomicReference<>(false); Executor recordingExecutor = command -> internalExecutor.execute(() -> { try { command.run(); } finally { - interruptStatusAfterAgent.set(Thread.currentThread().isInterrupted()); + interruptStatusAtPoolReturn.set(Thread.currentThread().isInterrupted()); + Thread.interrupted(); agentFinished.countDown(); } }); @@ -1808,14 +1809,14 @@ public void onTaskFinalized(String taskId) { ExecutorService controller = Executors.newSingleThreadExecutor(); try { - Future interruptWasRestored = controller.submit(() -> { + Future poolInterruptStatus = controller.submit(() -> { assertTrue(workingEventProcessing.await(5, TimeUnit.SECONDS), "Processor should hold the queue capacity after the WORKING event"); assertTrue(executorRan.await(5, TimeUnit.SECONDS), "Executor should have run"); Thread worker = agentThread.get(); assertNotNull(worker); if (interruptBlockedTerminalEnqueue) { - assertTrue(awaitSemaphoreAcquire(worker, false), + assertTrue(awaitThreadWaiting(worker, false), "Terminal enqueue should wait while the queue is full"); } else { assertNotNull(agentReadyToBeInterrupted); @@ -1825,11 +1826,11 @@ public void onTaskFinalized(String taskId) { worker.interrupt(); - assertTrue(awaitSemaphoreAcquire(worker, false), + assertTrue(awaitThreadWaiting(worker, false), "Error enqueue should wait with the interrupt temporarily cleared"); releaseWorkingEvent.countDown(); assertTrue(agentFinished.await(5, TimeUnit.SECONDS), "Agent error reporting should finish"); - return interruptStatusAfterAgent.get(); + return interruptStatusAtPoolReturn.get(); }); String taskId = ""; @@ -1849,24 +1850,19 @@ public void onTaskFinalized(String taskId) { assertNotNull(storedTask); assertEquals(TaskState.TASK_STATE_FAILED, storedTask.status().state()); } - assertTrue(interruptWasRestored.get(5, TimeUnit.SECONDS), - "The agent worker's interrupt status should be restored after reporting the error"); + assertFalse(poolInterruptStatus.get(5, TimeUnit.SECONDS), + "The agent worker should return to the executor pool without an interrupt flag"); } finally { releaseWorkingEvent.countDown(); controller.shutdownNow(); } } - private static boolean awaitSemaphoreAcquire(Thread thread, boolean interrupted) throws InterruptedException { + private static boolean awaitThreadWaiting(Thread thread, boolean interrupted) throws InterruptedException { long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); do { if (thread.getState() == Thread.State.WAITING && thread.isInterrupted() == interrupted) { - for (StackTraceElement frame : thread.getStackTrace()) { - if (frame.getClassName().equals("java.util.concurrent.Semaphore") - && frame.getMethodName().equals("acquire")) { - return true; - } - } + return true; } TimeUnit.MILLISECONDS.sleep(10); } while (System.nanoTime() < deadline); From a35a7523b5d2fa7f6918f0e7721aefb28c124074 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Mon, 28 Sep 2026 12:53:31 +0100 Subject: [PATCH 5/7] More review --- .../sdk/server/events/EventQueue.java | 48 ++++++++++++------- 1 file changed, 32 insertions(+), 16 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 92b93a741..8510a0614 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 @@ -1,5 +1,6 @@ package org.a2aproject.sdk.server.events; +import java.util.ArrayList; import java.util.List; import java.util.Objects; import java.util.concurrent.BlockingQueue; @@ -9,6 +10,7 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; +import java.util.concurrent.atomic.AtomicInteger; import org.a2aproject.sdk.server.tasks.TaskStateProvider; import org.a2aproject.sdk.spec.A2AError; @@ -513,20 +515,31 @@ public void enqueueItem(EventQueueItem item) { boolean finalEvent = isFinalEvent(event); + // Notify current children before waiting for capacity. A consumer may already be + // marked complete (for example, while waiting for a replicated final event) and + // otherwise close its queue before this producer is able to submit the event. + List awaitingChildren = new ArrayList<>(); + if (finalEvent) { + awaitingChildren.addAll(children); + awaitingChildren.forEach(ChildQueue::expectFinalEvent); + } + // Acquire semaphore for backpressure try { semaphore.acquire(); } catch (InterruptedException e) { + awaitingChildren.forEach(ChildQueue::cancelExpectedFinalEvent); Thread.currentThread().interrupt(); throw new RuntimeException("Unable to acquire the semaphore to enqueue the event", e); } - // Notify children only after acquiring capacity. An interrupted acquire must not - // leave them waiting for an event that was never submitted. + // Include children that subscribed while this producer was waiting for capacity. if (finalEvent) { - LOGGER.debug("Final event detected, notifying {} children to expect it", children.size()); for (ChildQueue child : children) { - child.expectFinalEvent(); + if (!awaitingChildren.contains(child)) { + child.expectFinalEvent(); + awaitingChildren.add(child); + } } } @@ -544,11 +557,7 @@ public void enqueueItem(EventQueueItem item) { // Release the permit here to avoid leaking it and eventually blocking // all event processing for this task. semaphore.release(); - if (finalEvent) { - for (ChildQueue child : children) { - child.clearAwaitingFinalEvent(); - } - } + awaitingChildren.forEach(ChildQueue::cancelExpectedFinalEvent); throw e; } } @@ -786,7 +795,7 @@ static class ChildQueue extends EventQueue { private final MainQueue parent; private final BlockingQueue queue; private volatile boolean immediateClose = false; - private volatile boolean awaitingFinalEvent = false; + private final AtomicInteger expectedFinalEvents = new AtomicInteger(); public ChildQueue(MainQueue parent) { this.parent = parent; @@ -824,8 +833,8 @@ private void internalEnqueueItem(EventQueueItem item) { LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this); // If we were awaiting a final event and this is it, clear the flag - if (awaitingFinalEvent && isFinalEvent(event)) { - awaitingFinalEvent = false; + if (isAwaitingFinalEvent() && isFinalEvent(event)) { + expectedFinalEvents.set(0); LOGGER.debug("ChildQueue {} received awaited final event", System.identityHashCode(this)); } } @@ -861,7 +870,7 @@ public EventQueueItem dequeueEventItem(int waitMilliSeconds) throws EventQueueCl // For immediate close: exit immediately even if queue is not empty (race with MainEventBusProcessor) // For graceful close: only exit when queue is empty (wait for all events to be consumed) // BUT: if awaiting final event, keep polling even if closed and empty - if (isClosed() && (queue.isEmpty() || immediateClose) && !awaitingFinalEvent) { + if (isClosed() && (queue.isEmpty() || immediateClose) && !isAwaitingFinalEvent()) { LOGGER.debug("ChildQueue is closed{}, sending termination message. {} (queueSize={})", immediateClose ? " (immediate)" : " and empty", this, @@ -906,7 +915,7 @@ public int size() { @Override public boolean isAwaitingFinalEvent() { - return awaitingFinalEvent; + return expectedFinalEvents.get() > 0; } @Override @@ -938,17 +947,24 @@ protected void doClose(boolean immediate) { * This ensures the ChildQueue keeps polling until the final event arrives (after MainEventBusProcessor). */ void expectFinalEvent() { - awaitingFinalEvent = true; + expectedFinalEvents.incrementAndGet(); LOGGER.debug("ChildQueue {} now awaiting final event", System.identityHashCode(this)); } + /** + * Removes one expectation when the corresponding final event could not be submitted. + */ + void cancelExpectedFinalEvent() { + expectedFinalEvents.updateAndGet(count -> Math.max(0, count - 1)); + } + /** * Called by EventConsumer when it has waited too long for the final event. * This allows normal timeout logic to proceed if the final event never arrives. */ @Override public void clearAwaitingFinalEvent() { - awaitingFinalEvent = false; + expectedFinalEvents.set(0); LOGGER.debug("ChildQueue {} cleared awaitingFinalEvent flag (timeout)", System.identityHashCode(this)); } From bdf3776ad361d84e21af33f7637b6029486ed373 Mon Sep 17 00:00:00 2001 From: Kabir Khan Date: Mon, 28 Sep 2026 13:17:31 +0100 Subject: [PATCH 6/7] More review fixes --- .../sdk/server/events/EventConsumer.java | 12 +- .../sdk/server/events/EventQueue.java | 176 +++++++++++++++--- .../DefaultRequestHandler.java | 20 +- .../sdk/server/tasks/AgentEmitter.java | 48 +++-- .../DefaultRequestHandlerTest.java | 10 +- 5 files changed, 208 insertions(+), 58 deletions(-) diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java index 531421aca..a4379fb09 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/EventConsumer.java @@ -179,10 +179,14 @@ public Flow.Publisher consumeAll() { if (pollTimeoutsAfterAgentCompleted >= MAX_POLL_TIMEOUTS_AFTER_AGENT_COMPLETED) { LOGGER.debug("Agent completed with {} consecutive poll timeouts and empty queue, closing for graceful completion (queue={})", pollTimeoutsAfterAgentCompleted, System.identityHashCode(queue)); - queue.close(); - completed = true; - tube.complete(); - return; + if (queue.closeIfNotAwaitingFinalEvent()) { + completed = true; + tube.complete(); + return; + } + // A final event started while this consumer was timing out. + // Keep polling until it is distributed or its producer rolls back. + pollTimeoutsAfterAgentCompleted = 0; } else { LOGGER.debug("Agent completed but grace period active ({}/{} timeouts), continuing to poll (queue={})", pollTimeoutsAfterAgentCompleted, MAX_POLL_TIMEOUTS_AFTER_AGENT_COMPLETED, System.identityHashCode(queue)); 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 8510a0614..58db0a866 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 @@ -1,8 +1,9 @@ package org.a2aproject.sdk.server.events; -import java.util.ArrayList; +import java.util.HashSet; import java.util.List; import java.util.Objects; +import java.util.Set; import java.util.concurrent.BlockingQueue; import java.util.concurrent.CopyOnWriteArrayList; import java.util.concurrent.CountDownLatch; @@ -10,7 +11,6 @@ import java.util.concurrent.Semaphore; import java.util.concurrent.TimeUnit; import java.util.concurrent.atomic.AtomicBoolean; -import java.util.concurrent.atomic.AtomicInteger; import org.a2aproject.sdk.server.tasks.TaskStateProvider; import org.a2aproject.sdk.spec.A2AError; @@ -224,6 +224,22 @@ public void enqueueEvent(Event event) { enqueueItem(new LocalEventQueueItem(event)); } + /** + * Enqueues an event using the queue's bounded backpressure operation. + * + * @param event the event to enqueue + * @param timeout the maximum time to wait for queue capacity; must be non-negative + * @param unit the timeout unit + * @throws IllegalArgumentException if timeout is negative + * @throws UnsupportedOperationException if this queue does not support bounded enqueues + */ + public void enqueueEvent(Event event, long timeout, TimeUnit unit) { + if (timeout < 0) { + throw new IllegalArgumentException("timeout must be non-negative"); + } + enqueueItem(new LocalEventQueueItem(event), timeout, Objects.requireNonNull(unit, "unit")); + } + /** * Enqueues an event queue item for processing. *

@@ -236,6 +252,29 @@ public void enqueueEvent(Event event) { */ public abstract void enqueueItem(EventQueueItem item); + /** Timed form of {@link #enqueueItem(EventQueueItem)}. */ + public void enqueueItem(EventQueueItem item, long timeout, TimeUnit unit) { + throw new UnsupportedOperationException("This event queue does not support bounded enqueues"); + } + + /** Runtime exception used when a queue acquire was interrupted. */ + public static class EnqueueInterruptedException extends RuntimeException { + private static final long serialVersionUID = 1L; + + public EnqueueInterruptedException(InterruptedException cause) { + super("Unable to acquire the semaphore to enqueue the event", cause); + } + } + + /** Runtime exception used when a timed queue acquire expires. */ + public static class EnqueueTimeoutException extends RuntimeException { + private static final long serialVersionUID = 1L; + + public EnqueueTimeoutException() { + super("Timed out waiting for queue capacity to enqueue the event"); + } + } + /** * Enqueues an event directly to this specific queue only, bypassing the MainEventBus. *

@@ -349,6 +388,20 @@ public void clearAwaitingFinalEvent() { // Default no-op implementation - overridden by ChildQueue } + /** + * Closes this queue if it is not currently waiting for an in-flight final event. + * ChildQueue overrides this with an atomic check-and-close operation. + * + * @return true if the queue was closed, false if a final event is still expected + */ + public boolean closeIfNotAwaitingFinalEvent() { + if (isAwaitingFinalEvent()) { + return false; + } + close(); + return true; + } + /** * Closes this event queue gracefully, allowing pending events to be consumed. */ @@ -502,6 +555,18 @@ public int size() { @Override public void enqueueItem(EventQueueItem item) { + enqueueItemInternal(item, -1, null); + } + + @Override + public void enqueueItem(EventQueueItem item, long timeout, TimeUnit unit) { + if (timeout < 0) { + throw new IllegalArgumentException("timeout must be non-negative"); + } + enqueueItemInternal(item, timeout, Objects.requireNonNull(unit, "unit")); + } + + private void enqueueItemInternal(EventQueueItem item, long timeout, @Nullable TimeUnit unit) { // MainQueue must accept events even when closed to support: // 1. Late-arriving replicated events for non-finalized tasks // 2. Events enqueued during onClose callbacks (before super.doClose()) @@ -518,29 +583,37 @@ public void enqueueItem(EventQueueItem item) { // Notify current children before waiting for capacity. A consumer may already be // marked complete (for example, while waiting for a replicated final event) and // otherwise close its queue before this producer is able to submit the event. - List awaitingChildren = new ArrayList<>(); + Set awaitingChildren = new HashSet<>(); if (finalEvent) { - awaitingChildren.addAll(children); - awaitingChildren.forEach(ChildQueue::expectFinalEvent); + for (ChildQueue child : children) { + if (child.expectFinalEvent()) { + awaitingChildren.add(child); + } + } } // Acquire semaphore for backpressure try { - semaphore.acquire(); + if (unit == null) { + semaphore.acquire(); + } else if (!semaphore.tryAcquire(timeout, unit)) { + awaitingChildren.forEach(ChildQueue::cancelPendingFinalEvent); + throw new EnqueueTimeoutException(); + } } catch (InterruptedException e) { - awaitingChildren.forEach(ChildQueue::cancelExpectedFinalEvent); + awaitingChildren.forEach(ChildQueue::cancelPendingFinalEvent); Thread.currentThread().interrupt(); - throw new RuntimeException("Unable to acquire the semaphore to enqueue the event", e); + throw new EnqueueInterruptedException(e); } // Include children that subscribed while this producer was waiting for capacity. if (finalEvent) { for (ChildQueue child : children) { - if (!awaitingChildren.contains(child)) { - child.expectFinalEvent(); + if (!awaitingChildren.contains(child) && child.expectFinalEvent()) { awaitingChildren.add(child); } } + awaitingChildren.forEach(ChildQueue::submitExpectedFinalEvent); } LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this); @@ -557,7 +630,7 @@ public void enqueueItem(EventQueueItem item) { // Release the permit here to avoid leaking it and eventually blocking // all event processing for this task. semaphore.release(); - awaitingChildren.forEach(ChildQueue::cancelExpectedFinalEvent); + awaitingChildren.forEach(ChildQueue::cancelSubmittedFinalEvent); throw e; } } @@ -795,7 +868,8 @@ static class ChildQueue extends EventQueue { private final MainQueue parent; private final BlockingQueue queue; private volatile boolean immediateClose = false; - private final AtomicInteger expectedFinalEvents = new AtomicInteger(); + private int pendingFinalEvents; + private int submittedFinalEvents; public ChildQueue(MainQueue parent) { this.parent = parent; @@ -816,6 +890,11 @@ public void enqueueItem(EventQueueItem item) { parent.enqueueItem(item); } + @Override + public void enqueueItem(EventQueueItem item, long timeout, TimeUnit unit) { + parent.enqueueItem(item, timeout, unit); + } + private void internalEnqueueItem(EventQueueItem item) { // Internal method called by MainEventBusProcessor to add to local queue // Note: Semaphore is managed by parent MainQueue (acquire/release), not ChildQueue @@ -833,9 +912,13 @@ private void internalEnqueueItem(EventQueueItem item) { LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this); // If we were awaiting a final event and this is it, clear the flag - if (isAwaitingFinalEvent() && isFinalEvent(event)) { - expectedFinalEvents.set(0); - LOGGER.debug("ChildQueue {} received awaited final event", System.identityHashCode(this)); + if (isFinalEvent(event)) { + synchronized (this) { + if (submittedFinalEvents > 0) { + submittedFinalEvents = 0; + LOGGER.debug("ChildQueue {} received awaited final event", System.identityHashCode(this)); + } + } } } } @@ -914,8 +997,8 @@ public int size() { } @Override - public boolean isAwaitingFinalEvent() { - return expectedFinalEvents.get() > 0; + public synchronized boolean isAwaitingFinalEvent() { + return pendingFinalEvents > 0 || submittedFinalEvents > 0; } @Override @@ -933,9 +1016,14 @@ protected void doClose(boolean immediate) { super.doClose(immediate); // Sets closed flag if (immediate) { // Immediate close: clear pending events from local queue - this.immediateClose = true; - int clearedCount = queue.size(); - queue.clear(); + int clearedCount; + synchronized (this) { + this.immediateClose = true; + pendingFinalEvents = 0; + submittedFinalEvents = 0; + clearedCount = queue.size(); + queue.clear(); + } LOGGER.debug("Cleared {} events from ChildQueue for immediate close: {}", clearedCount, this); } // For graceful close, let the queue drain naturally through normal consumption @@ -946,16 +1034,34 @@ protected void doClose(boolean immediate) { * Called by MainQueue when it enqueues a final event, BEFORE submitting to MainEventBus. * This ensures the ChildQueue keeps polling until the final event arrives (after MainEventBusProcessor). */ - void expectFinalEvent() { - expectedFinalEvents.incrementAndGet(); + synchronized boolean expectFinalEvent() { + if (isClosed()) { + return false; + } + pendingFinalEvents++; LOGGER.debug("ChildQueue {} now awaiting final event", System.identityHashCode(this)); + return true; } - /** - * Removes one expectation when the corresponding final event could not be submitted. - */ - void cancelExpectedFinalEvent() { - expectedFinalEvents.updateAndGet(count -> Math.max(0, count - 1)); + synchronized void submitExpectedFinalEvent() { + if (pendingFinalEvents > 0) { + pendingFinalEvents--; + if (!isClosed()) { + submittedFinalEvents++; + } + } + } + + synchronized void cancelPendingFinalEvent() { + if (pendingFinalEvents > 0) { + pendingFinalEvents--; + } + } + + synchronized void cancelSubmittedFinalEvent() { + if (submittedFinalEvents > 0) { + submittedFinalEvents--; + } } /** @@ -963,11 +1069,23 @@ void cancelExpectedFinalEvent() { * This allows normal timeout logic to proceed if the final event never arrives. */ @Override - public void clearAwaitingFinalEvent() { - expectedFinalEvents.set(0); + public synchronized void clearAwaitingFinalEvent() { + submittedFinalEvents = 0; LOGGER.debug("ChildQueue {} cleared awaitingFinalEvent flag (timeout)", System.identityHashCode(this)); } + @Override + public boolean closeIfNotAwaitingFinalEvent() { + synchronized (this) { + if (pendingFinalEvents > 0 || submittedFinalEvents > 0) { + return false; + } + doClose(false); + } + parent.childClosing(this, false); + return true; + } + @Override public void close() { close(false); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index f34788c49..a0f4cee46 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -187,6 +187,9 @@ */ @ApplicationScoped public class DefaultRequestHandler implements RequestHandler { + private static final int MAX_INTERRUPTED_ERROR_ENQUEUE_RETRIES = 1; + private static final long ERROR_ENQUEUE_TIMEOUT_SECONDS = 5; + private static final Logger LOGGER = LoggerFactory.getLogger(DefaultRequestHandler.class); @@ -1319,15 +1322,17 @@ public void run() { private void enqueueErrorPreservingInterrupt(AgentEmitter emitter, A2AError error) { boolean wasInterrupted = Thread.interrupted(); + int retries = 0; try { while (true) { try { - emitter.tryFail(error); + emitter.tryFailWithTimeout(error, ERROR_ENQUEUE_TIMEOUT_SECONDS, SECONDS); return; - } catch (RuntimeException e) { - if (!causedByInterruptedEnqueue(e)) { + } catch (EventQueue.EnqueueInterruptedException e) { + if (retries >= MAX_INTERRUPTED_ERROR_ENQUEUE_RETRIES) { throw e; } + retries++; wasInterrupted |= Thread.interrupted(); } } @@ -1338,15 +1343,6 @@ private void enqueueErrorPreservingInterrupt(AgentEmitter emitter, A2AError erro } } - private boolean causedByInterruptedEnqueue(Throwable error) { - for (Throwable cause = error; cause != null; cause = cause.getCause()) { - if (cause instanceof InterruptedException) { - return true; - } - } - return false; - } - private CompletableFuture cleanupProducer(@Nullable CompletableFuture agentFuture, @Nullable CompletableFuture consumptionFuture, String taskId, EventQueue queue, boolean isStreaming) { LOGGER.debug("Starting cleanup for task {} (streaming={})", taskId, isStreaming); logThreadStats("CLEANUP START"); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java index 36e61e897..2f5dcad8a 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/AgentEmitter.java @@ -3,6 +3,8 @@ import java.util.List; import java.util.Map; import java.util.UUID; +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicBoolean; import org.a2aproject.sdk.server.agentexecution.RequestContext; import org.slf4j.Logger; @@ -100,7 +102,7 @@ public class AgentEmitter { private final EventQueue eventQueue; private final String taskId; private final String contextId; - private boolean terminalStateReached; + private final AtomicBoolean terminalStateReached = new AtomicBoolean(false); /** * Creates a new AgentEmitter for the given request context and event queue. @@ -120,16 +122,16 @@ public AgentEmitter(RequestContext context, EventQueue eventQueue) { * @param taskState the new task state * @param message optional message to include with the status update */ - public synchronized void updateStatus(TaskState taskState, @Nullable Message message) { + public void updateStatus(TaskState taskState, @Nullable Message message) { boolean isFinal = taskState.isFinal(); // Check terminal state first (fail fast) - if (terminalStateReached) { + if (terminalStateReached.get()) { throw new IllegalStateException("Cannot update task status - terminal state already reached"); } - // For final states, record the terminal transition while holding this emitter's lock. - if (isFinal) { - terminalStateReached = true; + // Claim the terminal transition atomically without holding a lock during queue backpressure. + if (isFinal && !terminalStateReached.compareAndSet(false, true)) { + throw new IllegalStateException("Cannot update task status - terminal state already reached"); } TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder() @@ -147,8 +149,17 @@ public synchronized void updateStatus(TaskState taskState, @Nullable Message mes private void enqueueTerminalEvent(Event event) { try { eventQueue.enqueueEvent(event); - } catch (RuntimeException e) { - terminalStateReached = false; + } catch (RuntimeException | Error e) { + terminalStateReached.compareAndSet(true, false); + throw e; + } + } + + private void enqueueTerminalEvent(Event event, long timeout, TimeUnit unit) { + try { + eventQueue.enqueueEvent(event, timeout, unit); + } catch (RuntimeException | Error e) { + terminalStateReached.compareAndSet(true, false); throw e; } } @@ -301,16 +312,31 @@ public void fail(A2AError error) { * @param error the A2A error to enqueue * @return {@code true} if the error was enqueued, or {@code false} if a terminal state was already reached */ - public synchronized boolean tryFail(A2AError error) { - if (terminalStateReached) { + public boolean tryFail(A2AError error) { + if (!terminalStateReached.compareAndSet(false, true)) { return false; } - terminalStateReached = true; enqueueTerminalEvent(error); // Status transition happens automatically in MainEventBusProcessor return true; } + /** + * Attempts to fail the task, waiting no longer than the supplied duration for queue capacity. + * + * @param error the A2A error to enqueue + * @param timeout maximum time to wait for queue capacity + * @param unit timeout unit + * @return {@code true} if the error was enqueued, or {@code false} if a terminal state was already reached + */ + public boolean tryFailWithTimeout(A2AError error, long timeout, TimeUnit unit) { + if (!terminalStateReached.compareAndSet(false, true)) { + return false; + } + enqueueTerminalEvent(error, timeout, unit); + return true; + } + /** * Marks the task as SUBMITTED. */ 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 73759b331..95ffd0f7c 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 @@ -10,6 +10,7 @@ import static org.mockito.Mockito.mock; import static org.mockito.Mockito.when; +import java.util.Arrays; import java.util.List; import java.util.Map; import java.util.Set; @@ -1859,9 +1860,14 @@ public void onTaskFinalized(String taskId) { } private static boolean awaitThreadWaiting(Thread thread, boolean interrupted) throws InterruptedException { - long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(5); + long deadline = System.nanoTime() + TimeUnit.SECONDS.toNanos(30); do { - if (thread.getState() == Thread.State.WAITING && thread.isInterrupted() == interrupted) { + boolean waitingOrAcquiring = thread.getState() == Thread.State.WAITING + || Arrays.stream(thread.getStackTrace()).anyMatch(frame -> + frame.getClassName().equals("java.util.concurrent.Semaphore") + && (frame.getMethodName().equals("acquire") + || frame.getMethodName().equals("tryAcquire"))); + if (waitingOrAcquiring && thread.isInterrupted() == interrupted) { return true; } TimeUnit.MILLISECONDS.sleep(10); From 9f14dc20dceeb37c637556a4a37bcba976f57f41 Mon Sep 17 00:00:00 2001 From: Emmanuel Hugonnet Date: Mon, 28 Sep 2026 15:52:44 +0200 Subject: [PATCH 7/7] fix(server): fix EventQueue close races and resilient error enqueue - Invoke onClose() inside synchronized block for atomic state updates - Deduplicate isFinalEvent() into parent EventQueue - Skip closed child queues during event dispatch - Decrement submittedFinalEvents instead of resetting to zero - Reset pendingFinalEvents in clearAwaitingFinalEvent() - Handle enqueue timeout and max-retries gracefully in DefaultRequestHandler Signed-off-by: Emmanuel Hugonnet --- .../sdk/server/events/EventQueue.java | 98 ++++++++++--------- .../DefaultRequestHandler.java | 13 ++- 2 files changed, 65 insertions(+), 46 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 58db0a866..c018bee5a 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 @@ -453,9 +453,37 @@ protected void doClose(boolean immediate) { return; } LOGGER.debug("Closing {} (immediate={})", this, immediate); + // onClose is invoked while the lock is still held so that subclasses can update + // their own state (e.g. immediateClose + queue.clear()) atomically with closed=true. + onClose(immediate); closed = true; } - // Subclasses handle immediate close logic (e.g., ChildQueue clears its local queue) + } + + /** + * Called from within the {@code synchronized(this)} block of {@link #doClose(boolean)}, + * before {@code closed} is set to {@code true}. Subclasses may override to perform + * state updates that must be atomic with the close. The default implementation is a no-op. + * + * @param immediate whether this is an immediate (non-draining) close + */ + protected void onClose(boolean immediate) { + // no-op — subclasses override + } + + /** + * Returns whether an event represents a final task state. + */ + static boolean isFinalEvent(Event event) { + if (event instanceof Task task) { + return task.status() != null && task.status().state() != null + && task.status().state().isFinal(); + } else if (event instanceof TaskStatusUpdateEvent statusUpdate) { + return statusUpdate.isFinal(); + } else if (event instanceof A2AError) { + return true; + } + return false; } static class MainQueue extends EventQueue { @@ -676,21 +704,6 @@ private void validateEventIds(Event event) { } } - /** - * Checks if an event represents a final task state. - */ - private boolean isFinalEvent(Event event) { - if (event instanceof Task task) { - return task.status() != null && task.status().state() != null - && task.status().state().isFinal(); - } else if (event instanceof TaskStatusUpdateEvent statusUpdate) { - return statusUpdate.isFinal(); - } else if (event instanceof A2AError) { - return true; - } - return false; - } - @Override public void awaitQueuePollerStart() throws InterruptedException { LOGGER.debug("Waiting for queue poller to start on {}", this); @@ -756,6 +769,18 @@ 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). + if (child.isClosed()) { + LOGGER.debug("MainQueue[{}]: Skipping closed child queue for event {}", + taskId, item.getEvent().getClass().getSimpleName()); + return; + } LOGGER.debug("MainQueue[{}]: Enqueueing event {} to child queue", taskId, item.getEvent().getClass().getSimpleName()); child.internalEnqueueItem(item); @@ -911,11 +936,11 @@ private void internalEnqueueItem(EventQueueItem item) { } else { LOGGER.debug("Enqueued event {} {}", event instanceof Throwable ? event.toString() : event, this); - // If we were awaiting a final event and this is it, clear the flag + // If we were awaiting a final event and this is it, decrement the counter if (isFinalEvent(event)) { synchronized (this) { if (submittedFinalEvents > 0) { - submittedFinalEvents = 0; + submittedFinalEvents--; LOGGER.debug("ChildQueue {} received awaited final event", System.identityHashCode(this)); } } @@ -923,21 +948,6 @@ private void internalEnqueueItem(EventQueueItem item) { } } - /** - * Checks if an event represents a final task state. - */ - private boolean isFinalEvent(Event event) { - if (event instanceof Task task) { - return task.status() != null && task.status().state() != null - && task.status().state().isFinal(); - } else if (event instanceof TaskStatusUpdateEvent statusUpdate) { - return statusUpdate.isFinal(); - } else if (event instanceof A2AError) { - return true; - } - return false; - } - @Override public void enqueueLocalOnly(EventQueueItem item) { internalEnqueueItem(item); @@ -1012,21 +1022,18 @@ public void signalQueuePollerStarted() { } @Override - protected void doClose(boolean immediate) { - super.doClose(immediate); // Sets closed flag + protected void onClose(boolean immediate) { + // Invoked by EventQueue.doClose() while synchronized(this) is held, so closed, + // immediateClose, and the queue clear all become visible atomically. if (immediate) { - // Immediate close: clear pending events from local queue - int clearedCount; - synchronized (this) { - this.immediateClose = true; - pendingFinalEvents = 0; - submittedFinalEvents = 0; - clearedCount = queue.size(); - queue.clear(); - } + this.immediateClose = true; + pendingFinalEvents = 0; + submittedFinalEvents = 0; + int clearedCount = queue.size(); + queue.clear(); LOGGER.debug("Cleared {} events from ChildQueue for immediate close: {}", clearedCount, this); } - // For graceful close, let the queue drain naturally through normal consumption + // For graceful close the queue drains naturally; nothing to do here. } /** @@ -1070,6 +1077,7 @@ synchronized void cancelSubmittedFinalEvent() { */ @Override public synchronized void clearAwaitingFinalEvent() { + pendingFinalEvents = 0; submittedFinalEvents = 0; LOGGER.debug("ChildQueue {} cleared awaitingFinalEvent flag (timeout)", System.identityHashCode(this)); } diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index a0f4cee46..f9958ad62 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -1328,9 +1328,20 @@ private void enqueueErrorPreservingInterrupt(AgentEmitter emitter, A2AError erro try { emitter.tryFailWithTimeout(error, ERROR_ENQUEUE_TIMEOUT_SECONDS, SECONDS); return; + } catch (EventQueue.EnqueueTimeoutException e) { + // Queue stayed full for the entire timeout window — the infrastructure is + // under severe backpressure and we cannot deliver the terminal error event. + // Log and return so the agent thread exits cleanly; the EventConsumer will + // eventually close the stream via its normal timeout path. + LOGGER.warn("Timed out after {} s enqueueing error event for task after {} attempt(s) — task may not receive terminal status", + ERROR_ENQUEUE_TIMEOUT_SECONDS, retries + 1); + return; } catch (EventQueue.EnqueueInterruptedException e) { + wasInterrupted = true; if (retries >= MAX_INTERRUPTED_ERROR_ENQUEUE_RETRIES) { - throw e; + LOGGER.warn("Interrupted {} time(s) enqueueing error event for task — task may not receive terminal status", + retries + 1); + return; } retries++; wasInterrupted |= Thread.interrupted();