Skip to content

Commit 4343831

Browse files
committed
fix: address unified client compatibility review findings
1 parent 97f5877 commit 4343831

21 files changed

Lines changed: 880 additions & 90 deletions

File tree

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

Lines changed: 83 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -6,11 +6,23 @@
66
import static org.junit.jupiter.api.Assertions.assertTrue;
77

88
import java.util.List;
9+
import java.util.Map;
10+
import java.util.concurrent.TimeUnit;
911
import java.util.concurrent.atomic.AtomicBoolean;
12+
import java.util.concurrent.atomic.AtomicReference;
1013

14+
import io.grpc.ManagedChannel;
15+
import io.grpc.Server;
16+
import io.grpc.Status;
17+
import io.grpc.inprocess.InProcessChannelBuilder;
18+
import io.grpc.inprocess.InProcessServerBuilder;
19+
import io.grpc.stub.StreamObserver;
1120
import org.a2aproject.sdk.client.transport.grpc.GrpcTransport;
1221
import org.a2aproject.sdk.client.transport.grpc.GrpcTransportConfigBuilder;
1322
import org.a2aproject.sdk.client.transport.spi.ClientTransport;
23+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallContext;
24+
import org.a2aproject.sdk.client.transport.spi.interceptors.ClientCallInterceptor;
25+
import org.a2aproject.sdk.client.transport.spi.interceptors.PayloadAndHeaders;
1426
import org.a2aproject.sdk.compat03.grpc.A2AServiceGrpc;
1527
import org.a2aproject.sdk.compat03.grpc.Message;
1628
import org.a2aproject.sdk.compat03.grpc.Part;
@@ -20,18 +32,83 @@
2032
import org.a2aproject.sdk.spec.AgentCard;
2133
import org.a2aproject.sdk.spec.AgentInterface;
2234
import org.a2aproject.sdk.spec.AgentSkill;
35+
import org.a2aproject.sdk.spec.GetTaskPushNotificationConfigParams;
2336
import org.a2aproject.sdk.spec.MessageSendParams;
2437
import org.a2aproject.sdk.spec.TextPart;
2538
import org.a2aproject.sdk.spec.TransportProtocol;
26-
import io.grpc.ManagedChannel;
27-
import io.grpc.Server;
28-
import io.grpc.inprocess.InProcessChannelBuilder;
29-
import io.grpc.inprocess.InProcessServerBuilder;
30-
import io.grpc.stub.StreamObserver;
31-
3239
import org.junit.jupiter.api.Test;
3340

