Upstream Merge

.release-please-manifest.json

@@ -1,3 +1,3 @@
 {
-    ".": "1.7.1"
+    ".": "1.8.0"
 }

CHANGELOG.md

@@ -1,5 +1,26 @@
 # Changelog
 
+## [1.8.0](https://github.com/google/adk-java/compare/v1.7.1...v1.8.0) (2026-08-13)
+
+
+### Features
+
+* Add onRunErrorCallback to ADK Plugin and Runner ([3e6b915](https://github.com/google/adk-java/commit/3e6b9154e089f24daf43c9ded7e4e40483fb7995))
+
+
+### Bug Fixes
+
+* **a2a:** drop unparseable A2A metadata instead of aborting conversion ([b75c916](https://github.com/google/adk-java/commit/b75c9169c630ab0450d16aa74898da6b953d0e78))
+* **a2a:** fail the A2A stream in the handler, not via the transport ([faa3482](https://github.com/google/adk-java/commit/faa3482fa70a5f750e4db04335ca5a5091be471e))
+* **a2a:** guard null DataPart metadata in ResponseConverter ([fcfd9bd](https://github.com/google/adk-java/commit/fcfd9bd8b1b5516932b9c5d72a62a191aea88e13))
+* **a2a:** require explicit adk_type metadata to convert A2A DataParts ([b704c5f](https://github.com/google/adk-java/commit/b704c5fc963d315c624b06d97a6a00d963e54cc0))
+* **core:** only resume tool confirmations for calls this agent emitted ([e5aba3a](https://github.com/google/adk-java/commit/e5aba3aa08c5b85a892e0c9164fa0ab8513786fa))
+* keep thought signature and tool call parts through streaming and history ([d7355a7](https://github.com/google/adk-java/commit/d7355a712345864682134762df890bf7b713b8c4))
+* **runner:** build a new message when saving input blobs, instead of writing into the caller's Content ([80c1a21](https://github.com/google/adk-java/commit/80c1a21da378f121fef3af065eb65dce0e0080c9))
+* stop returning exception text to remote A2A peers ([a3df463](https://github.com/google/adk-java/commit/a3df4632c19d857552af3d2c38373aabea9069de))
+* Update default BigQueryLoggerConfig table name and remove default dataset ID ([723a2ef](https://github.com/google/adk-java/commit/723a2ef0c4929879a6bd831287c34002ef02ef00))
+* update stream completion check in A2A SDK to handle all terminal and interrupted task states ([2b87d65](https://github.com/google/adk-java/commit/2b87d65d9704a61ff4668b8c9482a79fef9fe0d4))
+
 ## [1.7.1](https://github.com/google/adk-java/compare/v1.7.0...v1.7.1) (2026-07-28)
 
 

README.md

@@ -50,13 +50,13 @@ If you're using Maven, add the following to your dependencies:
 <dependency>
   <groupId>com.google.adk</groupId>
   <artifactId>google-adk</artifactId>
-  <version>1.7.1</version>
+  <version>1.8.0</version>
 </dependency>
 <!-- Dev UI -->
 <dependency>
     <groupId>com.google.adk</groupId>
     <artifactId>google-adk-dev</artifactId>
-    <version>1.7.1</version>
+    <version>1.8.0</version>
 </dependency>
 ```
 

a2a/pom.xml

@@ -5,7 +5,7 @@
   <parent>
     <groupId>com.google.adk</groupId>
     <artifactId>google-adk-parent</artifactId>
-    <version>1.7.2-SNAPSHOT</version><!-- {x-version-update:google-adk:current} -->
+    <version>1.8.1-SNAPSHOT</version><!-- {x-version-update:google-adk:current} -->
   </parent>
 
   <artifactId>google-adk-a2a</artifactId>

a2a/src/main/java/com/google/adk/a2a/agent/RemoteA2AAgent.java

@@ -36,6 +36,7 @@
 import com.google.genai.types.Part;
 import io.a2a.client.Client;
 import io.a2a.client.ClientEvent;
+import io.a2a.client.MessageEvent;
 import io.a2a.client.TaskEvent;
 import io.a2a.client.TaskUpdateEvent;
 import io.a2a.spec.A2AClientException;
@@ -228,14 +229,25 @@ protected Flowable<Event> runAsyncImpl(InvocationContext invocationContext) {
         emitter -> {
           StreamHandler handler =
               new StreamHandler(
-                  emitter.serialize(), invocationContext, requestJson, streaming, name());
+                  emitter.serialize(),
+                  invocationContext,
+                  requestJson,
+                  name(),
+                  /* subscribeThread= */ Thread.currentThread());
           ImmutableList<BiConsumer<ClientEvent, AgentCard>> consumers =
               ImmutableList.of(handler::handleEvent);
-          a2aClient.sendMessage(originalMessage, consumers, handler::handleError, null);
+          handler.dispatchSynchronously(
+              () -> a2aClient.sendMessage(originalMessage, consumers, handler::handleError, null));
         },
         BackpressureStrategy.BUFFER);
   }
 
+  /** A {@link Runnable} that may throw, so the a2a client's checked exception can propagate. */
+  @FunctionalInterface
+  private interface ThrowingRunnable {
+    void run() throws Exception;
+  }
+
   private @Nullable String serializeMessageToJson(Message message) {
     try {
       return objectMapper.writeValueAsString(message);
@@ -249,8 +261,26 @@ private static class StreamHandler {
     private final FlowableEmitter<Event> emitter;
     private final InvocationContext invocationContext;
     private final String requestJson;
-    private final boolean streaming;
     private final String agentName;
+    private final Thread subscribeThread;
+
+    /** True while {@code sendMessage} is running on {@link #subscribeThread}. */
+    private volatile boolean dispatchingSynchronously = false;
+
+    /**
+     * Runs the a2a client call that this handler is the consumer for, recording that any event
+     * delivered on {@link #subscribeThread} while it is on the stack came from synchronous
+     * dispatch. Owned here so the window cannot be left open by an edit at the call site.
+     */
+    void dispatchSynchronously(ThrowingRunnable dispatch) throws Exception {
+      dispatchingSynchronously = true;
+      try {
+        dispatch.run();
+      } finally {
+        dispatchingSynchronously = false;
+      }
+    }
+
     private boolean done = false;
     private final StringBuilder textBuffer = new StringBuilder();
     private final StringBuilder thoughtsBuffer = new StringBuilder();
@@ -259,16 +289,20 @@ private static class StreamHandler {
         FlowableEmitter<Event> emitter,
         InvocationContext invocationContext,
         String requestJson,
-        boolean streaming,
-        String agentName) {
+        String agentName,
+        Thread subscribeThread) {
       this.emitter = emitter;
       this.invocationContext = invocationContext;
       this.requestJson = requestJson;
-      this.streaming = streaming;
       this.agentName = agentName;
+      this.subscribeThread = subscribeThread;
     }
 
     synchronized void handleError(Throwable e) {
+      handleError("Failed to communicate with the remote agent", e);
+    }
+
+    synchronized void handleError(String message, Throwable e) {
       // Mark the flow as done if it is already cancelled.
       if (!done) {
         done = emitter.isCancelled();
@@ -280,7 +314,7 @@ synchronized void handleError(Throwable e) {
       }
       // If the error is raised, complete the flow with an error.
       done = true;
-      emitter.tryOnError(new A2AClientError("Failed to communicate with the remote agent", e));
+      emitter.tryOnError(new A2AClientError(message, e));
     }
 
     // TODO: b/483038527 - The synchronized block might block the thread, we should optimize for
@@ -296,8 +330,33 @@ synchronized void handleEvent(ClientEvent clientEvent, AgentCard unused) {
         return;
       }
 
-      Optional<Event> eventOpt =
-          ResponseConverter.clientEventToEvent(clientEvent, invocationContext);
+      Optional<Event> eventOpt;
+      try {
+        eventOpt = ResponseConverter.clientEventToEvent(clientEvent, invocationContext);
+      } catch (Throwable t) {
+        logger.warn("Failed to convert A2A event", t);
+        handleError("Failed to convert the remote agent's response", t);
+        if (!dispatchingSynchronously || Thread.currentThread() != subscribeThread) {
+          // Delivered by the transport rather than from inside sendMessage: rethrow so the
+          // transport tears the connection down. Synchronous dispatch is the one case where
+          // Flowable.create has already failed the flow, and rethrowing into it would only add
+          // an undeliverable via RxJavaPlugins.
+          throw t;
+        }
+        return;
+      }
+      emit(clientEvent, eventOpt);
+    }
+
+    /**
+     * Emits a converted event.
+     *
+     * <p>Outside the conversion guard above only because of the {@code emitter} calls: RxJava's
+     * contract is that a downstream {@code onNext} does not throw, so such a throw is the
+     * subscriber's bug and must not be fed back to that same subscriber as {@code onError}. The
+     * surrounding aggregation bookkeeping is ADK's own; none of it throws on peer input today.
+     */
+    private void emit(ClientEvent clientEvent, Optional<Event> eventOpt) {
       eventOpt.ifPresent(
           event -> {
             addMetadata(event, clientEvent);
@@ -522,13 +581,15 @@ private Event createAggregatedEvent(Content content, @Nullable ClientEvent trigg
   }
 
   private static boolean isCompleted(ClientEvent event) {
-    TaskState executionState = TaskState.UNKNOWN;
+    TaskState state;
     if (event instanceof TaskEvent taskEvent) {
-      executionState = taskEvent.getTask().getStatus().state();
+      state = taskEvent.getTask().getStatus().state();
     } else if (event instanceof TaskUpdateEvent updateEvent) {
-      executionState = updateEvent.getTask().getStatus().state();
+      state = updateEvent.getTask().getStatus().state();
+    } else {
+      return false;
     }
-    return executionState.equals(TaskState.COMPLETED);
+    return state.isFinal() || state == TaskState.INPUT_REQUIRED || state == TaskState.AUTH_REQUIRED;
   }
 
   private static ImmutableList<Part> eventParts(Event event) {

a2a/src/main/java/com/google/adk/a2a/converters/ResponseConverter.java

@@ -150,7 +150,9 @@ private static Optional<Event> handleTaskUpdate(
                     return messageToEvent(value, context, PENDING_STATES.contains(taskState));
                   });
 
-      if (statusEvent.isFinal()) {
+      if (statusEvent.isFinal()
+          || taskState == TaskState.INPUT_REQUIRED
+          || taskState == TaskState.AUTH_REQUIRED) {
         messageEvent =
             messageEvent
                 .map(Event::toBuilder)
@@ -256,15 +258,18 @@ public static Event taskToEvent(Task task, InvocationContext invocationContext)
 
     ImmutableList<Part> finalParts = genaiParts.build();
     boolean isFinal =
-        task.getStatus().state().isFinal() || task.getStatus().state() == TaskState.INPUT_REQUIRED;
+        task.getStatus().state().isFinal()
+            || task.getStatus().state() == TaskState.INPUT_REQUIRED
+            || task.getStatus().state() == TaskState.AUTH_REQUIRED;
 
     if (finalParts.isEmpty() && !isFinal) {
       return emptyEvent(invocationContext);
     }
     if (!finalParts.isEmpty()) {
       eventBuilder.content(fromModelParts(finalParts));
     }
-    if (task.getStatus().state() == TaskState.INPUT_REQUIRED) {
+    if (task.getStatus().state() == TaskState.INPUT_REQUIRED
+        || task.getStatus().state() == TaskState.AUTH_REQUIRED) {
       eventBuilder.longRunningToolIds(longRunningToolIds.build());
     }
     eventBuilder.turnComplete(isFinal);