fix(a2a): stop a converter failure from stalling the A2A event stream

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<Event> eventOpt =
           ResponseConverter.clientEventToEvent(clientEvent, invocationContext);
       eventOpt.ifPresent(

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).
    *
+   * <p>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<Event> clientEventToEvent(
@@ -91,6 +93,11 @@ private static boolean isPartial(@Nullable Map<String, Object> metadata) {
     return Objects.equals(metadata.getOrDefault(A2AMetadataKey.PARTIAL.getType(), false), true);
   }
 
+  private static boolean isLongRunning(@Nullable Map<String, Object> 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.
+   *
+   * <p>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.
+   *
+   * <p>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<Part> genaiParts = ImmutableList.builder();
@@ -266,9 +279,8 @@ private static ImmutableSet<String> getLongRunningToolIds(
               if (!(part instanceof DataPart dataPart)) {
                 return Optional.<String>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.<String>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<CustomMetadata> customMetadataList = ImmutableList.builder();
     customMetadataList
@@ -314,43 +326,76 @@ 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<CustomMetadata>>() {}));
+    List<CustomMetadata> parsedCustomMetadata =
+        parseMetadata(
+            clientMetadata,
+            A2AMetadataKey.CUSTOM_METADATA,
+            new TypeReference<List<CustomMetadata>>() {});
+    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 <T> @Nullable T parseMetadata(@Nullable Object metadata, Class<T> type) {
+  /**
+   * Reads {@code key} out of the peer-supplied {@code clientMetadata} and deserializes it.
+   *
+   * <p>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 <T> @Nullable T parseMetadata(
+      Map<String, Object> clientMetadata, A2AMetadataKey key, Class<T> type) {
+    Object metadata = clientMetadata.get(key.getType());
     try {
       if (metadata instanceof String jsonString) {
         return objectMapper.readValue(jsonString, type);
       } else {
         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 <T> @Nullable T parseMetadata(@Nullable Object metadata, TypeReference<T> type) {
+  /** Overload of {@link #parseMetadata(Map, A2AMetadataKey, Class)} for generic target types. */
+  private static <T> @Nullable T parseMetadata(
+      Map<String, Object> clientMetadata, A2AMetadataKey key, TypeReference<T> type) {
+    Object metadata = clientMetadata.get(key.getType());
     try {
       if (metadata instanceof String jsonString) {
         return objectMapper.readValue(jsonString, type);
       } else {
         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.
+   *
+   * <p>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()

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<Event> 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();