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

- Per-event child spans and terminal closing span for streaming methods, linked to the initial method span
- Duration histograms: gen_ai.client.operation.duration (client) and gen_ai.agent.a2a.streaming.duration (server)

Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>

Author: ehsavoie

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/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();
+    }
+}

docs/content/dev/extra/opentelemetry.md

@@ -14,6 +14,7 @@ Adds distributed tracing and context propagation to A2A servers and clients usin
 - **Context Propagation**: OpenTelemetry trace context propagation across async operations
 - **Request/Response Logging**: Optional extraction of request and response data into spans
 - **Error Tracking**: Automatic error status and error type attributes on failures
+- **Metrics**: `gen_ai.client.operation.duration` histogram on the client side; `gen_ai.agent.a2a.streaming.duration` histogram for full streaming lifecycle on the server side
 
 ## Modules
 
@@ -65,6 +66,76 @@ Async agent execution (maintains trace context)
 Response (with trace headers)
 ```
 
+### Streaming Span Lifecycle
+
+Streaming methods (`message/stream` and `tasks/resubscribe`) follow a two-span model to correctly represent the full duration of a streaming response.
+
+**Span 1 — method span** is created when the request arrives and ended as soon as the publisher is returned to the transport layer. It captures request attributes and the initial outcome (success or synchronous error).
+
+**Per-event child spans** (`<method>-event`) are created for each streaming event published via `onNext()`. Each child span carries a [link](https://opentelemetry.io/docs/concepts/signals/traces/#span-links) back to Span 1 and is made the current span during delegate execution, so any nested telemetry is correctly parented.
+
+**Span 2 — end span** (`<method>-end`) is created when the stream terminates — either on completion or error. It also carries a link to Span 1 for correlation.
+
+<pre class="mermaid">
+sequenceDiagram
+    participant C as Client
+    participant D as OTel Decorator
+    participant H as RequestHandler
+    participant W as WrappedPublisher
+
+    C->>D: onMessageSendStream(params)
+    D->>D: create Span₁ (SERVER)
+    D->>H: onMessageSendStream(params)
+    H-->>D: publisher
+    D->>D: capture Span₁.context
+    D->>D: Span₁.setStatus(OK) · Span₁.end()
+    D-->>C: WrappedPublisher
+
+    note over D: Span₁ closed immediately
+
+    loop for each streaming event
+        W->>W: create child span (link→Span₁)
+        W-->>C: onNext(event)
+        W->>W: child span.end()
+    end
+
+    alt stream completes normally
+        W->>W: create Span₂ (SERVER, link→Span₁)
+        W->>W: Span₂.setStatus(OK) · Span₂.end()
+        W-->>C: onComplete()
+    else stream errors
+        W->>W: create Span₂ (SERVER, link→Span₁)
+        W->>W: Span₂.setStatus(ERROR) · Span₂.end()
+        W-->>C: onError(throwable)
+    end
+</pre>
+
+Each per-event child span carries:
+
+| Attribute | Value |
+|-----------|-------|
+| `gen_ai.agent.a2a.streaming-event` | Event kind identifier (`"message"`, `"task"`, `"statusUpdate"`, `"artifactUpdate"`) |
+| `gen_ai.agent.a2a.task.status.state` | Task status state (for `TaskStatusUpdateEvent` and `Task` events) |
+| `gen_ai.agent.a2a.task_id` | Task identifier (when available from the event) |
+
+### Metrics
+
+The server decorator records the following histogram instruments:
+
+| Instrument | Type | Unit | Description |
+|------------|------|------|-------------|
+| `gen_ai.agent.a2a.streaming.duration` | Histogram | `s` | Duration from stream initiation to last event, for `message/stream` and `tasks/resubscribe` |
+
+Attributes on `gen_ai.agent.a2a.streaming.duration`:
+
+| Attribute | Value |
+|-----------|-------|
+| `gen_ai.agent.a2a.operation.name` | A2A method name (`message/stream` or `tasks/resubscribe`) |
+| `gen_ai.system` | `a2a` |
+| `error.type` | `error` (only on stream failure) |
+
+Meter injection uses CDI `Instance<Meter>`, so the histogram is silently skipped if no `Meter` bean is available in the container.
+
 ### Context-Aware Async Executor
 
 > **Note:** The `AsyncManagedExecutorProducer` is provided by the **Quarkus reference server** ([`reference/common`](https://github.com/a2aproject/a2a-java/blob/main/reference/common/src/main/java/org/a2aproject/sdk/server/common/quarkus/AsyncManagedExecutorProducer.java)), not the OpenTelemetry module. It is documented here because it enables context propagation (including trace context) across async boundaries.
@@ -133,6 +204,30 @@ Adds OpenTelemetry spans to A2A client operations:
 </dependency>
 ```
 
+### Metrics
+
+The client transport records the following histogram:
+
+| Instrument | Type | Unit | Description |
+|------------|------|------|-------------|
+| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of A2A client operations from invocation to completion or error |
+
+Attributes on `gen_ai.client.operation.duration`:
+
+| Attribute | Value |
+|-----------|-------|
+| `gen_ai.agent.a2a.operation.name` | A2A method name (e.g., `message/send`, `tasks/get`) |
+| `gen_ai.system` | `a2a` |
+| `error.type` | `error` (only on failure) |
+
+Metrics require passing a `Meter` to the constructor:
+
+```java
+OpenTelemetryClientTransport transport = new OpenTelemetryClientTransport(delegate, tracer, meter);
+```
+
+If `meter` is `null`, the histogram is silently skipped.
+
 ### Context Propagation
 
 For trace context propagation across async client boundaries:

