fix(server): preserve interrupts for protocol errors

This fixes #1183

Author: kabir

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();

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<? extends A2AError> 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<A2AError> 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 {