> fetchWebhookRequests() {
+ String json = given().get("/test/webhook/requests")
+ .then().statusCode(HTTP_OK)
+ .extract().body().asString();
+ return new Gson().fromJson(json, WEBHOOK_LIST_TYPE);
+ }
+
protected void ensureQueueForTask(String taskId) throws Exception {
HttpClient httpClient = HttpClient.newBuilder()
.version(HttpClient.Version.HTTP_2)
diff --git a/extras/opentelemetry/integration-tests/src/test/resources/application.properties b/extras/opentelemetry/integration-tests/src/test/resources/application.properties
index 932acdd58..f9ec66409 100644
--- a/extras/opentelemetry/integration-tests/src/test/resources/application.properties
+++ b/extras/opentelemetry/integration-tests/src/test/resources/application.properties
@@ -26,3 +26,7 @@ quarkus.log.category."org.a2aproject.sdk".level=DEBUG
# Authorization is not configured in test profiles
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
diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClient.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClient.java
new file mode 100644
index 000000000..2245724d3
--- /dev/null
+++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClient.java
@@ -0,0 +1,179 @@
+package org.a2aproject.sdk.extras.opentelemetry.server;
+
+import java.io.IOException;
+import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.propagation.TextMapSetter;
+import org.a2aproject.sdk.client.http.A2AHttpClient;
+import org.a2aproject.sdk.client.http.A2AHttpResponse;
+import org.a2aproject.sdk.client.http.ServerSentEvent;
+import org.a2aproject.sdk.util.Assert;
+
+/**
+ * Delegating {@link A2AHttpClient} that injects W3C trace-context headers
+ * ({@code traceparent}, {@code tracestate}) into every outbound HTTP request.
+ *
+ * The propagation uses the {@link io.opentelemetry.context.propagation.TextMapPropagator}
+ * registered with the supplied {@link OpenTelemetry} instance, reading the current
+ * {@link Context} at call time. This is the same mechanism the client-side
+ * {@code OpenTelemetryClientPropagatorTransport} uses, applied here to
+ * server-initiated outbound calls such as push-notification delivery.
+ *
+ * @see OpenTelemetryHttpClientProducer
+ */
+class OpenTelemetryA2AHttpClient implements A2AHttpClient {
+
+ private final A2AHttpClient delegate;
+ private final OpenTelemetry openTelemetry;
+
+ OpenTelemetryA2AHttpClient(A2AHttpClient delegate, OpenTelemetry openTelemetry) {
+ this.delegate = Assert.checkNotNullParam("delegate", delegate);
+ this.openTelemetry = Assert.checkNotNullParam("openTelemetry", openTelemetry);
+ }
+
+ @Override
+ public GetBuilder createGet() {
+ return new PropagatingGetBuilder(delegate.createGet(), openTelemetry);
+ }
+
+ @Override
+ public PostBuilder createPost() {
+ return new PropagatingPostBuilder(delegate.createPost(), openTelemetry);
+ }
+
+ @Override
+ public DeleteBuilder createDelete() {
+ return new PropagatingDeleteBuilder(delegate.createDelete(), openTelemetry);
+ }
+
+ @SuppressWarnings("rawtypes")
+ private static final TextMapSetter BUILDER_SETTER = (carrier, key, value) -> {
+ if (carrier != null) {
+ carrier.addHeader(key, value);
+ }
+ };
+
+ /**
+ * Abstract base for propagating builder wrappers. Consolidates the common
+ * {@link Builder} delegation ({@code url}, {@code addHeaders}, {@code addHeader})
+ * and the context injection logic into one place.
+ *
+ * @param the concrete propagating builder type (for method chaining)
+ * @param the delegate builder type
+ */
+ private abstract static class AbstractPropagatingBuilder, D extends Builder> {
+
+ final D delegate;
+ private final OpenTelemetry openTelemetry;
+
+ AbstractPropagatingBuilder(D delegate, OpenTelemetry openTelemetry) {
+ this.delegate = delegate;
+ this.openTelemetry = openTelemetry;
+ }
+
+ @SuppressWarnings("unchecked")
+ protected S self() {
+ return (S) this;
+ }
+
+ public S url(String s) {
+ delegate.url(s);
+ return self();
+ }
+
+ public S addHeaders(Map headers) {
+ delegate.addHeaders(headers);
+ return self();
+ }
+
+ public S addHeader(String name, String value) {
+ delegate.addHeader(name, value);
+ return self();
+ }
+
+ @SuppressWarnings("rawtypes")
+ protected void injectContext() {
+ openTelemetry.getPropagators().getTextMapPropagator()
+ .inject(Context.current(), (Builder) delegate, BUILDER_SETTER);
+ }
+ }
+
+ private static final class PropagatingGetBuilder
+ extends AbstractPropagatingBuilder
+ implements GetBuilder {
+
+ PropagatingGetBuilder(GetBuilder delegate, OpenTelemetry openTelemetry) {
+ super(delegate, openTelemetry);
+ }
+
+ @Override
+ public A2AHttpResponse get() throws IOException, InterruptedException {
+ injectContext();
+ return delegate.get();
+ }
+
+ @Override
+ public CompletableFuture getAsyncSSE(
+ Consumer messageConsumer,
+ Consumer errorConsumer,
+ Runnable completeRunnable) throws IOException, InterruptedException {
+ injectContext();
+ return delegate.getAsyncSSE(messageConsumer, errorConsumer, completeRunnable);
+ }
+ }
+
+ private static final class PropagatingPostBuilder
+ extends AbstractPropagatingBuilder
+ implements PostBuilder {
+
+ PropagatingPostBuilder(PostBuilder delegate, OpenTelemetry openTelemetry) {
+ super(delegate, openTelemetry);
+ }
+
+ @Override
+ public PostBuilder body(String body) {
+ delegate.body(body);
+ return this;
+ }
+
+ @Override
+ public PostBuilder followRedirects(boolean follow) {
+ delegate.followRedirects(follow);
+ return this;
+ }
+
+ @Override
+ public A2AHttpResponse post() throws IOException, InterruptedException {
+ injectContext();
+ return delegate.post();
+ }
+
+ @Override
+ public CompletableFuture postAsyncSSE(
+ Consumer messageConsumer,
+ Consumer errorConsumer,
+ Runnable completeRunnable) throws IOException, InterruptedException {
+ injectContext();
+ return delegate.postAsyncSSE(messageConsumer, errorConsumer, completeRunnable);
+ }
+ }
+
+ private static final class PropagatingDeleteBuilder
+ extends AbstractPropagatingBuilder
+ implements DeleteBuilder {
+
+ PropagatingDeleteBuilder(DeleteBuilder delegate, OpenTelemetry openTelemetry) {
+ super(delegate, openTelemetry);
+ }
+
+ @Override
+ public A2AHttpResponse delete() throws IOException, InterruptedException {
+ injectContext();
+ return delegate.delete();
+ }
+ }
+}
diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducer.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducer.java
new file mode 100644
index 000000000..344f6e918
--- /dev/null
+++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryHttpClientProducer.java
@@ -0,0 +1,73 @@
+package org.a2aproject.sdk.extras.opentelemetry.server;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.context.Context;
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Inject;
+import jakarta.inject.Named;
+import org.a2aproject.sdk.client.http.A2AHttpClient;
+import org.a2aproject.sdk.client.http.A2AHttpClientFactory;
+import org.a2aproject.sdk.server.tasks.BasePushNotificationSender;
+
+/**
+ * CDI producer that creates an {@link OpenTelemetryA2AHttpClient} wrapping the
+ * factory-resolved {@link A2AHttpClient} with W3C trace-context propagation.
+ *
+ * When this module is on the classpath and CDI discovers this bean,
+ * {@link org.a2aproject.sdk.server.tasks.BasePushNotificationSender} will use the
+ * produced client instead of calling {@link A2AHttpClientFactory#create()} directly,
+ * ensuring that outbound push-notification HTTP requests carry {@code traceparent}
+ * and {@code tracestate} headers.
+ *
+ *
CDI ambiguity: The {@link A2AHttpClient} bean produced by
+ * {@link #createPropagatingClient()} has no qualifier. If the application also
+ * produces an unqualified {@code A2AHttpClient} bean (e.g. a custom Vert.x or
+ * Android client), CDI will report an {@code AmbiguousResolutionException} at
+ * startup. In that case, use {@code @Alternative @Priority} on one of the
+ * producers to resolve the conflict, or inject the custom client with a qualifier.
+ */
+@ApplicationScoped
+public class OpenTelemetryHttpClientProducer {
+
+ private final OpenTelemetry openTelemetry;
+
+ @SuppressWarnings("NullAway")
+ protected OpenTelemetryHttpClientProducer() {
+ this.openTelemetry = null;
+ }
+
+ @Inject
+ public OpenTelemetryHttpClientProducer(OpenTelemetry openTelemetry) {
+ this.openTelemetry = openTelemetry;
+ }
+
+ /**
+ * Produces an {@link A2AHttpClient} that injects trace-context headers into
+ * every outbound request.
+ *
+ * @return a propagating HTTP client wrapping the default factory client
+ */
+ @Produces
+ @ApplicationScoped
+ A2AHttpClient createPropagatingClient() {
+ return new OpenTelemetryA2AHttpClient(A2AHttpClientFactory.create(), openTelemetry);
+ }
+
+ /**
+ * Produces an {@link Executor} that propagates the current OpenTelemetry
+ * {@link Context} to submitted tasks, ensuring trace context survives
+ * {@code CompletableFuture.supplyAsync()} boundaries in push notification dispatch.
+ *
+ * @return a context-propagating executor wrapping the common ForkJoinPool
+ */
+ @Produces
+ @ApplicationScoped
+ @Named(BasePushNotificationSender.DISPATCH_EXECUTOR_NAME)
+ Executor createContextPropagatingExecutor() {
+ return Context.taskWrapping(ForkJoinPool.commonPool());
+ }
+}
diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java
new file mode 100644
index 000000000..c8bc84881
--- /dev/null
+++ b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecorator.java
@@ -0,0 +1,149 @@
+package org.a2aproject.sdk.extras.opentelemetry.server;
+
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_EVENT_KIND;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import jakarta.annotation.Priority;
+import jakarta.decorator.Decorator;
+import jakarta.decorator.Delegate;
+import jakarta.enterprise.inject.Any;
+import jakarta.inject.Inject;
+import org.a2aproject.sdk.server.tasks.PushNotificationSender;
+import org.a2aproject.sdk.spec.Message;
+import org.a2aproject.sdk.spec.StreamingEventKind;
+import org.a2aproject.sdk.spec.Task;
+import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
+import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * OpenTelemetry CDI Decorator for {@link PushNotificationSender}.
+ *
+ * Creates a {@code CLIENT} span named {@code SendPushNotification} for every push
+ * notification delivery attempt, capturing the task ID, context ID, and event kind
+ * as span attributes.
+ *
+ * To enable this decorator, add it to your beans.xml:
+ *
{@code
+ *
+ * org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator
+ *
+ * }
+ */
+@Decorator
+@Priority(100)
+public abstract class OpenTelemetryPushNotificationSenderDecorator implements PushNotificationSender {
+
+ static final String SEND_PUSH_NOTIFICATION = "SendPushNotification";
+
+ @Inject
+ @Delegate
+ @Any
+ private PushNotificationSender delegate;
+
+ @Inject
+ private Tracer tracer;
+
+ /**
+ * Default constructor for CDI.
+ */
+ public OpenTelemetryPushNotificationSenderDecorator() {
+ }
+
+ /**
+ * Constructor for unit testing.
+ *
+ * @param delegate the delegate push notification sender
+ * @param tracer the tracer to use
+ */
+ OpenTelemetryPushNotificationSenderDecorator(PushNotificationSender delegate, Tracer tracer) {
+ this.delegate = delegate;
+ this.tracer = tracer;
+ }
+
+ @Override
+ public void sendNotification(StreamingEventKind event, @Nullable Task taskSnapshot) {
+ var spanBuilder = tracer.spanBuilder(SEND_PUSH_NOTIFICATION)
+ .setSpanKind(SpanKind.CLIENT)
+ .setAttribute(GENAI_OPERATION_NAME, SEND_PUSH_NOTIFICATION)
+ .setAttribute(PUSH_NOTIFICATION_EVENT_KIND, event.getClass().getSimpleName());
+
+ String taskId = extractTaskId(event, taskSnapshot);
+ if (taskId != null) {
+ spanBuilder.setAttribute(GENAI_TASK_ID, taskId);
+ }
+ String contextId = extractContextId(event, taskSnapshot);
+ if (contextId != null) {
+ spanBuilder.setAttribute(GENAI_CONTEXT_ID, contextId);
+ }
+ if (extractResponse()) {
+ spanBuilder.setAttribute(GENAI_RESPONSE, event.toString());
+ }
+
+ Span span = spanBuilder.startSpan();
+ try (Scope scope = span.makeCurrent()) {
+ delegate.sendNotification(event, taskSnapshot);
+ span.setStatus(StatusCode.OK);
+ } catch (Exception ex) {
+ span.setAttribute(ERROR_TYPE, ex.getMessage() != null ? ex.getMessage() : ex.getClass().getName());
+ span.setStatus(StatusCode.ERROR, ex.getMessage());
+ throw ex;
+ } finally {
+ span.end();
+ }
+ }
+
+ @Nullable
+ private static String extractTaskId(StreamingEventKind event, @Nullable Task taskSnapshot) {
+ if (taskSnapshot != null) {
+ return taskSnapshot.id();
+ }
+ if (event instanceof Task task) {
+ return task.id();
+ }
+ if (event instanceof Message message) {
+ return message.taskId();
+ }
+ if (event instanceof TaskStatusUpdateEvent statusUpdate) {
+ return statusUpdate.taskId();
+ }
+ if (event instanceof TaskArtifactUpdateEvent artifactUpdate) {
+ return artifactUpdate.taskId();
+ }
+ return null;
+ }
+
+ @Nullable
+ private static String extractContextId(StreamingEventKind event, @Nullable Task taskSnapshot) {
+ if (taskSnapshot != null) {
+ return taskSnapshot.contextId();
+ }
+ if (event instanceof Task task) {
+ return task.contextId();
+ }
+ if (event instanceof Message message) {
+ return message.contextId();
+ }
+ if (event instanceof TaskStatusUpdateEvent statusUpdate) {
+ return statusUpdate.contextId();
+ }
+ if (event instanceof TaskArtifactUpdateEvent artifactUpdate) {
+ return artifactUpdate.contextId();
+ }
+ return null;
+ }
+
+ private boolean extractResponse() {
+ return Boolean.getBoolean(EXTRACT_RESPONSE_SYS_PROPERTY);
+ }
+}
diff --git a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml
index e4652bc27..f775403c9 100644
--- a/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml
+++ b/extras/opentelemetry/server/src/main/resources/META-INF/beans.xml
@@ -5,5 +5,6 @@
bean-discovery-mode="annotated">
org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryRequestHandlerDecorator
+ org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator
-
\ No newline at end of file
+
diff --git a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClientTest.java b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClientTest.java
new file mode 100644
index 000000000..e4cf86cb5
--- /dev/null
+++ b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryA2AHttpClientTest.java
@@ -0,0 +1,189 @@
+package org.a2aproject.sdk.extras.opentelemetry.server;
+
+import static org.junit.jupiter.api.Assertions.assertNotNull;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.inOrder;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.io.IOException;
+
+import io.opentelemetry.api.OpenTelemetry;
+import io.opentelemetry.context.Context;
+import io.opentelemetry.context.Scope;
+import io.opentelemetry.context.propagation.ContextPropagators;
+import io.opentelemetry.context.propagation.TextMapPropagator;
+import io.opentelemetry.context.propagation.TextMapSetter;
+import org.a2aproject.sdk.client.http.A2AHttpClient;
+import org.a2aproject.sdk.client.http.A2AHttpResponse;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.ArgumentCaptor;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class OpenTelemetryA2AHttpClientTest {
+
+ @Mock
+ private A2AHttpClient delegate;
+
+ @Mock
+ private OpenTelemetry openTelemetry;
+
+ @Mock
+ private ContextPropagators propagators;
+
+ @Mock
+ private TextMapPropagator textMapPropagator;
+
+ @Mock
+ private A2AHttpClient.PostBuilder delegatePostBuilder;
+
+ @Mock
+ private A2AHttpClient.GetBuilder delegateGetBuilder;
+
+ @Mock
+ private A2AHttpClient.DeleteBuilder delegateDeleteBuilder;
+
+ @Mock
+ private A2AHttpResponse response;
+
+ private OpenTelemetryA2AHttpClient client;
+
+ @BeforeEach
+ void setUp() {
+ when(openTelemetry.getPropagators()).thenReturn(propagators);
+ when(propagators.getTextMapPropagator()).thenReturn(textMapPropagator);
+
+ when(delegate.createPost()).thenReturn(delegatePostBuilder);
+ when(delegate.createGet()).thenReturn(delegateGetBuilder);
+ when(delegate.createDelete()).thenReturn(delegateDeleteBuilder);
+
+ when(delegatePostBuilder.url(any())).thenReturn(delegatePostBuilder);
+ when(delegatePostBuilder.body(any())).thenReturn(delegatePostBuilder);
+ when(delegatePostBuilder.addHeader(any(), any())).thenReturn(delegatePostBuilder);
+ when(delegatePostBuilder.addHeaders(any())).thenReturn(delegatePostBuilder);
+ when(delegatePostBuilder.followRedirects(any(boolean.class))).thenReturn(delegatePostBuilder);
+
+ when(delegateGetBuilder.url(any())).thenReturn(delegateGetBuilder);
+ when(delegateGetBuilder.addHeader(any(), any())).thenReturn(delegateGetBuilder);
+ when(delegateGetBuilder.addHeaders(any())).thenReturn(delegateGetBuilder);
+
+ when(delegateDeleteBuilder.url(any())).thenReturn(delegateDeleteBuilder);
+ when(delegateDeleteBuilder.addHeader(any(), any())).thenReturn(delegateDeleteBuilder);
+ when(delegateDeleteBuilder.addHeaders(any())).thenReturn(delegateDeleteBuilder);
+
+ client = new OpenTelemetryA2AHttpClient(delegate, openTelemetry);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void post_injectsContextBeforeDelegating() throws IOException, InterruptedException {
+ when(delegatePostBuilder.post()).thenReturn(response);
+
+ A2AHttpClient.PostBuilder builder = client.createPost();
+ A2AHttpResponse result = builder
+ .url("http://example.com/push")
+ .addHeader("Content-Type", "application/json")
+ .body("{}")
+ .post();
+
+ assertSame(response, result);
+
+ InOrder order = inOrder(textMapPropagator, delegatePostBuilder);
+ order.verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class));
+ order.verify(delegatePostBuilder).post();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void post_propagatesCurrentSpanContext() throws IOException, InterruptedException {
+ when(delegatePostBuilder.post()).thenReturn(response);
+
+ Context testContext = Context.current().with(io.opentelemetry.context.ContextKey.named("test"), "value");
+ try (Scope ignored = testContext.makeCurrent()) {
+ client.createPost()
+ .url("http://example.com/push")
+ .body("{}")
+ .post();
+ }
+
+ verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class));
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void get_injectsContextBeforeDelegating() throws IOException, InterruptedException {
+ when(delegateGetBuilder.get()).thenReturn(response);
+
+ A2AHttpClient.GetBuilder builder = client.createGet();
+ A2AHttpResponse result = builder
+ .url("http://example.com/api")
+ .get();
+
+ assertSame(response, result);
+
+ InOrder order = inOrder(textMapPropagator, delegateGetBuilder);
+ order.verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class));
+ order.verify(delegateGetBuilder).get();
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void delete_injectsContextBeforeDelegating() throws IOException, InterruptedException {
+ when(delegateDeleteBuilder.delete()).thenReturn(response);
+
+ A2AHttpClient.DeleteBuilder builder = client.createDelete();
+ A2AHttpResponse result = builder
+ .url("http://example.com/resource")
+ .delete();
+
+ assertSame(response, result);
+
+ InOrder order = inOrder(textMapPropagator, delegateDeleteBuilder);
+ order.verify(textMapPropagator).inject(any(Context.class), any(), any(TextMapSetter.class));
+ order.verify(delegateDeleteBuilder).delete();
+ }
+
+ @Test
+ void postBuilder_chainsCorrectly() {
+ A2AHttpClient.PostBuilder builder = client.createPost();
+
+ A2AHttpClient.PostBuilder chained = builder
+ .url("http://example.com")
+ .addHeader("X-Custom", "value")
+ .body("{}")
+ .followRedirects(false);
+
+ assertNotNull(chained);
+ }
+
+ @Test
+ @SuppressWarnings("unchecked")
+ void textMapSetter_callsAddHeaderOnBuilder() throws IOException, InterruptedException {
+ when(delegatePostBuilder.post()).thenReturn(response);
+
+ @SuppressWarnings("rawtypes")
+ ArgumentCaptor setterCaptor = ArgumentCaptor.forClass(TextMapSetter.class);
+
+ client.createPost()
+ .url("http://example.com")
+ .body("{}")
+ .post();
+
+ verify(textMapPropagator).inject(any(Context.class), any(), setterCaptor.capture());
+
+ @SuppressWarnings("rawtypes")
+ TextMapSetter setter = setterCaptor.getValue();
+ setter.set(delegatePostBuilder, "traceparent", "00-abc-def-01");
+
+ verify(delegatePostBuilder).addHeader("traceparent", "00-abc-def-01");
+ }
+}
diff --git a/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecoratorTest.java b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecoratorTest.java
new file mode 100644
index 000000000..3b2d8b71b
--- /dev/null
+++ b/extras/opentelemetry/server/src/test/java/org/a2aproject/sdk/extras/opentelemetry/server/OpenTelemetryPushNotificationSenderDecoratorTest.java
@@ -0,0 +1,187 @@
+package org.a2aproject.sdk.extras.opentelemetry.server;
+
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.ERROR_TYPE;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.EXTRACT_RESPONSE_SYS_PROPERTY;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_CONTEXT_ID;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_OPERATION_NAME;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_RESPONSE;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.GENAI_TASK_ID;
+import static org.a2aproject.sdk.extras.opentelemetry.A2AObservabilityNames.PUSH_NOTIFICATION_EVENT_KIND;
+import static org.a2aproject.sdk.extras.opentelemetry.server.OpenTelemetryPushNotificationSenderDecorator.SEND_PUSH_NOTIFICATION;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyString;
+import static org.mockito.ArgumentMatchers.eq;
+import static org.mockito.Mockito.doThrow;
+import static org.mockito.Mockito.lenient;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.SpanBuilder;
+import io.opentelemetry.api.trace.SpanKind;
+import io.opentelemetry.api.trace.StatusCode;
+import io.opentelemetry.api.trace.Tracer;
+import io.opentelemetry.context.Scope;
+import java.util.Collections;
+import java.util.List;
+import org.a2aproject.sdk.server.tasks.PushNotificationSender;
+import org.a2aproject.sdk.spec.Task;
+import org.a2aproject.sdk.spec.TaskArtifactUpdateEvent;
+import org.a2aproject.sdk.spec.TaskState;
+import org.a2aproject.sdk.spec.TaskStatus;
+import org.a2aproject.sdk.spec.TaskStatusUpdateEvent;
+import org.junit.jupiter.api.BeforeEach;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+import org.mockito.junit.jupiter.MockitoSettings;
+import org.mockito.quality.Strictness;
+
+@ExtendWith(MockitoExtension.class)
+@MockitoSettings(strictness = Strictness.LENIENT)
+class OpenTelemetryPushNotificationSenderDecoratorTest {
+
+ @Mock
+ private PushNotificationSender delegate;
+
+ @Mock
+ private Tracer tracer;
+
+ @Mock
+ private SpanBuilder spanBuilder;
+
+ @Mock
+ private Span span;
+
+ @Mock
+ private Scope scope;
+
+ private OpenTelemetryPushNotificationSenderDecorator decorator;
+
+ @BeforeEach
+ void setUp() {
+ lenient().when(tracer.spanBuilder(anyString())).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.setSpanKind(any(SpanKind.class))).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.setAttribute(anyString(), anyString())).thenReturn(spanBuilder);
+ lenient().when(spanBuilder.startSpan()).thenReturn(span);
+ lenient().when(span.makeCurrent()).thenReturn(scope);
+
+ decorator = new TestablePushNotificationSenderDecorator(delegate, tracer);
+ }
+
+ @Test
+ void sendNotification_createsSpanWithTaskIdFromSnapshot() {
+ Task taskSnapshot = Task.builder()
+ .id("task-1")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.TASK_STATE_WORKING))
+ .history(Collections.emptyList())
+ .artifacts(Collections.emptyList())
+ .build();
+ TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder()
+ .taskId("task-1")
+ .contextId("ctx-1")
+ .status(new TaskStatus(TaskState.TASK_STATE_WORKING))
+ .build();
+
+ decorator.sendNotification(event, taskSnapshot);
+
+ verify(tracer).spanBuilder(SEND_PUSH_NOTIFICATION);
+ verify(spanBuilder).setSpanKind(SpanKind.CLIENT);
+ verify(spanBuilder).setAttribute(GENAI_OPERATION_NAME, SEND_PUSH_NOTIFICATION);
+ verify(spanBuilder).setAttribute(GENAI_TASK_ID, "task-1");
+ verify(spanBuilder).setAttribute(GENAI_CONTEXT_ID, "ctx-1");
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ verify(delegate).sendNotification(event, taskSnapshot);
+ }
+
+ @Test
+ void sendNotification_createsSpanWithTaskIdFromEvent() {
+ org.a2aproject.sdk.spec.Artifact artifact = org.a2aproject.sdk.spec.Artifact.builder()
+ .artifactId("artifact-id")
+ .parts(List.of(new org.a2aproject.sdk.spec.TextPart("result")))
+ .build();
+ TaskArtifactUpdateEvent event = new TaskArtifactUpdateEvent(
+ "task-event-1", artifact, "ctx-event-1", null, null, null);
+
+ decorator.sendNotification(event, null);
+
+ verify(spanBuilder).setAttribute(GENAI_TASK_ID, "task-event-1");
+ verify(spanBuilder).setAttribute(GENAI_CONTEXT_ID, "ctx-event-1");
+ verify(spanBuilder, never()).setAttribute(eq(GENAI_TASK_ID), eq((String) null));
+ verify(span).setStatus(StatusCode.OK);
+ verify(span).end();
+ }
+
+ @Test
+ void sendNotification_setsErrorStatusOnException() {
+ TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder()
+ .taskId("task-err")
+ .contextId("ctx-err")
+ .status(new TaskStatus(TaskState.TASK_STATE_FAILED))
+ .build();
+ RuntimeException ex = new RuntimeException("webhook unreachable");
+ doThrow(ex).when(delegate).sendNotification(any(), any());
+
+ assertThrows(RuntimeException.class, () -> decorator.sendNotification(event, null));
+
+ verify(span).setStatus(StatusCode.ERROR, "webhook unreachable");
+ verify(span).setAttribute(ERROR_TYPE, "webhook unreachable");
+ verify(span).end();
+ }
+
+ @Test
+ void sendNotification_setsEventKindAttribute() {
+ TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder()
+ .taskId("task-kind")
+ .contextId("ctx-kind")
+ .status(new TaskStatus(TaskState.TASK_STATE_COMPLETED))
+ .build();
+
+ decorator.sendNotification(event, null);
+
+ verify(spanBuilder).setAttribute(PUSH_NOTIFICATION_EVENT_KIND, "TaskStatusUpdateEvent");
+ }
+
+ @Test
+ void sendNotification_extractsResponseWhenEnabled() {
+ System.setProperty(EXTRACT_RESPONSE_SYS_PROPERTY, "true");
+ try {
+ TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder()
+ .taskId("task-resp")
+ .contextId("ctx-resp")
+ .status(new TaskStatus(TaskState.TASK_STATE_WORKING))
+ .build();
+
+ decorator.sendNotification(event, null);
+
+ verify(spanBuilder).setAttribute(eq(GENAI_RESPONSE), anyString());
+ } finally {
+ System.clearProperty(EXTRACT_RESPONSE_SYS_PROPERTY);
+ }
+ }
+
+ @Test
+ void sendNotification_noResponseAttributeWhenDisabled() {
+ System.clearProperty(EXTRACT_RESPONSE_SYS_PROPERTY);
+ TaskStatusUpdateEvent event = TaskStatusUpdateEvent.builder()
+ .taskId("task-noresp")
+ .contextId("ctx-noresp")
+ .status(new TaskStatus(TaskState.TASK_STATE_WORKING))
+ .build();
+
+ decorator.sendNotification(event, null);
+
+ verify(spanBuilder, never()).setAttribute(eq(GENAI_RESPONSE), anyString());
+ }
+
+ static class TestablePushNotificationSenderDecorator extends OpenTelemetryPushNotificationSenderDecorator {
+ TestablePushNotificationSenderDecorator(PushNotificationSender delegate, Tracer tracer) {
+ super(delegate, tracer);
+ }
+ }
+}
diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java
index 9dff9da60..e6671c48c 100644
--- a/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java
+++ b/server-common/src/main/java/org/a2aproject/sdk/server/tasks/BasePushNotificationSender.java
@@ -7,6 +7,7 @@
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.enterprise.inject.Instance;
import jakarta.inject.Inject;
+import jakarta.inject.Named;
import java.io.IOException;
import java.util.ArrayList;
@@ -15,6 +16,8 @@
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ForkJoinPool;
import org.a2aproject.sdk.client.http.A2AHttpClient;
import org.a2aproject.sdk.client.http.A2AHttpClientFactory;
@@ -38,6 +41,13 @@ public class BasePushNotificationSender implements PushNotificationSender {
private static final Logger LOGGER = LoggerFactory.getLogger(BasePushNotificationSender.class);
public static final int DEFAULT_PAGE_SIZE = 100;
+ /**
+ * Named qualifier for the optional {@link Executor} used to dispatch push notifications
+ * asynchronously. When an OpenTelemetry context-propagating executor is produced under
+ * this name, trace context is automatically carried across the async boundary.
+ */
+ public static final String DISPATCH_EXECUTOR_NAME = "a2a.pushNotificationDispatch";
+
// Fields set by constructor injection cannot be final. We need a noargs constructor for
// Jakarta compatibility, and it seems that making fields set by constructor injection
// final, is not proxyable in all runtimes
@@ -45,6 +55,7 @@ public class BasePushNotificationSender implements PushNotificationSender {
private PushNotificationConfigStore configStore;
private Map formattersByVersion;
private PushNotificationUrlValidator urlValidator;
+ private Executor dispatchExecutor = ForkJoinPool.commonPool();
/**
@@ -74,11 +85,14 @@ public BasePushNotificationSender(PushNotificationConfigStore configStore,
@Inject
public BasePushNotificationSender(PushNotificationConfigStore configStore,
Instance formatters,
- PushNotificationUrlValidator urlValidator) {
- this.httpClient = A2AHttpClientFactory.create();
+ PushNotificationUrlValidator urlValidator,
+ Instance httpClient,
+ @Named(DISPATCH_EXECUTOR_NAME) Instance dispatchExecutor) {
+ this.httpClient = httpClient.isResolvable() ? httpClient.get() : A2AHttpClientFactory.create();
this.configStore = configStore;
this.formattersByVersion = toFormatterMap(formatters);
this.urlValidator = urlValidator;
+ this.dispatchExecutor = dispatchExecutor.isResolvable() ? dispatchExecutor.get() : ForkJoinPool.commonPool();
}
public BasePushNotificationSender(PushNotificationConfigStore configStore, A2AHttpClient httpClient,
@@ -171,7 +185,8 @@ private CompletableFuture dispatch(StreamingEventKind event,
@Nullable Task taskSnapshot,
TaskPushNotificationConfig pushInfo,
Map versionsByConfigId) {
- return CompletableFuture.supplyAsync(() -> dispatchNotification(event, taskSnapshot, pushInfo, versionsByConfigId));
+ return CompletableFuture.supplyAsync(
+ () -> dispatchNotification(event, taskSnapshot, pushInfo, versionsByConfigId), dispatchExecutor);
}
private boolean dispatchNotification(StreamingEventKind event,