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

Please ensure you have read the contribution guide before creating a pull request.

Link to Issue or Description of Change

1. Link to an existing issue (if applicable):

• Closes: runLive passes partial events to appendEvent and never aggregates text chunks, unlike ADK Python #1648

2. Or, if no issue exists, describe the change:

Problem:

In live mode, what a turn leaves in the session depends on the session service. GeminiLlmConnection turns each server message into responses without aggregating text, and marks a response partial only when its message carries turnComplete: false. Runner.runLive passes every event to appendEvent, while runAsync skips partial events. So a turn's text is stored as one model event per chunk, or, when the chunks are partial, dropped by InMemorySessionService and stored chunk by chunk by FirestoreSessionService and VertexAiSessionService. ADK Python stores it once. #1648 has the details and a reproduction.

Solution:

Follow ADK Python's live connection (gemini_llm_connection.py) and runner (runners.py):

• In GeminiLlmConnection, only the responses that stream text are partial. The connection also collects that text in a PendingText and emits it as one non-partial response before the response that completes the turn or reports an interruption (:550-559, :600-610), before a tool call (:619-627), and when thought and non-thought text alternate or a part that is neither text nor inline data arrives. Text that arrives in the same message as the end of the turn or an interruption is emitted only in that full response, as ADK Python does.
• The full response carries only the text. turnComplete, interrupted and the transcriptions stay on the message's own response, where callers read them today. ADK Python marks the full-text response interrupted when an interruption ends the text, and then sends no separate interruption response unless the turn also completes. When a message streams text and also carries a transcription, the transcription gets a non-partial response of its own, so that it is still stored.
• Runner.runLive skips partial events when persisting, as runAsync does.

runLive now hands only the non-partial events to the session service. Since every text chunk is partial, a turn that completes, is interrupted or calls a tool stores its text once, whether or not the Live API sends turnComplete: false on the intermediate messages, the question #1648 left open.

Responses that do not stream text, such as audio, transcriptions and interruptions, are no longer partial; on main they were partial whenever their message carried turnComplete: false. If the Live API sends that flag, InMemorySessionService now stores them, as the other session services already did; otherwise nothing changes for them. ADK Python does not mark audio or interruptions partial either. Two differences from ADK Python remain:

• Live model audio is stored like any other event, as on main when the messages do not carry turnComplete: false. When they do, InMemorySessionService now stores it too, and the stored audio is sent back with the history of the next live connection. ADK Python keeps live model media with inline data out of the session in its runner (_should_append_event) and out of the history it sends (send_history). runLive passes partial events to appendEvent and never aggregates text chunks, unlike ADK Python #1648 lists this as a separate difference, and this PR adds neither check.
• Transcriptions are not aggregated yet, so each chunk is stored as its own event. ADK Python streams the chunks as partial responses and stores one finished transcription. I can send that separately.

Testing Plan

Unit Tests:

• I have added or updated unit tests for my change.
• All unit tests pass locally.

./mvnw -pl core clean test (Windows 11, Microsoft Build of OpenJDK 17.0.19): the default-test and basic executions each run 2,214 tests with no failures (24 skipped), and the other four executions pass (apigee-llm-proxy-url runs no tests, see #1557).

Four existing GeminiLlmConnectionTest tests now expect the full-text response before the message's own response: withInterruptedTrue_mapsInterruptedField, withoutInterruptedField_mapsEmptyOptional, withTurnCompleteTrue_mapsPartialFalse and withContentAndUsageMetadata_emitsMultiple. withInterruptedTrue_mapsInterruptedField also checks that the interruption response, whose message sets turnComplete: false, is not partial. The other convertToServerResponse tests only pass a new PendingText.

Manual End-to-End (E2E) Tests:

Not run against the Gemini Live API. The connection tests drive the real GeminiLlmConnection and its conversion, and the Runner test uses TestLlm's live connection.

Checklist

• I have read the CONTRIBUTING.md document.
• My pull request contains a single commit.
• I have performed a self-review of my own code.
• I have commented my code, particularly in hard-to-understand areas.
• I have added tests that prove my fix is effective or that my feature works.
• New and existing unit tests pass locally with my changes.
• I have manually tested my changes end-to-end.
• Any dependent changes have been merged and published in downstream modules.

Additional context

• Callers of runLive now receive a turn's text twice, as partial chunks and then as one non-partial event, as ADK Python's callers do. A caller that shows the text of every model event needs to skip the partial events or the full one.
• If the connection closes or fails before the turn ends, the pending text is not emitted, as in ADK Python. On main, its chunks were stored when they were not partial.
• FirestoreSessionService.appendEvent still has no partial check of its own, the smaller alignment mentioned in runLive passes partial events to appendEvent and never aggregates text chunks, unlike ADK Python #1648. With this change, neither runAsync nor runLive passes it partial events.