diff --git a/a2a/src/main/java/com/google/adk/a2a/converters/ResponseConverter.java b/a2a/src/main/java/com/google/adk/a2a/converters/ResponseConverter.java index ef9318a26..cebdc4b52 100644 --- a/a2a/src/main/java/com/google/adk/a2a/converters/ResponseConverter.java +++ b/a2a/src/main/java/com/google/adk/a2a/converters/ResponseConverter.java @@ -16,8 +16,6 @@ package com.google.adk.a2a.converters; import static com.google.common.collect.ImmutableList.toImmutableList; -import static com.google.common.collect.ImmutableSet.toImmutableSet; -import static com.google.common.collect.Streams.zip; import com.fasterxml.jackson.core.JsonProcessingException; import com.fasterxml.jackson.databind.JavaType; @@ -31,6 +29,7 @@ import com.google.genai.types.Content; import com.google.genai.types.CustomMetadata; import com.google.genai.types.FinishReason; +import com.google.genai.types.FunctionCall; import com.google.genai.types.GenerateContentResponseUsageMetadata; import com.google.genai.types.GroundingMetadata; import com.google.genai.types.Part; @@ -55,7 +54,12 @@ import org.slf4j.Logger; import org.slf4j.LoggerFactory; -/** Utility for converting ADK events to A2A spec messages (and back). */ +/** + * Utility for converting ADK events to A2A spec messages (and back). + * + *

Unparseable ADK metadata and unconvertible parts from a remote agent are logged and dropped; + * the rest of the event is still converted. + */ public final class ResponseConverter { private static final ObjectMapper objectMapper = new ObjectMapper(); private static final Logger logger = LoggerFactory.getLogger(ResponseConverter.class); @@ -69,16 +73,16 @@ private ResponseConverter() {} /** * Converts a A2A {@link ClientEvent} to an ADK {@link Event}, based on the event type. Returns an * empty optional if the event should be ignored (e.g. if the event is not a final update for - * TaskArtifactUpdateEvent or if the message is empty for TaskStatusUpdateEvent). - * - *

Unparseable ADK metadata is logged and dropped; the rest of the event is still converted. + * TaskArtifactUpdateEvent, if the message is empty for TaskStatusUpdateEvent, or if no part could + * be converted from a message or from the message of a status update that does not end the turn). * * @throws IllegalArgumentException if the event type is not supported. */ public static Optional clientEventToEvent( ClientEvent event, InvocationContext invocationContext) { if (event instanceof MessageEvent messageEvent) { - return Optional.of(messageToEvent(messageEvent.getMessage(), invocationContext)); + return Optional.of(messageToEvent(messageEvent.getMessage(), invocationContext)) + .filter(ResponseConverter::hasParts); } else if (event instanceof TaskEvent taskEvent) { return Optional.of(taskToEvent(taskEvent.getTask(), invocationContext)); } else if (event instanceof TaskUpdateEvent updateEvent) { @@ -101,9 +105,10 @@ private static boolean isLongRunning(@Nullable Map metadata) { } /** - * Converts a A2A {@link TaskUpdateEvent} to an ADK {@link Event}, if applicable. Returns null if - * the event is not a final update for TaskArtifactUpdateEvent or if the message is empty for - * TaskStatusUpdateEvent. + * Converts a A2A {@link TaskUpdateEvent} to an ADK {@link Event}, if applicable. Returns empty if + * the event is not a final update for TaskArtifactUpdateEvent, if the message is empty for + * TaskStatusUpdateEvent, or if no part could be converted from the message of a status update + * that does not end the turn. * * @throws IllegalArgumentException if the task update type is not supported. */ @@ -120,7 +125,7 @@ private static Optional handleTaskUpdate( } Event eventPart = artifactToEvent(artifactEvent.getArtifact(), context); - if (eventPart.content().flatMap(Content::parts).orElse(ImmutableList.of()).isEmpty()) { + if (!hasParts(eventPart)) { return Optional.empty(); } eventPart.setPartial(isAppend || !isLastChunk); @@ -148,7 +153,8 @@ private static Optional handleTaskUpdate( return messageToFailedEvent(value, context); } return messageToEvent(value, context, PENDING_STATES.contains(taskState)); - }); + }) + .filter(converted -> taskState == TaskState.FAILED || hasParts(converted)); if (statusEvent.isFinal() || taskState == TaskState.INPUT_REQUIRED @@ -176,10 +182,10 @@ private static Optional handleTaskUpdate( /** Converts an artifact to an ADK event. */ public static Event artifactToEvent(Artifact artifact, InvocationContext invocationContext) { Event.Builder eventBuilder = remoteAgentEventBuilder(invocationContext); - ImmutableList genaiParts = PartConverter.toGenaiParts(artifact.parts()); + ConvertedParts converted = convertParts(artifact.parts()); eventBuilder - .content(fromModelParts(genaiParts)) - .longRunningToolIds(getLongRunningToolIds(artifact.parts(), genaiParts)); + .content(fromModelParts(converted.parts())) + .longRunningToolIds(converted.longRunningToolIds()); return eventBuilder.build(); } @@ -193,15 +199,11 @@ public static Event messageToFailedEvent(Message message, InvocationContext invo return builder.build(); } - /** - * Converts an A2A message back to ADK events. - * - *

Unparseable ADK metadata is logged and dropped; the rest of the event is still converted. - */ + /** Converts an A2A message back to ADK events. */ public static Event messageToEvent(Message message, InvocationContext invocationContext) { return updateEventMetadata( remoteAgentEventBuilder(invocationContext) - .content(fromModelParts(PartConverter.toGenaiParts(message.getParts()))) + .content(fromModelParts(convertParts(message.getParts()).parts())) .build(), message.getMetadata(), message.getTaskId(), @@ -216,7 +218,7 @@ public static Event messageToEvent( Message message, InvocationContext invocationContext, boolean isPending) { ImmutableList genaiParts = - PartConverter.toGenaiParts(message.getParts()).stream() + convertParts(message.getParts()).parts().stream() .map(part -> part.toBuilder().thought(isPending).build()) .collect(toImmutableList()); @@ -227,26 +229,23 @@ public static Event messageToEvent( * Converts an A2A {@link Task} to an ADK {@link Event}. If the artifacts are present, the last * artifact is used. If not, the status message is used. If not, the last history message is used. * If none of these are present, an empty event is returned. - * - *

Unparseable ADK metadata is logged and dropped; the rest of the event is still converted. */ public static Event taskToEvent(Task task, InvocationContext invocationContext) { ImmutableList.Builder genaiParts = ImmutableList.builder(); ImmutableSet.Builder longRunningToolIds = ImmutableSet.builder(); for (Artifact artifact : task.getArtifacts()) { - ImmutableList converted = PartConverter.toGenaiParts(artifact.parts()); - longRunningToolIds.addAll(getLongRunningToolIds(artifact.parts(), converted)); - genaiParts.addAll(converted); + ConvertedParts converted = convertParts(artifact.parts()); + longRunningToolIds.addAll(converted.longRunningToolIds()); + genaiParts.addAll(converted.parts()); } Event.Builder eventBuilder = remoteAgentEventBuilder(invocationContext); if (task.getStatus().message() != null) { - ImmutableList msgParts = - PartConverter.toGenaiParts(task.getStatus().message().getParts()); - longRunningToolIds.addAll( - getLongRunningToolIds(task.getStatus().message().getParts(), msgParts)); + ConvertedParts convertedMessage = convertParts(task.getStatus().message().getParts()); + ImmutableList msgParts = convertedMessage.parts(); + longRunningToolIds.addAll(convertedMessage.longRunningToolIds()); if (task.getStatus().state() == TaskState.FAILED && msgParts.size() == 1 && msgParts.get(0).text().isPresent()) { @@ -277,26 +276,36 @@ public static Event taskToEvent(Task task, InvocationContext invocationContext) eventBuilder.build(), task.getMetadata(), task.getId(), task.getContextId()); } - private static ImmutableSet getLongRunningToolIds( - List> parts, List convertedParts) { - return zip( - parts.stream(), - convertedParts.stream(), - (part, convertedPart) -> { - if (!(part instanceof DataPart dataPart)) { - return Optional.empty(); - } - // A2A peers may omit metadata entirely, which deserializes to null. - if (!isLongRunning(dataPart.getMetadata())) { - return Optional.empty(); - } - if (convertedPart.functionCall().isEmpty()) { - return Optional.empty(); - } - return convertedPart.functionCall().get().id(); - }) - .flatMap(Optional::stream) - .collect(toImmutableSet()); + /** GenAI parts converted from A2A parts, and the ids of the long-running calls among them. */ + private record ConvertedParts( + ImmutableList parts, ImmutableSet longRunningToolIds) {} + + /** + * Converts peer-supplied parts, skipping any part that cannot be converted so that one malformed + * part does not fail the whole event. + */ + private static ConvertedParts convertParts(List> a2aParts) { + ImmutableList.Builder parts = ImmutableList.builder(); + ImmutableSet.Builder longRunningToolIds = ImmutableSet.builder(); + for (io.a2a.spec.Part a2aPart : a2aParts) { + Part part; + try { + part = PartConverter.toGenaiPart(a2aPart); + } catch (RuntimeException e) { + // The exception message can quote the peer's bytes, so log only the types. + logger.warn( + "Skipping A2A part that could not be converted: {} ({})", + a2aPart == null ? "null" : a2aPart.getClass().getSimpleName(), + e.getClass().getSimpleName()); + continue; + } + parts.add(part); + // A2A peers may omit metadata entirely, which deserializes to null. + if (a2aPart instanceof DataPart dataPart && isLongRunning(dataPart.getMetadata())) { + part.functionCall().flatMap(FunctionCall::id).ifPresent(longRunningToolIds::add); + } + } + return new ConvertedParts(parts.build(), longRunningToolIds.build()); } private static Event updateEventMetadata( @@ -406,6 +415,10 @@ private static Content fromModelParts(List parts) { return Content.builder().role("model").parts(parts).build(); } + private static boolean hasParts(Event event) { + return !event.content().flatMap(Content::parts).orElse(ImmutableList.of()).isEmpty(); + } + private static Event.Builder remoteAgentEventBuilder(InvocationContext invocationContext) { return Event.builder() .id(UUID.randomUUID().toString()) diff --git a/a2a/src/test/java/com/google/adk/a2a/agent/RemoteA2AAgentTest.java b/a2a/src/test/java/com/google/adk/a2a/agent/RemoteA2AAgentTest.java index 8edc25684..66a1f0cde 100644 --- a/a2a/src/test/java/com/google/adk/a2a/agent/RemoteA2AAgentTest.java +++ b/a2a/src/test/java/com/google/adk/a2a/agent/RemoteA2AAgentTest.java @@ -67,6 +67,7 @@ import io.reactivex.rxjava3.core.Maybe; import io.reactivex.rxjava3.plugins.RxJavaPlugins; import io.reactivex.rxjava3.subscribers.TestSubscriber; +import java.util.AbstractList; import java.util.ArrayList; import java.util.Collections; import java.util.HashMap; @@ -153,6 +154,44 @@ public void createAgent_streaming_true_returnsStreamingAgent() { assertThat(agent.isStreaming()).isTrue(); } + @Test + public void runAsync_withUnconvertiblePartInStream_skipsItAndCompletes() { + RemoteA2AAgent agent = createAgent(); + mockStreamResponse( + consumer -> { + consumer.accept(createPartialEvent("Hello", true, false), agentCard); + TaskStatus working = + new TaskStatus( + TaskState.WORKING, + new Message.Builder() + .role(Message.Role.AGENT) + .parts( + ImmutableList.of( + new FilePart(new FileWithBytes("text/plain", "bad.txt", "!!!")))) + .build(), + null); + consumer.accept( + new TaskUpdateEvent( + new Task.Builder().id("task-1").contextId("context-1").status(working).build(), + new TaskStatusUpdateEvent.Builder() + .taskId("task-1") + .contextId("context-1") + .status(working) + .isFinal(false) + .build()), + agentCard); + consumer.accept(createFinalEvent("Done"), agentCard); + }); + + List events = agent.runAsync(invocationContext).toList().blockingGet(); + + assertThat(events).hasSize(3); + assertText(events.get(0), "Hello"); + assertText(events.get(1), "Hello"); + assertAggregated(events.get(1)); + assertText(events.get(2), "Done"); + } + @Test public void runAsync_aggregatesPartialEvents() { RemoteA2AAgent agent = createAgent(); @@ -1060,13 +1099,25 @@ private RemoteA2AAgent createAgent() { return getAgentBuilder().streaming(true).build(); } - /** An event whose file part carries invalid base64, so {@code PartConverter} cannot decode it. */ + /** + * An event whose conversion throws. Conversion skips malformed peer parts and metadata, so the + * failure is injected through a parts list that throws when read. + */ private ClientEvent unconvertibleEvent() { - return createTestEvent( - new FilePart(new FileWithBytes("text/plain", "bad.txt", "!!!")), - TaskState.WORKING, - true, - false); + List> unreadableParts = + new AbstractList<>() { + @Override + public io.a2a.spec.Part get(int index) { + throw new IllegalArgumentException("injected"); + } + + @Override + public int size() { + return 1; + } + }; + return new MessageEvent( + new Message.Builder().role(Message.Role.AGENT).parts(unreadableParts).build()); } @SuppressWarnings("unchecked") // cast for Mockito diff --git a/a2a/src/test/java/com/google/adk/a2a/converters/ResponseConverterTest.java b/a2a/src/test/java/com/google/adk/a2a/converters/ResponseConverterTest.java index c57866c99..684009df9 100644 --- a/a2a/src/test/java/com/google/adk/a2a/converters/ResponseConverterTest.java +++ b/a2a/src/test/java/com/google/adk/a2a/converters/ResponseConverterTest.java @@ -39,6 +39,8 @@ import io.a2a.client.TaskUpdateEvent; import io.a2a.spec.Artifact; import io.a2a.spec.DataPart; +import io.a2a.spec.FilePart; +import io.a2a.spec.FileWithBytes; import io.a2a.spec.Message; import io.a2a.spec.Task; import io.a2a.spec.TaskArtifactUpdateEvent; @@ -47,6 +49,7 @@ import io.a2a.spec.TaskStatusUpdateEvent; import io.a2a.spec.TextPart; import io.reactivex.rxjava3.core.Flowable; +import java.util.Arrays; import java.util.List; import java.util.Optional; import org.junit.Before; @@ -89,6 +92,16 @@ private static TaskStatusUpdateEvent.Builder testTaskStatusUpdateEvent() { return new TaskStatusUpdateEvent.Builder().taskId("task-1").contextId("context-1"); } + private TaskUpdateEvent statusUpdateEvent( + TaskState state, boolean isFinal, io.a2a.spec.Part... parts) { + Message statusMessage = + new Message.Builder().role(Message.Role.AGENT).parts(ImmutableList.copyOf(parts)).build(); + TaskStatus status = new TaskStatus(state, statusMessage, null); + return new TaskUpdateEvent( + testTask().status(status).build(), + testTaskStatusUpdateEvent().isFinal(isFinal).status(status).build()); + } + @Test public void clientEventToEvent_withMessageEvent_returnsEvent() { Message a2aMessage = @@ -485,6 +498,141 @@ public void taskToEvent_withMixedMetadataParts_keepsLongRunningId() { assertThat(event.content().get().parts().get()).hasSize(2); } + @Test + public void messageToEvent_withUnconvertiblePart_skipsOnlyThatPart() { + Message a2aMessage = + new Message.Builder() + .messageId("msg-1") + .role(Message.Role.AGENT) + .parts( + ImmutableList.of( + new TextPart("before"), invalidBase64Part(), new TextPart("after"))) + .build(); + + Event event = ResponseConverter.messageToEvent(a2aMessage, invocationContext); + + assertThat(event.content().get().parts().get().stream().map(p -> p.text().orElse(null))) + .containsExactly("before", "after") + .inOrder(); + } + + @Test + public void messageToEvent_withNullPart_skipsIt() { + // A peer can send a null element in "parts"; the v0.3 Message keeps the list as given. + Message a2aMessage = + new Message.Builder() + .messageId("msg-1") + .role(Message.Role.AGENT) + .parts( + Arrays.>asList( + new TextPart("before"), null, new TextPart("after"))) + .build(); + + Event event = ResponseConverter.messageToEvent(a2aMessage, invocationContext); + + assertThat(event.content().get().parts().get().stream().map(p -> p.text().orElse(null))) + .containsExactly("before", "after") + .inOrder(); + } + + @Test + public void clientEventToEvent_withUnconvertiblePartInStatusUpdate_skipsOnlyThatPart() { + TaskUpdateEvent event = + statusUpdateEvent(TaskState.WORKING, false, new TextPart("thought-1"), invalidBase64Part()); + + Optional optionalEvent = ResponseConverter.clientEventToEvent(event, invocationContext); + + assertThat(optionalEvent).isPresent(); + assertThat( + optionalEvent.get().content().get().parts().get().stream() + .map(p -> p.text().orElse(null))) + .containsExactly("thought-1"); + } + + @Test + public void clientEventToEvent_withOnlyUnconvertibleMessagePart_returnsEmpty() { + Message a2aMessage = + new Message.Builder() + .messageId("msg-1") + .role(Message.Role.AGENT) + .parts(ImmutableList.of(invalidBase64Part())) + .build(); + + Optional optionalEvent = + ResponseConverter.clientEventToEvent(new MessageEvent(a2aMessage), invocationContext); + + assertThat(optionalEvent).isEmpty(); + } + + @Test + public void clientEventToEvent_withOnlyUnconvertiblePartsInWorkingStatusUpdate_returnsEmpty() { + TaskUpdateEvent event = statusUpdateEvent(TaskState.WORKING, false, invalidBase64Part()); + + Optional optionalEvent = ResponseConverter.clientEventToEvent(event, invocationContext); + + assertThat(optionalEvent).isEmpty(); + } + + @Test + public void + clientEventToEvent_withOnlyUnconvertiblePartsInFinalStatusUpdate_returnsTurnComplete() { + TaskUpdateEvent event = statusUpdateEvent(TaskState.COMPLETED, true, invalidBase64Part()); + + Optional optionalEvent = ResponseConverter.clientEventToEvent(event, invocationContext); + + assertThat(optionalEvent).isPresent(); + Event result = optionalEvent.get(); + assertThat(result.turnComplete()).hasValue(true); + assertThat(result.partial()).hasValue(false); + assertThat(result.content()).isEmpty(); + } + + @Test + public void taskToEvent_withUnconvertiblePartBeforeLongRunningCall_keepsLongRunningId() { + DataPart longRunningPart = + new DataPart( + ImmutableMap.of("name", "lrTool", "id", "call_lr", "args", ImmutableMap.of()), + ImmutableMap.of( + A2AMetadataKey.TYPE.getType(), + "function_call", + A2AMetadataKey.IS_LONG_RUNNING.getType(), + true)); + Message statusMessage = + new Message.Builder() + .role(Message.Role.AGENT) + .parts(ImmutableList.of(invalidBase64Part(), longRunningPart)) + .build(); + Task task = + testTask().status(new TaskStatus(TaskState.INPUT_REQUIRED, statusMessage, null)).build(); + + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.longRunningToolIds().get()).containsExactly("call_lr"); + assertThat(event.content().get().parts().get()).hasSize(1); + } + + @Test + public void clientEventToEvent_withOnlyUnconvertibleArtifactPart_returnsEmpty() { + Artifact artifact = + new Artifact.Builder() + .artifactId("artifact-1") + .parts(ImmutableList.of(invalidBase64Part())) + .build(); + TaskArtifactUpdateEvent updateEvent = + new TaskArtifactUpdateEvent.Builder() + .append(true) + .lastChunk(false) + .contextId("context-1") + .artifact(artifact) + .taskId("task-1") + .build(); + TaskUpdateEvent event = + new TaskUpdateEvent( + testTask().status(new TaskStatus(TaskState.WORKING)).build(), updateEvent); + + assertThat(ResponseConverter.clientEventToEvent(event, invocationContext)).isEmpty(); + } + @Test public void taskToEvent_withFailedState_setsErrorCode() { Message statusMessage = @@ -801,6 +949,11 @@ public void clientEventToEvent_withTaskArtifactUpdateEventAndPartialTrue_returns assertThat(optionalEvent).isEmpty(); } + /** A file part whose bytes are not valid base64, so {@link PartConverter} cannot convert it. */ + private static FilePart invalidBase64Part() { + return new FilePart(new FileWithBytes("text/plain", "bad.txt", "!!!")); + } + private static final class TestAgent extends BaseAgent { TestAgent() { super("test_agent", "test", ImmutableList.of(), null, null);