Merge pull request #61 from kabir/align-main

Align with main, main testsuite passing

Author: kabir

.github/workflows/run-tck.yml

@@ -13,9 +13,12 @@ on:
 
 env:
   # Ref (branch/tag/SHA) of the TCK
-  TCK_VERSION: main
+  TCK_VERSION: 1.0-dev
   # Tells uv to not need a venv, and instead use system
   UV_SYSTEM_PYTHON: 1
+  # Env vars to configure SUT hosts
+  SUT_HOST: localhost:8080
+  SUT_GRPC_HOST: localhost:9555
 
 # Only run the latest job
 concurrency:
@@ -83,13 +86,14 @@ jobs:
         working-directory: a2a-tck
       - name: Start the WildFly SUT
         run: |
-          SUT_JSONRPC_URL=http://localhost:8080 SUT_REST_URL=http://localhost:8080 SUT_GRPC_URL=http://localhost:9555 mvn wildfly:start -B -Dversion.sdk=${SDK_VERSION} -pl tck -Dstartup-timeout=120 -Dwildfly.serverArgs="--stability=preview"
+          mvn wildfly:start -B -Dversion.sdk=${SDK_VERSION} -pl tck -Dstartup-timeout=120 -Dwildfly.serverArgs="--stability=preview"
       - name: Run TCK
-        env:
-          SUT_JSONRPC_URL: http://localhost:8080
-          TCK_STREAMING_TIMEOUT: 4.0
+        # Might not be needed any longer
+        # env:
+        #  TCK_STREAMING_TIMEOUT: 4.0
         run: |
-          ./run_tck.py --sut-url ${SUT_JSONRPC_URL} --category all --transports jsonrpc,grpc,rest --compliance-report report.json
+          set -o pipefail
+          uv run ./run_tck.py --sut-host http://${{ env.SUT_HOST }} -v 2>&1 | tee tck-output.log
         working-directory: a2a-tck
       - name: Shutdown the WildFly SUT
         run: |

README.md

@@ -46,7 +46,7 @@ In the `tck/pom.xml` we enable both JSON-RPC and gRPC, and have the following de
 * `org.wildfly.a2a:a2a-java-sdk-server-jakarta-grpc` - this is the dependency for **gRPC** support.
     * We exclude the gRPC core libraries (`io.grpc` and `com.google.protobuf:protobuf-java`). This is because when deploying to WildFly with gRPC support, the server is provisioned with the WildFly gRPC feature-pack, which already provides these libraries. Including them in the `.war` would lead to conflicts. If you only want to support JSON-RPC, you can omit this dependency.
 * `jakarta.ws.rs:jakarta.ws.rs-api` - this is not part of the dependencies brought in via the A2A dependencies but is needed to compile the TCK module. Since it is provided by WildFly, we make the scope `provided` so it is not included in the `.war`.
-* `org.a2aproject.sdk:a2a-tck-server` - this is the application, which contains the `AgentExecutor` and `AgentCard` implementations for the TCK. In your case, they will most likely be implemented in the project you use to create the `.war`.
+* `org.a2aproject.sdk:a2a-java-sdk-tck-sut` - this is the application, which contains the `AgentExecutor` and `AgentCard` implementations for the TCK. In your case, they will most likely be implemented in the project you use to create the `.war`.
     * In this case we exclude all transitive dependencies, since we are doing the main dependency management via the `org.wildfly.a2a:a2a-java-sdk-jakarta-jsonrpc` and `org.wildfly.a2a:a2a-java-sdk-server-jakarta-grpc` dependencies.
 
 If you are deploying to WildFly and want to use gRPC, you will also need to provision the server with the gRPC feature pack. You can see how this is done in the `wildfly-maven-plugin` configuration in the `tck/pom.xml`. Since the gRPC subsystem and feature pack are currently at the `preview` stability level, you will need to start the server with the `--stability=preview` argument.

impl/grpc/src/main/java/org/wildfly/extras/a2a/server/apps/grpc/A2AExtensionsInterceptor.java

