diff --git a/a2a/src/main/java/com/google/adk/a2a/agent/RemoteA2AAgent.java b/a2a/src/main/java/com/google/adk/a2a/agent/RemoteA2AAgent.java index d4e094710..1f130416b 100644 --- a/a2a/src/main/java/com/google/adk/a2a/agent/RemoteA2AAgent.java +++ b/a2a/src/main/java/com/google/adk/a2a/agent/RemoteA2AAgent.java @@ -296,6 +296,17 @@ synchronized void handleEvent(ClientEvent clientEvent, AgentCard unused) { return; } + try { + convertAndEmit(clientEvent); + } catch (RuntimeException e) { + // On the streaming path nothing between here and the transport's read loop catches this: + // the SSE subscriber skips its next request() call, so an escaping exception stalls the + // stream instead of failing it. Route it to handleError so the caller sees an error. + handleError(e); + } + } + + private void convertAndEmit(ClientEvent clientEvent) { Optional eventOpt = ResponseConverter.clientEventToEvent(clientEvent, invocationContext); eventOpt.ifPresent( 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 cffd76983..a58549318 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 @@ -69,6 +69,8 @@ private ResponseConverter() {} * 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. + * * @throws IllegalArgumentException if the event type is not supported. */ public static Optional clientEventToEvent( @@ -91,6 +93,11 @@ private static boolean isPartial(@Nullable Map metadata) { return Objects.equals(metadata.getOrDefault(A2AMetadataKey.PARTIAL.getType(), false), true); } + private static boolean isLongRunning(@Nullable Map metadata) { + return metadata != null + && Objects.equals(metadata.get(A2AMetadataKey.IS_LONG_RUNNING.getType()), true); + } + /** * 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 @@ -182,7 +189,11 @@ public static Event messageToFailedEvent(Message message, InvocationContext invo return builder.build(); } - /** Converts an A2A message back to ADK events. */ + /** + * Converts an A2A message back to ADK events. + * + *

Unparseable ADK metadata is logged and dropped; the rest of the event is still converted. + */ public static Event messageToEvent(Message message, InvocationContext invocationContext) { return updateEventMetadata( remoteAgentEventBuilder(invocationContext) @@ -212,6 +223,8 @@ 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(); @@ -266,9 +279,8 @@ private static ImmutableSet getLongRunningToolIds( if (!(part instanceof DataPart dataPart)) { return Optional.empty(); } - Object isLongRunning = - dataPart.getMetadata().get(A2AMetadataKey.IS_LONG_RUNNING.getType()); - if (!Objects.equals(isLongRunning, true)) { + // A2A peers may omit metadata entirely, which deserializes to null. + if (!isLongRunning(dataPart.getMetadata())) { return Optional.empty(); } if (convertedPart.functionCall().isEmpty()) { @@ -294,13 +306,13 @@ private static Event updateEventMetadata( clientMetadata = ImmutableMap.of(); } Event.Builder eventBuilder = event.toBuilder(); - Object groundingMetadata = clientMetadata.get(A2AMetadataKey.GROUNDING_METADATA.getType()); - // if groundingMetadata is null, parseMetadata will return null as well. - eventBuilder.groundingMetadata(parseMetadata(groundingMetadata, GroundingMetadata.class)); - Object usageMetadata = clientMetadata.get(A2AMetadataKey.USAGE_METADATA.getType()); - // if usageMetadata is null, parseMetadata will return null as well. + eventBuilder.groundingMetadata( + parseMetadata(clientMetadata, A2AMetadataKey.GROUNDING_METADATA, GroundingMetadata.class)); eventBuilder.usageMetadata( - parseMetadata(usageMetadata, GenerateContentResponseUsageMetadata.class)); + parseMetadata( + clientMetadata, + A2AMetadataKey.USAGE_METADATA, + GenerateContentResponseUsageMetadata.class)); ImmutableList.Builder customMetadataList = ImmutableList.builder(); customMetadataList @@ -314,20 +326,32 @@ private static Event updateEventMetadata( .key(AdkMetadataKey.CONTEXT_ID.getType()) .stringValue(contextId) .build()); - Object customMetadata = clientMetadata.get(A2AMetadataKey.CUSTOM_METADATA.getType()); - if (customMetadata != null) { - customMetadataList.addAll( - parseMetadata(customMetadata, new TypeReference>() {})); + List parsedCustomMetadata = + parseMetadata( + clientMetadata, + A2AMetadataKey.CUSTOM_METADATA, + new TypeReference>() {}); + if (parsedCustomMetadata != null) { + customMetadataList.addAll(parsedCustomMetadata); } eventBuilder.customMetadata(customMetadataList.build()); - Object errorCode = clientMetadata.get(A2AMetadataKey.ERROR_CODE.getType()); - eventBuilder.errorCode(parseMetadata(errorCode, FinishReason.class)); + eventBuilder.errorCode( + parseMetadata(clientMetadata, A2AMetadataKey.ERROR_CODE, FinishReason.class)); return eventBuilder.build(); } - private static @Nullable T parseMetadata(@Nullable Object metadata, Class type) { + /** + * Reads {@code key} out of the peer-supplied {@code clientMetadata} and deserializes it. + * + *

Returns null when the key is absent, and also when its value cannot be parsed: metadata is + * peer-controlled, so a malformed value is logged and dropped rather than failing the whole + * conversion. + */ + private static @Nullable T parseMetadata( + Map clientMetadata, A2AMetadataKey key, Class type) { + Object metadata = clientMetadata.get(key.getType()); try { if (metadata instanceof String jsonString) { return objectMapper.readValue(jsonString, type); @@ -335,11 +359,15 @@ private static Event updateEventMetadata( return objectMapper.convertValue(metadata, type); } } catch (IllegalArgumentException | JsonProcessingException e) { - throw new IllegalArgumentException("Failed to parse metadata of type " + type, e); + logDroppedMetadata(key, e); + return null; } } - private static @Nullable T parseMetadata(@Nullable Object metadata, TypeReference type) { + /** Overload of {@link #parseMetadata(Map, A2AMetadataKey, Class)} for generic target types. */ + private static @Nullable T parseMetadata( + Map clientMetadata, A2AMetadataKey key, TypeReference type) { + Object metadata = clientMetadata.get(key.getType()); try { if (metadata instanceof String jsonString) { return objectMapper.readValue(jsonString, type); @@ -347,10 +375,27 @@ private static Event updateEventMetadata( return objectMapper.convertValue(metadata, type); } } catch (IllegalArgumentException | JsonProcessingException e) { - throw new IllegalArgumentException("Failed to parse metadata of type " + type.getType(), e); + logDroppedMetadata(key, e); + return null; } } + /** + * Reports a dropped metadata value. + * + *

The parser's message quotes the peer's bytes, so the warning carries only the key and the + * exception type. A peer that streams malformed metadata would otherwise be able to write + * arbitrary content and a stack trace into the log on every event. The full exception is + * available at debug level. + */ + private static void logDroppedMetadata(A2AMetadataKey key, Exception e) { + logger.warn( + "Dropping unparseable A2A metadata for key {} ({})", + key.getType(), + e.getClass().getSimpleName()); + logger.debug("Unparseable A2A metadata for key {}", key.getType(), e); + } + private static Event emptyEvent(InvocationContext invocationContext) { Event.Builder builder = Event.builder() 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 8d6c9b062..6f4b705c0 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 @@ -25,6 +25,7 @@ import static org.mockito.Mockito.verifyNoInteractions; import static org.mockito.Mockito.when; +import com.google.adk.a2a.common.A2AClientError; import com.google.adk.a2a.common.A2AMetadata; import com.google.adk.agents.BaseAgent; import com.google.adk.agents.CallbackContext; @@ -51,6 +52,7 @@ import io.a2a.spec.Artifact; import io.a2a.spec.DataPart; import io.a2a.spec.FilePart; +import io.a2a.spec.FileWithBytes; import io.a2a.spec.FileWithUri; import io.a2a.spec.Message; import io.a2a.spec.Task; @@ -61,10 +63,12 @@ import io.a2a.spec.TextPart; import io.reactivex.rxjava3.core.Flowable; import io.reactivex.rxjava3.core.Maybe; +import io.reactivex.rxjava3.subscribers.TestSubscriber; import java.util.HashMap; import java.util.List; import java.util.Map; import java.util.Objects; +import java.util.concurrent.CountDownLatch; import java.util.concurrent.atomic.AtomicBoolean; import java.util.function.BiConsumer; import java.util.function.Consumer; @@ -300,6 +304,58 @@ public void runAsync_handlesTasksWithMultipartArtifact() { assertResponseMetadata(events.get(0)); } + @Test + public void runAsync_whenConversionThrows_reportsError() { + RemoteA2AAgent agent = createAgent(); + mockStreamResponse(consumer -> consumer.accept(unconvertibleEvent(), agentCard)); + + agent + .runAsync(invocationContext) + .test() + .awaitDone(5, SECONDS) + .assertError(A2AClientError.class) + .assertError(e -> e.getCause() instanceof IllegalArgumentException); + } + + @Test + public void runAsync_whenConversionThrowsOffThread_terminatesInsteadOfStalling() + throws InterruptedException { + RemoteA2AAgent agent = createAgent(); + // Deliver off-thread and do not join, mimicking a transport that pushes events after + // sendMessage returns. A throw that escapes handleEvent there reaches no RxJava boundary, so + // without the guard nothing ever terminates the flow and awaitDone below times out. + CountDownLatch delivered = new CountDownLatch(1); + mockStreamResponse( + consumer -> { + Thread thread = + new Thread( + () -> { + try { + consumer.accept(unconvertibleEvent(), agentCard); + } finally { + delivered.countDown(); + } + }); + // The throw is the behaviour under test; keep it off the test's stderr. + thread.setUncaughtExceptionHandler((t, e) -> {}); + thread.start(); + }); + + TestSubscriber subscriber = agent.runAsync(invocationContext).test(); + assertThat(delivered.await(5, SECONDS)).isTrue(); + + subscriber.awaitDone(5, SECONDS).assertError(A2AClientError.class); + } + + /** An event whose file part carries invalid base64, so {@code PartConverter} cannot decode it. */ + private ClientEvent unconvertibleEvent() { + return createTestEvent( + new FilePart(new FileWithBytes("text/plain", "bad.txt", "!!!")), + TaskState.WORKING, + true, + false); + } + @Test public void runAsync_handlesNonFinalStatusUpdatesAsThoughts() { RemoteA2AAgent agent = createAgent(); 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 20f2d3c10..4cf441aae 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 @@ -18,7 +18,6 @@ import static com.google.common.truth.Truth.assertThat; import static java.util.stream.Collectors.joining; -import static org.junit.Assert.assertThrows; import com.google.adk.agents.BaseAgent; import com.google.adk.agents.InvocationContext; @@ -47,6 +46,7 @@ import io.a2a.spec.TaskStatusUpdateEvent; import io.a2a.spec.TextPart; import io.reactivex.rxjava3.core.Flowable; +import java.util.List; import java.util.Optional; import org.junit.Before; import org.junit.Test; @@ -212,6 +212,92 @@ public void taskToEvent_withCustomMetadata_returnsEvent() { .inOrder(); } + @Test + public void taskToEvent_withMalformedMetadata_dropsFieldsAndConverts() { + Message statusMessage = + new Message.Builder() + .role(Message.Role.AGENT) + .parts(ImmutableList.of(new TextPart("Status message"))) + .build(); + TaskStatus status = new TaskStatus(TaskState.WORKING, statusMessage, null); + Task task = + testTask() + .status(status) + .artifacts(null) + .metadata( + ImmutableMap.of( + A2AMetadataKey.GROUNDING_METADATA.getType(), "not-valid-json", + A2AMetadataKey.USAGE_METADATA.getType(), "not-valid-json", + A2AMetadataKey.CUSTOM_METADATA.getType(), "not-valid-json", + A2AMetadataKey.ERROR_CODE.getType(), "not-valid-json")) + .build(); + + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.content().get().parts().get().get(0).text()).hasValue("Status message"); + assertThat(event.groundingMetadata()).isEmpty(); + assertThat(event.usageMetadata()).isEmpty(); + assertThat(event.errorCode()).isEmpty(); + assertThat(event.customMetadata().get()) + .containsExactly( + CustomMetadata.builder().key("a2a:task_id").stringValue("task-1").build(), + CustomMetadata.builder().key("a2a:context_id").stringValue("context-1").build()); + } + + @Test + public void taskToEvent_withUnrecognizedMetadataField_dropsOnlyThatField() { + Message statusMessage = + new Message.Builder() + .role(Message.Role.AGENT) + .parts(ImmutableList.of(new TextPart("Status message"))) + .build(); + TaskStatus status = new TaskStatus(TaskState.WORKING, statusMessage, null); + Task task = + testTask() + .status(status) + .artifacts(null) + .metadata( + ImmutableMap.of( + // A nested object takes the convertValue branch rather than readValue. The + // genai builders reject unknown fields, so snake_case fails to convert. + A2AMetadataKey.GROUNDING_METADATA.getType(), + ImmutableMap.of("web_search_queries", ImmutableList.of("test-query")))) + .build(); + + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.groundingMetadata()).isEmpty(); + assertThat(event.content().get().parts().get().get(0).text()).hasValue("Status message"); + } + + @Test + public void taskToEvent_withOneMalformedMetadataField_keepsTheValidFields() { + GroundingMetadata groundingMetadata = + GroundingMetadata.builder().webSearchQueries("test-query").build(); + Message statusMessage = + new Message.Builder() + .role(Message.Role.AGENT) + .parts(ImmutableList.of(new TextPart("Status message"))) + .build(); + TaskStatus status = new TaskStatus(TaskState.WORKING, statusMessage, null); + Task task = + testTask() + .status(status) + .artifacts(null) + .metadata( + ImmutableMap.of( + A2AMetadataKey.GROUNDING_METADATA.getType(), + groundingMetadata.toJson(), + A2AMetadataKey.USAGE_METADATA.getType(), + "not-valid-json")) + .build(); + + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.groundingMetadata()).hasValue(groundingMetadata); + assertThat(event.usageMetadata()).isEmpty(); + } + @Test public void messageToEvent_withMissingTaskId_returnsEvent() { Message a2aMessage = @@ -272,6 +358,79 @@ public void taskToEvent_withInputRequired_parsesLongRunningToolIds() { assertThat(event.longRunningToolIds().get()).containsExactly("call_123", "msg_123"); } + @Test + public void taskToEvent_withDataPartWithoutMetadata_doesNotThrow() { + DataPart dataPart = + new DataPart( + ImmutableMap.of("name", "myTool", "id", "call_123", "args", ImmutableMap.of())); + DataPart statusDataPart = + new DataPart( + ImmutableMap.of("name", "messageTool", "id", "msg_123", "args", ImmutableMap.of())); + Message statusMessage = + new Message.Builder() + .role(Message.Role.AGENT) + .parts(ImmutableList.of(statusDataPart)) + .build(); + TaskStatus status = new TaskStatus(TaskState.INPUT_REQUIRED, statusMessage, null); + Artifact artifact = + new Artifact.Builder().artifactId("artifact-1").parts(ImmutableList.of(dataPart)).build(); + Task task = testTask().status(status).artifacts(ImmutableList.of(artifact)).build(); + + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.longRunningToolIds().get()).isEmpty(); + List parts = event.content().get().parts().get(); + assertThat(parts).hasSize(2); + assertThat(parts.get(0).functionCall().get().id()).hasValue("call_123"); + assertThat(parts.get(1).functionCall().get().id()).hasValue("msg_123"); + } + + @Test + public void artifactToEvent_withDataPartWithoutMetadata_doesNotThrow() { + DataPart dataPart = + new DataPart( + ImmutableMap.of("name", "myTool", "id", "call_123", "args", ImmutableMap.of())); + Artifact artifact = + new Artifact.Builder().artifactId("artifact-1").parts(ImmutableList.of(dataPart)).build(); + + Event event = ResponseConverter.artifactToEvent(artifact, invocationContext); + + assertThat(event.longRunningToolIds().get()).isEmpty(); + List parts = event.content().get().parts().get(); + assertThat(parts).hasSize(1); + assertThat(parts.get(0).functionCall().get().id()).hasValue("call_123"); + } + + @Test + public void taskToEvent_withMixedMetadataParts_keepsLongRunningId() { + DataPart noMetadataPart = + new DataPart( + ImmutableMap.of("name", "plainTool", "id", "call_plain", "args", ImmutableMap.of())); + 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)); + Artifact artifact = + new Artifact.Builder() + .artifactId("artifact-1") + .parts(ImmutableList.of(noMetadataPart, longRunningPart)) + .build(); + Task task = + testTask() + .status(new TaskStatus(TaskState.INPUT_REQUIRED, null, null)) + .artifacts(ImmutableList.of(artifact)) + .build(); + + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.longRunningToolIds().get()).containsExactly("call_lr"); + assertThat(event.content().get().parts().get()).hasSize(2); + } + @Test public void taskToEvent_withFailedState_setsErrorCode() { Message statusMessage = @@ -452,7 +611,7 @@ public void clientEventToEvent_withFailedTaskStatusUpdateEvent_returnsErrorEvent } @Test - public void taskToEvent_withInvalidMetadata_throwsException() { + public void taskToEvent_withInvalidMetadata_dropsFieldInsteadOfThrowing() { Message statusMessage = new Message.Builder() .role(Message.Role.AGENT) @@ -467,12 +626,10 @@ public void taskToEvent_withInvalidMetadata_throwsException() { ImmutableMap.of(A2AMetadataKey.GROUNDING_METADATA.getType(), "{ invalid json ]")) .build(); - IllegalArgumentException exception = - assertThrows( - IllegalArgumentException.class, - () -> ResponseConverter.taskToEvent(task, invocationContext)); - assertThat(exception).hasMessageThat().contains("Failed to parse metadata"); - assertThat(exception).hasMessageThat().contains("GroundingMetadata"); + Event event = ResponseConverter.taskToEvent(task, invocationContext); + + assertThat(event.groundingMetadata()).isEmpty(); + assertThat(event.content().get().parts().get().get(0).text()).hasValue("Status message"); } @Test