feat(opentelemetry): add streaming span lifecycle and operation metrics

client/transport/grpc/src/main/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserver.java

@@ -9,14 +9,15 @@
 import org.a2aproject.sdk.grpc.StreamResponse;
 import org.a2aproject.sdk.spec.StreamingEventKind;
 import io.grpc.stub.StreamObserver;
+import org.jspecify.annotations.Nullable;
 
 public class EventStreamObserver implements StreamObserver<StreamResponse> {
 
     private static final Logger LOGGER = Logger.getLogger(EventStreamObserver.class.getName());
     private final Consumer<StreamingEventKind> eventHandler;
-    private final Consumer<Throwable> errorHandler;
+    @Nullable private final Consumer<Throwable> errorHandler;
 
-    public EventStreamObserver(Consumer<StreamingEventKind> eventHandler, Consumer<Throwable> errorHandler) {
+    public EventStreamObserver(Consumer<StreamingEventKind> eventHandler, @Nullable Consumer<Throwable> errorHandler) {
         this.eventHandler = eventHandler;
         this.errorHandler = errorHandler;
     }
@@ -39,7 +40,9 @@ public void onNext(StreamResponse response) {
                 break;
             default:
                 LOGGER.warning("Invalid stream response " + response.getPayloadCase());
-                errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
+                if (errorHandler != null) {
+                    errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
+                }
                 return;
         }
         eventHandler.accept(event);
@@ -59,6 +62,8 @@ public void onError(Throwable t) {
 
     @Override
     public void onCompleted() {
-        // done
+        if (errorHandler != null) {
+            errorHandler.accept(null);
+        }
     }
 }

client/transport/grpc/src/test/java/org/a2aproject/sdk/client/transport/grpc/EventStreamObserverTest.java

@@ -0,0 +1,37 @@
+package org.a2aproject.sdk.client.transport.grpc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import org.a2aproject.sdk.spec.StreamingEventKind;
+import org.junit.jupiter.api.Test;
+
+class EventStreamObserverTest {
+
+    @Test
+    void onCompleted_callsErrorHandlerWithNull() {
+        List<StreamingEventKind> events = new ArrayList<>();
+        List<Throwable> errors = new ArrayList<>();
+
+        Consumer<StreamingEventKind> eventHandler = events::add;
+        Consumer<Throwable> errorHandler = errors::add;
+        EventStreamObserver observer = new EventStreamObserver(eventHandler, errorHandler);
+
+        observer.onCompleted();
+
+        assertEquals(1, errors.size(), "errorHandler should have been called once");
+        assertNull(errors.get(0), "errorHandler should be called with null to signal clean completion");
+        assertEquals(0, events.size(), "eventHandler should not be called");
+    }
+
+    @Test
+    void onCompleted_withNullErrorHandler_doesNotThrow() {
+        Consumer<StreamingEventKind> eventHandler = e -> {};
+        EventStreamObserver observer = new EventStreamObserver(eventHandler, null);
+
+        observer.onCompleted();
+    }
+}

client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/RestTransport.java

@@ -121,7 +121,8 @@ public void sendMessageStreaming(MessageSendParams messageSendParams, Consumer<S
                     event -> sseEventListener.onMessage(event, ref.get()),
                     throwable -> sseEventListener.onError(throwable, ref.get()),
                     () -> {
-                        // We don't need to do anything special on completion
+                        // Signal normal stream completion to error handler (null error means success)
+                        sseEventListener.onComplete();
                     }));
         } catch (IOException e) {
             throw new A2AClientException("Failed to send streaming message request: " + e, e);
@@ -381,7 +382,8 @@ public void subscribeToTask(TaskIdParams request, Consumer<StreamingEventKind> e
                     event -> sseEventListener.onMessage(event, ref.get()),
                     throwable -> sseEventListener.onError(throwable, ref.get()),
                     () -> {
-                        // We don't need to do anything special on completion
+                        // Signal normal stream completion to error handler (null error means success)
+                        sseEventListener.onComplete();
                     }));
         } catch (IOException e) {
             throw new A2AClientException("Failed to send streaming message request: " + e, e);

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

@@ -27,6 +27,11 @@ public SSEEventListener(Consumer<StreamingEventKind> eventHandler,
         super(eventHandler, errorHandler);
     }
 
+    public void onComplete() {
+        LOGGER.fine("SSEEventListener.onComplete() called - signaling successful stream completion");
+        signalTerminal(null);
+    }
+
     @Override
     public void onMessage(ServerSentEvent event, @Nullable Future<Void> completableFuture) {
         try {
@@ -35,9 +40,7 @@ public void onMessage(ServerSentEvent event, @Nullable Future<Void> completableF
             JsonFormat.parser().merge(event.data(), builder);
             parseAndHandleMessage(builder.build(), completableFuture);
         } catch (InvalidProtocolBufferException e) {
-            if (getErrorHandler() != null) {
-                getErrorHandler().accept(RestErrorMapper.mapRestError(event.data(), 500));
-            }
+            signalTerminal(RestErrorMapper.mapRestError(event.data(), 500));
         }
     }
 
@@ -60,9 +63,7 @@ private void parseAndHandleMessage(StreamResponse response, @Nullable Future<Voi
                 event = ProtoUtils.FromProto.taskArtifactUpdateEvent(response.getArtifactUpdate());
             default -> {
                 LOGGER.warning("Invalid stream response " + response.getPayloadCase());
-                if (getErrorHandler() != null) {
-                    getErrorHandler().accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
-                }
+                signalTerminal(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
                 return;
             }
         }

client/transport/rest/src/test/java/org/a2aproject/sdk/client/transport/rest/RestTransportTest.java

@@ -69,6 +69,7 @@
 public class RestTransportTest {
 
     private static final Logger LOGGER = Logger.getLogger(RestTransportTest.class.getName());
+    private static final Throwable SENTINEL = new RuntimeException("not yet called");
     private ClientAndServer server;
     private static final AgentCard CARD = AgentCard.builder()
                 .name("Hello World Agent")
@@ -474,6 +475,89 @@ public void testSubscribe() throws Exception {
         assertEquals("Why did the chicken cross the road? To get to the other side!", ((TextPart) part).text());
     }
 
+    /**
+     * Test that the error handler is called with null on normal stream completion for sendMessageStreaming.
+     */
+    @Test
+    public void testSendMessageStreamingCompletion() throws Exception {
+        this.server.when(
+                request()
+                        .withMethod("POST")
+                        .withPath("/message:stream")
+                        .withBody(JsonBody.json(SEND_MESSAGE_STREAMING_TEST_REQUEST, MatchType.ONLY_MATCHING_FIELDS))
+        )
+                .respond(
+                        response()
+                                .withStatusCode(200)
+                                .withHeader("Content-Type", "text/event-stream")
+                                .withBody(SEND_MESSAGE_STREAMING_TEST_RESPONSE)
+                );
+
+        RestTransport client = new RestTransport(CARD);
+        Message message = Message.builder()
+                .role(Message.Role.ROLE_USER)
+                .parts(Collections.singletonList(new TextPart("tell me some jokes")))
+                .contextId("context-1234")
+                .messageId("message-1234")
+                .build();
+        MessageSendConfiguration configuration = MessageSendConfiguration.builder()
+                .acceptedOutputModes(List.of("text"))
+                .returnImmediately(true)
+                .build();
+        MessageSendParams params = MessageSendParams.builder()
+                .message(message)
+                .configuration(configuration)
+                .build();
+        CountDownLatch eventLatch = new CountDownLatch(1);
+        CountDownLatch completionLatch = new CountDownLatch(1);
+        AtomicReference<Throwable> completionValue = new AtomicReference<>(SENTINEL);
+        Consumer<StreamingEventKind> eventHandler = event -> eventLatch.countDown();
+        Consumer<Throwable> errorHandler = error -> {
+            completionValue.set(error);
+            completionLatch.countDown();
+        };
+        client.sendMessageStreaming(params, eventHandler, errorHandler, null);
+
+        assertTrue(eventLatch.await(10, TimeUnit.SECONDS), "Should receive streaming event");
+        assertTrue(completionLatch.await(10, TimeUnit.SECONDS), "Error handler should be called on stream completion");
+        assertNull(completionValue.get(), "Error handler should be called with null for successful completion");
+    }
+
+    /**
+     * Test that the error handler is called with null on normal stream completion for subscribeToTask.
+     */
+    @Test
+    public void testSubscribeToTaskCompletion() throws Exception {
+        this.server.when(
+                        request()
+                                .withMethod("POST")
+                                .withPath("/tasks/task-1234:subscribe")
+                )
+                .respond(
+                        response()
+                                .withStatusCode(200)
+                                .withHeader("Content-Type", "text/event-stream")
+                                .withBody(TASK_RESUBSCRIPTION_REQUEST_TEST_RESPONSE)
+                );
+
+        RestTransport client = new RestTransport(CARD);
+        TaskIdParams taskIdParams = new TaskIdParams("task-1234");
+
+        CountDownLatch eventLatch = new CountDownLatch(1);
+        CountDownLatch completionLatch = new CountDownLatch(1);
+        AtomicReference<Throwable> completionValue = new AtomicReference<>(SENTINEL);
+        Consumer<StreamingEventKind> eventHandler = event -> eventLatch.countDown();
+        Consumer<Throwable> errorHandler = error -> {
+            completionValue.set(error);
+            completionLatch.countDown();
+        };
+        client.subscribeToTask(taskIdParams, eventHandler, errorHandler, null);
+
+        assertTrue(eventLatch.await(10, TimeUnit.SECONDS), "Should receive streaming event");
+        assertTrue(completionLatch.await(10, TimeUnit.SECONDS), "Error handler should be called on stream completion");
+        assertNull(completionValue.get(), "Error handler should be called with null for successful completion");
+    }
+
     /**
      * Test that ExtensionSupportRequiredError is properly unmarshalled from REST error response.
      */

client/transport/spi/src/main/java/org/a2aproject/sdk/client/transport/spi/ClientTransport.java

@@ -43,7 +43,8 @@ EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext con
      *
      * @param request       the message send parameters
      * @param eventConsumer consumer that will receive streaming events as they arrive
-     * @param errorConsumer consumer that will be called if an error occurs during streaming
+     * @param errorConsumer consumer called when the stream ends; called with {@code null} to signal
+     *                      clean completion, or with a non-null {@link Throwable} if an error occurs
      * @param context       optional client call context for the request (may be {@code null})
      * @throws A2AClientException if setting up the streaming connection fails
      */
@@ -131,7 +132,8 @@ void deleteTaskPushNotificationConfigurations(
      *
      * @param request       the task ID parameters specifying which task to subscribe to
      * @param eventConsumer consumer that will receive streaming events as they arrive
-     * @param errorConsumer consumer that will be called if an error occurs during streaming
+     * @param errorConsumer consumer called when the stream ends; called with {@code null} to signal
+     *                      clean completion, or with a non-null {@link Throwable} if an error occurs
      * @param context       optional client call context for the request (may be {@code null})
      * @throws A2AClientException if subscribing to the task fails for any reason
      */

compat-0.3/client/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3.java

@@ -8,15 +8,16 @@
 import java.util.function.Consumer;
 import java.util.logging.Logger;
 
+import org.jspecify.annotations.Nullable;
 import static org.a2aproject.sdk.compat03.grpc.utils.ProtoUtils_v0_3.FromProto;
 
 public class EventStreamObserver_v0_3 implements StreamObserver<StreamResponse> {
 
     private static final Logger LOGGER = Logger.getLogger(EventStreamObserver_v0_3.class.getName());
     private final Consumer<StreamingEventKind_v0_3> eventHandler;
-    private final Consumer<Throwable> errorHandler;
+    @Nullable private final Consumer<Throwable> errorHandler;
 
-    public EventStreamObserver_v0_3(Consumer<StreamingEventKind_v0_3> eventHandler, Consumer<Throwable> errorHandler) {
+    public EventStreamObserver_v0_3(Consumer<StreamingEventKind_v0_3> eventHandler, @Nullable Consumer<Throwable> errorHandler) {
         this.eventHandler = eventHandler;
         this.errorHandler = errorHandler;
     }
@@ -39,7 +40,9 @@ public void onNext(StreamResponse response) {
                 break;
             default:
                 LOGGER.warning("Invalid stream response " + response.getPayloadCase());
-                errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
+                if (errorHandler != null) {
+                    errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
+                }
                 return;
         }
         eventHandler.accept(event);
@@ -59,6 +62,8 @@ public void onError(Throwable t) {
 
     @Override
     public void onCompleted() {
-        // done
+        if (errorHandler != null) {
+            errorHandler.accept(null);
+        }
     }
 }

compat-0.3/client/transport/grpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/grpc/EventStreamObserver_v0_3_Test.java

@@ -0,0 +1,37 @@
+package org.a2aproject.sdk.compat03.client.transport.grpc;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNull;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.function.Consumer;
+import org.a2aproject.sdk.compat03.spec.StreamingEventKind_v0_3;
+import org.junit.jupiter.api.Test;
+
+class EventStreamObserver_v0_3_Test {
+
+    @Test
+    void onCompleted_callsErrorHandlerWithNull() {
+        List<StreamingEventKind_v0_3> events = new ArrayList<>();
+        List<Throwable> errors = new ArrayList<>();
+
+        Consumer<StreamingEventKind_v0_3> eventHandler = events::add;
+        Consumer<Throwable> errorHandler = errors::add;
+        EventStreamObserver_v0_3 observer = new EventStreamObserver_v0_3(eventHandler, errorHandler);
+
+        observer.onCompleted();
+
+        assertEquals(1, errors.size(), "errorHandler should have been called once");
+        assertNull(errors.get(0), "errorHandler should be called with null to signal clean completion");
+        assertEquals(0, events.size(), "eventHandler should not be called");
+    }
+
+    @Test
+    void onCompleted_withNullErrorHandler_doesNotThrow() {
+        Consumer<StreamingEventKind_v0_3> eventHandler = e -> {};
+        EventStreamObserver_v0_3 observer = new EventStreamObserver_v0_3(eventHandler, null);
+
+        observer.onCompleted();
+    }
+}