fix: Encapsulate the payload of Push notifications

The payload of push notifications is encapsulated by the
`kind` of payload as specified in [§ 4.3.3. Push Notification Payload](https://a2a-protocol.org/latest/specification/#433-push-notification-payload)

This commit only works for `task` payload and will required more work to support additional payloads (tracked by #490).

This fixes #491
Signed-off-by: Jeff Mesnil <[email protected]>

Author: jmesnil

server-common/src/main/java/io/a2a/server/tasks/BasePushNotificationSender.java

@@ -3,25 +3,25 @@
 import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON;
 import static io.a2a.client.http.A2AHttpClient.CONTENT_TYPE;
 import static io.a2a.common.A2AHeaders.X_A2A_NOTIFICATION_TOKEN;
-import jakarta.enterprise.context.ApplicationScoped;
-import jakarta.inject.Inject;
 
 import java.io.IOException;
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutionException;
 
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.inject.Inject;
+
 import com.fasterxml.jackson.core.JsonProcessingException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import io.a2a.client.http.A2AHttpClient;
 import io.a2a.client.http.JdkA2AHttpClient;
 import io.a2a.spec.PushNotificationConfig;
 import io.a2a.spec.Task;
 import io.a2a.util.Utils;
 
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
 @ApplicationScoped
 public class BasePushNotificationSender implements PushNotificationSender {
 
@@ -80,7 +80,7 @@ private boolean dispatchNotification(Task task, PushNotificationConfig pushInfo)
 
         String body;
         try {
-            body = Utils.OBJECT_MAPPER.writeValueAsString(task);
+            body = Utils.marshalFrom(task);
         } catch (JsonProcessingException e) {
             LOGGER.debug("Error writing value as string: {}", e.getMessage(), e);
             return false;

server-common/src/test/java/io/a2a/server/requesthandlers/AbstractA2ARequestHandlerTest.java

@@ -6,6 +6,7 @@
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.Map;
 import java.util.Properties;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
@@ -15,6 +16,12 @@
 
 import jakarta.enterprise.context.Dependent;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import io.quarkus.arc.profile.IfBuildProfile;
+import org.junit.jupiter.api.AfterEach;
+import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
+
 import io.a2a.client.http.A2AHttpClient;
 import io.a2a.client.http.A2AHttpResponse;
 import io.a2a.server.agentexecution.AgentExecutor;
@@ -30,20 +37,14 @@
 import io.a2a.server.tasks.TaskStore;
 import io.a2a.spec.AgentCapabilities;
 import io.a2a.spec.AgentCard;
+import io.a2a.spec.Event;
 import io.a2a.spec.JSONRPCError;
 import io.a2a.spec.Message;
 import io.a2a.spec.Task;
 import io.a2a.spec.TaskState;
 import io.a2a.spec.TaskStatus;
-import io.a2a.spec.Event;
 import io.a2a.spec.TextPart;
 import io.a2a.util.Utils;
-import io.quarkus.arc.profile.IfBuildProfile;
-import java.util.Map;
-
-import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
 
 public class AbstractA2ARequestHandlerTest {
 
@@ -199,7 +200,11 @@ public PostBuilder body(String body) {
 
             @Override
             public A2AHttpResponse post() throws IOException, InterruptedException {
-                tasks.add(Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE));
+                JsonNode root = Utils.OBJECT_MAPPER.readTree(body);
+                // This will need to be updated for #490 to unmarshall based on the kind of payload
+                JsonNode taskNode = root.elements().next();
+                Task task = Utils.OBJECT_MAPPER.treeToValue(taskNode, Task.TYPE_REFERENCE);
+                tasks.add(task);
                 try {
                     return new A2AHttpResponse() {
                         @Override

server-common/src/test/java/io/a2a/server/tasks/PushNotificationSenderTest.java

@@ -1,6 +1,6 @@
 package io.a2a.server.tasks;
 
-import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON;
+    import static io.a2a.client.http.A2AHttpClient.APPLICATION_JSON;
 import static io.a2a.client.http.A2AHttpClient.CONTENT_TYPE;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertFalse;
@@ -16,17 +16,18 @@
 import java.util.concurrent.TimeUnit;
 import java.util.function.Consumer;
 
+import com.fasterxml.jackson.databind.JsonNode;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 
 import io.a2a.client.http.A2AHttpClient;
 import io.a2a.client.http.A2AHttpResponse;
 import io.a2a.common.A2AHeaders;
-import io.a2a.util.Utils;
 import io.a2a.spec.PushNotificationConfig;
 import io.a2a.spec.Task;
 import io.a2a.spec.TaskState;
 import io.a2a.spec.TaskStatus;
+import io.a2a.util.Utils;
 
 public class PushNotificationSenderTest {
 
@@ -77,7 +78,11 @@ public A2AHttpResponse post() throws IOException, InterruptedException {
                 }
 
                 try {
-                    Task task = Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE);
+                    JsonNode root = Utils.OBJECT_MAPPER.readTree(body);
+                    // This assumes there is always one field in the outer JSON object.
+                    // This will need to be updated for #490 to unmarshall based on the kind of payload
+                    JsonNode taskNode = root.elements().next();
+                    Task task = Utils.OBJECT_MAPPER.treeToValue(taskNode, Task.TYPE_REFERENCE);
                     tasks.add(task);
                     urls.add(url);
                     headers.add(new java.util.HashMap<>(requestHeaders));

spec/src/main/java/io/a2a/util/Utils.java

@@ -2,6 +2,7 @@
 
 import java.util.ArrayList;
 import java.util.List;
+import java.util.Map;
 import java.util.logging.Logger;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
@@ -11,6 +12,7 @@
 
 import io.a2a.spec.Artifact;
 import io.a2a.spec.Part;
+import io.a2a.spec.StreamingEventKind;
 import io.a2a.spec.Task;
 import io.a2a.spec.TaskArtifactUpdateEvent;
 
@@ -69,6 +71,21 @@ public static <T> T unmarshalFrom(String data, TypeReference<T> typeRef) throws
         return OBJECT_MAPPER.readValue(data, typeRef);
     }
 
+    /**
+     * Serializes a StreamingEventKind in a JSON string
+     * <p>
+     * The StreamingEventKind object is wrapped in a JSON field named from its kind (e.g. "task") before
+     * it is serialized
+     *
+     * @param kind the StreamingEventKind to deserialize
+     * @return a JSON String
+     * @throws JsonProcessingException if JSON parsing fails
+     */
+    public static String marshalFrom(StreamingEventKind kind) throws JsonProcessingException {
+        Map<String, StreamingEventKind> wrapper = Map.of(kind.getKind(), kind);
+        return OBJECT_MAPPER.writeValueAsString(wrapper);
+    }
+
     /**
      * Returns the provided value if non-null, otherwise returns the default value.
      * <p>