feat: add a GeminiLiveTransport seam to GeminiLlmConnection

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

@@ -313,7 +313,17 @@ public BaseLlmConnection connect(LlmRequest llmRequest) {
     logger.debug("Connecting to model {}", effectiveModelName);
     logger.trace("Connection Config: {}", liveConnectConfig);
 
-    return new GeminiLlmConnection(apiClient, effectiveModelName, liveConnectConfig);
+    return new GeminiLlmConnection(connectLiveTransport(effectiveModelName, liveConnectConfig));
+  }
+
+  /**
+   * Opens the live transport the connection drives. Overridable so a test can supply an in-process
+   * {@link GeminiLiveTransport} double and exercise the real {@link GeminiLlmConnection} without a
+   * network.
+   */
+  protected CompletableFuture<GeminiLiveTransport> connectLiveTransport(
+      String modelName, LiveConnectConfig config) {
+    return apiClient.async.live.connect(modelName, config).thenApply(GenAiLiveTransport::new);
   }
 
   private static final class StreamingResponseAggregator {

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

@@ -0,0 +1,53 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.models;
+
+import com.google.genai.types.LiveSendClientContentParameters;
+import com.google.genai.types.LiveSendRealtimeInputParameters;
+import com.google.genai.types.LiveSendToolResponseParameters;
+import com.google.genai.types.LiveServerMessage;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+/**
+ * The bidirectional live transport that {@link GeminiLlmConnection} drives.
+ *
+ * <p>{@link GeminiLlmConnection} holds the translation logic; this is the transport beneath it, so
+ * the connection can drive any implementation. The default one delegates to a genai live session.
+ */
+public interface GeminiLiveTransport {
+
+  /** Sends a client-content turn to the transport. */
+  CompletableFuture<Void> sendClientContent(LiveSendClientContentParameters params);
+
+  /** Sends realtime input (audio, video, or text) to the transport. */
+  CompletableFuture<Void> sendRealtimeInput(LiveSendRealtimeInputParameters params);
+
+  /** Sends a tool response to the transport. */
+  CompletableFuture<Void> sendToolResponse(LiveSendToolResponseParameters params);
+
+  /**
+   * Registers the callback for messages the transport yields and a callback for when its receive
+   * stream ends. An implementation whose stream ends only when the client closes never invokes
+   * {@code onStreamEnd}; one backed by a finite script invokes it when the script is exhausted, so
+   * the run can end.
+   */
+  CompletableFuture<Void> receive(Consumer<LiveServerMessage> onMessage, Runnable onStreamEnd);
+
+  /** Closes the transport. */
+  CompletableFuture<Void> close();
+}

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

@@ -19,13 +19,10 @@
 import static com.google.common.collect.ImmutableList.toImmutableList;
 
 import com.google.common.collect.ImmutableList;
-import com.google.genai.AsyncSession;
-import com.google.genai.Client;
 import com.google.genai.types.Blob;
 import com.google.genai.types.Content;
 import com.google.genai.types.FinishReason;
 import com.google.genai.types.FunctionResponse;
-import com.google.genai.types.LiveConnectConfig;
 import com.google.genai.types.LiveSendClientContentParameters;
 import com.google.genai.types.LiveSendRealtimeInputParameters;
 import com.google.genai.types.LiveSendToolResponseParameters;
@@ -61,60 +58,57 @@ public final class GeminiLlmConnection implements BaseLlmConnection {
 
   private static final Logger logger = LoggerFactory.getLogger(GeminiLlmConnection.class);
 
-  private final Client apiClient;
-  private final String modelName;
-  private final LiveConnectConfig connectConfig;
-  private final CompletableFuture<AsyncSession> sessionFuture;
+  private final CompletableFuture<GeminiLiveTransport> transportFuture;
   private final PublishProcessor<LlmResponse> responseProcessor = PublishProcessor.create();
   private final Flowable<LlmResponse> responseFlowable = responseProcessor.serialize();
   private final CompositeDisposable disposables = new CompositeDisposable();
   private final AtomicBoolean closed = new AtomicBoolean(false);
 
   /**
-   * Establishes a new connection.
+   * Establishes a new connection over the given live transport.
    *
-   * @param apiClient The API client for communication.
-   * @param modelName The specific Gemini model endpoint (e.g., "gemini-2.0-flash).
-   * @param connectConfig Configuration parameters for the live session.
+   * @param transportFuture The live transport the connection drives, once established.
    */
-  GeminiLlmConnection(Client apiClient, String modelName, LiveConnectConfig connectConfig) {
-    this.apiClient = Objects.requireNonNull(apiClient);
-    this.modelName = Objects.requireNonNull(modelName);
-    this.connectConfig = Objects.requireNonNull(connectConfig);
-
-    this.sessionFuture =
-        this.apiClient
-            .async
-            .live
-            .connect(this.modelName, this.connectConfig)
+  GeminiLlmConnection(CompletableFuture<GeminiLiveTransport> transportFuture) {
+    this.transportFuture =
+        Objects.requireNonNull(transportFuture)
             .whenCompleteAsync(
-                (session, throwable) -> {
+                (transport, throwable) -> {
                   if (throwable != null) {
                     handleConnectionError(throwable);
-                  } else if (session != null) {
-                    setupReceiver(session);
+                  } else if (transport != null) {
+                    setupReceiver(transport);
                   } else if (!closed.get()) {
                     handleConnectionError(
                         new SocketException("WebSocket connection failed without explicit error."));
                   }
                 });
   }
 
-  /** Configures the session to forward incoming messages to the response processor. */
-  private void setupReceiver(AsyncSession session) {
+  /** Configures the transport to forward incoming messages to the response processor. */
+  private void setupReceiver(GeminiLiveTransport transport) {
     if (closed.get()) {
-      closeSessionIgnoringErrors(session);
+      closeTransportIgnoringErrors(transport);
       return;
     }
-    session
-        .receive(this::handleServerMessage)
+    transport
+        .receive(this::handleServerMessage, this::completeReceive)
         .exceptionally(
             error -> {
               handleReceiveError(error);
               return null;
             });
   }
 
+  /** Completes the response stream when the transport's receive stream ends, ending the run. */
+  private void completeReceive() {
+    // Only a transport that ends its own stream reaches this, so it owns its close; none is issued.
+    if (closed.compareAndSet(false, true)) {
+      responseProcessor.onComplete();
+      disposables.dispose();
+    }
+  }
+
   /** Processes messages received from the WebSocket server. */
   private void handleServerMessage(LiveServerMessage message) {
     if (closed.get()) {
@@ -241,7 +235,9 @@ private void handleReceiveError(Throwable throwable) {
     if (closed.compareAndSet(false, true)) {
       logger.error("Error during WebSocket receive operation", throwable);
       responseProcessor.onError(throwable);
-      sessionFuture.thenAccept(this::closeSessionIgnoringErrors).exceptionally(unusedError -> null);
+      transportFuture
+          .thenAccept(this::closeTransportIgnoringErrors)
+          .exceptionally(unusedError -> null);
     }
   }
 
@@ -286,22 +282,22 @@ private List<FunctionResponse> extractFunctionResponses(Content content) {
   @Override
   public Completable sendRealtime(Blob blob) {
     return Completable.fromFuture(
-        sessionFuture.thenCompose(
-            session ->
-                session.sendRealtimeInput(
+        transportFuture.thenCompose(
+            transport ->
+                transport.sendRealtimeInput(
                     LiveSendRealtimeInputParameters.builder().media(blob).build())));
   }
 
   /** Helper to send client content parameters. */
   private Completable sendClientContentInternal(LiveSendClientContentParameters parameters) {
     return Completable.fromFuture(
-        sessionFuture.thenCompose(session -> session.sendClientContent(parameters)));
+        transportFuture.thenCompose(transport -> transport.sendClientContent(parameters)));
   }
 
   /** Helper to send tool response parameters. */
   private Completable sendToolResponseInternal(LiveSendToolResponseParameters parameters) {
     return Completable.fromFuture(
-        sessionFuture.thenCompose(session -> session.sendToolResponse(parameters)));
+        transportFuture.thenCompose(transport -> transport.sendToolResponse(parameters)));
   }
 
   @Override
@@ -331,26 +327,26 @@ private void closeInternal(Throwable throwable) {
         responseProcessor.onError(throwable);
       }
 
-      if (sessionFuture.isDone()) {
-        sessionFuture
-            .thenAccept(this::closeSessionIgnoringErrors)
+      if (transportFuture.isDone()) {
+        transportFuture
+            .thenAccept(this::closeTransportIgnoringErrors)
             .exceptionally(unusedError -> null);
       } else {
-        sessionFuture.cancel(false);
+        transportFuture.cancel(false);
       }
 
       disposables.dispose();
     }
   }
 
-  /** Closes the AsyncSession safely, logging any errors. */
-  private void closeSessionIgnoringErrors(AsyncSession session) {
-    if (session != null) {
-      session
+  /** Closes the transport safely, logging any errors. */
+  private void closeTransportIgnoringErrors(GeminiLiveTransport transport) {
+    if (transport != null) {
+      transport
           .close()
           .exceptionally(
               closeError -> {
-                logger.warn("Error occurred while closing AsyncSession", closeError);
+                logger.warn("Error occurred while closing live transport", closeError);
                 return null; // Suppress error during close
               });
     }

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

@@ -0,0 +1,63 @@
+/*
+ * Copyright 2025 Google LLC
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.google.adk.models;
+
+import com.google.genai.AsyncSession;
+import com.google.genai.types.LiveSendClientContentParameters;
+import com.google.genai.types.LiveSendRealtimeInputParameters;
+import com.google.genai.types.LiveSendToolResponseParameters;
+import com.google.genai.types.LiveServerMessage;
+import java.util.Objects;
+import java.util.concurrent.CompletableFuture;
+import java.util.function.Consumer;
+
+/** The default {@link GeminiLiveTransport}, delegating to a genai {@link AsyncSession}. */
+final class GenAiLiveTransport implements GeminiLiveTransport {
+
+  private final AsyncSession session;
+
+  GenAiLiveTransport(AsyncSession session) {
+    this.session = Objects.requireNonNull(session);
+  }
+
+  @Override
+  public CompletableFuture<Void> sendClientContent(LiveSendClientContentParameters params) {
+    return session.sendClientContent(params);
+  }
+
+  @Override
+  public CompletableFuture<Void> sendRealtimeInput(LiveSendRealtimeInputParameters params) {
+    return session.sendRealtimeInput(params);
+  }
+
+  @Override
+  public CompletableFuture<Void> sendToolResponse(LiveSendToolResponseParameters params) {
+    return session.sendToolResponse(params);
+  }
+
+  @Override
+  public CompletableFuture<Void> receive(
+      Consumer<LiveServerMessage> onMessage, Runnable onStreamEnd) {
+    // The genai session has no end-of-stream signal, so onStreamEnd never fires here.
+    return session.receive(onMessage);
+  }
+
+  @Override
+  public CompletableFuture<Void> close() {
+    return session.close();
+  }
+}