3441
class GrpcCompat03ClientTransportProviderTest {
42+
@Test
43+
void sendsCompletePushConfigurationResourceNamesIncludingDefaultAndClearedIds() throws Exception {
44+
String name = InProcessServerBuilder.generateName();
45+
AtomicReference<String> receivedName = new AtomicReference<>();
46+
Server server = InProcessServerBuilder.forName(name).directExecutor()
47+
.addService(new A2AServiceGrpc.A2AServiceImplBase() {
48+
@Override
49+
public void getTaskPushNotificationConfig(
50+
org.a2aproject.sdk.compat03.grpc.GetTaskPushNotificationConfigRequest request,
51+
StreamObserver<org.a2aproject.sdk.compat03.grpc.TaskPushNotificationConfig> observer) {
52+
receivedName.set(request.getName());
53+
// Validate the contract directly rather than using the permissive reference parser.
54+
if (!request.getName().matches("tasks/task-123/pushNotificationConfigs/(task-123|specific)")) {
55+
observer.onError(Status.INVALID_ARGUMENT.withDescription("Invalid config resource name")
56+
.asRuntimeException());
57+
return;
58+
}
59+
String configId = request.getName().substring(request.getName().lastIndexOf('/') + 1);
60+
observer.onNext(org.a2aproject.sdk.compat03.grpc.TaskPushNotificationConfig.newBuilder()
61+
.setName(request.getName())
62+
.setPushNotificationConfig(org.a2aproject.sdk.compat03.grpc.PushNotificationConfig.newBuilder()
63+
.setId(configId).setUrl("https://example.test/callback"))
64+
.build());
65+
observer.onCompleted();
66+
}
67+
}).build().start();
68+
ManagedChannel channel = InProcessChannelBuilder.forName(name).directExecutor().build();
69+
try {
70+
AgentCard card = card(name);
71+
var provider = new GrpcCompat03ClientTransportProvider();
72+
ClientTransport transport = provider.create(
73+
new GrpcTransportConfigBuilder().channelFactory(ignored -> channel).build(), card,
74+
card.supportedInterfaces().get(0));
75+
try {
76+
assertEquals("task-123", transport.getTaskPushNotificationConfiguration(
77+
new GetTaskPushNotificationConfigParams("task-123"), null).id());
78+
assertEquals("tasks/task-123/pushNotificationConfigs/task-123", receivedName.get());
79+
assertEquals("specific", transport.getTaskPushNotificationConfiguration(
80+
new GetTaskPushNotificationConfigParams("task-123", "specific"), null).id());
81+
assertEquals("tasks/task-123/pushNotificationConfigs/specific", receivedName.get());
82+
} finally {
83+
transport.close();
84+
}
85+
86+
ClientCallInterceptor clearingInterceptor = new ClientCallInterceptor() {
87+
@Override
88+
public PayloadAndHeaders intercept(String method, Object payload, Map<String, String> headers,
89+
AgentCard agentCard, ClientCallContext context) {
90+
var request = (org.a2aproject.sdk.grpc.GetTaskPushNotificationConfigRequest) payload;
91+
return new PayloadAndHeaders(request.toBuilder().clearId().build(), headers);
92+
}
93+
};
94+
ClientTransport intercepted = provider.create(new GrpcTransportConfigBuilder()
95+
.channelFactory(ignored -> channel).addInterceptor(clearingInterceptor).build(), card,
96+
card.supportedInterfaces().get(0));
97+
try {
98+
assertEquals("task-123", intercepted.getTaskPushNotificationConfiguration(
99+
new GetTaskPushNotificationConfigParams("task-123", "specific"), null).id());
100+
assertEquals("tasks/task-123/pushNotificationConfigs/task-123", receivedName.get());
101+
} finally {
102+
intercepted.close();
103+
}
104+
} finally {
105+
channel.shutdownNow();
106+
server.shutdownNow();
107+
assertTrue(channel.awaitTermination(5, TimeUnit.SECONDS));
108+
assertTrue(server.awaitTermination(5, TimeUnit.SECONDS));
109+
}
110+
}
111+
35112
@Test
36113
void providerTargetsLegacyGrpcBindingAndVersion() {
37114
GrpcCompat03ClientTransportProvider provider = new GrpcCompat03ClientTransportProvider();

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

Lines changed: 33 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -13,6 +13,8 @@
1313
import org.a2aproject.sdk.compat03.client.adapter.Compat03InterceptorSupport;
1414
import org.a2aproject.sdk.compat03.client.transport.spi.ClientTransport_v0_3;
1515
import org.a2aproject.sdk.compat03.conversion.mappers.domain.TaskMapper_v0_3;
16+
import org.a2aproject.sdk.compat03.spec.MessageSendParams_v0_3;
17+
import org.a2aproject.sdk.compat03.spec.TaskIdParams_v0_3;
1618
import org.a2aproject.sdk.grpc.utils.ProtoUtils;
1719
import org.a2aproject.sdk.spec.A2AMethods;
1820
import org.a2aproject.sdk.spec.AgentCard;
@@ -40,23 +42,23 @@ public JSONRPCCompat03ClientTransport(ClientTransport_v0_3 delegate, AgentCard c
4042
@Override
4143
public EventKind sendMessage(MessageSendParams request, @Nullable ClientCallContext context) {
4244
Compat03ClientTransportSupport.validateMessageSend(request);
45+
org.a2aproject.sdk.grpc.SendMessageRequest original = ProtoUtils.ToProto.sendMessageRequest(request);
4346
PayloadAndHeaders payload = apply(A2AMethods.SEND_MESSAGE_METHOD,
44-
ProtoUtils.ToProto.sendMessageRequest(request), org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
47+
original, org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
4548
return Compat03ClientTransportSupport.call(() -> Compat03ClientTransportSupport.toV10(delegate.sendMessage(
46-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(
47-
(org.a2aproject.sdk.grpc.SendMessageRequest) payload.getPayload())),
49+
adaptMessage(request, original, (org.a2aproject.sdk.grpc.SendMessageRequest) payload.getPayload()),
4850
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
4951
}
5052

5153
@Override
5254
public void sendMessageStreaming(MessageSendParams request, Consumer<StreamingEventKind> events,
5355
Consumer<Throwable> errors, @Nullable ClientCallContext context) {
5456
Compat03ClientTransportSupport.validateMessageSend(request);
57+
org.a2aproject.sdk.grpc.SendMessageRequest original = ProtoUtils.ToProto.sendMessageRequest(request);
5558
PayloadAndHeaders payload = apply(A2AMethods.SEND_STREAMING_MESSAGE_METHOD,
56-
ProtoUtils.ToProto.sendMessageRequest(request), org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
59+
original, org.a2aproject.sdk.grpc.SendMessageRequest.class, context);
5760
Compat03ClientTransportSupport.run(() -> delegate.sendMessageStreaming(
58-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(
59-
(org.a2aproject.sdk.grpc.SendMessageRequest) payload.getPayload())),
61+
adaptMessage(request, original, (org.a2aproject.sdk.grpc.SendMessageRequest) payload.getPayload()),
6062
event -> events.accept(Compat03ClientTransportSupport.toV10(event)),
6163
Compat03ClientTransportSupport.mapAsyncError(errors),
6264
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload))));
@@ -76,11 +78,11 @@ public Task getTask(TaskQueryParams request, @Nullable ClientCallContext context
7678
@Override
7779
public Task cancelTask(CancelTaskParams request, @Nullable ClientCallContext context) {
7880
Compat03ClientTransportSupport.validateCancel(request);
79-
PayloadAndHeaders payload = apply(A2AMethods.CANCEL_TASK_METHOD, ProtoUtils.ToProto.cancelTaskRequest(request),
81+
org.a2aproject.sdk.grpc.CancelTaskRequest original = ProtoUtils.ToProto.cancelTaskRequest(request);
82+
PayloadAndHeaders payload = apply(A2AMethods.CANCEL_TASK_METHOD, original,
8083
org.a2aproject.sdk.grpc.CancelTaskRequest.class, context);
8184
return Compat03ClientTransportSupport.call(() -> TaskMapper_v0_3.INSTANCE.toV10(delegate.cancelTask(
82-
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.cancelTaskParams(
83-
(org.a2aproject.sdk.grpc.CancelTaskRequest) payload.getPayload())),
85+
adaptCancel(request, original, (org.a2aproject.sdk.grpc.CancelTaskRequest) payload.getPayload()),
8486
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload)))));
8587
}
8688

@@ -146,6 +148,28 @@ public void subscribeToTask(TaskIdParams request, Consumer<StreamingEventKind> e
146148
Compat03ClientCallContextMapper.toV03(contextWithHeaders(context, payload))));
147149
}
148150

151+
private static MessageSendParams_v0_3 adaptMessage(MessageSendParams request,
152+
org.a2aproject.sdk.grpc.SendMessageRequest original, org.a2aproject.sdk.grpc.SendMessageRequest modified) {
153+
MessageSendParams_v0_3 legacy = Compat03ClientTransportSupport.toV03(request);
154+
if (original.equals(modified)) {
155+
return legacy;
156+
}
157+
return JSONRPCCompat03PayloadSupport.preserveUnchangedValues(legacy,
158+
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(original)),
159+
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.messageSendParams(modified)));
160+
}
161+
162+
private static TaskIdParams_v0_3 adaptCancel(CancelTaskParams request,
163+
org.a2aproject.sdk.grpc.CancelTaskRequest original, org.a2aproject.sdk.grpc.CancelTaskRequest modified) {
164+
TaskIdParams_v0_3 legacy = Compat03ClientTransportSupport.toV03(request);
165+
if (original.equals(modified)) {
166+
return legacy;
167+
}
168+
return JSONRPCCompat03PayloadSupport.preserveUnchangedValues(legacy,
169+
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.cancelTaskParams(original)),
170+
Compat03ClientTransportSupport.toV03(ProtoUtils.FromProto.cancelTaskParams(modified)));
171+
}
172+
149173
private PayloadAndHeaders apply(String method, Object payload, Class<?> expected,
150174
@Nullable ClientCallContext context) {
151175
return Compat03InterceptorSupport.apply(interceptors, method, payload, agentCard, context, expected);
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,148 @@
1+
package org.a2aproject.sdk.compat03.client.adapter.jsonrpc;
2+
3+
import java.math.BigDecimal;
4+
import java.util.ArrayList;
5+
import java.util.LinkedHashMap;
6+
import java.util.List;
7+
import java.util.Map;
8+
import java.util.Objects;
9+
10+
import org.a2aproject.sdk.compat03.spec.DataPart_v0_3;
11+
import org.a2aproject.sdk.compat03.spec.FilePart_v0_3;
12+
import org.a2aproject.sdk.compat03.spec.MessageSendParams_v0_3;
13+
import org.a2aproject.sdk.compat03.spec.Message_v0_3;
14+
import org.a2aproject.sdk.compat03.spec.Part_v0_3;
15+
import org.a2aproject.sdk.compat03.spec.TaskIdParams_v0_3;
16+
import org.a2aproject.sdk.compat03.spec.TextPart_v0_3;
17+
import org.a2aproject.sdk.spec.A2AClientException;
18+
import org.jspecify.annotations.Nullable;
19+
20+
/** Retains original JSON values in fields that native protobuf interceptors have not changed. */
21+
final class JSONRPCCompat03PayloadSupport {
22+
private JSONRPCCompat03PayloadSupport() {
23+
}
24+
25+
static MessageSendParams_v0_3 preserveUnchangedValues(MessageSendParams_v0_3 original,
26+
MessageSendParams_v0_3 before, MessageSendParams_v0_3 after) {
27+
Message_v0_3 source = original.message();
28+
Message_v0_3 baseline = before.message();
29+
Message_v0_3 modified = after.message();
30+
List<Part_v0_3<?>> parts = modified.parts();
31+
if (baseline.parts().equals(modified.parts())) {
32+
parts = source.parts();
33+
} else if (source.parts().size() == 1 && baseline.parts().size() == 1 && modified.parts().size() == 1) {
34+
parts = List.of(restorePart(source.parts().get(0), baseline.parts().get(0), modified.parts().get(0)));
35+
} else if (hasLossyNumbers(source.parts(), baseline.parts())) {
36+
throw ambiguousArray();
37+
}
38+
Message_v0_3 message = new Message_v0_3(modified.role(), parts, modified.messageId(), modified.contextId(),
39+
modified.taskId(), modified.referenceTaskIds(),
40+
restoreMap(source.metadata(), baseline.metadata(), modified.metadata()), modified.extensions());
41+
return new MessageSendParams_v0_3(message, after.configuration(),
42+
restoreMap(original.metadata(), before.metadata(), after.metadata()));
43+
}
44+
45+
static TaskIdParams_v0_3 preserveUnchangedValues(TaskIdParams_v0_3 original,
46+
TaskIdParams_v0_3 before, TaskIdParams_v0_3 after) {
47+
return new TaskIdParams_v0_3(after.id(), restoreMap(original.metadata(), before.metadata(), after.metadata()));
48+
}
49+
50+
private static Part_v0_3<?> restorePart(Part_v0_3<?> original, Part_v0_3<?> before, Part_v0_3<?> after) {
51+
if (before.equals(after)) {
52+
return original;
53+
}
54+
@Nullable Map<String, Object> metadata = restoreMap(partMetadata(original), partMetadata(before), partMetadata(after));
55+
if (after instanceof DataPart_v0_3 modified) {
56+
Map<String, Object> data = modified.data();
57+
if (original instanceof DataPart_v0_3 source && before instanceof DataPart_v0_3 baseline) {
58+
data = Objects.requireNonNull(restoreMap(source.data(), baseline.data(), modified.data()));
59+
}
60+
return new DataPart_v0_3(data, metadata);
61+
}
62+
if (after instanceof TextPart_v0_3 modified) {
63+
return new TextPart_v0_3(modified.text(), metadata);
64+
}
65+
FilePart_v0_3 modified = (FilePart_v0_3) after;
66+
return new FilePart_v0_3(modified.file(), metadata);
67+
}
68+
69+
private static @Nullable Map<String, Object> partMetadata(Part_v0_3<?> part) {
70+
if (part instanceof DataPart_v0_3 data) {
71+
return data.metadata();
72+
}
73+
if (part instanceof TextPart_v0_3 text) {
74+
return text.metadata();
75+
}
76+
return ((FilePart_v0_3) part).metadata();
77+
}
78+
79+
@SuppressWarnings("unchecked")
80+
private static @Nullable Map<String, Object> restoreMap(@Nullable Map<String, Object> original,
81+
@Nullable Map<String, Object> before, @Nullable Map<String, Object> after) {
82+
return (Map<String, Object>) restore(original, before, after);
83+
}
84+
85+
private static @Nullable Object restore(@Nullable Object original, @Nullable Object before, @Nullable Object after) {
86+
// Struct/Value uses doubles. Restore unchanged values directly from the original objects,
87+
// avoiding another JSON parse that could itself round large integers or decimal numbers.
88+
if (Objects.equals(before, after)) {
89+
return original;
90+
}
91+
if (original instanceof Map<?, ?> source && before instanceof Map<?, ?> baseline
92+
&& after instanceof Map<?, ?> modified) {
93+
Map<Object, @Nullable Object> result = new LinkedHashMap<>();
94+
modified.forEach((key, value) -> result.put(key, source.containsKey(key) && baseline.containsKey(key)
95+
? restore(source.get(key), baseline.get(key), value) : value));
96+
return result;
97+
}
98+
if (original instanceof List<?> source && before instanceof List<?> baseline && after instanceof List<?> modified) {
99+
if (source.size() == 1 && baseline.size() == 1 && modified.size() == 1) {
100+
// Array slots have no identity. Only a singleton has an unambiguous slot.
101+
List<@Nullable Object> result = new ArrayList<>();
102+
result.add(restore(source.get(0), baseline.get(0), modified.get(0)));
103+
return result;
104+
}
105+
if (hasLossyNumbers(source, baseline)) {
106+
throw ambiguousArray();
107+
}
108+
}
109+
return after;
110+
}
111+
112+
private static boolean hasLossyNumbers(@Nullable Object original, @Nullable Object projected) {
113+
if (original instanceof Number source && projected instanceof Number baseline) {
114+
return !Double.isFinite(baseline.doubleValue())
115+
|| new BigDecimal(source.toString()).compareTo(new BigDecimal(baseline.toString())) != 0;
116+
}
117+
if (original instanceof DataPart_v0_3 source && projected instanceof DataPart_v0_3 baseline) {
118+
return hasLossyNumbers(source.metadata(), baseline.metadata())
119+
|| hasLossyNumbers(source.data(), baseline.data());
120+
}
121+
if (original instanceof TextPart_v0_3 source && projected instanceof TextPart_v0_3 baseline) {
122+
return hasLossyNumbers(source.metadata(), baseline.metadata());
123+
}
124+
if (original instanceof FilePart_v0_3 source && projected instanceof FilePart_v0_3 baseline) {
125+
return hasLossyNumbers(source.metadata(), baseline.metadata());
126+
}
127+
if (original instanceof Map<?, ?> source && projected instanceof Map<?, ?> baseline) {
128+
for (var entry : source.entrySet()) {
129+
if (hasLossyNumbers(entry.getValue(), baseline.get(entry.getKey()))) {
130+
return true;
131+
}
132+
}
133+
}
134+
if (original instanceof List<?> source && projected instanceof List<?> baseline) {
135+
for (int i = 0; i < source.size() && i < baseline.size(); i++) {
136+
if (hasLossyNumbers(source.get(i), baseline.get(i))) {
137+
return true;
138+
}
139+
}
140+
}
141+
return false;
142+
}
143+
144+
private static A2AClientException ambiguousArray() {
145+
return new A2AClientException("Cannot preserve exact JSON numbers after a protobuf interceptor edits "
146+
+ "a non-singleton array. Keep that array unchanged or use strings for exact numeric identifiers.");
147+
}
148+
}

0 commit comments

Comments
 (0)