fix: aggregate live text chunks and skip partial events in runLive

core/src/main/java/com/google/adk/models/GeminiLlmConnection.java

@@ -34,9 +34,11 @@
 import io.reactivex.rxjava3.core.Completable;
 import io.reactivex.rxjava3.core.Flowable;
 import io.reactivex.rxjava3.core.Observable;
+import io.reactivex.rxjava3.core.ObservableEmitter;
 import io.reactivex.rxjava3.disposables.CompositeDisposable;
 import io.reactivex.rxjava3.processors.PublishProcessor;
 import java.net.SocketException;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
@@ -63,6 +65,8 @@ public final class GeminiLlmConnection implements BaseLlmConnection {
   private final Flowable<LlmResponse> responseFlowable = responseProcessor.serialize();
   private final CompositeDisposable disposables = new CompositeDisposable();
   private final AtomicBoolean closed = new AtomicBoolean(false);
+  // Used only by handleServerMessage, which genai's WebSocket listener calls one message at a time.
+  private final PendingText pendingText = new PendingText();
 
   /**
    * Establishes a new connection over the given live transport.
@@ -117,16 +121,21 @@ private void handleServerMessage(LiveServerMessage message) {
 
     logger.debug("Received server message: {}", message.toJson());
 
-    Observable<LlmResponse> llmResponse = convertToServerResponse(message);
+    Observable<LlmResponse> llmResponse = convertToServerResponse(message, pendingText);
     if (!disposables.add(
         llmResponse.subscribe(responseProcessor::onNext, responseProcessor::onError))) {
       logger.warn(
           "disposables container already disposed, the subscription will be disposed immediately");
     }
   }
 
-  /** Converts a server message into the standardized LlmResponse format. */
-  static Observable<LlmResponse> convertToServerResponse(LiveServerMessage message) {
+  /**
+   * Converts a server message into the standardized LlmResponse format.
+   *
+   * @param pendingText the text streamed by earlier messages and not yet emitted in full
+   */
+  static Observable<LlmResponse> convertToServerResponse(
+      LiveServerMessage message, PendingText pendingText) {
     return Observable.create(
         emitter -> {
           // AtomicBoolean is used to modify state from within lambdas, which
@@ -136,13 +145,14 @@ static Observable<LlmResponse> convertToServerResponse(LiveServerMessage message
               .serverContent()
               .ifPresent(
                   serverContent -> {
-                    emitter.onNext(createServerContentResponse(serverContent));
+                    emitServerContentResponses(serverContent, pendingText, emitter);
                     handled.set(true);
                   });
           message
               .toolCall()
               .ifPresent(
                   toolCall -> {
+                    pendingText.flush().ifPresent(emitter::onNext);
                     emitter.onNext(createToolCallResponse(toolCall));
                     handled.set(true);
                   });
@@ -178,16 +188,74 @@ static Observable<LlmResponse> convertToServerResponse(LiveServerMessage message
         });
   }
 
-  private static LlmResponse createServerContentResponse(LiveServerContent serverContent) {
+  /**
+   * Emits the responses for a server content message. Text parts are streamed in partial responses
+   * and collected in {@code pendingText}, which is emitted as one non-partial response when other
+   * content ends it and before the end of the turn or an interruption. Text that arrives with the
+   * end of the turn or an interruption is emitted only in full. The full text reports an
+   * interruption, and no response is emitted just to report it again.
+   */
+  private static void emitServerContentResponses(
+      LiveServerContent serverContent,
+      PendingText pendingText,
+      ObservableEmitter<LlmResponse> emitter) {
+    List<Part> parts = serverContent.modelTurn().flatMap(Content::parts).orElse(ImmutableList.of());
+    List<Part> streamedParts = new ArrayList<>();
+    for (Part part : parts) {
+      if (pendingText.isEndedBy(part)) {
+        pendingText.flush().ifPresent(emitter::onNext);
+        // The full response now carries the text parts of this message so far.
+        streamedParts.removeIf(GeminiLlmConnection::hasText);
+      }
+      if (hasText(part)) {
+        pendingText.append(part);
+      }
+      streamedParts.add(part);
+    }
+    boolean turnComplete = serverContent.turnComplete().orElse(false);
+    boolean interrupted = serverContent.interrupted().orElse(false);
+    boolean interruptionInFullText = false;
+    if (turnComplete || interrupted) {
+      streamedParts.removeIf(GeminiLlmConnection::hasText);
+      Optional<LlmResponse> fullText = pendingText.flush(interrupted);
+      fullText.ifPresent(emitter::onNext);
+      interruptionInFullText =
+          interrupted && fullText.isPresent() && !turnComplete && streamedParts.isEmpty();
+    }
+    boolean hasTranscription =
+        serverContent.inputTranscription().isPresent()
+            || serverContent.outputTranscription().isPresent();
+    if (interruptionInFullText && !hasTranscription) {
+      return;
+    }
+
     LlmResponse.Builder builder = LlmResponse.builder();
-    serverContent.modelTurn().ifPresent(builder::content);
-    return builder
-        .partial(serverContent.turnComplete().map(completed -> !completed).orElse(false))
-        .turnComplete(serverContent.turnComplete().orElse(false))
-        .interrupted(serverContent.interrupted().orElse(null))
-        .inputTranscription(serverContent.inputTranscription().orElse(null))
-        .outputTranscription(serverContent.outputTranscription().orElse(null))
-        .build();
+    if (streamedParts.size() == parts.size()) {
+      serverContent.modelTurn().ifPresent(builder::content);
+    } else if (!streamedParts.isEmpty()) {
+      builder.content(serverContent.modelTurn().get().toBuilder().parts(streamedParts).build());
+    }
+    // Only text is partial: it is emitted again in full.
+    boolean streamsText = streamedParts.stream().anyMatch(GeminiLlmConnection::hasText);
+    builder
+        .partial(streamsText)
+        .turnComplete(turnComplete)
+        .interrupted(interruptionInFullText ? null : serverContent.interrupted().orElse(null));
+    if (streamsText && hasTranscription) {
+      // A partial response is not persisted, so the transcriptions get a response of their own.
+      emitter.onNext(builder.build());
+      builder = LlmResponse.builder().partial(false).turnComplete(turnComplete);
+    }
+    emitter.onNext(
+        builder
+            .inputTranscription(serverContent.inputTranscription().orElse(null))
+            .outputTranscription(serverContent.outputTranscription().orElse(null))
+            .build());
+  }
+
+  /** Returns whether {@code part} has text; an empty string does not count. */
+  private static boolean hasText(Part part) {
+    return part.text().filter(text -> !text.isEmpty()).isPresent();
   }
 
   private static LlmResponse createToolCallResponse(LiveServerToolCall toolCall) {
@@ -356,4 +424,53 @@ private void closeTransportIgnoringErrors(GeminiLiveTransport transport) {
               });
     }
   }
+
+  /** Model text that has been streamed in partial responses but not yet emitted in full. */
+  static final class PendingText {
+    private final StringBuilder text = new StringBuilder();
+    private boolean thought;
+
+    /**
+     * Returns whether {@code part} ends the pending text: text of the other thought kind, or a part
+     * that is neither text nor inline data.
+     */
+    private boolean isEndedBy(Part part) {
+      if (text.length() == 0) {
+        return false;
+      }
+      return hasText(part) ? part.thought().orElse(false) != thought : part.inlineData().isEmpty();
+    }
+
+    private void append(Part part) {
+      text.append(part.text().get());
+      thought = part.thought().orElse(false);
+    }
+
+    /** Returns the pending text as one non-partial response and clears it, or empty if none. */
+    private Optional<LlmResponse> flush() {
+      return flush(/* interrupted= */ false);
+    }
+
+    /**
+     * Returns the pending text as one non-partial response and clears it, or empty if none.
+     *
+     * @param interrupted whether an interruption ended the text, which the response then reports
+     */
+    private Optional<LlmResponse> flush(boolean interrupted) {
+      if (text.length() == 0) {
+        return Optional.empty();
+      }
+      Part.Builder part = Part.builder().text(text.toString());
+      if (thought) {
+        part.thought(true);
+      }
+      text.setLength(0);
+      return Optional.of(
+          LlmResponse.builder()
+              .content(Content.builder().role("model").parts(part.build()).build())
+              .partial(false)
+              .interrupted(interrupted ? true : null)
+              .build());
+    }
+  }
 }

core/src/main/java/com/google/adk/runner/Runner.java

@@ -1609,8 +1609,12 @@ protected Flowable<Event> runLiveImpl(
                       updatedInvocationContext
                           .agent()
                           .runLive(updatedInvocationContext)
+                          // As in runAsync, partial events are streamed but not persisted.
                           .concatMapSingle(
-                              event -> this.sessionService.appendEvent(session, event)))
+                              event ->
+                                  event.partial().orElse(false)
+                                      ? Single.just(event)
+                                      : this.sessionService.appendEvent(session, event)))
               .doOnError(
                   throwable -> {
                     Span span = Span.current();