examples/helloworld/client/README.md

@@ -161,11 +161,11 @@ The Java client (`HelloWorldClient.java`) performs the following actions:
 
 ## Enable OpenTelemetry (Optional)
 
-The client includes support for distributed tracing with OpenTelemetry. To enable it:
+The client includes support for distributed tracing and metrics with OpenTelemetry. To enable it:
 
 ### Prerequisites
 
-**IMPORTANT**: The client expects an OpenTelemetry collector to be ready and accepting traces. You have two options:
+**IMPORTANT**: The client expects an OpenTelemetry collector to be ready and accepting traces and metrics. You have two options:
 
 #### Option 1: Use the Java Server Example (Recommended)
 

examples/helloworld/client/src/main/java/org/a2aproject/sdk/examples/helloworld/client/HelloWorldClient.java

@@ -1,5 +1,6 @@
 package org.a2aproject.sdk.examples.helloworld.client;
 
+import static org.a2aproject.sdk.extras.opentelemetry.client.OpenTelemetryClientTransportWrapper.OTEL_METER_KEY;
 import static org.a2aproject.sdk.extras.opentelemetry.client.OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY;
 import static org.a2aproject.sdk.extras.opentelemetry.client.propagation.OpenTelemetryClientPropagatorTransportWrapper.OTEL_OPEN_TELEMETRY_KEY;
 
@@ -34,8 +35,11 @@
 import io.grpc.ManagedChannelBuilder;
 import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
 import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.exporter.otlp.metrics.OtlpGrpcMetricExporter;
 import io.opentelemetry.exporter.otlp.trace.OtlpGrpcSpanExporter;
 import io.opentelemetry.sdk.OpenTelemetrySdk;
+import io.opentelemetry.sdk.metrics.SdkMeterProvider;
+import io.opentelemetry.sdk.metrics.export.PeriodicMetricReader;
 import io.opentelemetry.sdk.resources.Resource;
 import io.opentelemetry.sdk.trace.SdkTracerProvider;
 import io.opentelemetry.sdk.trace.export.BatchSpanProcessor;
@@ -136,20 +140,32 @@ public static void main(String[] args) {
     }
 
     static OpenTelemetrySdk initOpenTelemetry() {
+        Resource resource = Resource.getDefault().toBuilder()
+                .put("service.version", "1.0")
+                .put("service.name", "helloworld-client")
+                .build();
+
         SdkTracerProvider sdkTracerProvider = SdkTracerProvider.builder()
                 .addSpanProcessor(BatchSpanProcessor.builder(
                         OtlpGrpcSpanExporter.builder()
                                 .setEndpoint("http://localhost:5317")
                                 .build()
                 ).build())
-                .setResource(Resource.getDefault().toBuilder()
-                        .put("service.version", "1.0")
-                        .put("service.name", "helloworld-client")
-                        .build())
+                .setResource(resource)
+                .build();
+
+        SdkMeterProvider sdkMeterProvider = SdkMeterProvider.builder()
+                .registerMetricReader(PeriodicMetricReader.builder(
+                        OtlpGrpcMetricExporter.builder()
+                                .setEndpoint("http://localhost:5317")
+                                .build()
+                ).build())
+                .setResource(resource)
                 .build();
 
         return OpenTelemetrySdk.builder()
                 .setTracerProvider(sdkTracerProvider)
+                .setMeterProvider(sdkMeterProvider)
                 .setPropagators(ContextPropagators.create(W3CTraceContextPropagator.getInstance()))
                 .build();
     }
@@ -186,6 +202,7 @@ private static void updateTransportConfig(ClientTransportConfig transportConfig,
         if (openTelemetrySdk != null) {
             Map<String, Object> parameters = new HashMap<>(transportConfig.getParameters());
             parameters.put(OTEL_TRACER_KEY, openTelemetrySdk.getTracer("helloworld-client"));
+            parameters.put(OTEL_METER_KEY, openTelemetrySdk.getMeter("helloworld-client"));
             parameters.put(OTEL_OPEN_TELEMETRY_KEY, openTelemetrySdk);
             transportConfig.setParameters(parameters);
         }

examples/helloworld/server/src/main/resources/application.properties

@@ -5,6 +5,7 @@ quarkus.agentcard.protocol=JSONRPC
 
 # OpenTelemetry configuration
 quarkus.otel.sdk.disabled=false
+quarkus.otel.metrics.enabled=true
 quarkus.observability.lgtm.grafana-port=3001
 quarkus.observability.lgtm.otel-grpc-port=5317
 quarkus.observability.lgtm.otel-http-port=5318

extras/opentelemetry/client/pom.xml

@@ -28,6 +28,12 @@
             <groupId>org.slf4j</groupId>
             <artifactId>slf4j-api</artifactId>
         </dependency>
+        <dependency>
+            <groupId>${project.groupId}</groupId>
+            <artifactId>a2a-java-sdk-opentelemetry-common</artifactId>
+            <classifier>tests</classifier>
+            <scope>test</scope>
+        </dependency>
         <dependency>
             <groupId>org.junit.jupiter</groupId>
             <artifactId>junit-jupiter</artifactId>