fix(client): deliver exactly one terminal SSE callback (#1170)

client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/sse/SSEEventListener.java

@@ -21,8 +21,6 @@
 public class SSEEventListener extends AbstractSSEEventListener {
 
     private static final Logger LOGGER = Logger.getLogger(SSEEventListener.class.getName());
-    private volatile boolean completed = false;
-
     public SSEEventListener(Consumer<StreamingEventKind> eventHandler,
             @Nullable Consumer<Throwable> errorHandler) {
         super(eventHandler, errorHandler);
@@ -34,21 +32,8 @@ public void onMessage(ServerSentEvent event, @Nullable Future<Void> completableF
     }
 
     public void onComplete() {
-        // Idempotent: only signal completion once, even if called multiple times
-        if (completed) {
-            LOGGER.fine("SSEEventListener.onComplete() called again - ignoring (already completed)");
-            return;
-        }
-        completed = true;
-
-        // Signal normal stream completion (null error means successful completion)
         LOGGER.fine("SSEEventListener.onComplete() called - signaling successful stream completion");
-        if (getErrorHandler() != null) {
-            LOGGER.fine("Calling errorHandler.accept(null) to signal successful completion");
-            getErrorHandler().accept(null);
-        } else {
-            LOGGER.warning("errorHandler is null, cannot signal completion");
-        }
+        signalTerminal(null);
     }
 
     /**
@@ -65,9 +50,7 @@ private void parseAndHandleMessage(String message, @Nullable Future<Void> future
             // Delegate to base class for common event handling and auto-close logic
             handleEvent(event, future);
         } catch (A2AError error) {
-            if (getErrorHandler() != null) {
-                getErrorHandler().accept(error);
-            }
+            signalTerminal(error);
         } catch (JsonProcessingException e) {
             throw new RuntimeException(e);
         }

client/transport/jsonrpc/src/test/java/org/a2aproject/sdk/client/transport/jsonrpc/sse/SSEEventListenerTest.java

@@ -4,12 +4,18 @@
 import static org.junit.jupiter.api.Assertions.assertFalse;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.ExecutionException;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
+import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicReference;
 
@@ -248,6 +254,128 @@ public void testOnEventWithFinalTaskStatusUpdateEventEventCancels() throws Excep
     }
 
 
+
+    @Test
+    public void testOnCompleteThenOnErrorDeliversSingleTerminalCallback() {
+        AtomicInteger terminalCount = new AtomicInteger(0);
+        AtomicReference<Throwable> lastArg = new AtomicReference<>();
+        SSEEventListener listener = new SSEEventListener(
+                event -> {},
+                error -> { terminalCount.incrementAndGet(); lastArg.set(error); });
+
+        listener.onComplete();
+        listener.onError(new RuntimeException("late error"), new CancelCapturingFuture());
+
+        assertEquals(1, terminalCount.get());
+        assertNull(lastArg.get());
+    }
+
+    @Test
+    public void testOnErrorThenOnCompleteDeliversSingleTerminalCallback() {
+        AtomicInteger terminalCount = new AtomicInteger(0);
+        AtomicReference<Throwable> lastArg = new AtomicReference<>();
+        SSEEventListener listener = new SSEEventListener(
+                event -> {},
+                error -> { terminalCount.incrementAndGet(); lastArg.set(error); });
+
+        RuntimeException boom = new RuntimeException("first error");
+        listener.onError(boom, new CancelCapturingFuture());
+        listener.onComplete();
+
+        assertEquals(1, terminalCount.get());
+        assertSame(boom, lastArg.get());
+    }
+
+    @Test
+    public void testRepeatedOnCompleteDeliversSingleTerminalCallback() {
+        AtomicInteger terminalCount = new AtomicInteger(0);
+        AtomicReference<Throwable> lastArg = new AtomicReference<>();
+        SSEEventListener listener = new SSEEventListener(
+                event -> {},
+                error -> { terminalCount.incrementAndGet(); lastArg.set(error); });
+
+        listener.onComplete();
+        listener.onComplete();
+        listener.onComplete();
+
+        assertEquals(1, terminalCount.get());
+    }
+
+    @Test
+    public void testConcurrentTerminalSignalsDeliverExactlyOneCallback() throws Exception {
+        AtomicInteger terminalCount = new AtomicInteger(0);
+        AtomicReference<Throwable> lastArg = new AtomicReference<>();
+        SSEEventListener listener = new SSEEventListener(
+                event -> {},
+                error -> { terminalCount.incrementAndGet(); lastArg.set(error); });
+
+        int threads = 32;
+        ExecutorService pool = Executors.newFixedThreadPool(threads);
+        CountDownLatch start = new CountDownLatch(1);
+        CountDownLatch done = new CountDownLatch(threads);
+        for (int i = 0; i < threads; i++) {
+            final boolean complete = (i % 2 == 0);
+            pool.submit(() -> {
+                try {
+                    start.await();
+                    if (complete) {
+                        listener.onComplete();
+                    } else {
+                        listener.onError(new RuntimeException("race"), new CancelCapturingFuture());
+                    }
+                } catch (InterruptedException e) {
+                    Thread.currentThread().interrupt();
+                } finally {
+                    done.countDown();
+                }
+            });
+        }
+        start.countDown();
+        assertTrue(done.await(10, TimeUnit.SECONDS));
+        pool.shutdownNow();
+
+        assertEquals(1, terminalCount.get());
+    }
+
+    @Test
+    public void testFinalEventThenOnCompleteDeliversSingleTerminalCallback() throws Exception {
+        AtomicInteger terminalCount = new AtomicInteger(0);
+        SSEEventListener listener = new SSEEventListener(
+                event -> {},
+                error -> terminalCount.incrementAndGet());
+
+        String eventData = JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.substring(
+                JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.indexOf("{"));
+        CancelCapturingFuture future = new CancelCapturingFuture();
+        listener.onMessage(new ServerSentEvent(eventData), future);
+        listener.onComplete();
+
+        assertTrue(future.cancelHandlerCalled);
+        assertEquals(1, terminalCount.get());
+    }
+
+    @Test
+    public void testFinalEventThenOnErrorDeliversNormalCompletionOnly() {
+        AtomicInteger terminalCount = new AtomicInteger(0);
+        AtomicReference<Throwable> terminalError = new AtomicReference<>();
+        SSEEventListener listener = new SSEEventListener(
+                event -> {},
+                error -> {
+                    terminalCount.incrementAndGet();
+                    terminalError.set(error);
+                });
+
+        String eventData = JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.substring(
+                JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.indexOf("{"));
+        CancelCapturingFuture future = new CancelCapturingFuture();
+        listener.onMessage(new ServerSentEvent(eventData), future);
+        listener.onError(new RuntimeException("cancelled after final event"), future);
+
+        assertTrue(future.cancelHandlerCalled);
+        assertEquals(1, terminalCount.get());
+        assertNull(terminalError.get());
+    }
+
     private static class CancelCapturingFuture implements Future<Void> {
         private boolean cancelHandlerCalled;
 
@@ -280,4 +408,4 @@ public Void get(long timeout, TimeUnit unit) throws InterruptedException, Execut
             return null;
         }
     }
-}
+}

client/transport/spi/src/main/java/org/a2aproject/sdk/client/transport/spi/sse/AbstractSSEEventListener.java

@@ -1,6 +1,7 @@
 package org.a2aproject.sdk.client.transport.spi.sse;
 
 import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.logging.Logger;
 
@@ -25,6 +26,7 @@ public abstract class AbstractSSEEventListener {
 
     private final Consumer<StreamingEventKind> eventHandler;
     private final @Nullable Consumer<Throwable> errorHandler;
+    private final AtomicBoolean terminalSignaled = new AtomicBoolean(false);
 
     /**
      * Creates a new SSE event listener with the specified handlers.
@@ -73,14 +75,33 @@ protected Consumer<StreamingEventKind> getEventHandler() {
      * @param future Optional future for closing the SSE connection
      */
     public void onError(Throwable throwable, @Nullable Future<Void> future) {
-        if (errorHandler != null) {
-            errorHandler.accept(throwable);
-        }
+        signalTerminal(throwable);
         if (future != null) {
             future.cancel(true); // close SSE channel
         }
     }
 
+    /**
+     * Delivers exactly one terminal callback for the stream. The first caller to win
+     * the atomic transition delivers its outcome to the error/completion consumer (a
+     * non-null {@code error} is a failure, {@code null} is normal completion); every
+     * later completion, error or post-cancellation signal is dropped, so a single
+     * streaming request yields exactly one terminal callback.
+     *
+     * @param error the failure to report, or {@code null} to signal normal completion
+     */
+    protected void signalTerminal(@Nullable Throwable error) {
+        if (!terminalSignaled.compareAndSet(false, true)) {
+            LOGGER.fine("Terminal callback already delivered, ignoring subsequent signal");
+            return;
+        }
+        if (errorHandler != null) {
+            errorHandler.accept(error);
+        } else if (error != null) {
+            LOGGER.warning("errorHandler is null, cannot report terminal error");
+        }
+    }
+
     /**
      * Processes a parsed streaming event and handles auto-close logic for final events.
      * This method encapsulates the common logic for handling events and determining
@@ -97,6 +118,7 @@ protected void handleEvent(StreamingEventKind event, @Nullable Future<Void> futu
         // This covers late subscriptions to completed tasks and ensures no connection leaks
         if (shouldAutoClose(event) && future != null) {
             LOGGER.fine("Auto-closing SSE connection for final event: " + event.getClass().getSimpleName());
+            signalTerminal(null);
             future.cancel(true); // close SSE channel
         }
     }

compat-0.3/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/jsonrpc/sse/SSEEventListener_v0_3.java

@@ -10,14 +10,15 @@
 import org.a2aproject.sdk.compat03.spec.TaskStatusUpdateEvent_v0_3;
 
 import java.util.concurrent.Future;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.function.Consumer;
 import java.util.logging.Logger;
 
 public class SSEEventListener_v0_3 {
     private static final Logger LOGGER = Logger.getLogger(SSEEventListener_v0_3.class.getName());
     private final Consumer<StreamingEventKind_v0_3> eventHandler;
     private final Consumer<Throwable> errorHandler;
-    private volatile boolean completed = false;
+    private final AtomicBoolean terminalSignaled = new AtomicBoolean(false);
 
     public SSEEventListener_v0_3(Consumer<StreamingEventKind_v0_3> eventHandler,
                                  Consumer<Throwable> errorHandler) {
@@ -34,50 +35,44 @@ public void onMessage(String message, Future<Void> completableFuture) {
             LOGGER.warning("Failed to process JSON message: " + message);
         } catch (IllegalArgumentException e) {
             LOGGER.warning("Invalid message format: " + message);
-            if (errorHandler != null) {
-                errorHandler.accept(e);
-            }
+            signalTerminal(e);
             completableFuture.cancel(true); // close SSE channel
         }
     }
 
     public void onError(Throwable throwable, Future<Void> future) {
-        if (errorHandler != null) {
-            errorHandler.accept(throwable);
-        }
+        signalTerminal(throwable);
         future.cancel(true); // close SSE channel
     }
 
-    public void onComplete() {
-        // Idempotent: only signal completion once, even if called multiple times
-        if (completed) {
-            LOGGER.fine("SSEEventListener.onComplete() called again - ignoring (already completed)");
+    private void signalTerminal(Throwable error) {
+        if (!terminalSignaled.compareAndSet(false, true)) {
+            LOGGER.fine("Terminal callback already delivered, ignoring subsequent signal");
             return;
         }
-        completed = true;
-
-        // Signal normal stream completion (null error means successful completion)
-        LOGGER.fine("SSEEventListener.onComplete() called - signaling successful stream completion");
         if (errorHandler != null) {
-            LOGGER.fine("Calling errorHandler.accept(null) to signal successful completion");
-            errorHandler.accept(null);
-        } else {
-            LOGGER.warning("errorHandler is null, cannot signal completion");
+            errorHandler.accept(error);
+        } else if (error != null) {
+            LOGGER.warning("errorHandler is null, cannot report terminal error");
         }
     }
 
+    public void onComplete() {
+        LOGGER.fine("SSEEventListener.onComplete() called - signaling successful stream completion");
+        signalTerminal(null);
+    }
+
     private void handleMessage(JsonObject jsonObject, Future<Void> future) throws JsonProcessingException_v0_3 {
         if (jsonObject.has("error")) {
             JSONRPCError_v0_3 error = JsonUtil_v0_3.fromJson(jsonObject.get("error").toString(), JSONRPCError_v0_3.class);
-            if (errorHandler != null) {
-                errorHandler.accept(error);
-            }
+            signalTerminal(error);
         } else if (jsonObject.has("result")) {
             // result can be a Task, Message, TaskStatusUpdateEvent, or TaskArtifactUpdateEvent
             String resultJson = jsonObject.get("result").toString();
             StreamingEventKind_v0_3 event = JsonUtil_v0_3.fromJson(resultJson, StreamingEventKind_v0_3.class);
             eventHandler.accept(event);
             if (event instanceof TaskStatusUpdateEvent_v0_3 tsue && tsue.isFinal()) {
+                signalTerminal(null);
                 future.cancel(true); // close SSE channel
             }
         } else {