Skip to content

Commit 4dda2ec

Browse files
kabirkabir
committed
fix: harden A2A 0.3 client compatibility
1 parent 3e8a927 commit 4dda2ec

13 files changed

Lines changed: 399 additions & 13 deletions

File tree

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

Lines changed: 31 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -435,6 +435,14 @@ && hasTransportProvider(iface))
435435
}
436436

437437
if (matchedInterface == null) {
438+
for (AgentInterface iface : serverInterfaces) {
439+
String adapter = missingCompatibilityAdapter(iface);
440+
if (adapter != null) {
441+
throw new A2AClientException(iface.protocolBinding() + " "
442+
+ normalizeInterfaceVersion(iface) + " requires "
443+
+ adapter);
444+
}
445+
}
438446
throw new A2AClientException("No compatible transport found");
439447
}
440448
String version = normalizeInterfaceVersionForClient(matchedInterface);
@@ -463,6 +471,29 @@ private static boolean hasTransportProvider(AgentInterface agentInterface) {
463471
new VersionKey(agentInterface.protocolBinding(), version));
464472
}
465473

474+
private @Nullable String missingCompatibilityAdapter(AgentInterface agentInterface) {
475+
String version = normalizeInterfaceVersion(agentInterface);
476+
if ("1.0".equals(version)
477+
|| hasTransportProvider(agentInterface)
478+
|| !isConfiguredBinding(agentInterface.protocolBinding())) {
479+
return null;
480+
}
481+
return switch (agentInterface.protocolBinding()) {
482+
case "JSONRPC" -> "a2a-java-sdk-compat-0.3-client-adapter-jsonrpc";
483+
case "HTTP+JSON" -> "a2a-java-sdk-compat-0.3-client-adapter-rest";
484+
case "GRPC" -> "a2a-java-sdk-compat-0.3-client-adapter-grpc";
485+
default -> "a2a-java-sdk-compat-0.3-client-adapter-" + agentInterface.protocolBinding().toLowerCase();
486+
};
487+
}
488+
489+
private boolean isConfiguredBinding(String binding) {
490+
if (clientTransports.isEmpty() && TransportProtocol.JSONRPC.asString().equals(binding)) {
491+
return true;
492+
}
493+
return clientTransports.keySet().stream()
494+
.anyMatch(clazz -> binding.equals(transportProtocolMapping.get(clazz)));
495+
}
496+
466497
private static String normalizeInterfaceVersionForClient(AgentInterface agentInterface) throws A2AClientException {
467498
try {
468499
return normalizeInterfaceVersion(agentInterface);

‎compat-0.3/client/adapter/src/main/java/org/a2aproject/sdk/compat03/client/adapter/Compat03ClientTransportSupport.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -103,7 +103,7 @@ public static void validateDeletePush(org.a2aproject.sdk.spec.DeleteTaskPushNoti
103103

104104
public static void validatePushList(ListTaskPushNotificationConfigsParams request) {
105105
validateTenant("listTaskPushNotificationConfigurations", request.tenant());
106-
if (request.pageSize() != 0 || (request.pageToken() != null && !request.pageToken().isEmpty())) {
106+
if (request.pageSize() > 0 || (request.pageToken() != null && !request.pageToken().isEmpty())) {
107107
throw unsupported("listTaskPushNotificationConfigurations pagination");
108108
}
109109
}

‎compat-0.3/client/adapter/src/main/java/org/a2aproject/sdk/compat03/client/adapter/Compat03InterceptorSupport.java‎

Lines changed: 12 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,6 @@
11
package org.a2aproject.sdk.compat03.client.adapter;
22

3+
import java.util.HashMap;
34
import java.util.List;
45
import java.util.Map;
56

@@ -18,11 +19,19 @@ private Compat03InterceptorSupport() {
1819

1920
public static PayloadAndHeaders apply(List<ClientCallInterceptor> interceptors, String method,
2021
Object payload, AgentCard card, @Nullable ClientCallContext context, Class<?> expectedType) {
21-
Map<String, String> headers = context == null ? Map.of() : context.getHeaders();
22+
Map<String, String> headers = new HashMap<>();
23+
if (context != null) {
24+
context.getHeaders().forEach((name, value) -> {
25+
if (!A2AHeaders.A2A_VERSION.equalsIgnoreCase(name)) {
26+
headers.put(name, value);
27+
}
28+
});
29+
}
30+
ClientCallContext sanitizedContext = context == null ? null
31+
: new ClientCallContext(context.getState(), headers);
2232
PayloadAndHeaders result = new PayloadAndHeaders(payload, headers);
23-
validateVersionHeader(result.getHeaders());
2433
for (ClientCallInterceptor interceptor : interceptors) {
25-
result = interceptor.intercept(method, result.getPayload(), result.getHeaders(), card, context);
34+
result = interceptor.intercept(method, result.getPayload(), result.getHeaders(), card, sanitizedContext);
2635
if (result == null || result.getPayload() == null) {
2736
throw new A2AClientException("0.3 interceptor returned a forbidden null payload for " + method);
2837
}

‎compat-0.3/client/adapter/src/test/java/org/a2aproject/sdk/compat03/client/adapter/Compat03ClientTransportSupportTest.java‎

Lines changed: 6 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -55,6 +55,12 @@ void acceptsDefaultPushListAndReturnsNoPageToken() {
5555
new ListTaskPushNotificationConfigsParams("task", 0, "", null));
5656
}
5757

58+
@Test
59+
void acceptsNegativeDefaultPushListPageSize() {
60+
Compat03ClientTransportSupport.validatePushList(
61+
new ListTaskPushNotificationConfigsParams("task", -1, "", null));
62+
}
63+
5864
@Test
5965
void mapsContextsAndRequestParametersInBothDirections() {
6066
ClientCallContext context = new ClientCallContext(

‎compat-0.3/client/adapter/src/test/java/org/a2aproject/sdk/compat03/client/adapter/Compat03InterceptorSupportTest.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
package org.a2aproject.sdk.compat03.client.adapter;
2+
3+
import static org.junit.jupiter.api.Assertions.assertFalse;
4+
import static org.junit.jupiter.api.Assertions.assertEquals;
5+
import static org.junit.jupiter.api.Assertions.assertSame;
6+
7+
import java.util.Map;
8+
import java.util.concurrent.atomic.AtomicReference;
9+
10+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext;
11+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallInterceptor;
12+
import org.a2aproject.sdk.client.transport.spi.interceptors.PayloadAndHeaders;
13+
import org.junit.jupiter.api.Test;
14+
15+
class Compat03InterceptorSupportTest {
16+
17+
@Test
18+
void stripsVersionHeaderFromInitialContextBeforeInterceptors() {
19+
AtomicReference<Map<String, String>> interceptedHeaders = new AtomicReference<>();
20+
AtomicReference<ClientCallContext> interceptedContext = new AtomicReference<>();
21+
Map<String, Object> state = Map.of("request-id", "request-1");
22+
ClientCallInterceptor interceptor = new ClientCallInterceptor() {
23+
@Override
24+
public PayloadAndHeaders intercept(String methodName, Object payload, Map<String, String> headers,
25+
org.a2aproject.sdk.spec.AgentCard agentCard, ClientCallContext clientCallContext) {
26+
interceptedHeaders.set(headers);
27+
interceptedContext.set(clientCallContext);
28+
return new PayloadAndHeaders(payload, headers);
29+
}
30+
};
31+
32+
Compat03InterceptorSupport.apply(
33+
java.util.List.of(interceptor), "message/send", "payload", null,
34+
new ClientCallContext(state, Map.of("a2a-version", "1.0", "Authorization", "Bearer token")),
35+
String.class);
36+
37+
assertFalse(interceptedHeaders.get().keySet().stream()
38+
.anyMatch(name -> name.equalsIgnoreCase("A2A-Version")));
39+
assertEquals("Bearer token", interceptedHeaders.get().get("Authorization"));
40+
assertSame(state, interceptedContext.get().getState());
41+
assertFalse(interceptedContext.get().getHeaders().keySet().stream()
42+
.anyMatch(name -> name.equalsIgnoreCase("A2A-Version")));
43+
assertEquals("Bearer token", interceptedContext.get().getHeaders().get("Authorization"));
44+
}
45+
}

‎compat-0.3/client/transport/jsonrpc/src/main/java/org/a2aproject/sdk/compat03/client/transport/jsonrpc/sse/SSEEventListener_v0_3.java‎

Lines changed: 16 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -7,7 +7,9 @@
77
import org.a2aproject.sdk.compat03.json.JsonUtil_v0_3;
88
import org.a2aproject.sdk.compat03.spec.JSONRPCError_v0_3;
99
import org.a2aproject.sdk.compat03.spec.StreamingEventKind_v0_3;
10+
import org.a2aproject.sdk.compat03.spec.Task_v0_3;
1011
import org.a2aproject.sdk.compat03.spec.TaskStatusUpdateEvent_v0_3;
12+
import org.jspecify.annotations.Nullable;
1113

1214
import java.util.concurrent.Future;
1315
import java.util.function.Consumer;
@@ -25,7 +27,7 @@ public SSEEventListener_v0_3(Consumer<StreamingEventKind_v0_3> eventHandler,
2527
this.errorHandler = errorHandler;
2628
}
2729

28-
public void onMessage(String message, Future<Void> completableFuture) {
30+
public void onMessage(String message, @Nullable Future<Void> completableFuture) {
2931
try {
3032
handleMessage(JsonParser.parseString(message).getAsJsonObject(), completableFuture);
3133
} catch (JsonSyntaxException e) {
@@ -34,15 +36,19 @@ public void onMessage(String message, Future<Void> completableFuture) {
3436
fail(e, completableFuture);
3537
} catch (IllegalArgumentException e) {
3638
fail(e, completableFuture);
39+
} catch (IllegalStateException e) {
40+
fail(e, completableFuture);
3741
}
3842
}
3943

40-
public void onError(Throwable throwable, Future<Void> future) {
44+
public void onError(Throwable throwable, @Nullable Future<Void> future) {
4145
completed = true;
4246
if (errorHandler != null) {
4347
errorHandler.accept(throwable);
4448
}
45-
future.cancel(true); // close SSE channel
49+
if (future != null) {
50+
future.cancel(true); // close SSE channel
51+
}
4652
}
4753

4854
public void onComplete() {
@@ -63,7 +69,7 @@ public void onComplete() {
6369
}
6470
}
6571

66-
private void handleMessage(JsonObject jsonObject, Future<Void> future) throws JsonProcessingException_v0_3 {
72+
private void handleMessage(JsonObject jsonObject, @Nullable Future<Void> future) throws JsonProcessingException_v0_3 {
6773
if (jsonObject.has("error")) {
6874
completed = true;
6975
JSONRPCError_v0_3 error = JsonUtil_v0_3.fromJson(jsonObject.get("error").toString(), JSONRPCError_v0_3.class);
@@ -78,15 +84,18 @@ private void handleMessage(JsonObject jsonObject, Future<Void> future) throws Js
7884
String resultJson = jsonObject.get("result").toString();
7985
StreamingEventKind_v0_3 event = JsonUtil_v0_3.fromJson(resultJson, StreamingEventKind_v0_3.class);
8086
eventHandler.accept(event);
81-
if (event instanceof TaskStatusUpdateEvent_v0_3 tsue && tsue.isFinal()) {
82-
future.cancel(true); // close SSE channel
87+
if ((event instanceof TaskStatusUpdateEvent_v0_3 tsue && tsue.isFinal())
88+
|| (event instanceof Task_v0_3 task && task.status().state().isFinal())) {
89+
if (future != null) {
90+
future.cancel(true); // close SSE channel
91+
}
8392
}
8493
} else {
8594
throw new IllegalArgumentException("Unknown message type");
8695
}
8796
}
8897

89-
private void fail(Throwable throwable, Future<Void> future) {
98+
private void fail(Throwable throwable, @Nullable Future<Void> future) {
9099
completed = true;
91100
if (errorHandler != null) {
92101
errorHandler.accept(throwable);

‎compat-0.3/client/transport/jsonrpc/src/test/java/org/a2aproject/sdk/compat03/client/transport/jsonrpc/sse/SSEEventListener_v0_3_Test.java‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -54,6 +54,21 @@ public void testOnEventWithTaskResult() throws Exception {
5454
assertEquals(TaskState_v0_3.WORKING, task.status().state());
5555
}
5656

57+
@Test
58+
public void testFinalTaskResultCancelsStream() {
59+
AtomicReference<StreamingEventKind_v0_3> receivedEvent = new AtomicReference<>();
60+
SSEEventListener_v0_3 listener = new SSEEventListener_v0_3(receivedEvent::set, error -> {});
61+
CancelCapturingFuture future = new CancelCapturingFuture();
62+
63+
String eventData = JsonStreamingMessages_v0_3.STREAMING_TASK_EVENT
64+
.replace("\"working\"", "\"completed\"")
65+
.substring(JsonStreamingMessages_v0_3.STREAMING_TASK_EVENT.indexOf("{"));
66+
listener.onMessage(eventData, future);
67+
68+
assertInstanceOf(Task_v0_3.class, receivedEvent.get());
69+
assertTrue(future.cancelHandlerCalled);
70+
}
71+
5772
@Test
5873
public void testOnEventWithMessageResult() throws Exception {
5974
// Set up event handler
@@ -180,6 +195,19 @@ public void testMalformedEventReportsAndCancels() {
180195
assertTrue(future.cancelHandlerCalled);
181196
}
182197

198+
@Test
199+
public void testNonObjectJsonEventReportsAndCancels() {
200+
AtomicReference<Throwable> receivedError = new AtomicReference<>();
201+
SSEEventListener_v0_3 listener = new SSEEventListener_v0_3(
202+
event -> {}, receivedError::set);
203+
CancelCapturingFuture future = new CancelCapturingFuture();
204+
205+
listener.onMessage("[]", future);
206+
207+
assertNotNull(receivedError.get());
208+
assertTrue(future.cancelHandlerCalled);
209+
}
210+
183211
@Test
184212
public void testOnFailure() {
185213
AtomicBoolean failureHandlerCalled = new AtomicBoolean(false);
@@ -198,6 +226,14 @@ public void testOnFailure() {
198226
assertTrue(future.cancelHandlerCalled);
199227
}
200228

229+
@Test
230+
public void testOnFailureWithNullFutureDoesNotThrow() {
231+
SSEEventListener_v0_3 listener = new SSEEventListener_v0_3(
232+
event -> {}, error -> {});
233+
234+
listener.onError(new RuntimeException("Test exception"), null);
235+
}
236+
201237
@Test
202238
public void testFinalTaskStatusUpdateEventCancels() {
203239
TaskStatusUpdateEvent_v0_3 tsue = new TaskStatusUpdateEvent_v0_3.Builder()
@@ -246,6 +282,17 @@ public void testOnEventWithFinalTaskStatusUpdateEventEventCancels() throws Excep
246282
assertTrue(future.cancelHandlerCalled);
247283
}
248284

285+
@Test
286+
public void testFinalTaskStatusUpdateWithNullFutureDoesNotThrow() {
287+
SSEEventListener_v0_3 listener = new SSEEventListener_v0_3(
288+
event -> {}, error -> {});
289+
290+
String eventData = JsonStreamingMessages_v0_3.STREAMING_STATUS_UPDATE_EVENT_FINAL.substring(
291+
JsonStreamingMessages_v0_3.STREAMING_STATUS_UPDATE_EVENT_FINAL.indexOf("{"));
292+
293+
listener.onMessage(eventData, null);
294+
}
295+
249296

250297
private static class CancelCapturingFuture implements Future<Void> {
251298
private boolean cancelHandlerCalled;

‎compat-0.3/client/transport/rest/src/main/java/org/a2aproject/sdk/compat03/client/transport/rest/sse/RestSSEEventListener_v0_3.java‎

Lines changed: 12 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -10,6 +10,8 @@
1010
import org.a2aproject.sdk.compat03.grpc.StreamResponse;
1111
import org.a2aproject.sdk.compat03.grpc.utils.ProtoUtils_v0_3;
1212
import org.a2aproject.sdk.compat03.spec.StreamingEventKind_v0_3;
13+
import org.a2aproject.sdk.compat03.spec.Task_v0_3;
14+
import org.a2aproject.sdk.compat03.spec.TaskStatusUpdateEvent_v0_3;
1315
import org.jspecify.annotations.Nullable;
1416

1517
public class RestSSEEventListener_v0_3 {
@@ -31,7 +33,9 @@ public void onMessage(String message, @Nullable Future<Void> completableFuture)
3133
JsonFormat.parser().merge(message, builder);
3234
handleMessage(builder.build(), completableFuture);
3335
} catch (InvalidProtocolBufferException e) {
34-
errorHandler.accept(RestErrorMapper_v0_3.mapRestError(message, 500));
36+
if (errorHandler != null) {
37+
errorHandler.accept(RestErrorMapper_v0_3.mapRestError(message, 500));
38+
}
3539
cancel(completableFuture);
3640
}
3741
}
@@ -58,12 +62,18 @@ private void handleMessage(StreamResponse response, @Nullable Future<Void> futur
5862
event = ProtoUtils_v0_3.FromProto.taskArtifactUpdateEvent(response.getArtifactUpdate());
5963
default -> {
6064
LOGGER.warning("Invalid stream response " + response.getPayloadCase());
61-
errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
65+
if (errorHandler != null) {
66+
errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
67+
}
6268
cancel(future);
6369
return;
6470
}
6571
}
6672
eventHandler.accept(event);
73+
if ((event instanceof TaskStatusUpdateEvent_v0_3 statusUpdate && statusUpdate.isFinal())
74+
|| (event instanceof Task_v0_3 task && task.status().state().isFinal())) {
75+
cancel(future);
76+
}
6777
}
6878

6979
private static void cancel(@Nullable Future<Void> future) {

‎compat-0.3/client/transport/rest/src/test/java/org/a2aproject/sdk/compat03/client/transport/rest/sse/RestSSEEventListener_v0_3_Test.java‎

Lines changed: 53 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -38,6 +38,59 @@ void invalidPayloadReportsAndCancels() {
3838
assertTrue(future.cancelled);
3939
}
4040

41+
@Test
42+
void finalStatusUpdateCancels() {
43+
RestSSEEventListener_v0_3 listener = new RestSSEEventListener_v0_3(
44+
event -> {}, error -> {});
45+
CancelCapturingFuture future = new CancelCapturingFuture();
46+
47+
listener.onMessage("""
48+
{
49+
"status_update": {
50+
"task_id": "task-1",
51+
"context_id": "context-1",
52+
"status": {"state": "TASK_STATE_COMPLETED"},
53+
"final": true
54+
}
55+
}""", future);
56+
57+
assertTrue(future.cancelled);
58+
}
59+
60+
@Test
61+
void finalTaskCancels() {
62+
RestSSEEventListener_v0_3 listener = new RestSSEEventListener_v0_3(
63+
event -> {}, error -> {});
64+
CancelCapturingFuture future = new CancelCapturingFuture();
65+
66+
listener.onMessage("""
67+
{
68+
"task": {
69+
"id": "task-1",
70+
"contextId": "context-1",
71+
"status": {"state": "TASK_STATE_COMPLETED"}
72+
}
73+
}""", future);
74+
75+
assertTrue(future.cancelled);
76+
}
77+
78+
@Test
79+
void malformedEventWithoutErrorHandlerDoesNotThrow() {
80+
RestSSEEventListener_v0_3 listener = new RestSSEEventListener_v0_3(
81+
event -> {}, null);
82+
83+
listener.onMessage("{not-json", null);
84+
}
85+
86+
@Test
87+
void invalidPayloadWithoutErrorHandlerDoesNotThrow() {
88+
RestSSEEventListener_v0_3 listener = new RestSSEEventListener_v0_3(
89+
event -> {}, null);
90+
91+
listener.onMessage("{}", null);
92+
}
93+
4194
private static final class CancelCapturingFuture implements Future<Void> {
4295
private boolean cancelled;
4396

‎pom.xml‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -644,6 +644,9 @@
644644
<!-- A2A Protocol v0.3 backward compatibility layer -->
645645
<module>compat-0.3</module>
646646

647+
<!-- Classpath fixture for optional 0.3 client adapter diagnostics -->
648+
<module>tests/client-builder-classpath</module>
649+
647650
<!-- Multiversion reference modules (depend on both v1.0 and v0.3) -->
648651
<module>reference/multiversion-jsonrpc</module>
649652
<module>reference/multiversion-rest</module>

0 commit comments

Comments
 (0)