fix: improve SSE error handling and add early task validation

ehsavoie · ehsavoie · commit 60b9261351a6 · 2026-04-15T15:10:19.000+02:00
- Detect SSE vs plain JSON responses in HTTP clients - Add early task validation before opening streams - Ensure JSON-RPC spec compliance for error responses - Apply validation consistently across all transports Fixes a2aproject#791 Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>
diff --git a/extras/http-client-vertx/src/main/java/org/a2aproject/sdk/client/http/VertxA2AHttpClient.java b/extras/http-client-vertx/src/main/java/org/a2aproject/sdk/client/http/VertxA2AHttpClient.java
diff --git a/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/OpenTelemetryRequestHandlerDecorator.java b/extras/opentelemetry/server/src/main/java/org/a2aproject/sdk/extras/opentelemetry/OpenTelemetryRequestHandlerDecorator.java
@@ -43,6 +43,7 @@
 import jakarta.enterprise.inject.Any;
 import jakarta.inject.Inject;
 import java.util.concurrent.Flow;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -455,7 +456,11 @@ public void onDeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigP
             span.end();
         }
     }
-    
+
+    public void validateRequestedTask(@Nullable String requestedTaskId) throws A2AError {
+        delegate.validateRequestedTask(requestedTaskId);
+    }
+
     private boolean extractRequest() {
         return Boolean.getBoolean(EXTRACT_REQUEST_SYS_PROPERTY);
     }
diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java
@@ -142,6 +142,8 @@ protected CompletableFuture<Void> asyncRequest(
             Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
                 private Flow.@Nullable Subscription subscription;
                 private volatile boolean errorRaised = false;
+                private boolean isSseStream = false;
+                private boolean firstMeaningfulLineSeen = false;
 
                 @Override
                 public void onSubscribe(Flow.Subscription subscription) {
@@ -151,10 +153,23 @@ public void onSubscribe(Flow.Subscription subscription) {
 
                 @Override
                 public void onNext(String item) {
-                    // SSE messages sometimes start with "data:". Strip that off
-                    if (item != null && item.startsWith("data:")) {
-                        item = item.substring(5).trim();
-                        if (!item.isEmpty()) {
+                    if (item != null && !item.isEmpty()) {
+                        if (!firstMeaningfulLineSeen) {
+                            firstMeaningfulLineSeen = true;
+                            isSseStream = item.startsWith("data:") || item.startsWith(":")
+                                    || item.startsWith("event:") || item.startsWith("id:")
+                                    || item.startsWith("retry:");
+                        }
+                        if (isSseStream) {
+                            if (item.startsWith("data:")) {
+                                String data = item.substring(5).trim();
+                                if (!data.isEmpty()) {
+                                    messageConsumer.accept(data);
+                                }
+                            }
+                            // Other SSE control lines (event:, id:, retry:, :) are ignored
+                        } else {
+                            // Plain error body: deliver so SSEEventListener can parse the typed error
                             messageConsumer.accept(item);
                         }
                     }
diff --git a/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutes.java b/reference/jsonrpc/src/main/java/org/a2aproject/sdk/server/apps/quarkus/A2AServerRoutes.java
@@ -438,7 +438,12 @@ private A2AResponse<?> processNonStreamingRequest(NonStreamingJSONRPCRequest<?>
      * @return a Multi stream of JSON-RPC responses
      */
     private Multi<? extends A2AResponse<?>> processStreamingRequest(
-            A2ARequest<?> request, ServerCallContext context) {
+            A2ARequest<?> request, ServerCallContext context) throws A2AError {
+        if (request instanceof SendStreamingMessageRequest req) {
+            jsonRpcHandler.validateRequestedTask(req.getParams().message().taskId());
+        } else if (request instanceof SubscribeToTaskRequest req) {
+            jsonRpcHandler.validateRequestedTask(req.getParams().id());
+        }
         try {
             Flow.Publisher<? extends A2AResponse<?>> publisher;
             if (request instanceof SendStreamingMessageRequest req) {
@@ -450,8 +455,6 @@ private Multi<? extends A2AResponse<?>> processStreamingRequest(
             }
             return Multi.createFrom().publisher(publisher);
         } catch (A2AError error) {
-            // For streaming endpoints, wrap immediate errors (like TaskNotFoundError,
-            // UnsupportedOperationError) in error response and send as first SSE event
             return Multi.createFrom().item(generateErrorResponse(request, error));
         }
     }
diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java
@@ -1061,6 +1061,23 @@ private MessageSendSetup initMessageSend(MessageSendParams params, ServerCallCon
         return new MessageSendSetup(taskManager, task, requestContext);
     }
 
+    @Override
+    public void validateRequestedTask(@Nullable String requestedTaskId) throws A2AError {
+        if (requestedTaskId == null) {
+            return;
+        }
+        Task task = taskStore.get(requestedTaskId);
+        if (task == null) {
+            throw new TaskNotFoundError();
+        }
+
+        if (task.status().state().isFinal()) {
+            throw new UnsupportedOperationError(null, String.format(
+                    "Cannot send message to task %s: task is in terminal state %s and cannot accept further messages",
+                    task.id(), task.status().state()), null);
+        }
+    }
+
     private @Nullable Task validateRequestedTask(MessageSendParams params) throws A2AError {
         String requestedTaskId = params.message().taskId();
         if (requestedTaskId == null) {
diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/RequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/RequestHandler.java
@@ -18,6 +18,7 @@
 import org.a2aproject.sdk.spec.TaskIdParams;
 import org.a2aproject.sdk.spec.TaskPushNotificationConfig;
 import org.a2aproject.sdk.spec.TaskQueryParams;
+import org.jspecify.annotations.Nullable;
 
 public interface RequestHandler {
     Task onGetTask(
@@ -59,4 +60,6 @@ ListTaskPushNotificationConfigsResult onListTaskPushNotificationConfigs(
     void onDeleteTaskPushNotificationConfig(
             DeleteTaskPushNotificationConfigParams params,
             ServerCallContext context) throws A2AError;
+
+    void validateRequestedTask(@Nullable String requestedTaskId) throws A2AError;
 }
diff --git a/spec-grpc/src/main/java/org/a2aproject/sdk/grpc/utils/JSONRPCUtils.java b/spec-grpc/src/main/java/org/a2aproject/sdk/grpc/utils/JSONRPCUtils.java
@@ -284,11 +284,12 @@ public static StreamResponse parseResponseEvent(String body) throws JsonMappingE
         JsonElement jelement = JsonParser.parseString(body);
         JsonObject jsonRpc = jelement.getAsJsonObject();
         String version = getAndValidateJsonrpc(jsonRpc);
-        Object id = getAndValidateId(jsonRpc);
-        JsonElement paramsNode = jsonRpc.get("result");
+        // Check for error before validating id: per JSON-RPC spec, error responses may have null id
         if (jsonRpc.has("error")) {
             throw processError(jsonRpc.getAsJsonObject("error"));
         }
+        Object id = getAndValidateId(jsonRpc);
+        JsonElement paramsNode = jsonRpc.get("result");
         StreamResponse.Builder builder = StreamResponse.newBuilder();
         parseRequestBody(paramsNode, builder, id);
         return builder.build();
diff --git a/transport/jsonrpc/src/main/java/org/a2aproject/sdk/transport/jsonrpc/handler/JSONRPCHandler.java b/transport/jsonrpc/src/main/java/org/a2aproject/sdk/transport/jsonrpc/handler/JSONRPCHandler.java
@@ -381,16 +381,13 @@ public Flow.Publisher<SendStreamingMessageResponse> onSubscribeToTask(
                             request.getId(),
                             new InvalidRequestError("Streaming is not supported by the agent")));
         }
-
+        requestHandler.validateRequestedTask(request.getParams().id());
         try {
             Flow.Publisher<StreamingEventKind> publisher =
                     requestHandler.onSubscribeToTask(request.getParams(), context);
             // We can't use the convertingProcessor convenience method since that propagates any errors as an error handled
             // via Subscriber.onError() rather than as part of the SendStreamingResponse payload
             return convertToSendStreamingMessageResponse(request.getId(), publisher);
-        } catch (TaskNotFoundError | UnsupportedOperationError e) {
-            // Re-throw initial validation errors for routing layer to wrap in SSE format
-            throw e;
         } catch (A2AError e) {
             // Other A2AError types - wrap inline as part of the stream
             return ZeroPublisher.fromItems(new SendStreamingMessageResponse(request.getId(), e));
@@ -745,4 +742,8 @@ public void onComplete() {
                 }, executor);
             });
     }
+
+    public void validateRequestedTask(String requestedTaskId) {
+        requestHandler.validateRequestedTask(requestedTaskId);
+    }
 }
diff --git a/transport/rest/src/main/java/org/a2aproject/sdk/transport/rest/handler/RestHandler.java b/transport/rest/src/main/java/org/a2aproject/sdk/transport/rest/handler/RestHandler.java
@@ -53,6 +53,7 @@
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsParams;
 import org.a2aproject.sdk.spec.ListTaskPushNotificationConfigsResult;
 import org.a2aproject.sdk.spec.ListTasksParams;
+import org.a2aproject.sdk.spec.MessageSendParams;
 import org.a2aproject.sdk.spec.PushNotificationNotSupportedError;
 import org.a2aproject.sdk.spec.StreamingEventKind;
 import org.a2aproject.sdk.spec.Task;
@@ -297,7 +298,13 @@ public HTTPRestResponse sendStreamingMessage(ServerCallContext context, String t
             org.a2aproject.sdk.grpc.SendMessageRequest.Builder request = org.a2aproject.sdk.grpc.SendMessageRequest.newBuilder();
             parseRequestBody(body, request);
             request.setTenant(tenant);
-            Flow.Publisher<StreamingEventKind> publisher = requestHandler.onMessageSendStream(ProtoUtils.FromProto.messageSendParams(request), context);
+            MessageSendParams params = ProtoUtils.FromProto.messageSendParams(request);
+            try {
+                requestHandler.validateRequestedTask(params.message().taskId());
+            } catch (A2AError e) {
+                return createErrorResponse(e);
+            }
+            Flow.Publisher<StreamingEventKind> publisher = requestHandler.onMessageSendStream(params, context);
             return createStreamingResponse(publisher);
         } catch (A2AError e) {
             return new HTTPRestStreamingResponse(ZeroPublisher.fromItems(new HTTPRestErrorResponse(e).toJson()));
@@ -370,7 +377,6 @@ public HTTPRestResponse createTaskPushNotificationConfiguration(ServerCallContex
             if (!taskIdFromBody.isEmpty() && !taskIdFromBody.equals(taskId)) {
                 throw new InvalidParamsError("Task ID in request body (" + taskIdFromBody + ") does not match task ID in URL path (" + taskId + ").");
             }
-            
             builder.setTenant(tenant);
             builder.setTaskId(taskId);
             TaskPushNotificationConfig result = requestHandler.onCreateTaskPushNotificationConfig(ProtoUtils.FromProto.createTaskPushNotificationConfig(builder), context);
@@ -420,6 +426,11 @@ public HTTPRestResponse subscribeToTask(ServerCallContext context, String tenant
                 return createErrorResponse(new InvalidRequestError("Streaming is not supported by the agent"));
             }
             TaskIdParams params = TaskIdParams.builder().id(taskId).tenant(tenant).build();
+            try {
+                requestHandler.validateRequestedTask(params.id());
+            } catch (A2AError e) {
+                return createErrorResponse(e);
+            }
             Flow.Publisher<StreamingEventKind> publisher = requestHandler.onSubscribeToTask(params, context);
             return createStreamingResponse(publisher);
         } catch (A2AError e) {

Original file line number	Diff line number	Diff line change
`@@ -43,6 +43,7 @@`
`43`	`43`	`import jakarta.enterprise.inject.Any;`
`44`	`44`	`import jakarta.inject.Inject;`
`45`	`45`	`import java.util.concurrent.Flow;`
	`46`	`+import org.jspecify.annotations.Nullable;`
`46`	`47`	`import org.slf4j.Logger;`
`47`	`48`	`import org.slf4j.LoggerFactory;`
`48`	`49`
`@@ -455,7 +456,11 @@ public void onDeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigP`
`455`	`456`	`span.end();`
`456`	`457`	`}`
`457`	`458`	`}`
`458`		`-`
	`459`	`+`
	`460`	`+ public void validateRequestedTask(@Nullable String requestedTaskId) throws A2AError {`
	`461`	`+ delegate.validateRequestedTask(requestedTaskId);`
	`462`	`+ }`
	`463`	`+`
`459`	`464`	`private boolean extractRequest() {`
`460`	`465`	`return Boolean.getBoolean(EXTRACT_REQUEST_SYS_PROPERTY);`
`461`	`466`	`}`