Skip to content

Commit 97f5877

Browse files
kabirkabir
committed
fix: harden unified client protocol compatibility
Preserve binary data, JSON nulls, empty objects, and message extensions when adapting A2A 0.3 transports. Reject unrepresentable REST/gRPC fields before and after interceptors. Align legacy REST subscription and push-config requests with the binding, and retain HTTP response details through the unified exception API. Preserve compatibility with older SDK request shapes on reference servers. Add independent wire and conversion regression coverage, and require lifecycle tests to observe specific updates and queue closure. Validation: staged diff passes whitespace checks. Tests and builds were not run at the user's request.
1 parent b14ba92 commit 97f5877

39 files changed

Lines changed: 1034 additions & 211 deletions

File tree

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -39,22 +39,22 @@ public GrpcCompat03ClientTransport(GrpcTransport_v0_3 delegate, AgentCard card,
3939

4040
@Override
4141
public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context) {
42-
Compat03ClientTransportSupport.validateMessageSend(request);
42+
Compat03ClientTransportSupport.validateProtobufMessageSend(request);
4343
PayloadAndHeaders payload = apply(A2AMethods.SEND_MESSAGE_METHOD,
4444
ProtoUtils.ToProto.sendMessageRequest(request), org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
4545
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10(delegate.sendMessage(
46-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(payloadPayload(payload))),
46+
Compat03ClientTransportSupport.toV03Protobuf(ProtoUtils.FromProto.messageSendParams(payloadPayload(payload))),
4747
legacyContext(context, payload))));
4848
}
4949

5050
@Override
5151
public void sendMessageStreaming(MessageSendParams request, Consumer<StreamingEventKind> events,
5252
Consumer<Throwable> errors, @Nullable ClientCallContext context) {
53-
Compat03ClientTransportSupport.validateMessageSend(request);
53+
Compat03ClientTransportSupport.validateProtobufMessageSend(request);
5454
PayloadAndHeaders payload = apply(A2AMethods.SEND_STREAMING_MESSAGE_METHOD,
5555
ProtoUtils.ToProto.sendMessageRequest(request), org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
5656
Compat03ClientTransportSupport.run(() -> delegate.sendMessageStreaming(
57-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(payloadPayload(payload))),
57+
Compat03ClientTransportSupport.toV03Protobuf(ProtoUtils.FromProto.messageSendParams(payloadPayload(payload))),
5858
event -> events.accept(Compat03ClientTransportSupport.toV10(event)),
5959
Compat03ClientTransportSupport.mapAsyncError(errors), legacyContext(context, payload)));
6060
}
@@ -71,11 +71,11 @@ public Task getTask(TaskQueryParams request, @Nullable ClientCallContext context
7171

7272
@Override
7373
public Task cancelTask(CancelTaskParams request, @Nullable ClientCallContext context) {
74-
Compat03ClientTransportSupport.validateCancel(request);
74+
Compat03ClientTransportSupport.validateProtobufCancel(request);
7575
PayloadAndHeaders payload = apply(A2AMethods.CANCEL_TASK_METHOD, ProtoUtils.ToProto.cancelTaskRequest(request),
7676
org.a2aproject.sdk.grpc.CancelTaskRequest.class, context);
7777
return Compat03ClientTransportSupport.call(() -> TaskMapper_v0_3.INSTANCE.toV10(delegate.cancelTask(
78-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.cancelTaskParams(cancelPayload(payload))),
78+
Compat03ClientTransportSupport.toV03Protobuf(ProtoUtils.FromProto.cancelTaskParams(cancelPayload(payload))),
7979
legacyContext(context, payload))));
8080
}
8181

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

