Skip to content

Commit 34fec0b

Browse files
kabirkabir
committed
Get TCK working
1 parent 91d8ba1 commit 34fec0b

4 files changed

Lines changed: 140 additions & 39 deletions

File tree

‎.github/workflows/run-tck.yml‎

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -13,9 +13,12 @@ on:
1313

1414
env:
1515
# Ref (branch/tag/SHA) of the TCK
16-
TCK_VERSION: main
16+
TCK_VERSION: 1.0-dev
1717
# Tells uv to not need a venv, and instead use system
1818
UV_SYSTEM_PYTHON: 1
19+
# Env vars to configure SUT hosts
20+
SUT_HOST: localhost:8080
21+
SUT_GRPC_HOST: localhost:9555
1922

2023
# Only run the latest job
2124
concurrency:
@@ -83,13 +86,14 @@ jobs:
8386
working-directory: a2a-tck
8487
- name: Start the WildFly SUT
8588
run: |
86-
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"
89+
mvn wildfly:start -B -Dversion.sdk=${SDK_VERSION} -pl tck -Dstartup-timeout=120 -Dwildfly.serverArgs="--stability=preview"
8790
- name: Run TCK
88-
env:
89-
SUT_JSONRPC_URL: http://localhost:8080
90-
TCK_STREAMING_TIMEOUT: 4.0
91+
# Might not be needed any longer
92+
# env:
93+
# TCK_STREAMING_TIMEOUT: 4.0
9194
run: |
92-
./run_tck.py --sut-url ${SUT_JSONRPC_URL} --category all --transports jsonrpc,grpc,rest --compliance-report report.json
95+
set -o pipefail
96+
uv run ./run_tck.py --sut-host http://${{ env.SUT_HOST }} -v 2>&1 | tee tck-output.log
9397
working-directory: a2a-tck
9498
- name: Shutdown the WildFly SUT
9599
run: |

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
package org.wildfly.extras.a2a.server.apps.grpc;
2+
3+
import org.a2aproject.sdk.common.A2AHeaders;
4+
import org.a2aproject.sdk.transport.grpc.context.GrpcContextKeys;
5+
6+
import io.grpc.Context;
7+
import io.grpc.Contexts;
8+
import io.grpc.Metadata;
9+
import io.grpc.ServerCall;
10+
import io.grpc.ServerCallHandler;
11+
import io.grpc.ServerInterceptor;
12+
13+
/**
14+
* gRPC server interceptor that extracts A2A protocol headers from request
15+
* metadata and stores them in the gRPC {@link Context} for use by
16+
* {@link org.a2aproject.sdk.transport.grpc.handler.GrpcHandler}.
17+
*
18+
* <p>WildFly's gRPC subsystem discovers {@link ServerInterceptor} implementations
19+
* in the deployment and applies them automatically.
20+
*/
21+
public class A2AExtensionsInterceptor implements ServerInterceptor {
22+
23+
private static final Metadata.Key<String> VERSION_KEY = Metadata.Key.of(
24+
A2AHeaders.A2A_VERSION.toLowerCase(), Metadata.ASCII_STRING_MARSHALLER);
25+
private static final Metadata.Key<String> EXTENSIONS_KEY = Metadata.Key.of(
26+
A2AHeaders.A2A_EXTENSIONS.toLowerCase(), Metadata.ASCII_STRING_MARSHALLER);
27+
28+
@Override
29+
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
30+
ServerCall<ReqT, RespT> call, Metadata metadata, ServerCallHandler<ReqT, RespT> next) {
31+
String version = metadata.get(VERSION_KEY);
32+
String extensions = metadata.get(EXTENSIONS_KEY);
33+
34+
Context context = Context.current()
35+
.withValue(GrpcContextKeys.METADATA_KEY, metadata)
36+
.withValue(GrpcContextKeys.GRPC_METHOD_NAME_KEY, call.getMethodDescriptor().getFullMethodName())
37+
.withValue(GrpcContextKeys.METHOD_NAME_KEY,
38+
GrpcContextKeys.METHOD_MAPPING.get(call.getMethodDescriptor().getBareMethodName()));
39+
40+
if (version != null) {
41+
context = context.withValue(GrpcContextKeys.VERSION_HEADER_KEY, version);
42+
}
43+
if (extensions != null) {
44+
context = context.withValue(GrpcContextKeys.EXTENSIONS_HEADER_KEY, extensions);
45+
}
46+
47+
return Contexts.interceptCall(context, call, metadata, next);
48+
}
49+
}

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

