feat(opentelemetry): add push notification tracing and trace-context propagation (#1229)

feat!(opentelemetry): add push notification tracing and trace-context propagation

BREAKING: PushNotificationSender.sendNotification() now returns boolean.

- OTel decorator creates CLIENT spans for push delivery with error tracking
- W3C traceparent/tracestate headers injected into outbound push notification requests
- @Instrumented CDI qualifier transparently wraps application HTTP client and dispatch executor
- Three-tier CDI resolution: @Instrumented → @Named/unqualified → built-in default

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

Author: ehsavoie

compat-0.3/server-conversion/src/test/java/org/a2aproject/sdk/compat03/conversion/AbstractA2ARequestHandlerTest_v0_3.java

@@ -30,7 +30,6 @@
 import org.a2aproject.sdk.server.tasks.InMemoryPushNotificationConfigStore;
 import org.a2aproject.sdk.server.tasks.InMemoryTaskStore;
 import org.a2aproject.sdk.server.tasks.PushNotificationConfigStore;
-import org.a2aproject.sdk.server.tasks.PushNotificationSender;
 import org.a2aproject.sdk.server.tasks.PushNotificationUrlValidator;
 import org.a2aproject.sdk.server.tasks.TaskStore;
 import org.a2aproject.sdk.spec.A2AError;
@@ -85,8 +84,6 @@ public abstract class AbstractA2ARequestHandlerTest_v0_3 {
             .parts(new TextPart_v0_3("test message"))
             .build();
 
-    private static final PushNotificationSender NOOP_PUSHNOTIFICATION_SENDER = (event, snapshot) -> {};
-
     // V1.0 backend infrastructure
     protected AgentExecutor agentExecutor;
     protected TaskStore taskStore;

docs/content/dev/extra/http-clients.md

@@ -266,3 +266,9 @@ public class MyHttpClientProducer {
 | 0 | JDK | Always (fallback) |
 
 If the CDI container is active but no `A2AHttpClient` bean is registered, this provider is skipped and the next available provider is used.
+
+### Compatibility with OpenTelemetry
+
+When the [OpenTelemetry server module](opentelemetry.md) is on the classpath alongside a custom `A2AHttpClient` CDI bean, the OpenTelemetry module wraps your bean with trace-context propagation rather than replacing it. Your custom configuration (mTLS, proxy, authentication, timeouts, etc.) is preserved for push notification delivery while `traceparent` and `tracestate` headers are injected automatically.
+
+This works through the `@Instrumented` CDI qualifier (defined in `server-common`): the OpenTelemetry module produces an `@Instrumented A2AHttpClient` that delegates to your unqualified bean. `BasePushNotificationSender` prefers the `@Instrumented` variant when available, so both your custom configuration and the trace-context headers are present on outbound requests. No changes to your producer code are required.

docs/content/dev/extra/opentelemetry.md

@@ -177,6 +177,70 @@ 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 span status reflects the actual delivery outcome: `OK` when all notifications succeed, `ERROR` when one or more deliveries fail (with `error.type` set to `delivery_failed`) or when an exception is thrown.
+
+The module also provides **W3C trace-context propagation** for outbound push notification HTTP requests via the `@Instrumented` CDI qualifier and `OpenTelemetryHttpClientProducer`.
+
+#### How it works
+
+The `OpenTelemetryHttpClientProducer` CDI bean produces an `@Instrumented A2AHttpClient` — a wrapper that injects `traceparent` and `tracestate` headers into every outbound HTTP request. `BasePushNotificationSender` resolves the HTTP client in this order:
+
+1. An `@Instrumented A2AHttpClient` bean (if available — e.g. the one produced by this module)
+2. An unqualified `A2AHttpClient` bean (if the application provides one)
+3. `A2AHttpClientFactory.create()` as last resort
+
+The `@Instrumented` qualifier (defined in `server-common`) marks HTTP client beans that have been wrapped with observability instrumentation. It ensures the instrumented client does not compete with the application's own `A2AHttpClient` bean in default CDI resolution — the two coexist, and the instrumented variant wraps the application's bean as its delegate.
+
+#### Custom HTTP client compatibility
+
+If your application provides a custom `A2AHttpClient` CDI bean (e.g. for mTLS, proxy, or custom timeouts), the OpenTelemetry wrapper **preserves that configuration** by delegating to your bean rather than replacing it. No additional annotations or configuration are needed — just produce your bean as usual:
+
+```java
+@Produces
+@ApplicationScoped
+public A2AHttpClient myHttpClient() {
+    // your custom client (mTLS, proxy, auth, etc.)
+}
+```
+
+The OpenTelemetry module automatically wraps it with trace-context injection.
+
+#### Context-propagating executor
+
+The producer also creates an `@Instrumented 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.
+
+`BasePushNotificationSender` resolves the dispatch executor in this order:
+
+1. An `@Instrumented Executor` bean (if available — e.g. the one produced by this module)
+2. A `@Named("a2a.pushNotificationDispatch") Executor` bean (if the application provides one)
+3. `ForkJoinPool.commonPool()` as last resort
+
+#### Custom dispatch executor compatibility
+
+If your application provides a custom `@Named("a2a.pushNotificationDispatch") Executor` CDI bean (e.g. for custom thread pool sizing), the OpenTelemetry module **preserves that configuration** by wrapping it with `Context.taskWrapping()` rather than replacing it. Just produce your bean as usual:
+
+```java
+@Produces
+@ApplicationScoped
+@Named("a2a.pushNotificationDispatch")
+public Executor myDispatchExecutor() {
+    // your custom thread pool
+}
+```
+
+The OpenTelemetry module automatically wraps it with context propagation.
+
+| 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` | `delivery_failed` when one or more deliveries fail, or the exception message when an exception is thrown |
+
 ### Request/Response Extraction
 
 Enable request and response data extraction in spans using JVM system properties:
@@ -195,7 +259,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>
@@ -206,11 +270,20 @@ Adds OpenTelemetry spans to A2A client operations:
 
 ### Metrics
 
-The client transport records the following histogram:
+The client transport wrapper is auto-discovered via ServiceLoader — just adding the dependency enables tracing. 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")
+));
+```
+
+If no `Meter` is provided, the histogram is silently skipped.
 
 | Instrument | Type | Unit | Description |
 |------------|------|------|-------------|
-| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of A2A client operations from invocation to completion or error |
+| `gen_ai.client.operation.duration` | Histogram | `s` | Duration of each A2A client operation call |
 
 Attributes on `gen_ai.client.operation.duration`:
 
@@ -220,14 +293,6 @@ Attributes on `gen_ai.client.operation.duration`:
 | `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:

extras/http-client-cdi/src/test/java/org/a2aproject/sdk/client/http/cdi/CdiA2AHttpClientFactoryTest.java

@@ -81,6 +81,17 @@ public void testCreateDoesNotLogWarningWhenCdiSucceeds() {
             assertFalse(records.stream().anyMatch(r -> r.getLevel() == Level.WARNING),
                     "No warning should be logged when CDI provider succeeds");
         }
+
+        @Test
+        public void testCreateExcludingCdiSkipsTheCdiBeanAndFallsBack() {
+            // Regression guard for the recursive CDI self-delegation bug: a CDI producer that
+            // produces an A2AHttpClient must call createExcluding("cdi") so the factory does not
+            // re-enter the container and return the proxy being constructed as its own delegate.
+            A2AHttpClient client = A2AHttpClientFactory.createExcluding("cdi");
+            assertNotNull(client);
+            assertFalse(client instanceof DummyA2AHttpClient,
+                    "createExcluding('cdi') must not return the CDI-registered bean");
+        }
     }
 
     @Nested

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

