diff --git a/core/src/main/java/com/google/adk/agents/LiveRequest.java b/core/src/main/java/com/google/adk/agents/LiveRequest.java index 84420b1ae..7cc081c92 100644 --- a/core/src/main/java/com/google/adk/agents/LiveRequest.java +++ b/core/src/main/java/com/google/adk/agents/LiveRequest.java @@ -45,6 +45,16 @@ public abstract class LiveRequest extends JsonBaseModel { @JsonProperty("content") public abstract Optional content(); + /** + * Returns whether sending {@link #content()} should complete the current turn. + * + *

If unset, content sends complete the turn to preserve the existing behavior. + * + * @return An optional boolean indicating whether the current turn should be completed. + */ + @JsonProperty("turnComplete") + public abstract Optional turnComplete(); + /** * Returns the blob of the request. * @@ -83,6 +93,9 @@ static LiveRequest.Builder jacksonBuilder() { @JsonProperty("content") public abstract Builder content(@Nullable Content content); + @JsonProperty("turnComplete") + public abstract Builder turnComplete(@Nullable Boolean turnComplete); + @JsonProperty("blob") public abstract Builder blob(@Nullable Blob blob); diff --git a/core/src/main/java/com/google/adk/agents/LiveRequestQueue.java b/core/src/main/java/com/google/adk/agents/LiveRequestQueue.java index 8bd2b4d64..5704847c5 100644 --- a/core/src/main/java/com/google/adk/agents/LiveRequestQueue.java +++ b/core/src/main/java/com/google/adk/agents/LiveRequestQueue.java @@ -38,7 +38,17 @@ public void close() { } public void content(Content content) { - processor.onNext(LiveRequest.builder().content(content).build()); + content(content, true); + } + + /** + * Sends content to the model with explicit control over whether it completes the current turn. + * + * @param content the content to send + * @param turnComplete whether the model should begin responding after receiving the content + */ + public void content(Content content, boolean turnComplete) { + processor.onNext(LiveRequest.builder().content(content).turnComplete(turnComplete).build()); } public void realtime(Blob blob) { diff --git a/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java b/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java index 91cc225f2..678987e9b 100644 --- a/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java +++ b/core/src/main/java/com/google/adk/flows/llmflows/BaseLlmFlow.java @@ -632,7 +632,9 @@ public Flowable runLive(InvocationContext invocationContext) { .concatMapCompletable( request -> { if (request.content().isPresent()) { - return connection.sendContent(request.content().get()); + return connection.sendContent( + request.content().get(), + request.turnComplete().orElse(true)); } else if (request.blob().isPresent()) { return connection.sendRealtime(request.blob().get()); } diff --git a/core/src/main/java/com/google/adk/models/BaseLlmConnection.java b/core/src/main/java/com/google/adk/models/BaseLlmConnection.java index c8093ff9c..bcf4f4708 100644 --- a/core/src/main/java/com/google/adk/models/BaseLlmConnection.java +++ b/core/src/main/java/com/google/adk/models/BaseLlmConnection.java @@ -41,6 +41,18 @@ public interface BaseLlmConnection { */ Completable sendContent(Content content); + /** + * Sends user content to the model and controls whether it completes the current turn. + * + *

Implementations that do not support incomplete turns retain the existing behavior. + * + * @param content the content to send + * @param turnComplete whether the model should begin responding after receiving the content + */ + default Completable sendContent(Content content, boolean turnComplete) { + return sendContent(content); + } + /** * Sends a chunk of audio or a frame of video to the model in realtime. * diff --git a/core/src/main/java/com/google/adk/models/GeminiLlmConnection.java b/core/src/main/java/com/google/adk/models/GeminiLlmConnection.java index 35dadd9fd..d9c592bff 100644 --- a/core/src/main/java/com/google/adk/models/GeminiLlmConnection.java +++ b/core/src/main/java/com/google/adk/models/GeminiLlmConnection.java @@ -253,6 +253,11 @@ public Completable sendHistory(List history) { @Override public Completable sendContent(Content content) { + return sendContent(content, true); + } + + @Override + public Completable sendContent(Content content, boolean turnComplete) { Objects.requireNonNull(content, "content cannot be null"); List functionResponses = extractFunctionResponses(content); @@ -260,7 +265,7 @@ public Completable sendContent(Content content) { return sendClientContentInternal( LiveSendClientContentParameters.builder() .turns(ImmutableList.of(content)) - .turnComplete(true) + .turnComplete(turnComplete) .build()); } return sendToolResponseInternal( diff --git a/core/src/test/java/com/google/adk/agents/LiveRequestQueueTest.java b/core/src/test/java/com/google/adk/agents/LiveRequestQueueTest.java new file mode 100644 index 000000000..9ac383487 --- /dev/null +++ b/core/src/test/java/com/google/adk/agents/LiveRequestQueueTest.java @@ -0,0 +1,68 @@ +/* + * Copyright 2026 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.agents; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.reactivex.rxjava3.subscribers.TestSubscriber; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class LiveRequestQueueTest { + + private static final Content CONTENT = Content.fromParts(Part.fromText("context")); + + @Test + public void content_defaultsToCompletingTurn() { + LiveRequestQueue queue = new LiveRequestQueue(); + TestSubscriber subscriber = queue.get().test(); + + queue.content(CONTENT); + + subscriber.assertValueCount(1); + LiveRequest request = subscriber.values().get(0); + assertThat(request.content()).hasValue(CONTENT); + assertThat(request.turnComplete()).hasValue(true); + } + + @Test + public void content_withTurnCompleteFalse_preservesValue() { + LiveRequestQueue queue = new LiveRequestQueue(); + TestSubscriber subscriber = queue.get().test(); + + queue.content(CONTENT, false); + + subscriber.assertValueCount(1); + LiveRequest request = subscriber.values().get(0); + assertThat(request.content()).hasValue(CONTENT); + assertThat(request.turnComplete()).hasValue(false); + } + + @Test + public void liveRequest_jsonRoundTrip_preservesTurnComplete() { + LiveRequest request = LiveRequest.builder().content(CONTENT).turnComplete(false).build(); + + LiveRequest restored = LiveRequest.fromJsonString(request.toJson()); + + assertThat(restored.content()).hasValue(CONTENT); + assertThat(restored.turnComplete()).hasValue(false); + } +}