@@ -0,0 +1,49 @@
+package org.wildfly.extras.a2a.server.apps.grpc;
+
+import org.a2aproject.sdk.common.A2AHeaders;
+import org.a2aproject.sdk.transport.grpc.context.GrpcContextKeys;
+
+import io.grpc.Context;
+import io.grpc.Contexts;
+import io.grpc.Metadata;
+import io.grpc.ServerCall;
+import io.grpc.ServerCallHandler;
+import io.grpc.ServerInterceptor;
+
+/**
+ * gRPC server interceptor that extracts A2A protocol headers from request
+ * metadata and stores them in the gRPC {@link Context} for use by
+ * {@link org.a2aproject.sdk.transport.grpc.handler.GrpcHandler}.
+ *
+ * <p>WildFly's gRPC subsystem discovers {@link ServerInterceptor} implementations
+ * in the deployment and applies them automatically.
+ */
+public class A2AExtensionsInterceptor implements ServerInterceptor {
+
+    private static final Metadata.Key<String> VERSION_KEY = Metadata.Key.of(
+            A2AHeaders.A2A_VERSION.toLowerCase(), Metadata.ASCII_STRING_MARSHALLER);
+    private static final Metadata.Key<String> EXTENSIONS_KEY = Metadata.Key.of(
+            A2AHeaders.A2A_EXTENSIONS.toLowerCase(), Metadata.ASCII_STRING_MARSHALLER);
+
+    @Override
+    public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
+            ServerCall<ReqT, RespT> call, Metadata metadata, ServerCallHandler<ReqT, RespT> next) {
+        String version = metadata.get(VERSION_KEY);
+        String extensions = metadata.get(EXTENSIONS_KEY);
+
+        Context context = Context.current()
+                .withValue(GrpcContextKeys.METADATA_KEY, metadata)
+                .withValue(GrpcContextKeys.GRPC_METHOD_NAME_KEY, call.getMethodDescriptor().getFullMethodName())
+                .withValue(GrpcContextKeys.METHOD_NAME_KEY,
+                        GrpcContextKeys.METHOD_MAPPING.get(call.getMethodDescriptor().getBareMethodName()));
+
+        if (version != null) {
+            context = context.withValue(GrpcContextKeys.VERSION_HEADER_KEY, version);
+        }
+        if (extensions != null) {
+            context = context.withValue(GrpcContextKeys.EXTENSIONS_HEADER_KEY, extensions);
+        }
+
+        return Contexts.interceptCall(context, call, metadata, next);
+    }
+}

impl/grpc/src/main/java/org/wildfly/extras/a2a/server/apps/grpc/WildFlyGrpcHandler.java

@@ -2,6 +2,7 @@
 
 import java.util.concurrent.Executor;
 import java.util.concurrent.Flow;
+import java.util.function.Supplier;
 
 import org.a2aproject.sdk.jsonrpc.common.wrappers.ListTasksResult;
 import org.a2aproject.sdk.server.ServerCallContext;
@@ -173,5 +174,13 @@ public void onDeleteTaskPushNotificationConfig(DeleteTaskPushNotificationConfigP
                 return null;
             });
         }
+
+        @Override
+        public void validateRequestedTask(String requestedTaskId) throws A2AError {
+            withDeploymentClassLoader(() -> {
+                delegate.validateRequestedTask(requestedTaskId);
+                return null;
+            });
+        }
     }
 }

impl/jsonrpc/src/main/java/org/wildfly/extras/a2a/server/apps/jsonrpc/A2AServerResource.java