@@ -42,4 +42,8 @@ public interface A2AObservabilityNames {
     String GENAI_STREAMING_DURATION = GENAI_PREFIX + ".streaming.duration";
     String GENAI_SYSTEM = "gen_ai.system";
     String GENAI_SYSTEM_VALUE = "a2a";
+
+    // Push notification
+    String PUSH_NOTIFICATION_EVENT_KIND = GENAI_PREFIX + ".push_notification.event_kind";
+    String PUSH_NOTIFICATION_DELIVERY_FAILED = "delivery_failed";
 }

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

@@ -1,8 +1,7 @@
 package org.a2aproject.sdk.extras.opentelemetry.it;
 
-
-
 import static io.vertx.core.http.HttpHeaders.CONTENT_TYPE;
+import static java.net.HttpURLConnection.HTTP_OK;
 
 import com.google.gson.Gson;
 import com.google.gson.GsonBuilder;
@@ -12,6 +11,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.OpenTelemetry;
 import io.opentelemetry.api.trace.Span;
@@ -32,7 +32,10 @@
 import jakarta.enterprise.inject.Produces;
 import jakarta.inject.Inject;
 import jakarta.inject.Singleton;
+import java.util.HashMap;
 import java.util.List;
+import java.util.Map;
+import java.util.concurrent.CopyOnWriteArrayList;
 
 /**
  * Test routes for OpenTelemetry integration testing.
@@ -45,6 +48,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
@@ -53,6 +61,8 @@ public class A2ATestRoutes {
     InMemoryMetricExporter inMemoryMetricExporter;
     @Inject
     OpenTelemetry openTelemetry;
+    @Inject
+    CustomDispatchExecutorProducer customDispatchExecutorProducer;
 
     @Inject
     Tracer tracer;
@@ -85,6 +95,37 @@ 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("/test/dispatch-executor/count")
+            .produces(TEXT_PLAIN)
+            .handler(ctx -> ctx.response().setStatusCode(HTTP_OK)
+                    .end(String.valueOf(customDispatchExecutorProducer.getExecutionCount())));
+
+        router.delete("/test/dispatch-executor/count")
+            .handler(ctx -> {
+                customDispatchExecutorProducer.resetExecutionCount();
+                ctx.response().setStatusCode(HTTP_OK).end();
+            });
+
         router.get("/hello")
             .produces(TEXT_PLAIN)
             .handler(ctx -> hello(ctx));
@@ -197,6 +238,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 HashMap<>();
+        rc.request().headers().forEach(entry -> headers.put(entry.getKey(), entry.getValue()));
+        webhookRequests.add(headers);
+        rc.response().setStatusCode(HTTP_OK).end();
+    }
+
+    private void getWebhookRequests(RoutingContext rc) {
+        rc.response()
+                .setStatusCode(HTTP_OK)
+                .putHeader(CONTENT_TYPE, APPLICATION_JSON)
+                .end(gson.toJson(webhookRequests));
+    }
+
+    private void clearWebhookRequests(RoutingContext rc) {
+        webhookRequests.clear();
+        rc.response().setStatusCode(HTTP_OK).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/CustomDispatchExecutorProducer.java

@@ -0,0 +1,70 @@
+package org.a2aproject.sdk.extras.opentelemetry.it;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Named;
+import org.a2aproject.sdk.server.tasks.BasePushNotificationSender;
+
+/**
+ * Produces a custom {@link Named @Named} dispatch {@link Executor} CDI bean
+ * that uses a distinctive thread name prefix. Used by integration tests to
+ * verify that the OpenTelemetry module wraps the application's executor with
+ * context propagation instead of replacing it.
+ *
+ * <p>An {@link AtomicInteger} counter tracks how many tasks have been
+ * submitted, allowing tests to confirm the custom executor was actually used.
+ */
+@ApplicationScoped
+public class CustomDispatchExecutorProducer {
+
+    static final String CUSTOM_THREAD_PREFIX = "custom-dispatch-";
+
+    private final AtomicInteger executionCount = new AtomicInteger();
+    private final ExecutorService pool;
+
+    public CustomDispatchExecutorProducer() {
+        AtomicInteger threadCounter = new AtomicInteger();
+        ThreadFactory threadFactory = r -> {
+            Thread t = new Thread(r, CUSTOM_THREAD_PREFIX + threadCounter.getAndIncrement());
+            t.setDaemon(true);
+            return t;
+        };
+        this.pool = Executors.newFixedThreadPool(2, threadFactory);
+    }
+
+    @Produces
+    @ApplicationScoped
+    @Named(BasePushNotificationSender.DISPATCH_EXECUTOR_NAME)
+    Executor customDispatchExecutor() {
+        return command -> {
+            executionCount.incrementAndGet();
+            pool.execute(command);
+        };
+    }
+
+    /**
+     * Returns the number of tasks submitted through this executor.
+     */
+    public int getExecutionCount() {
+        return executionCount.get();
+    }
+
+    /**
+     * Resets the execution counter.
+     */
+    public void resetExecutionCount() {
+        executionCount.set(0);
+    }
+
+    @PreDestroy
+    void shutdown() {
+        pool.shutdownNow();
+    }
+}