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..a5ecc7e2e 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 {@link #content()} is a partial update that leaves the current turn open. + * + *

Defaults to false, so content sends complete the turn unless marked as partial. + * + * @return An optional boolean indicating whether the content is a partial turn update. + */ + @JsonProperty("partial") + public abstract Optional partial(); + /** * Returns the blob of the request. * @@ -83,6 +93,9 @@ static LiveRequest.Builder jacksonBuilder() { @JsonProperty("content") public abstract Builder content(@Nullable Content content); + @JsonProperty("partial") + public abstract Builder partial(@Nullable Boolean partial); + @JsonProperty("blob") public abstract Builder blob(@Nullable Blob blob); @@ -103,7 +116,7 @@ public final LiveRequest build() { } public static Builder builder() { - return new AutoValue_LiveRequest.Builder().close(false); + return new AutoValue_LiveRequest.Builder().close(false).partial(false); } public abstract Builder toBuilder(); 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..325991202 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, false); + } + + /** + * Sends content to the model, optionally as a partial update that leaves the current turn open. + * + * @param content the content to send + * @param partial whether this content is a partial update that does not complete the current turn + */ + public void content(Content content, boolean partial) { + processor.onNext(LiveRequest.builder().content(content).partial(partial).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 43faef5cf..cbfc9d168 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 @@ -657,7 +657,8 @@ public Flowable runLive(InvocationContext invocationContext) { .concatMapCompletable( request -> { if (request.content().isPresent()) { - return connection.sendContent(request.content().get()); + return connection.sendContent( + request.content().get(), request.partial().orElse(false)); } 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..b2a29157e 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,19 @@ public interface BaseLlmConnection { */ Completable sendContent(Content content); + /** + * Sends user content to the model, optionally leaving the current turn open for more content. + * + *

Implementations that do not support incomplete turns retain the existing behavior. Function + * responses use the existing tool-response path, regardless of {@code partial}. + * + * @param content the content to send + * @param partial whether this content is a partial update that does not complete the current turn + */ + default Completable sendContent(Content content, boolean partial) { + 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 5b2b26814..1f1ec1c12 100644 --- a/core/src/main/java/com/google/adk/models/GeminiLlmConnection.java +++ b/core/src/main/java/com/google/adk/models/GeminiLlmConnection.java @@ -249,6 +249,11 @@ public Completable sendHistory(List history) { @Override public Completable sendContent(Content content) { + return sendContent(content, false); + } + + @Override + public Completable sendContent(Content content, boolean partial) { Objects.requireNonNull(content, "content cannot be null"); List functionResponses = extractFunctionResponses(content); @@ -256,7 +261,7 @@ public Completable sendContent(Content content) { return sendClientContentInternal( LiveSendClientContentParameters.builder() .turns(ImmutableList.of(content)) - .turnComplete(true) + .turnComplete(!partial) .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..a019f7b8f --- /dev/null +++ b/core/src/test/java/com/google/adk/agents/LiveRequestQueueTest.java @@ -0,0 +1,123 @@ +/* + * 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.fasterxml.jackson.databind.JsonNode; +import com.google.adk.JsonBaseModel; +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.partial()).hasValue(false); + } + + @Test + public void content_withPartialTrue_preservesValue() { + LiveRequestQueue queue = new LiveRequestQueue(); + TestSubscriber subscriber = queue.get().test(); + + queue.content(CONTENT, true); + + subscriber.assertValueCount(1); + LiveRequest request = subscriber.values().get(0); + assertThat(request.content()).hasValue(CONTENT); + assertThat(request.partial()).hasValue(true); + } + + @Test + public void content_withPartialFalse_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.partial()).hasValue(false); + } + + @Test + public void liveRequest_builderDefaultsToNonPartial() { + LiveRequest request = LiveRequest.builder().content(CONTENT).build(); + + assertThat(request.partial()).hasValue(false); + } + + @Test + public void liveRequest_jsonRoundTrip_preservesPartialTrue() { + LiveRequest request = LiveRequest.builder().content(CONTENT).partial(true).build(); + + JsonNode json = JsonBaseModel.getMapper().valueToTree(request); + LiveRequest restored = LiveRequest.fromJsonString(request.toJson()); + + assertThat(json.get("partial").booleanValue()).isTrue(); + assertThat(json.has("turnComplete")).isFalse(); + assertThat(restored.content()).hasValue(CONTENT); + assertThat(restored.partial()).hasValue(true); + } + + @Test + public void liveRequest_jsonRoundTrip_preservesPartialFalse() { + LiveRequest request = LiveRequest.builder().content(CONTENT).partial(false).build(); + + JsonNode json = JsonBaseModel.getMapper().valueToTree(request); + LiveRequest restored = LiveRequest.fromJsonString(request.toJson()); + + assertThat(json.get("partial").booleanValue()).isFalse(); + assertThat(json.has("turnComplete")).isFalse(); + assertThat(restored.content()).hasValue(CONTENT); + assertThat(restored.partial()).hasValue(false); + } + + @Test + public void liveRequest_jsonWithoutPartial_defaultsToNonPartial() { + LiveRequest request = + LiveRequest.fromJsonString( + "{\"content\":{\"role\":\"user\",\"parts\":[{\"text\":\"context\"}]}}"); + + assertThat(request.content()).hasValue(CONTENT); + assertThat(request.partial()).hasValue(false); + } + + @Test + public void liveRequest_toBuilder_preservesPartial() { + LiveRequest request = LiveRequest.builder().content(CONTENT).partial(true).build(); + + assertThat(request.toBuilder().build().partial()).hasValue(true); + } +} diff --git a/core/src/test/java/com/google/adk/flows/llmflows/BaseLlmFlowTest.java b/core/src/test/java/com/google/adk/flows/llmflows/BaseLlmFlowTest.java index b097dc7f3..1bc449351 100644 --- a/core/src/test/java/com/google/adk/flows/llmflows/BaseLlmFlowTest.java +++ b/core/src/test/java/com/google/adk/flows/llmflows/BaseLlmFlowTest.java @@ -26,15 +26,24 @@ import static com.google.common.collect.Iterables.getOnlyElement; import static com.google.common.truth.Truth.assertThat; import static org.junit.Assert.assertThrows; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.ArgumentMatchers.anyBoolean; +import static org.mockito.Mockito.inOrder; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.when; import com.google.adk.agents.Callbacks; import com.google.adk.agents.InvocationContext; +import com.google.adk.agents.LiveRequest; +import com.google.adk.agents.LiveRequestQueue; import com.google.adk.agents.LlmAgent; import com.google.adk.agents.ReadonlyContext; import com.google.adk.agents.RunConfig; import com.google.adk.events.Event; import com.google.adk.flows.llmflows.RequestProcessor.RequestProcessingResult; import com.google.adk.flows.llmflows.ResponseProcessor.ResponseProcessingResult; +import com.google.adk.models.BaseLlm; +import com.google.adk.models.BaseLlmConnection; import com.google.adk.models.CacheMetadata; import com.google.adk.models.LlmCallsLimitExceededException; import com.google.adk.models.LlmRequest; @@ -76,11 +85,54 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; +import org.mockito.InOrder; /** Unit tests for {@link BaseLlmFlow}. */ @RunWith(JUnit4.class) public final class BaseLlmFlowTest { + @Test + public void runLive_contentForwardsPartialFlagsInOrder() { + Content firstContent = Content.fromParts(Part.fromText("first context")); + Content secondContent = Content.fromParts(Part.fromText("second context")); + Content finalContent = Content.fromParts(Part.fromText("final context")); + Content defaultContent = Content.fromParts(Part.fromText("complete input")); + Content unsetContent = Content.fromParts(Part.fromText("unset partial")); + BaseLlm llm = mock(BaseLlm.class); + BaseLlmConnection connection = mock(BaseLlmConnection.class); + when(llm.model()).thenReturn("test-model"); + when(llm.connect(any())).thenReturn(connection); + when(connection.sendContent(any(), anyBoolean())).thenReturn(Completable.complete()); + when(connection.receive()).thenReturn(Flowable.never()); + LiveRequestQueue queue = new LiveRequestQueue(); + InvocationContext context = + createInvocationContext(createTestAgentBuilder(llm).executor(Runnable::run).build()) + .toBuilder() + .liveRequestQueue(queue) + .build(); + TestSubscriber subscriber = createBaseLlmFlowWithoutProcessors().runLive(context).test(); + + try { + subscriber.assertNoErrors(); + queue.content(firstContent, true); + queue.content(secondContent, true); + queue.content(finalContent, false); + queue.content(defaultContent); + queue.send(LiveRequest.builder().content(unsetContent).partial(null).build()); + + InOrder order = inOrder(connection); + order.verify(connection).sendContent(firstContent, true); + order.verify(connection).sendContent(secondContent, true); + order.verify(connection).sendContent(finalContent, false); + order.verify(connection).sendContent(defaultContent, false); + order.verify(connection).sendContent(unsetContent, false); + subscriber.assertNoErrors(); + } finally { + queue.close(); + subscriber.cancel(); + } + } + @Test public void run_singleTextResponse_returnsSingleEvent() { Content content = Content.fromParts(Part.fromText("LLM response")); diff --git a/core/src/test/java/com/google/adk/models/BaseLlmConnectionTest.java b/core/src/test/java/com/google/adk/models/BaseLlmConnectionTest.java new file mode 100644 index 000000000..0a249ab98 --- /dev/null +++ b/core/src/test/java/com/google/adk/models/BaseLlmConnectionTest.java @@ -0,0 +1,76 @@ +/* + * 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.models; + +import static com.google.common.truth.Truth.assertThat; + +import com.google.genai.types.Blob; +import com.google.genai.types.Content; +import com.google.genai.types.Part; +import io.reactivex.rxjava3.core.Completable; +import io.reactivex.rxjava3.core.Flowable; +import java.util.ArrayList; +import java.util.List; +import org.junit.Test; +import org.junit.runner.RunWith; +import org.junit.runners.JUnit4; + +@RunWith(JUnit4.class) +public final class BaseLlmConnectionTest { + + @Test + public void sendContent_withPartialFlag_delegatesToLegacyImplementation() { + LegacyConnection connection = new LegacyConnection(); + Content content = Content.fromParts(Part.fromText("context")); + + connection.sendContent(content, true).test().assertComplete().assertNoErrors(); + connection.sendContent(content, false).test().assertComplete().assertNoErrors(); + + assertThat(connection.sentContent).containsExactly(content, content).inOrder(); + } + + /** Implements only the original connection API, not the new two-argument overload. */ + private static final class LegacyConnection implements BaseLlmConnection { + final List sentContent = new ArrayList<>(); + + @Override + public Completable sendHistory(List history) { + return Completable.complete(); + } + + @Override + public Completable sendContent(Content content) { + return Completable.fromAction(() -> sentContent.add(content)); + } + + @Override + public Completable sendRealtime(Blob blob) { + return Completable.complete(); + } + + @Override + public Flowable receive() { + return Flowable.empty(); + } + + @Override + public void close() {} + + @Override + public void close(Throwable throwable) {} + } +} diff --git a/core/src/test/java/com/google/adk/models/GeminiLlmConnectionTest.java b/core/src/test/java/com/google/adk/models/GeminiLlmConnectionTest.java index d212d536f..d256bfdf9 100644 --- a/core/src/test/java/com/google/adk/models/GeminiLlmConnectionTest.java +++ b/core/src/test/java/com/google/adk/models/GeminiLlmConnectionTest.java @@ -17,6 +17,12 @@ package com.google.adk.models; import static com.google.common.truth.Truth.assertThat; +import static org.mockito.ArgumentMatchers.any; +import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.never; +import static org.mockito.Mockito.times; +import static org.mockito.Mockito.verify; +import static org.mockito.Mockito.when; import com.google.common.collect.ImmutableList; import com.google.common.collect.ImmutableMap; @@ -43,10 +49,145 @@ import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.JUnit4; +import org.mockito.ArgumentCaptor; @RunWith(JUnit4.class) public final class GeminiLlmConnectionTest { + @Test + public void sendContent_withoutPartial_completesTurn() { + Content content = Content.fromParts(Part.fromText("complete input")); + try (LiveConnectionFixture fixture = new LiveConnectionFixture()) { + fixture + .connection + .sendContent(content) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + + ArgumentCaptor parameters = + ArgumentCaptor.forClass(LiveSendClientContentParameters.class); + verify(fixture.transport).sendClientContent(parameters.capture()); + assertThat(parameters.getValue().turns()).hasValue(ImmutableList.of(content)); + assertThat(parameters.getValue().turnComplete()).hasValue(true); + } + } + + @Test + public void sendContent_withPartialTrue_leavesTurnOpen() { + Content content = Content.fromParts(Part.fromText("partial input")); + try (LiveConnectionFixture fixture = new LiveConnectionFixture()) { + fixture + .connection + .sendContent(content, true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + + ArgumentCaptor parameters = + ArgumentCaptor.forClass(LiveSendClientContentParameters.class); + verify(fixture.transport).sendClientContent(parameters.capture()); + assertThat(parameters.getValue().turns()).hasValue(ImmutableList.of(content)); + assertThat(parameters.getValue().turnComplete()).hasValue(false); + } + } + + @Test + public void sendContent_withPartialFalse_completesTurn() { + Content content = Content.fromParts(Part.fromText("final input")); + try (LiveConnectionFixture fixture = new LiveConnectionFixture()) { + fixture + .connection + .sendContent(content, false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + + ArgumentCaptor parameters = + ArgumentCaptor.forClass(LiveSendClientContentParameters.class); + verify(fixture.transport).sendClientContent(parameters.capture()); + assertThat(parameters.getValue().turns()).hasValue(ImmutableList.of(content)); + assertThat(parameters.getValue().turnComplete()).hasValue(true); + } + } + + @Test + public void sendContent_multiplePartialUpdates_onlyFinalContentCompletesTurn() { + Content first = Content.fromParts(Part.fromText("first context")); + Content second = Content.fromParts(Part.fromText("second context")); + Content last = Content.fromParts(Part.fromText("final context")); + try (LiveConnectionFixture fixture = new LiveConnectionFixture()) { + fixture + .connection + .sendContent(first, true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + fixture + .connection + .sendContent(second, true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + fixture + .connection + .sendContent(last, false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + + ArgumentCaptor parameters = + ArgumentCaptor.forClass(LiveSendClientContentParameters.class); + verify(fixture.transport, times(3)).sendClientContent(parameters.capture()); + List sent = parameters.getAllValues(); + assertThat(sent.get(0).turns()).hasValue(ImmutableList.of(first)); + assertThat(sent.get(1).turns()).hasValue(ImmutableList.of(second)); + assertThat(sent.get(2).turns()).hasValue(ImmutableList.of(last)); + assertThat(sent.stream().map(p -> p.turnComplete().orElseThrow()).toList()) + .containsExactly(false, false, true) + .inOrder(); + } + } + + @Test + public void sendContent_functionResponses_useToolResponseRegardlessOfPartial() { + Content content = + Content.fromParts( + Part.fromFunctionResponse("firstTool", ImmutableMap.of("result", "first")), + Part.fromFunctionResponse("secondTool", ImmutableMap.of("result", "second"))); + try (LiveConnectionFixture fixture = new LiveConnectionFixture()) { + fixture + .connection + .sendContent(content, true) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + fixture + .connection + .sendContent(content, false) + .test() + .awaitDone(5, TimeUnit.SECONDS) + .assertComplete() + .assertNoErrors(); + + ArgumentCaptor parameters = + ArgumentCaptor.forClass(LiveSendToolResponseParameters.class); + verify(fixture.transport, times(2)).sendToolResponse(parameters.capture()); + verify(fixture.transport, never()).sendClientContent(any()); + for (LiveSendToolResponseParameters sent : parameters.getAllValues()) { + assertThat(sent.functionResponses()) + .hasValue(content.parts().get().stream().map(p -> p.functionResponse().get()).toList()); + } + } + } + @Test public void convertToServerResponse_withInterruptedTrue_mapsInterruptedField() { LiveServerContent serverContent = @@ -408,4 +549,23 @@ public CompletableFuture close() { return CompletableFuture.completedFuture(null); } } + + /** Exercises the real connection's send path without opening a WebSocket or using credentials. */ + private static final class LiveConnectionFixture implements AutoCloseable { + final GeminiLiveTransport transport = mock(GeminiLiveTransport.class); + final GeminiLlmConnection connection; + + LiveConnectionFixture() { + when(transport.receive(any(), any())).thenReturn(CompletableFuture.completedFuture(null)); + when(transport.close()).thenReturn(CompletableFuture.completedFuture(null)); + when(transport.sendClientContent(any())).thenReturn(CompletableFuture.completedFuture(null)); + when(transport.sendToolResponse(any())).thenReturn(CompletableFuture.completedFuture(null)); + connection = new GeminiLlmConnection(CompletableFuture.completedFuture(transport)); + } + + @Override + public void close() { + connection.close(); + } + } }