@@ -176,56 +176,66 @@ public void handleStreamingRequests(
         ServerCallContext context = createCallContext(httpRequest, securityContext);
         LOGGER.debug("Handling streaming request with custom SSE response");
 
-        // Set SSE headers manually for proper streaming
+        // Parse and validate before committing to SSE response format.
+        // Validation errors (e.g. terminal task) must be returned as plain
+        // JSON-RPC error responses, not SSE events.
+        A2ARequest<?> request = null;
+        try {
+            request = JSONRPCUtils.parseRequestBody(body, null);
+            validateStreamingRequest((StreamingJSONRPCRequest<?>) request);
+        } catch (A2AError e) {
+            LOGGER.debug("A2AError validating streaming request: {}", e.getMessage());
+            sendJsonRpcError(response, request != null ? request.getId() : null, e);
+            return;
+        } catch (InvalidParamsJsonMappingException e) {
+            LOGGER.warn("Invalid params in streaming request: {}", e.getMessage());
+            sendJsonRpcError(response, e.getId(), new InvalidParamsError(null, e.getMessage(), null));
+            return;
+        } catch (MethodNotFoundJsonMappingException e) {
+            LOGGER.warn("Method not found in streaming request: {}", e.getMessage());
+            sendJsonRpcError(response, e.getId(), new MethodNotFoundError(null, e.getMessage(), null));
+            return;
+        } catch (IdJsonMappingException e) {
+            LOGGER.warn("Invalid request ID in streaming request: {}", e.getMessage());
+            sendJsonRpcError(response, e.getId(), new InvalidRequestError(null, e.getMessage(), null));
+            return;
+        } catch (JsonMappingException e) {
+            LOGGER.warn("JSON mapping error in streaming request: {}", e.getMessage(), e);
+            sendJsonRpcError(response, null, new InvalidRequestError(null, e.getMessage(), null));
+            return;
+        } catch (JsonSyntaxException e) {
+            LOGGER.warn("JSON syntax error in streaming request: {}", e.getMessage());
+            sendJsonRpcError(response, null, new JSONParseError(e.getMessage()));
+            return;
+        } catch (JsonProcessingException e) {
+            LOGGER.warn("JSON processing error in streaming request: {}", e.getMessage());
+            sendJsonRpcError(response, null, new JSONParseError(e.getMessage()));
+            return;
+        } catch (Throwable e) {
+            LOGGER.error("Unexpected error processing streaming request: {}", e.getMessage(), e);
+            sendJsonRpcError(response, null, new InternalError(e.getMessage()));
+            return;
+        }
+
+        // Validation passed — now commit to SSE response format
         response.setContentType(MediaType.SERVER_SENT_EVENTS);
         response.setCharacterEncoding("UTF-8");
         response.setHeader(HttpHeaders.CACHE_CONTROL, "no-cache");
 
-        A2ARequest<?> request = null;
         try {
-            // Parse the request body
-            request = JSONRPCUtils.parseRequestBody(body, null);
-
-            // Get the publisher synchronously to avoid connection closure issues
             Flow.Publisher<? extends A2AResponse<?>> publisher = createStreamingPublisher((StreamingJSONRPCRequest<?>) request, context);
             LOGGER.debug("Created streaming publisher: {}", publisher);
 
             if (publisher != null) {
-                // Handle the streaming response with custom SSE formatting
                 LOGGER.debug("Handling custom SSE response for publisher: {}", publisher);
                 handleCustomSSEResponse(publisher, response, context);
             } else {
-                // Handle unsupported request types
                 LOGGER.debug("Unsupported streaming request type: {}", request.getClass().getSimpleName());
                 response.sendError(HttpServletResponse.SC_BAD_REQUEST, "Unsupported streaming request type");
             }
-        } catch (MethodNotFoundJsonMappingException e) {
-            LOGGER.warn("Method not found in streaming request: {}", e.getMessage());
-            sendErrorSSE(response, e.getId(), new MethodNotFoundError());
-        } catch (InvalidParamsJsonMappingException e) {
-            LOGGER.warn("Invalid params in streaming request: {}", e.getMessage());
-            sendErrorSSE(response, e.getId(), new InvalidParamsError());
-        } catch (IdJsonMappingException e) {
-            LOGGER.warn("Invalid request ID in streaming request: {}", e.getMessage());
-            sendErrorSSE(response, e.getId(), new InvalidRequestError());
-        } catch (JsonMappingException e) {
-            LOGGER.warn("JSON mapping error in streaming request: {}", e.getMessage(), e);
-            // Check if this is a parse error wrapped in a mapping exception
-            if (e.getCause() instanceof JsonProcessingException) {
-                sendErrorSSE(response, null, new JSONParseError());
-            } else {
-                // Otherwise it's an invalid request (valid JSON but doesn't match schema)
-                sendErrorSSE(response, null, new InvalidRequestError());
-            }
-        } catch (JsonSyntaxException e) {
-            LOGGER.warn("JSON syntax error in streaming request: {}", e.getMessage());
-            sendErrorSSE(response, null, new JSONParseError());
-        } catch (JsonProcessingException e) {
-            LOGGER.warn("JSON processing error in streaming request: {}", e.getMessage());
-            sendErrorSSE(response, null, new JSONParseError());
         } catch (A2AError e) {
             LOGGER.debug("A2AError in streaming request: {}", e.getMessage());
-            sendErrorSSE(response, request != null ? request.getId() : null, e);
+            sendErrorSSE(response, request.getId(), e);
         } catch (Throwable e) {
             LOGGER.error("Unexpected error processing streaming request: {}", e.getMessage(), e);
             sendErrorSSE(response, null, new InternalError(e.getMessage()));
@@ -293,6 +303,20 @@ private A2AResponse<?> processNonStreamingRequest(NonStreamingJSONRPCRequest<?>
         }
     }
 
+    /**
+     * Validates a streaming request before entering SSE mode.
+     * Throws A2AError if the task is in a terminal state or not found.
+     * This must be called before setting SSE headers so that errors
+     * are returned as plain JSON-RPC error responses, not SSE events.
+     */
+    private void validateStreamingRequest(StreamingJSONRPCRequest<?> request) throws A2AError {
+        if (request instanceof SendStreamingMessageRequest req) {
+            jsonRpcHandler.validateRequestedTask(req.getParams().message().taskId());
+        } else if (request instanceof SubscribeToTaskRequest req) {
+            jsonRpcHandler.validateRequestedTask(req.getParams().id());
+        }
+    }
+
     /**
      * Creates a streaming publisher for the given request.
      * This method runs synchronously to avoid connection closure issues.
@@ -415,6 +439,23 @@ private A2AResponse<?> generateErrorResponse(A2ARequest<?> request, A2AError err
         return new A2AErrorResponse(request.getId(), error);
     }
 
+    /**
+     * Sends a plain JSON-RPC error response (Content-Type: application/json).
+     * Used for pre-streaming validation errors that should not be sent as SSE.
+     */
+    private void sendJsonRpcError(HttpServletResponse response, Object id, A2AError error) {
+        try {
+            A2AErrorResponse errorResponse = new A2AErrorResponse(id, error);
+            String jsonData = serializeResponse(errorResponse);
+            response.setStatus(HttpServletResponse.SC_OK);
+            response.setContentType(org.a2aproject.sdk.common.MediaType.APPLICATION_JSON);
+            response.getWriter().write(jsonData);
+            response.getWriter().flush();
+        } catch (Exception e) {
+            LOGGER.error("Error sending JSON-RPC error response: {}", e.getMessage(), e);
+        }
+    }
+
     /**
      * Sends an error response as a Server-Sent Event.
      */
@@ -465,7 +506,7 @@ public String getUsername() {
             Map<String, String> headers = new HashMap<>();
             for (Enumeration<String> headerNames = request.getHeaderNames(); headerNames.hasMoreElements() ; ) {
                 String name = headerNames.nextElement();
-                headers.put(name, headers.get(name));
+                headers.put(name, request.getHeader(name));
             }
 
             state.put(HEADERS_KEY, headers);
@@ -478,8 +519,9 @@ public String getUsername() {
             Set<String> requestedExtensions = A2AExtensions.getRequestedExtensions(extensionHeaderValues);
             state.put(TENANT_KEY, extractTenant(request));
             state.put(TRANSPORT_KEY, TransportProtocol.JSONRPC);
-            
-            return new ServerCallContext(user, state, requestedExtensions);
+
+            String requestedVersion = request.getHeader(A2AHeaders.A2A_VERSION);
+            return new ServerCallContext(user, state, requestedExtensions, requestedVersion);
         } else {
             CallContextFactory builder = callContextFactory.get();
             return builder.build(request);

impl/rest/src/main/java/org/wildfly/extras/a2a/server/apps/rest/A2ARestServerResource.java

@@ -124,8 +124,7 @@ public void sendMessageStreaming(String body, @Context HttpServletRequest httpRe
             }
         } finally {
             if (error != null) {
-                httpResponse.setHeader(CONTENT_TYPE, APPLICATION_JSON);
-                httpResponse.sendError(error.getStatusCode(), error.getBody());
+                sendErrorResponse(httpResponse, error);
             } else {
                 handleCustomSSEResponse(streamingResponse.getPublisher(), httpResponse, context);
             }
@@ -149,8 +148,7 @@ public void resubscribeTask(@PathParam("taskId") String taskId, @Context HttpSer
             }
         } finally {
             if (error != null) {
-                httpResponse.setHeader(CONTENT_TYPE, APPLICATION_JSON);
-                httpResponse.sendError(error.getStatusCode(), error.getBody());
+                sendErrorResponse(httpResponse, error);
             } else {
                 handleCustomSSEResponse(streamingResponse.getPublisher(), httpResponse, context);
             }
@@ -420,6 +418,13 @@ public Response deleteTaskPushNotificationConfiguration(@PathParam("taskId") Str
         }
     }
 
+    private void sendErrorResponse(HttpServletResponse httpResponse, RestHandler.HTTPRestResponse error) throws IOException {
+        httpResponse.setStatus(error.getStatusCode());
+        httpResponse.setHeader(CONTENT_TYPE, error.getContentType());
+        httpResponse.getWriter().write(error.getBody());
+        httpResponse.getWriter().flush();
+    }
+
     /**
      * Handles the streaming response using custom SSE formatting.
      * This approach avoids JAX-RS SSE compatibility issues with async publishers.
@@ -428,10 +433,15 @@ public Response deleteTaskPushNotificationConfiguration(@PathParam("taskId") Str
     private void handleCustomSSEResponse(Flow.Publisher<String> publisher,
             HttpServletResponse response,
             ServerCallContext context) throws IOException {
+        response.setHeader(CONTENT_TYPE, MediaType.SERVER_SENT_EVENTS);
+        response.setHeader("Cache-Control", "no-cache");
+        response.setHeader("X-Accel-Buffering", "no");
+
         CompletableFuture<Void> streamingComplete = new CompletableFuture<>();
         try (PrintWriter writer = response.getWriter()) {
+            writer.write(": SSE stream started\n\n");
+            writer.flush();
             publisher.subscribe(new SSESubscriber(streamingComplete, writer, context));
-            // Wait for streaming to complete before method returns
             streamingComplete.get();
         } catch (Exception e) {
             LOGGER.error("Error waiting for streaming completion: {}", e.getMessage(), e);
@@ -485,7 +495,8 @@ public String getUsername() {
                 extensionHeaderValues.add(en.nextElement());
             }
             Set<String> requestedExtensions = A2AExtensions.getRequestedExtensions(extensionHeaderValues);
-            return new ServerCallContext(user, state, requestedExtensions);
+            String requestedVersion = request.getHeader(A2AHeaders.A2A_VERSION);
+            return new ServerCallContext(user, state, requestedExtensions, requestedVersion);
         } else {
             CallContextFactory builder = callContextFactory.get();
             return builder.build(request);

pom.xml

@@ -45,7 +45,7 @@
     <properties>
         <jboss.home>${project.build.directory}${file.separator}wildfly</jboss.home>
         <version.wildfly>39.0.1.Final</version.wildfly>
-        <version.sdk>1.0.0.Beta1-SNAPSHOT</version.sdk>
+        <version.sdk>1.0.0.Beta2-SNAPSHOT</version.sdk>
 
         <version.hamcrest>2.2</version.hamcrest>
         <version.junit>5.12.2</version.junit>
@@ -127,7 +127,7 @@
             </dependency>
             <dependency>
                 <groupId>org.a2aproject.sdk</groupId>
-                <artifactId>a2a-tck-server</artifactId>
+                <artifactId>a2a-java-sdk-tck-sut</artifactId>
                 <version>${version.sdk}</version>
             </dependency>
             <dependency>
@@ -220,8 +220,8 @@
         <module>impl/jsonrpc</module>
         <module>impl/grpc</module>
         <module>impl/rest</module>
-        <module>tck</module>
         <module>tests</module>
+        <module>tck</module>
     </modules>
     <profiles>
         <profile>