feat(opentelemetry): add push notification tracing with CLIENT spans (#1228)

Add OpenTelemetryPushNotificationSenderDecorator that wraps
PushNotificationSender to create a CLIENT span for every push delivery.
The span carries task_id, context_id, push_notification.event_kind, and
optional response payload attributes, with error.type on delivery failure.

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

Author: ehsavoie

docs/content/dev/extra/opentelemetry.md

@@ -106,6 +106,21 @@ The following attributes are automatically added to spans (defined in `A2AObserv
 | `gen_ai.agent.a2a.response` | Full response data (only if extraction enabled) |
 | `error.type` | Error message (on failures) |
 
+### Push Notification Delivery Tracing
+
+When the server delivers a push notification to a client webhook, `OpenTelemetryPushNotificationSenderDecorator` creates a `CLIENT` span named `SendPushNotification`. This span is independent of the original request span but carries the `gen_ai.agent.a2a.task_id` attribute, allowing correlation by task ID in your observability tool.
+
+The module also provides **W3C trace-context propagation** for outbound push notification HTTP requests. When the OpenTelemetry server module is on the classpath, the `OpenTelemetryHttpClientProducer` CDI bean wraps the HTTP client used by `BasePushNotificationSender` with `OpenTelemetryA2AHttpClient`, which injects `traceparent` and `tracestate` headers into every push notification delivery. It also produces a context-propagating `Executor` (via `Context.taskWrapping()`) that ensures the OTel trace context survives the async boundary when push notifications are dispatched to multiple recipients in parallel. This allows the receiving webhook to join the distributed trace.
+
+| Attribute | Description |
+|-----------|-------------|
+| `gen_ai.agent.a2a.operation.name` | `SendPushNotification` |
+| `gen_ai.agent.a2a.task_id` | Task identifier |
+| `gen_ai.agent.a2a.context_id` | Context identifier (when available) |
+| `gen_ai.agent.a2a.push_notification.event_kind` | The type of streaming event delivered (e.g. `TaskStatusUpdateEvent`) |
+| `gen_ai.agent.a2a.response` | Full event payload (only if extraction enabled) |
+| `error.type` | Error message (on delivery failure) |
+
 ### Request/Response Extraction
 
 Enable request and response data extraction in spans using JVM system properties:
@@ -124,7 +139,7 @@ Enable request and response data extraction in spans using JVM system properties
 
 ### Instrumentation
 
-Adds OpenTelemetry spans to A2A client operations:
+Adds OpenTelemetry spans and metrics to A2A client operations:
 
 ```xml
 <dependency>
@@ -133,6 +148,27 @@ Adds OpenTelemetry spans to A2A client operations:
 </dependency>
 ```
 
+To also enable `gen_ai.client.operation.duration` metrics, pass a `Meter` via the transport config:
+
+```java
+config.setParameters(Map.of(
+    OpenTelemetryClientTransportWrapper.OTEL_TRACER_KEY, openTelemetry.getTracer("my-service"),
+    OpenTelemetryClientTransportWrapper.OTEL_METER_KEY,  openTelemetry.getMeter("my-service")
+));
+```
+
+| Instrument | Type | Unit | Description |
+|------------|------|------|-------------|
+| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of each A2A client operation call |
+
+Attributes on `gen_ai.client.operation.duration`:
+
+| Attribute | Value |
+|-----------|-------|
+| `gen_ai.agent.a2a.operation.name` | A2A method name (e.g. `message/send`) |
+| `gen_ai.system` | `a2a` |
+| `error.type` | `error` (only on failure) |
+
 ### Context Propagation
 
 For trace context propagation across async client boundaries:

extras/opentelemetry/common/src/main/java/org/a2aproject/sdk/extras/opentelemetry/A2AObservabilityNames.java

@@ -20,4 +20,7 @@ public interface A2AObservabilityNames {
     String GENAI_RESPONSE = GENAI_PREFIX + ".response"; // gen_ai.output.messages ?
     String GENAI_ROLE = GENAI_PREFIX + ".role";
     String GENAI_TASK_ID = GENAI_PREFIX + ".task_id";
+
+    // Push notification
+    String PUSH_NOTIFICATION_EVENT_KIND = GENAI_PREFIX + ".push_notification.event_kind";
 }

extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/A2ATestRoutes.java

@@ -1,7 +1,5 @@
 package org.a2aproject.sdk.extras.opentelemetry.it;
 
-
-
 import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
 
 import com.google.gson.Gson;
@@ -12,6 +10,7 @@
 import org.a2aproject.sdk.jsonrpc.common.json.JsonUtil;
 import org.a2aproject.sdk.spec.Task;
 import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
+import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
 import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
 import io.opentelemetry.api.trace.Span;
 import io.opentelemetry.api.trace.Tracer;
@@ -27,6 +26,9 @@
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Collectors;
 
 /**
@@ -40,6 +42,11 @@ public class A2ATestRoutes {
     private static final String TEXT_PLAIN = "text/plain";
     private static final Gson gson = new GsonBuilder().create();
 
+    /**
+     * Captured webhook requests, each entry is a map of header name to value.
+     */
+    private final List<Map<String, String>> webhookRequests = new CopyOnWriteArrayList<>();
+
     @Inject
     TestUtilsBean testUtilsBean;
     @Inject
@@ -76,6 +83,26 @@ void setupRoutes(@Observes Router router) {
             .produces(TEXT_PLAIN)
             .handler(ctx -> getChildQueueCount(ctx.pathParam("taskId"), ctx));
 
+        router.post("/test/pushconfig/:taskId")
+            .consumes(APPLICATION_JSON)
+            .handler(BodyHandler.create())
+            .blockingHandler(ctx -> savePushNotificationConfig(ctx.pathParam("taskId"), ctx.body().asString(), ctx));
+
+        router.delete("/test/pushconfig/:taskId/:configId")
+            .blockingHandler(ctx -> deletePushNotificationConfig(
+                    ctx.pathParam("taskId"), ctx.pathParam("configId"), ctx));
+
+        router.post("/test/webhook")
+            .handler(BodyHandler.create())
+            .handler(ctx -> captureWebhook(ctx));
+
+        router.get("/test/webhook/requests")
+            .produces(APPLICATION_JSON)
+            .handler(ctx -> getWebhookRequests(ctx));
+
+        router.delete("/test/webhook/requests")
+            .handler(ctx -> clearWebhookRequests(ctx));
+
         router.get("/hello")
             .produces(TEXT_PLAIN)
             .handler(ctx -> hello(ctx));
@@ -180,6 +207,48 @@ public void getChildQueueCount(String taskId, RoutingContext rc) {
                 .end(String.valueOf(count));
     }
 
+    public void savePushNotificationConfig(String taskId, String body, RoutingContext rc) {
+        try {
+            TaskPushNotificationConfig config = JsonUtil.fromJson(body, TaskPushNotificationConfig.class);
+            testUtilsBean.savePushNotificationConfig(taskId, config);
+            rc.response()
+                    .setStatusCode(200)
+                    .end();
+        } catch (Throwable t) {
+            errorResponse(t, rc);
+        }
+    }
+
+    public void deletePushNotificationConfig(String taskId, String configId, RoutingContext rc) {
+        try {
+            testUtilsBean.deletePushNotificationConfig(taskId, configId);
+            rc.response()
+                    .setStatusCode(200)
+                    .end();
+        } catch (Throwable t) {
+            errorResponse(t, rc);
+        }
+    }
+
+    private void captureWebhook(RoutingContext rc) {
+        Map<String, String> headers = new ConcurrentHashMap<>();
+        rc.request().headers().forEach(entry -> headers.put(entry.getKey(), entry.getValue()));
+        webhookRequests.add(headers);
+        rc.response().setStatusCode(200).end();
+    }
+
+    private void getWebhookRequests(RoutingContext rc) {
+        rc.response()
+                .setStatusCode(200)
+                .putHeader(CONTENT_TYPE, APPLICATION_JSON)
+                .end(gson.toJson(webhookRequests));
+    }
+
+    private void clearWebhookRequests(RoutingContext rc) {
+        webhookRequests.clear();
+        rc.response().setStatusCode(200).end();
+    }
+
     public void hello(RoutingContext rc) {
         Span span = tracer.spanBuilder("hello").startSpan();
         try (Scope scope = span.makeCurrent()) {

extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestAgentCardProducer.java

@@ -31,7 +31,7 @@ public AgentCard agentCard() {
                 .documentationUrl("http://example.com/test")
                 .capabilities(AgentCapabilities.builder()
                         .streaming(true)
-                        .pushNotifications(false)
+                        .pushNotifications(true)
                         .build())
                 .defaultInputModes(Collections.singletonList("text"))
                 .defaultOutputModes(Collections.singletonList("text"))

extras/opentelemetry/integration-tests/src/main/java/org/a2aproject/sdk/extras/opentelemetry/it/TestUtilsBean.java

@@ -1,15 +1,17 @@
 package org.a2aproject.sdk.extras.opentelemetry.it;
 
 import org.a2aproject.sdk.server.events.QueueManager;
+import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore;
 import org.a2aproject.sdk.server.tasks.TaskStore;
 import org.a2aproject.sdk.spec.Event;
 import org.a2aproject.sdk.spec.Task;
+import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 
 /**
  * Test utilities for OpenTelemetry integration tests.
- * Allows direct manipulation of tasks and queues for testing.
+ * Allows direct manipulation of tasks, queues, and push notification configs for testing.
  */
 @ApplicationScoped
 public class TestUtilsBean {
@@ -20,6 +22,9 @@ public class TestUtilsBean {
     @Inject
     QueueManager queueManager;
 
+    @Inject
+    PushNotificationConfigStore pushNotificationConfigStore;
+
     public void saveTask(Task task) {
         taskStore.save(task, false);
     }
@@ -43,4 +48,14 @@ public void enqueueEvent(String taskId, Event event) {
     public int getChildQueueCount(String taskId) {
         return queueManager.getActiveChildQueueCount(taskId);
     }
+
+    public void savePushNotificationConfig(String taskId, TaskPushNotificationConfig config) {
+        pushNotificationConfigStore.setInfo(
+                TaskPushNotificationConfig.builder(config).taskId(taskId).build(),
+                null);
+    }
+
+    public void deletePushNotificationConfig(String taskId, String configId) {
+        pushNotificationConfigStore.deleteInfo(taskId, configId);
+    }
 }

extras/opentelemetry/integration-tests/src/main/resources/application.properties

@@ -29,3 +29,7 @@ quarkus.log.category."io.opentelemetry".level=DEBUG
 
 # Authorization is not configured in this test application
 a2a.authorization.required=false
+
+# Allow HTTP and localhost for push notification testing
+a2a.push-notification.url.allowed-schemes=http,https
+a2a.push-notification.url.allow-private-network-targets=true