fix: address protocol compliance issues across transports (#923) · a2aproject/a2a-java@a0ee1c1 · GitHub
Skip to content

Commit a0ee1c1

Browse files
kabirclaude
kabir
and
claude
authored
fix: address protocol compliance issues across transports (#923)
* fix: address protocol compliance issues across transports Fix SSE event listener, gRPC blocking offload handling, JSON-RPC route consistency, multiversion route alignment, and file content mapping to ensure correct behavior for cross-SDK interoperability. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: catch handler exceptions in BlockingOffloadInterceptor If super.onHalfClose() throws inside the executor lambda, the exception propagates uncaught on the worker thread and the gRPC call never closes properly. Add try-catch around the handler invocation to close the call with INTERNAL status on failure. Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> * fix: address review feedback for PR 923 - Add null check on task parameter in ClientTaskManager_v0_3.saveTaskEvent() - Extract duplicated gRPC context forking code into installForkedContextWrapper() in both GrpcHandler and GrpcHandler_v0_3 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com> --------- Co-authored-by: Claude Opus 4.6 <noreply@anthropic.com>
1 parent afe254d commit a0ee1c1

17 files changed

Lines changed: 256 additions & 44 deletions

File tree

‎client/base/src/main/java/org/a2aproject/sdk/client/ClientTaskManager.java‎

Lines changed: 7 additions & 7 deletions

‎client/transport/rest/src/main/java/org/a2aproject/sdk/client/transport/rest/sse/SSEEventListener.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -30,7 +30,7 @@ public SSEEventListener(Consumer<StreamingEventKind> eventHandler,
3030
@Override
3131
public void onMessage(ServerSentEvent event, @Nullable Future<Void> completableFuture) {
3232
try {
33-
log.fine("Streaming message received: " + event.data());
33+
log.fine("REST SSE raw data: " + event.data());
3434
org.a2aproject.sdk.grpc.StreamResponse.Builder builder = org.a2aproject.sdk.grpc.StreamResponse.newBuilder();
3535
JsonFormat.parser().merge(event.data(), builder);
3636
parseAndHandleMessage(builder.build(), completableFuture);

‎compat-0.3/client/base/src/main/java/org/a2aproject/sdk/compat03/client/ClientTaskManager_v0_3.java‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -11,46 +11,46 @@
1111
import org.a2aproject.sdk.compat03.spec.A2AClientInvalidArgsError_v0_3;
1212
import org.a2aproject.sdk.compat03.spec.A2AClientInvalidStateError_v0_3;
1313
import org.a2aproject.sdk.compat03.spec.Message_v0_3;
14-
import org.a2aproject.sdk.compat03.spec.Task_v0_3;
1514
import org.a2aproject.sdk.compat03.spec.TaskArtifactUpdateEvent_v0_3;
1615
import org.a2aproject.sdk.compat03.spec.TaskState_v0_3;
17-
import org.a2aproject.sdk.compat03.spec.TaskStatus_v0_3;
1816
import org.a2aproject.sdk.compat03.spec.TaskStatusUpdateEvent_v0_3;
17+
import org.a2aproject.sdk.compat03.spec.TaskStatus_v0_3;
18+
import org.a2aproject.sdk.compat03.spec.Task_v0_3;
1919
import org.jspecify.annotations.Nullable;
2020

2121
/**
2222
* Helps manage a task's lifecycle during the execution of a request.
2323
* Responsible for retrieving, saving, and updating the task based on
2424
* events received from the agent.
2525
*/
26-
public class ClientTaskManager_v0_3 {
26+
class ClientTaskManager_v0_3 {
2727

2828
private @Nullable Task_v0_3 currentTask;
2929
private @Nullable String taskId;
3030
private @Nullable String contextId;
3131

32-
public ClientTaskManager_v0_3() {
32+
ClientTaskManager_v0_3() {
3333
this.currentTask = null;
3434
this.taskId = null;
3535
this.contextId = null;
3636
}
3737

38-
public Task_v0_3 getCurrentTask() throws A2AClientInvalidStateError_v0_3 {
38+
Task_v0_3 getCurrentTask() throws A2AClientInvalidStateError_v0_3 {
3939
if (currentTask == null) {
4040
throw new A2AClientInvalidStateError_v0_3("No current task");
4141
}
4242
return currentTask;
4343
}
4444

45-
public Task_v0_3 saveTaskEvent(Task_v0_3 task) throws A2AClientInvalidArgsError_v0_3 {
46-
if (currentTask != null) {
45+
Task_v0_3 saveTaskEvent(Task_v0_3 task) throws A2AClientInvalidArgsError_v0_3 {
46+
if (currentTask != null && !currentTask.id().equals(task.id())) {
4747
throw new A2AClientInvalidArgsError_v0_3("Task is already set, create new manager for new tasks.");
4848
}
4949
saveTask(task);
5050
return task;
5151
}
5252

53-
public Task_v0_3 saveTaskEvent(TaskStatusUpdateEvent_v0_3 taskStatusUpdateEvent) throws A2AClientError_v0_3 {
53+
Task_v0_3 saveTaskEvent(TaskStatusUpdateEvent_v0_3 taskStatusUpdateEvent) throws A2AClientError_v0_3 {
5454
if (taskId == null) {
5555
taskId = taskStatusUpdateEvent.taskId();
5656
}
@@ -82,7 +82,7 @@ public Task_v0_3 saveTaskEvent(TaskStatusUpdateEvent_v0_3 taskStatusUpdateEvent)
8282
return currentTask;
8383
}
8484

85-
public Task_v0_3 saveTaskEvent(TaskArtifactUpdateEvent_v0_3 taskArtifactUpdateEvent) {
85+
Task_v0_3 saveTaskEvent(TaskArtifactUpdateEvent_v0_3 taskArtifactUpdateEvent) {
8686
if (taskId == null) {
8787
taskId = taskArtifactUpdateEvent.taskId();
8888
}
@@ -109,7 +109,7 @@ public Task_v0_3 saveTaskEvent(TaskArtifactUpdateEvent_v0_3 taskArtifactUpdateEv
109109
* @param task the task to update
110110
* @return the updated task
111111
*/
112-
public Task_v0_3 updateWithMessage(Message_v0_3 message, Task_v0_3 task) {
112+
Task_v0_3 updateWithMessage(Message_v0_3 message, Task_v0_3 task) {
113113
Task_v0_3.Builder taskBuilder = new Task_v0_3.Builder(task);
114114
List<Message_v0_3> history = new ArrayList<>(task.history());
115115
if (task.status().message() != null) {

‎compat-0.3/reference/grpc/src/main/java/org/a2aproject/sdk/compat03/server/grpc/quarkus/BlockingOffloadInterceptor_v0_3.java‎

Lines changed: 66 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,66 @@
1+
package org.a2aproject.sdk.compat03.server.grpc.quarkus;
2+
3+
import java.util.concurrent.Executor;
4+
5+
import jakarta.enterprise.context.ApplicationScoped;
6+
import jakarta.inject.Inject;
7+
8+
import org.a2aproject.sdk.server.util.async.Internal;
9+
import io.grpc.Context;
10+
import io.grpc.ForwardingServerCallListener.SimpleForwardingServerCallListener;
11+
import io.grpc.Metadata;
12+
import io.grpc.MethodDescriptor;
13+
import io.grpc.ServerCall;
14+
import io.grpc.ServerCallHandler;
15+
import io.grpc.ServerInterceptor;
16+
import io.grpc.Status;
17+
18+
/**
19+
* v0.3 variant of {@code BlockingOffloadInterceptor} in the reference/grpc module.
20+
*/
21+
@ApplicationScoped
22+
public class BlockingOffloadInterceptor_v0_3 implements ServerInterceptor {
23+
24+
private final Executor executor;
25+
26+
@Inject
27+
public BlockingOffloadInterceptor_v0_3(@Internal Executor executor) {
28+
this.executor = executor;
29+
}
30+
31+
@Override
32+
public <ReqT, RespT> ServerCall.Listener<ReqT> interceptCall(
33+
ServerCall<ReqT, RespT> call,
34+
Metadata headers,
35+
ServerCallHandler<ReqT, RespT> next) {
36+
37+
MethodDescriptor.MethodType type = call.getMethodDescriptor().getType();
38+
if (type == MethodDescriptor.MethodType.CLIENT_STREAMING
39+
|| type == MethodDescriptor.MethodType.BIDI_STREAMING) {
40+
return next.startCall(call, headers);
41+
}
42+
43+
ServerCall.Listener<ReqT> delegate = next.startCall(call, headers);
44+
45+
return new SimpleForwardingServerCallListener<ReqT>(delegate) {
46+
@Override
47+
public void onHalfClose() {
48+
Context grpcContext = Context.current().fork();
49+
try {
50+
executor.execute(() -> {
51+
Context previous = grpcContext.attach();
52+
try {
53+
super.onHalfClose();
54+
} catch (Exception e) {
55+
call.close(Status.INTERNAL.withDescription("Error during execution: " + e.getMessage()), new Metadata());
56+
} finally {
57+
grpcContext.detach(previous);
58+
}
59+
});
60+
} catch (Exception e) {
61+
call.close(Status.INTERNAL.withDescription("Failed to offload to worker thread: " + e.getMessage()), new Metadata());
62+
}
63+
}
64+
};
65+
}
66+
}

‎compat-0.3/reference/grpc/src/main/java/org/a2aproject/sdk/compat03/server/grpc/quarkus/QuarkusGrpcHandler_v0_3.java‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -8,6 +8,7 @@
88
import io.quarkus.grpc.GrpcService;
99
import io.quarkus.grpc.RegisterInterceptor;
1010
import io.quarkus.security.Authenticated;
11+
import io.smallrye.common.annotation.Blocking;
1112
import org.a2aproject.sdk.compat03.conversion.Convert_v0_3_To10RequestHandler;
1213
import org.a2aproject.sdk.compat03.spec.AgentCard_v0_3;
1314
import org.a2aproject.sdk.compat03.transport.grpc.handler.CallContextFactory_v0_3;
@@ -17,7 +18,9 @@
1718

1819
@GrpcService
1920
@RegisterInterceptor(A2AExtensionsInterceptor_v0_3.class)
21+
@RegisterInterceptor(BlockingOffloadInterceptor_v0_3.class)
2022
@Authenticated
23+
@Blocking
2124
public class QuarkusGrpcHandler_v0_3 extends GrpcHandler_v0_3 {
2225

2326
private final AgentCard_v0_3 agentCard;

‎compat-0.3/reference/jsonrpc/src/main/java/org/a2aproject/sdk/compat03/server/apps/quarkus/A2AServerRoutes_v0_3.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -97,7 +97,7 @@ void setupRoutes(@Observes Router router) {
9797
} catch (Exception e) {
9898
VertxSecurityHelper.handleGenericError(ctx);
9999
}
100-
});
100+
}, false);
101101

102102
// Only register v0.3 agent card if no real v1.0 agent card producer exists.
103103
// DefaultProducers provides a @DefaultBean AgentCard fallback that is always

‎compat-0.3/reference/rest/src/main/java/org/a2aproject/sdk/compat03/server/rest/quarkus/A2AServerRoutes_v0_3.java‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -75,52 +75,52 @@ void setupRouter(@Observes @Priority(10) Router router) {
7575
.handler(BodyHandler.create())
7676
.blockingHandler(authenticated(ctx -> {
7777
sendMessage(extractBody(ctx), ctx);
78-
}));
78+
}), false);
7979

8080
// POST /v1/message:stream
8181
router.postWithRegex("^\\/v1\\/message:stream$")
8282
.handler(BodyHandler.create())
8383
.blockingHandler(authenticatedStreaming(ctx -> {
8484
sendMessageStreaming(extractBody(ctx), ctx);
85-
}));
85+
}), false);
8686

8787
// GET /v1/tasks/:id
8888
router.get("/v1/tasks/:id")
8989
.order(1)
90-
.blockingHandler(authenticated(this::getTask));
90+
.blockingHandler(authenticated(this::getTask), false);
9191

9292
// POST /v1/tasks/{id}:cancel
9393
router.postWithRegex("^\\/v1\\/tasks\\/([^/]+):cancel$")
9494
.order(1)
95-
.blockingHandler(authenticated(this::cancelTask));
95+
.blockingHandler(authenticated(this::cancelTask), false);
9696

9797
// POST /v1/tasks/{id}:subscribe
9898
router.postWithRegex("^\\/v1\\/tasks\\/([^/]+):subscribe$")
9999
.order(1)
100-
.blockingHandler(authenticatedStreaming(this::resubscribeTask));
100+
.blockingHandler(authenticatedStreaming(this::resubscribeTask), false);
101101

102102
// POST /v1/tasks/:id/pushNotificationConfigs
103103
router.post("/v1/tasks/:id/pushNotificationConfigs")
104104
.order(1)
105105
.handler(BodyHandler.create())
106106
.blockingHandler(authenticated(ctx -> {
107107
setTaskPushNotificationConfiguration(extractBody(ctx), ctx);
108-
}));
108+
}), false);
109109

110110
// GET /v1/tasks/:id/pushNotificationConfigs/:configId
111111
router.get("/v1/tasks/:id/pushNotificationConfigs/:configId")
112112
.order(1)
113-
.blockingHandler(authenticated(this::getTaskPushNotificationConfiguration));
113+
.blockingHandler(authenticated(this::getTaskPushNotificationConfiguration), false);
114114

115115
// GET /v1/tasks/:id/pushNotificationConfigs
116116
router.get("/v1/tasks/:id/pushNotificationConfigs")
117117
.order(2)
118-
.blockingHandler(authenticated(this::listTaskPushNotificationConfigurations));
118+
.blockingHandler(authenticated(this::listTaskPushNotificationConfigurations), false);
119119

120120
// DELETE /v1/tasks/:id/pushNotificationConfigs/:configId
121121
router.delete("/v1/tasks/:id/pushNotificationConfigs/:configId")
122122
.order(1)
123-
.blockingHandler(authenticated(this::deleteTaskPushNotificationConfiguration));
123+
.blockingHandler(authenticated(this::deleteTaskPushNotificationConfiguration), false);
124124

125125
// Only register v0.3 agent card if no real v1.0 agent card producer exists.
126126
// DefaultProducers provides a @DefaultBean AgentCard fallback that is always
@@ -136,7 +136,7 @@ void setupRouter(@Observes @Priority(10) Router router) {
136136
router.get("/v1/card")
137137
.order(1)
138138
.produces(APPLICATION_JSON)
139-
.blockingHandler(authenticated(this::getAuthenticatedExtendedCard));
139+
.blockingHandler(authenticated(this::getAuthenticatedExtendedCard), false);
140140
}
141141

142142
private Handler<RoutingContext> authenticated(Consumer<RoutingContext> action) {

‎compat-0.3/server-conversion/src/main/java/org/a2aproject/sdk/compat03/conversion/mappers/domain/FileContentMapper_v0_3.java‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -51,9 +51,11 @@ default FileContent toV10(FileContent_v0_3 v03) {
5151
}
5252

5353
if (v03 instanceof FileWithBytes_v0_3 v03Bytes) {
54-
return new FileWithBytes(v03Bytes.mimeType(), v03Bytes.name(), v03Bytes.bytes());
54+
String name = v03Bytes.name() != null ? v03Bytes.name() : "";
55+
return new FileWithBytes(v03Bytes.mimeType(), name, v03Bytes.bytes());
5556
} else if (v03 instanceof FileWithUri_v0_3 v03Uri) {
56-
return new FileWithUri(v03Uri.mimeType(), v03Uri.name(), v03Uri.uri());
57+
String name = v03Uri.name() != null ? v03Uri.name() : "";
58+
return new FileWithUri(v03Uri.mimeType(), name, v03Uri.uri());
5759
}
5860

5961
throw new InvalidRequestError(null, "Unrecognized FileContent type: " + v03.getClass().getName(), null);

‎compat-0.3/transport/grpc/src/main/java/org/a2aproject/sdk/compat03/transport/grpc/handler/GrpcHandler_v0_3.java‎

Lines changed: 17 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -43,12 +43,14 @@
4343
import org.a2aproject.sdk.server.auth.UnauthenticatedUser;
4444
import org.a2aproject.sdk.server.auth.User;
4545
import org.a2aproject.sdk.spec.A2AError;
46+
import io.grpc.Context;
4647
import io.grpc.Status;
4748
import io.grpc.stub.StreamObserver;
4849

4950
import java.util.HashSet;
5051
import java.util.List;
5152
import java.util.Set;
53+
import java.util.function.UnaryOperator;
5254

5355
/**
5456
* Abstract gRPC handler for v0.3 protocol with translation layer to v1.0.
@@ -235,6 +237,7 @@ public void sendStreamingMessage(org.a2aproject.sdk.compat03.grpc.SendMessageReq
235237

236238
try {
237239
ServerCallContext context = createCallContext(responseObserver);
240+
installForkedContextWrapper(context);
238241
MessageSendParams_v0_3 params = FromProto.messageSendParams(request);
239242
Flow.Publisher<StreamingEventKind_v0_3> publisher = requestHandler.onMessageSendStream(params, context);
240243
convertToStreamResponse(publisher, responseObserver);
@@ -259,6 +262,7 @@ public void taskSubscription(org.a2aproject.sdk.compat03.grpc.TaskSubscriptionRe
259262

260263
try {
261264
ServerCallContext context = createCallContext(responseObserver);
265+
installForkedContextWrapper(context);
262266
TaskIdParams_v0_3 params = FromProto.taskIdParams(request);
263267
Flow.Publisher<StreamingEventKind_v0_3> publisher = requestHandler.onResubscribeToTask(params, context);
264268
convertToStreamResponse(publisher, responseObserver);
@@ -273,6 +277,19 @@ public void taskSubscription(org.a2aproject.sdk.compat03.grpc.TaskSubscriptionRe
273277
}
274278
}
275279

280+
private void installForkedContextWrapper(ServerCallContext context) {
281+
Context forked = Context.current().fork();
282+
context.getState().put(ServerCallContext.EXECUTION_WRAPPER_KEY,
283+
(UnaryOperator<Runnable>) (runnable -> () -> {
284+
Context prev = forked.attach();
285+
try {
286+
runnable.run();
287+
} finally {
288+
forked.detach(prev);
289+
}
290+
}));
291+
}
292+
276293
private void convertToStreamResponse(Flow.Publisher<StreamingEventKind_v0_3> publisher,
277294
StreamObserver<org.a2aproject.sdk.compat03.grpc.StreamResponse> responseObserver) {
278295
CompletableFuture.runAsync(() -> {

‎jsonrpc-common/src/main/java/org/a2aproject/sdk/jsonrpc/common/json/JsonUtil.java‎

Lines changed: 6 additions & 1 deletion

0 commit comments

Comments
 (0)