Lines changed: 105 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,105 @@
1+
package org.a2aproject.sdk.compat03.client.adapter.grpc;
2+
3+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
4+
import static org.junit.jupiter.api.Assertions.assertThrows;
5+
import static org.junit.jupiter.api.Assertions.assertTrue;
6+
7+
import java.util.List;
8+
import java.util.Map;
9+
import java.util.function.Consumer;
10+
11+
import com.google.protobuf.Struct;
12+
import com.google.protobuf.Value;
13+
import io.grpc.ManagedChannel;
14+
import io.grpc.inprocess.InProcessChannelBuilder;
15+
import io.grpc.inprocess.InProcessServerBuilder;
16+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext;
17+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallInterceptor;
18+
import org.a2aproject.sdk.client.transport.spi.interceptors.PayloadAndHeaders;
19+
import org.a2aproject.sdk.compat03.client.adapter.Compat03ClientTransportBase;
20+
import org.a2aproject.sdk.compat03.client.transport.grpc.GrpcTransport_v0_3;
21+
import org.a2aproject.sdk.compat03.client.transport.spi.interceptors.ClientCallContext_v0_3;
22+
import org.a2aproject.sdk.compat03.spec.EventKind_v0_3;
23+
import org.a2aproject.sdk.compat03.spec.MessageSendParams_v0_3;
24+
import org.a2aproject.sdk.compat03.spec.StreamingEventKind_v0_3;
25+
import org.a2aproject.sdk.compat03.spec.TaskIdParams_v0_3;
26+
import org.a2aproject.sdk.compat03.spec.Task_v0_3;
27+
import org.a2aproject.sdk.spec.A2AClientException;
28+
import org.a2aproject.sdk.spec.AgentCapabilities;
29+
import org.a2aproject.sdk.spec.AgentCard;
30+
import org.a2aproject.sdk.spec.AgentInterface;
31+
import org.a2aproject.sdk.spec.CancelTaskParams;
32+
import org.a2aproject.sdk.spec.Message;
33+
import org.a2aproject.sdk.spec.MessageSendParams;
34+
import org.a2aproject.sdk.spec.TextPart;
35+
import org.a2aproject.sdk.spec.UnsupportedOperationError;
36+
import org.jspecify.annotations.Nullable;
37+
import org.junit.jupiter.api.Test;
38+
39+
class GrpcCompat03UnsupportedFieldsTest {
40+
@Test
41+
void rejectsOriginalFieldsAndInterceptorMutationsBeforeDelegation() {
42+
AgentCard card = AgentCard.builder().name("legacy").description("legacy").version("1")
43+
.url("https://example.test").capabilities(AgentCapabilities.builder().build())
44+
.defaultInputModes(List.of("text")).defaultOutputModes(List.of("text")).skills(List.of())
45+
.supportedInterfaces(List.of(new AgentInterface("GRPC", "https://example.test", null, "0.3"))).build();
46+
ManagedChannel channel = InProcessChannelBuilder.forName(InProcessServerBuilder.generateName())
47+
.directExecutor().build();
48+
var delegate = new GrpcTransport_v0_3(channel, Compat03ClientTransportBase.legacyCard(card)) {
49+
@Override
50+
public EventKind_v0_3 sendMessage(MessageSendParams_v0_3 request, @Nullable ClientCallContext_v0_3 context) {
51+
throw new AssertionError("Unsupported request reached the delegate");
52+
}
53+
54+
@Override
55+
public void sendMessageStreaming(MessageSendParams_v0_3 request, Consumer<StreamingEventKind_v0_3> events,
56+
Consumer<Throwable> errors, @Nullable ClientCallContext_v0_3 context) {
57+
throw new AssertionError("Unsupported streaming request reached the delegate");
58+
}
59+
60+
@Override
61+
public Task_v0_3 cancelTask(TaskIdParams_v0_3 request, @Nullable ClientCallContext_v0_3 context) {
62+
throw new AssertionError("Unsupported cancellation reached the delegate");
63+
}
64+
};
65+
try {
66+
for (boolean intercepted : List.of(false, true)) {
67+
var transport = new GrpcCompat03ClientTransport(delegate, card,
68+
intercepted ? List.of(unsupportedFieldsInterceptor()) : List.of());
69+
var request = new MessageSendParams(new Message(Message.Role.ROLE_USER, List.of(new TextPart("hello")),
70+
"request", null, null, intercepted ? null : List.of("referenced-task"), null, null), null, null);
71+
assertUnsupported("referenceTaskIds", () -> transport.sendMessage(request, null));
72+
assertUnsupported("referenceTaskIds", () -> transport.sendMessageStreaming(request,
73+
event -> { }, error -> { throw new AssertionError(error); }, null));
74+
var cancel = new CancelTaskParams("task", null, intercepted ? Map.of() : Map.of("reason", "stop"));
75+
assertUnsupported("metadata", () -> transport.cancelTask(cancel, null));
76+
}
77+
} finally {
78+
channel.shutdownNow();
79+
}
80+
}
81+
82+
private static void assertUnsupported(String field, Runnable call) {
83+
A2AClientException error = assertThrows(A2AClientException.class, call::run);
84+
assertInstanceOf(UnsupportedOperationError.class, error.getCause());
85+
assertTrue(error.getMessage().contains(field));
86+
}
87+
88+
private static ClientCallInterceptor unsupportedFieldsInterceptor() {
89+
return new ClientCallInterceptor() {
90+
@Override
91+
public PayloadAndHeaders intercept(String method, @Nullable Object payload, Map<String, String> headers,
92+
@Nullable AgentCard card, @Nullable ClientCallContext context) {
93+
if (payload instanceof org.a2aproject.sdk.grpc.SendMessageRequest) {
94+
var request = ((org.a2aproject.sdk.grpc.SendMessageRequest) payload).toBuilder();
95+
request.getMessageBuilder().addReferenceTaskIds("referenced-task");
96+
return new PayloadAndHeaders(request.build(), headers);
97+
}
98+
Struct metadata = Struct.newBuilder().putFields("reason", Value.newBuilder().setStringValue("stop").build()).build();
99+
var request = ((org.a2aproject.sdk.grpc.CancelTaskRequest) payload).toBuilder();
100+
request.setMetadata(metadata);
101+
return new PayloadAndHeaders(request.build(), headers);
102+
}
103+
};
104+
}
105+
}

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