Lines changed: 74 additions & 33 deletions
Original file line numberDiff line numberDiff line change
@@ -176,56 +176,66 @@ public void handleStreamingRequests(
176176
ServerCallContext context = createCallContext(httpRequest, securityContext);
177177
LOGGER.debug("Handling streaming request with custom SSE response");
178178

179-
// Set SSE headers manually for proper streaming
179+
// Parse and validate before committing to SSE response format.
180+
// Validation errors (e.g. terminal task) must be returned as plain
181+
// JSON-RPC error responses, not SSE events.
182+
A2ARequest<?> request = null;
183+
try {
184+
request = JSONRPCUtils.parseRequestBody(body, null);
185+
validateStreamingRequest((StreamingJSONRPCRequest<?>) request);
186+
} catch (A2AError e) {
187+
LOGGER.debug("A2AError validating streaming request: {}", e.getMessage());
188+
sendJsonRpcError(response, request != null ? request.getId() : null, e);
189+
return;
190+
} catch (InvalidParamsJsonMappingException e) {
191+
LOGGER.warn("Invalid params in streaming request: {}", e.getMessage());
192+
sendJsonRpcError(response, e.getId(), new InvalidParamsError(null, e.getMessage(), null));
193+
return;
194+
} catch (MethodNotFoundJsonMappingException e) {
195+
LOGGER.warn("Method not found in streaming request: {}", e.getMessage());
196+
sendJsonRpcError(response, e.getId(), new MethodNotFoundError(null, e.getMessage(), null));
197+
return;
198+
} catch (IdJsonMappingException e) {
199+
LOGGER.warn("Invalid request ID in streaming request: {}", e.getMessage());
200+
sendJsonRpcError(response, e.getId(), new InvalidRequestError(null, e.getMessage(), null));
201+
return;
202+
} catch (JsonMappingException e) {
203+
LOGGER.warn("JSON mapping error in streaming request: {}", e.getMessage(), e);
204+
sendJsonRpcError(response, null, new InvalidRequestError(null, e.getMessage(), null));
205+
return;
206+
} catch (JsonSyntaxException e) {
207+
LOGGER.warn("JSON syntax error in streaming request: {}", e.getMessage());
208+
sendJsonRpcError(response, null, new JSONParseError(e.getMessage()));
209+
return;
210+
} catch (JsonProcessingException e) {
211+
LOGGER.warn("JSON processing error in streaming request: {}", e.getMessage());
212+
sendJsonRpcError(response, null, new JSONParseError(e.getMessage()));
213+
return;
214+
} catch (Throwable e) {
215+
LOGGER.error("Unexpected error processing streaming request: {}", e.getMessage(), e);
216+
sendJsonRpcError(response, null, new InternalError(e.getMessage()));
217+
return;
218+
}
219+
220+
// Validation passed — now commit to SSE response format
180221
response.setContentType(MediaType.SERVER_SENT_EVENTS);
181222
response.setCharacterEncoding("UTF-8");
182223
response.setHeader(HttpHeaders.CACHE_CONTROL, "no-cache");
183224

184-
A2ARequest<?> request = null;
185225
try {
186-
// Parse the request body
187-
request = JSONRPCUtils.parseRequestBody(body, null);
188-
189-
// Get the publisher synchronously to avoid connection closure issues
190226
Flow.Publisher<? extends A2AResponse<?>> publisher = createStreamingPublisher((StreamingJSONRPCRequest<?>) request, context);
191227
LOGGER.debug("Created streaming publisher: {}", publisher);
192228

193229
if (publisher != null) {
194-
// Handle the streaming response with custom SSE formatting
195230
LOGGER.debug("Handling custom SSE response for publisher: {}", publisher);
196231
handleCustomSSEResponse(publisher, response, context);
197232
} else {
198-
// Handle unsupported request types
199233
LOGGER.debug("Unsupported streaming request type: {}", request.getClass().getSimpleName());
200234
response.sendError(HttpServletResponse.SC_BAD_REQUEST, "Unsupported streaming request type");
201235
}
202-
} catch (MethodNotFoundJsonMappingException e) {
203-
LOGGER.warn("Method not found in streaming request: {}", e.getMessage());
204-
sendErrorSSE(response, e.getId(), new MethodNotFoundError());
205-
} catch (InvalidParamsJsonMappingException e) {
206-
LOGGER.warn("Invalid params in streaming request: {}", e.getMessage());
207-
sendErrorSSE(response, e.getId(), new InvalidParamsError());
208-
} catch (IdJsonMappingException e) {
209-
LOGGER.warn("Invalid request ID in streaming request: {}", e.getMessage());
210-
sendErrorSSE(response, e.getId(), new InvalidRequestError());
211-
} catch (JsonMappingException e) {
212-
LOGGER.warn("JSON mapping error in streaming request: {}", e.getMessage(), e);
213-
// Check if this is a parse error wrapped in a mapping exception
214-
if (e.getCause() instanceof JsonProcessingException) {
215-
sendErrorSSE(response, null, new JSONParseError());
216-
} else {
217-
// Otherwise it's an invalid request (valid JSON but doesn't match schema)
218-
sendErrorSSE(response, null, new InvalidRequestError());
219-
}
220-
} catch (JsonSyntaxException e) {
221-
LOGGER.warn("JSON syntax error in streaming request: {}", e.getMessage());
222-
sendErrorSSE(response, null, new JSONParseError());
223-
} catch (JsonProcessingException e) {
224-
LOGGER.warn("JSON processing error in streaming request: {}", e.getMessage());
225-
sendErrorSSE(response, null, new JSONParseError());
226236
} catch (A2AError e) {
227237
LOGGER.debug("A2AError in streaming request: {}", e.getMessage());
228-
sendErrorSSE(response, request != null ? request.getId() : null, e);
238+
sendErrorSSE(response, request.getId(), e);
229239
} catch (Throwable e) {
230240
LOGGER.error("Unexpected error processing streaming request: {}", e.getMessage(), e);
231241
sendErrorSSE(response, null, new InternalError(e.getMessage()));
@@ -293,6 +303,20 @@ private A2AResponse<?> processNonStreamingRequest(NonStreamingJSONRPCRequest<?>
293303
}
294304
}
295305

306+
/**
307+
* Validates a streaming request before entering SSE mode.
308+
* Throws A2AError if the task is in a terminal state or not found.
309+
* This must be called before setting SSE headers so that errors
310+
* are returned as plain JSON-RPC error responses, not SSE events.
311+
*/
312+
private void validateStreamingRequest(StreamingJSONRPCRequest<?> request) throws A2AError {
313+
if (request instanceof SendStreamingMessageRequest req) {
314+
jsonRpcHandler.validateRequestedTask(req.getParams().message().taskId());
315+
} else if (request instanceof SubscribeToTaskRequest req) {
316+
jsonRpcHandler.validateRequestedTask(req.getParams().id());
317+
}
318+
}
319+
296320
/**
297321
* Creates a streaming publisher for the given request.
298322
* This method runs synchronously to avoid connection closure issues.
@@ -415,6 +439,23 @@ private A2AResponse<?> generateErrorResponse(A2ARequest<?> request, A2AError err
415439
return new A2AErrorResponse(request.getId(), error);
416440
}
417441

442+
/**
443+
* Sends a plain JSON-RPC error response (Content-Type: application/json).
444+
* Used for pre-streaming validation errors that should not be sent as SSE.
445+
*/
446+
private void sendJsonRpcError(HttpServletResponse response, Object id, A2AError error) {
447+
try {
448+
A2AErrorResponse errorResponse = new A2AErrorResponse(id, error);
449+
String jsonData = serializeResponse(errorResponse);
450+
response.setStatus(HttpServletResponse.SC_OK);
451+
response.setContentType(org.a2aproject.sdk.common.MediaType.APPLICATION_JSON);
452+
response.getWriter().write(jsonData);
453+
response.getWriter().flush();
454+
} catch (Exception e) {
455+
LOGGER.error("Error sending JSON-RPC error response: {}", e.getMessage(), e);
456+
}
457+
}
458+
418459
/**
419460
* Sends an error response as a Server-Sent Event.
420461
*/

‎tck/pom.xml‎

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -64,6 +64,13 @@
6464
<scope>provided</scope>
6565
</dependency>
6666

67+
<!--
68+
Needed for the AgentExecutor which uses some utility classes from here
69+
-->
70+
<dependency>
71+
<groupId>org.a2aproject.sdk</groupId>
72+
<artifactId>a2a-java-sdk-client</artifactId>
73+
</dependency>
6774

6875
<!--
6976
Include the TCK server from the a2a-java project.

0 commit comments

Comments
 (0)