Lines changed: 6 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -39,25 +39,25 @@ public RestCompat03ClientTransport(RestTransport_v0_3 delegate, AgentCard card,
3939

4040
@Override
4141
public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context) {
42-
Compat03ClientTransportSupport.validateMessageSend(request);
42+
Compat03ClientTransportSupport.validateProtobufMessageSend(request);
4343
PayloadAndHeaders payload = apply(A2AMethods.SEND_MESSAGE_METHOD,
4444
org.a2aproject.sdk.grpc.SendMessageRequest.newBuilder(ProtoUtils.ToProto.sendMessageRequest(request)),
4545
org.a2aproject.sdk.grpc.SendMessageRequest.Builder.class, context);
4646
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10(delegate.sendMessage(
47-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(
47+
Compat03ClientTransportSupport.toV03Protobuf(ProtoUtils.FromProto.messageSendParams(
4848
((org.a2aproject.sdk.grpc.SendMessageRequest.Builder) payload.getPayload()).build())),
4949
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
5050
}
5151

5252
@Override
5353
public void sendMessageStreaming(MessageSendParams request, Consumer<StreamingEventKind> events,
5454
Consumer<Throwable> errors, @Nullable ClientCallContext context) {
55-
Compat03ClientTransportSupport.validateMessageSend(request);
55+
Compat03ClientTransportSupport.validateProtobufMessageSend(request);
5656
PayloadAndHeaders payload = apply(A2AMethods.SEND_STREAMING_MESSAGE_METHOD,
5757
org.a2aproject.sdk.grpc.SendMessageRequest.newBuilder(ProtoUtils.ToProto.sendMessageRequest(request)),
5858
org.a2aproject.sdk.grpc.SendMessageRequest.Builder.class, context);
5959
Compat03ClientTransportSupport.run(() -> delegate.sendMessageStreaming(
60-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(
60+
Compat03ClientTransportSupport.toV03Protobuf(ProtoUtils.FromProto.messageSendParams(
6161
((org.a2aproject.sdk.grpc.SendMessageRequest.Builder) payload.getPayload()).build())),
6262
event -> events.accept(Compat03ClientTransportSupport.toV10(event)),
6363
Compat03ClientTransportSupport.mapAsyncError(errors),
@@ -78,12 +78,12 @@ public Task getTask(TaskQueryParams request, @Nullable ClientCallContext context
7878

7979
@Override
8080
public Task cancelTask(CancelTaskParams request, @Nullable ClientCallContext context) {
81-
Compat03ClientTransportSupport.validateCancel(request);
81+
Compat03ClientTransportSupport.validateProtobufCancel(request);
8282
PayloadAndHeaders payload = apply(A2AMethods.CANCEL_TASK_METHOD,
8383
org.a2aproject.sdk.grpc.CancelTaskRequest.newBuilder(ProtoUtils.ToProto.cancelTaskRequest(request)),
8484
org.a2aproject.sdk.grpc.CancelTaskRequest.Builder.class, context);
8585
return Compat03ClientTransportSupport.call(() -> TaskMapper_v0_3.INSTANCE.toV10(delegate.cancelTask(
86-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.cancelTaskParams(
86+
Compat03ClientTransportSupport.toV03Protobuf(ProtoUtils.FromProto.cancelTaskParams(
8787
((org.a2aproject.sdk.grpc.CancelTaskRequest.Builder) payload.getPayload()).build())),
8888
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
8989
}

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

Lines changed: 96 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,96 @@
1+
package org.a2aproject.sdk.compat03.client.adapter.rest;
2+
3+
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
4+
import static org.junit.jupiter.api.Assertions.assertThrows;
5+
import static org.junit.jupiter.api.Assertions.assertTrue;
6+
7+
import java.util.List;
8+
import java.util.Map;
9+
import java.util.function.Consumer;
10+
11+
import com.google.protobuf.Struct;
12+
import com.google.protobuf.Value;
13+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext;
14+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallInterceptor;
15+
import org.a2aproject.sdk.client.transport.spi.interceptors.PayloadAndHeaders;
16+
import org.a2aproject.sdk.compat03.client.adapter.Compat03ClientTransportBase;
17+
import org.a2aproject.sdk.compat03.client.transport.rest.RestTransport_v0_3;
18+
import org.a2aproject.sdk.compat03.client.transport.spi.interceptors.ClientCallContext_v0_3;
19+
import org.a2aproject.sdk.compat03.spec.EventKind_v0_3;
20+
import org.a2aproject.sdk.compat03.spec.MessageSendParams_v0_3;
21+
import org.a2aproject.sdk.compat03.spec.StreamingEventKind_v0_3;
22+
import org.a2aproject.sdk.compat03.spec.TaskIdParams_v0_3;
23+
import org.a2aproject.sdk.compat03.spec.Task_v0_3;
24+
import org.a2aproject.sdk.spec.A2AClientException;
25+
import org.a2aproject.sdk.spec.AgentCapabilities;
26+
import org.a2aproject.sdk.spec.AgentCard;
27+
import org.a2aproject.sdk.spec.AgentInterface;
28+
import org.a2aproject.sdk.spec.CancelTaskParams;
29+
import org.a2aproject.sdk.spec.Message;
30+
import org.a2aproject.sdk.spec.MessageSendParams;
31+
import org.a2aproject.sdk.spec.TextPart;
32+
import org.a2aproject.sdk.spec.UnsupportedOperationError;
33+
import org.jspecify.annotations.Nullable;
34+
import org.junit.jupiter.api.Test;
35+
36+
class RestCompat03UnsupportedFieldsTest {
37+
@Test
38+
void rejectsOriginalFieldsAndInterceptorMutationsBeforeDelegation() {
39+
AgentCard card = AgentCard.builder().name("legacy").description("legacy").version("1")
40+
.url("https://example.test").capabilities(AgentCapabilities.builder().build())
41+
.defaultInputModes(List.of("text")).defaultOutputModes(List.of("text")).skills(List.of())
42+
.supportedInterfaces(List.of(new AgentInterface("HTTP+JSON", "https://example.test", null, "0.3"))).build();
43+
var delegate = new RestTransport_v0_3(null, Compat03ClientTransportBase.legacyCard(card), "https://example.test", List.of()) {
44+
@Override
45+
public EventKind_v0_3 sendMessage(MessageSendParams_v0_3 request, @Nullable ClientCallContext_v0_3 context) {
46+
throw new AssertionError("Unsupported request reached the delegate");
47+
}
48+
49+
@Override
50+
public void sendMessageStreaming(MessageSendParams_v0_3 request, Consumer<StreamingEventKind_v0_3> events,
51+
Consumer<Throwable> errors, @Nullable ClientCallContext_v0_3 context) {
52+
throw new AssertionError("Unsupported streaming request reached the delegate");
53+
}
54+
55+
@Override
56+
public Task_v0_3 cancelTask(TaskIdParams_v0_3 request, @Nullable ClientCallContext_v0_3 context) {
57+
throw new AssertionError("Unsupported cancellation reached the delegate");
58+
}
59+
};
60+
for (boolean intercepted : List.of(false, true)) {
61+
var transport = new RestCompat03ClientTransport(delegate, card,
62+
intercepted ? List.of(unsupportedFieldsInterceptor()) : List.of());
63+
var request = new MessageSendParams(new Message(Message.Role.ROLE_USER, List.of(new TextPart("hello")),
64+
"request", null, null, intercepted ? null : List.of("referenced-task"), null, null), null, null);
65+
assertUnsupported("referenceTaskIds", () -> transport.sendMessage(request, null));
66+
assertUnsupported("referenceTaskIds", () -> transport.sendMessageStreaming(request,
67+
event -> { }, error -> { throw new AssertionError(error); }, null));
68+
var cancel = new CancelTaskParams("task", null, intercepted ? Map.of() : Map.of("reason", "stop"));
69+
assertUnsupported("metadata", () -> transport.cancelTask(cancel, null));
70+
}
71+
}
72+
73+
private static void assertUnsupported(String field, Runnable call) {
74+
A2AClientException error = assertThrows(A2AClientException.class, call::run);
75+
assertInstanceOf(UnsupportedOperationError.class, error.getCause());
76+
assertTrue(error.getMessage().contains(field));
77+
}
78+
79+
private static ClientCallInterceptor unsupportedFieldsInterceptor() {
80+
return new ClientCallInterceptor() {
81+
@Override
82+
public PayloadAndHeaders intercept(String method, @Nullable Object payload, Map<String, String> headers,
83+
@Nullable AgentCard card, @Nullable ClientCallContext context) {
84+
if (payload instanceof org.a2aproject.sdk.grpc.SendMessageRequest.Builder) {
85+
var request = (org.a2aproject.sdk.grpc.SendMessageRequest.Builder) payload;
86+
request.getMessageBuilder().addReferenceTaskIds("referenced-task");
87+
return new PayloadAndHeaders(request, headers);
88+
}
89+
Struct metadata = Struct.newBuilder().putFields("reason", Value.newBuilder().setStringValue("stop").build()).build();
90+
var request = (org.a2aproject.sdk.grpc.CancelTaskRequest.Builder) payload;
91+
request.setMetadata(metadata);
92+
return new PayloadAndHeaders(request, headers);
93+
}
94+
};
95+
}
96+
}

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

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

3+
import java.io.IOException;
34
import java.util.LinkedHashMap;
45
import java.util.Map;
56

67
import org.a2aproject.sdk.compat03.spec.A2AClientException_v0_3;
8+
import org.a2aproject.sdk.compat03.spec.A2AClientHTTPError_v0_3;
79
import org.a2aproject.sdk.compat03.spec.A2AErrorCodes_v0_3;
810
import org.a2aproject.sdk.compat03.spec.JSONRPCError_v0_3;
911
import org.a2aproject.sdk.spec.A2AClientException;
12+
import org.a2aproject.sdk.spec.A2AClientHTTPError;
1013
import org.a2aproject.sdk.spec.A2AError;
1114
import org.a2aproject.sdk.spec.ContentTypeNotSupportedError;
1215
import org.a2aproject.sdk.spec.ExtendedAgentCardNotConfiguredError;
@@ -30,6 +33,13 @@ private Compat03ClientErrorMapper() {
3033
public static A2AClientException toV10(A2AClientException_v0_3 exception) {
3134
Throwable cause = exception.getCause();
3235
String message = exception.getMessage() == null ? "A2A 0.3 client operation failed" : exception.getMessage();
36+
if (cause instanceof A2AClientHTTPError_v0_3 error) {
37+
return new A2AClientException(message, new A2AClientHTTPError(error.getCode(), error.getMessage(),
38+
error.getResponseBody(), error.getResponseHeaders()));
39+
}
40+
if (cause instanceof IOException && cause.getCause() instanceof A2AClientHTTPError error) {
41+
return new A2AClientException(message, error);
42+
}
3343
if (cause instanceof JSONRPCError_v0_3 error) {
3444
return new A2AClientException(message, toV10(error));
3545
}

0 commit comments

Comments
 (0)