diff --git a/core/src/agents/invocation_context.ts b/core/src/agents/invocation_context.ts index 7d236140..0511742d 100644 --- a/core/src/agents/invocation_context.ts +++ b/core/src/agents/invocation_context.ts @@ -16,6 +16,7 @@ import {randomUUID} from '../utils/env_aware_utils.js'; import {ActiveStreamingTool} from './active_streaming_tool.js'; import {BaseAgent} from './base_agent.js'; +import {LiveRequestQueue} from './live_request_queue.js'; import {RunConfig} from './run_config.js'; import {TranscriptionEntry} from './transcription_entry.js'; @@ -38,6 +39,8 @@ export interface InvocationContextParams { activeStreamingTools?: Record; pluginManager: PluginManager; abortSignal?: AbortSignal; + liveRequestQueue?: LiveRequestQueue; + liveSessionResumptionHandle?: string; } /** @@ -181,6 +184,19 @@ export class InvocationContext { readonly abortSignal?: AbortSignal; + /** + * The live request queue feeding the model on the bidirectional (live) path. + * Set only for invocations started via `runner.runLive`. + */ + readonly liveRequestQueue?: LiveRequestQueue; + + /** + * The most recent session resumption handle observed on the live path. + * Updated as the server emits resumption updates so a reconnect can restore + * server-side state instead of replaying history. Mutable by design. + */ + liveSessionResumptionHandle?: string; + /** * @param params The parameters for creating an invocation context. */ @@ -199,6 +215,8 @@ export class InvocationContext { this.activeStreamingTools = params.activeStreamingTools; this.pluginManager = params.pluginManager; this.abortSignal = params.abortSignal; + this.liveRequestQueue = params.liveRequestQueue; + this.liveSessionResumptionHandle = params.liveSessionResumptionHandle; } /** diff --git a/core/src/agents/live_request_queue.ts b/core/src/agents/live_request_queue.ts index 5e70a69f..5966ff3e 100644 --- a/core/src/agents/live_request_queue.ts +++ b/core/src/agents/live_request_queue.ts @@ -55,17 +55,38 @@ export class LiveRequestQueue { /** * Retrieves a request from the queue. If the queue is empty, it will * wait until a request is available. + * + * @param abortSignal When provided, a parked (pending) read is released by + * rejecting with an abort error once the signal fires. The live send + * loop passes the connection's abort signal so a waiter is not stranded + * across reconnect/teardown, where it could otherwise steal a request + * from the next connection's send loop. * @returns A promise that resolves with the next available request. */ - async get(): Promise { + async get(abortSignal?: AbortSignal): Promise { if (this.queue.length > 0) { return this.queue.shift()!; } if (this.isClosed) { return {close: true}; } - return new Promise((resolve) => { - this.resolveFnFifoQueue.push(resolve); + if (abortSignal?.aborted) { + throw new Error('LiveRequestQueue.get() was aborted.'); + } + return new Promise((resolve, reject) => { + const onAbort = () => { + const index = this.resolveFnFifoQueue.indexOf(resolveFn); + if (index !== -1) { + this.resolveFnFifoQueue.splice(index, 1); + } + reject(new Error('LiveRequestQueue.get() was aborted.')); + }; + const resolveFn: PromiseResolveFn = (req: LiveRequest) => { + abortSignal?.removeEventListener('abort', onAbort); + resolve(req); + }; + this.resolveFnFifoQueue.push(resolveFn); + abortSignal?.addEventListener('abort', onAbort, {once: true}); }); } diff --git a/core/src/agents/llm_agent.ts b/core/src/agents/llm_agent.ts index db48f537..624c6c7e 100644 --- a/core/src/agents/llm_agent.ts +++ b/core/src/agents/llm_agent.ts @@ -24,6 +24,7 @@ import { import {BaseExampleProvider} from '../examples/base_example_provider.js'; import {Example} from '../examples/example.js'; import {BaseLlm, isBaseLlm} from '../models/base_llm.js'; +import {BaseLlmConnection} from '../models/base_llm_connection.js'; import {LlmRequest} from '../models/llm_request.js'; import {LlmResponse} from '../models/llm_response.js'; import {LLMRegistry} from '../models/registry.js'; @@ -56,6 +57,7 @@ import { import {BaseContextCompactor} from '../context/base_context_compactor.js'; import {InvocationContext} from './invocation_context.js'; +import {LiveRequest, LiveRequestQueue} from './live_request_queue.js'; import {AGENT_TRANSFER_LLM_REQUEST_PROCESSOR} from './processors/agent_transfer_llm_request_processor.js'; import {BASIC_LLM_REQUEST_PROCESSOR} from './processors/basic_llm_request_processor.js'; import {CODE_EXECUTION_REQUEST_PROCESSOR} from './processors/code_execution_request_processor.js'; @@ -68,6 +70,111 @@ import {TOOL_FILTER_REQUEST_PROCESSOR} from './processors/tool_filter_request_pr import {ReadonlyContext} from './readonly_context.js'; import {StreamingMode} from './run_config.js'; +/** + * Maximum number of reconnect attempts on transient live connection failure + * when a session resumption handle is available. + */ +const MAX_LIVE_RECONNECT_ATTEMPTS = 5; + +/** + * Delay before closing the parent connection on agent transfer. Gives the + * server-side model a moment to flush any pending audio for the final turn + * before teardown. Mirrors `DEFAULT_TRANSFER_AGENT_DELAY` (1.0s) in the Python + * ADK live flow; the value is an empirical heuristic, not a guarantee. + */ +const TRANSFER_AGENT_DELAY_MS = 1000; + +/** + * Sentinel thrown from `runReceiveLoop` to break out of the receive iterator + * and signal `runLiveFlow` to reconnect using the stored resumption handle. + * Used when the server sends `goAway` or any other recoverable terminal. + */ +class LiveReconnectSignal extends Error { + constructor(readonly reason: string) { + super(`live reconnect requested: ${reason}`); + this.name = 'LiveReconnectSignal'; + } +} + +/** + * Classifies errors that should trigger a reconnect attempt instead of + * propagating. Matches the Python flow's allowlist of recoverable codes. + */ +function isRecoverableLiveError(err: unknown): boolean { + if (err instanceof LiveReconnectSignal) return true; + if (!(err instanceof Error)) return false; + const code = (err as {code?: unknown}).code; + // Standard WebSocket close codes treated as transient by the Python flow. + if (code === 1000 || code === 1006 || code === 1011 || code === 1012) { + return true; + } + const message = err.message ?? ''; + return /ConnectionClosed|connection closed|ECONNRESET|socket hang up/i.test( + message, + ); +} + +async function closeQuietly(connection: BaseLlmConnection): Promise { + try { + await connection.close(); + } catch (error) { + logger.warn('Error closing live connection:', error); + } +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * Copies the live-relevant fields from the run config onto the live connect + * config so the model connection is opened with the caller's modalities, + * speech, transcription, and proactivity settings. + */ +function applyLiveRunConfig( + runConfig: InvocationContext['runConfig'], + llmRequest: LlmRequest, +): void { + if (!runConfig) return; + const liveConfig = (llmRequest.liveConnectConfig ??= {}); + if (runConfig.responseModalities) { + liveConfig.responseModalities = runConfig.responseModalities; + } + if (runConfig.speechConfig) { + liveConfig.speechConfig = runConfig.speechConfig; + } + if (runConfig.outputAudioTranscription) { + liveConfig.outputAudioTranscription = runConfig.outputAudioTranscription; + } + if (runConfig.inputAudioTranscription) { + liveConfig.inputAudioTranscription = runConfig.inputAudioTranscription; + } + if (runConfig.realtimeInputConfig) { + liveConfig.realtimeInputConfig = runConfig.realtimeInputConfig; + } + if (runConfig.contextWindowCompression) { + liveConfig.contextWindowCompression = runConfig.contextWindowCompression; + } + if (runConfig.proactivity) { + liveConfig.proactivity = runConfig.proactivity; + } + if (runConfig.enableAffectiveDialog) { + liveConfig.enableAffectiveDialog = runConfig.enableAffectiveDialog; + } +} + +/** + * Whether a live response should be attributed to the user side of the + * transcript. Input transcriptions are the user speaking; echoed user-role + * content (e.g. function responses) likewise belongs to the user side. + */ +function isUserAuthoredResponse(llmResponse: LlmResponse): boolean { + if (llmResponse.inputTranscription) { + return true; + } + return llmResponse.content?.role === 'user'; +} + /** * Input/output schema type for agent. */ @@ -741,13 +848,453 @@ export class LlmAgent extends BaseAgent { * times. Subsequent reconnects skip `sendHistory` because the server * already holds the conversation state associated with the handle. */ - // eslint-disable-next-line require-yield private async *runLiveFlow( - _invocationContext: InvocationContext, + invocationContext: InvocationContext, + ): AsyncGenerator { + if (!invocationContext.liveRequestQueue) { + throw new Error('liveRequestQueue is required for LlmAgent.runLiveFlow.'); + } + + const llmRequest: LlmRequest = { + contents: [], + toolsDict: {}, + liveConnectConfig: {}, + }; + + // ========================================================================= + // Preprocess: same processors as runAsync. Yields agent-emitted events + // (e.g. instruction injection metadata events) to the caller. + // ========================================================================= + for await (const event of this.runLivePreprocess( + invocationContext, + llmRequest, + )) { + yield event; + } + + if ( + invocationContext.endInvocation || + invocationContext.abortSignal?.aborted + ) { + return; + } + + // ========================================================================= + // Apply live-only request config from the run config. + // ========================================================================= + applyLiveRunConfig(invocationContext.runConfig, llmRequest); + + const llm = this.canonicalModel; + let attempt = 1; + + // Outer reconnect loop. Re-enters on recoverable failures when a session + // resumption handle is available; the server restores state on the new + // connection so we skip history replay. + while (true) { + if (invocationContext.abortSignal?.aborted) { + return; + } + + // Apply the latest resumption handle before each connect attempt. + const handle = invocationContext.liveSessionResumptionHandle; + if (handle) { + llmRequest.liveConnectConfig ??= {}; + llmRequest.liveConnectConfig.sessionResumption = { + handle, + transparent: true, + }; + } + + let connection: BaseLlmConnection; + try { + connection = await llm.connect(llmRequest); + } catch (err) { + if ( + isRecoverableLiveError(err) && + invocationContext.liveSessionResumptionHandle + ) { + if (attempt > MAX_LIVE_RECONNECT_ATTEMPTS) { + logger.error( + `Max live reconnection attempts reached (${attempt}).`, + err, + ); + throw err; + } + logger.info( + `Live connect failed (attempt ${attempt}); retrying with session handle.`, + err, + ); + attempt += 1; + continue; + } + throw err; + } + + // Skip history replay when resuming -- server already has the state. + if ( + llmRequest.contents.length > 0 && + !invocationContext.liveSessionResumptionHandle + ) { + await connection.sendHistory(llmRequest.contents); + } + + const sendAbort = new AbortController(); + // Stop the send loop when either the invocation aborts or this + // attempt's connection is torn down. AbortSignal.any cleans up its + // listeners on the input signals automatically. + const combinedAbort = invocationContext.abortSignal + ? AbortSignal.any([invocationContext.abortSignal, sendAbort.signal]) + : sendAbort.signal; + const sendTask = this.runSendLoop( + connection, + invocationContext.liveRequestQueue, + combinedAbort, + ); + sendTask.catch((error) => { + logger.error('Error in live send loop:', error); + }); + + let reconnect = false; + try { + yield* this.runReceiveLoop( + invocationContext, + connection, + llmRequest, + sendAbort, + ); + } catch (err) { + const canReconnect = + !!invocationContext.liveSessionResumptionHandle && + (err instanceof LiveReconnectSignal || isRecoverableLiveError(err)); + if (canReconnect) { + reconnect = true; + logger.info( + 'Live connection closed; will reconnect with session handle.', + err, + ); + } else { + // Tear down before rethrowing. + await this.teardownLiveConnection(sendAbort, connection, sendTask); + throw err; + } + } + + // Cancel send loop for this attempt; receive loop has exited. + await this.teardownLiveConnection(sendAbort, connection, sendTask); + + if (!reconnect) { + return; + } + + if (attempt > MAX_LIVE_RECONNECT_ATTEMPTS) { + throw new Error(`Max live reconnection attempts reached (${attempt}).`); + } + attempt += 1; + } + } + + private async *runLivePreprocess( + invocationContext: InvocationContext, + llmRequest: LlmRequest, ): AsyncGenerator { - // TODO - b/425992518: remove dummy logic, implement this. - await Promise.resolve(); - throw new Error('LlmAgent.runLiveFlow not implemented'); + for (const processor of this.requestProcessors) { + for await (const event of processor.runAsync( + invocationContext, + llmRequest, + )) { + if (invocationContext.abortSignal?.aborted) { + return; + } + yield event; + } + } + for (const toolUnion of this.tools) { + const toolContext = new Context({invocationContext}); + const tools = ( + await convertToolUnionToTools( + toolUnion, + new ReadonlyContext(invocationContext), + ) + ).filter( + (tool) => + !llmRequest.allowedTools || + llmRequest.allowedTools.includes(tool.name), + ); + for (const tool of tools) { + await tool.processLlmRequest({toolContext, llmRequest}); + if (invocationContext.abortSignal?.aborted) { + return; + } + } + } + } + + private async runSendLoop( + connection: BaseLlmConnection, + liveRequestQueue: LiveRequestQueue, + abortSignal?: AbortSignal, + ): Promise { + while (true) { + if (abortSignal?.aborted) { + return; + } + let liveRequest: LiveRequest; + try { + // Pass the abort signal so a parked read is released on teardown + // (reconnect, agent transfer) instead of stranding a waiter that + // would later steal a request from the next connection's send loop. + liveRequest = await liveRequestQueue.get(abortSignal); + } catch (error) { + if (abortSignal?.aborted) { + return; + } + throw error; + } + try { + await this.dispatchLiveRequest(connection, liveRequest); + } catch (error) { + logger.error('Error dispatching live request to model:', error); + throw error; + } + // Cooperative yield: avoid starving the event loop when the queue is + // backlogged so receive-loop events get a chance to interleave. + await Promise.resolve(); + if (liveRequest.close) { + return; + } + } + } + + private async dispatchLiveRequest( + connection: BaseLlmConnection, + liveRequest: LiveRequest, + ): Promise { + if (liveRequest.close) { + await connection.close(); + return; + } + if (liveRequest.activityStart) { + await connection.sendActivityStart?.(); + return; + } + if (liveRequest.activityEnd) { + await connection.sendActivityEnd?.(); + return; + } + if (liveRequest.blob) { + await connection.sendRealtime(liveRequest.blob); + return; + } + if (liveRequest.content) { + await connection.sendContent(liveRequest.content); + } + } + + /** + * Tears down a live attempt: stops the send loop, closes the connection + * (swallowing close errors), and waits for the send task to settle. Used + * before reconnecting or propagating an error. + */ + private async teardownLiveConnection( + sendAbort: AbortController, + connection: BaseLlmConnection, + sendTask: Promise, + ): Promise { + sendAbort.abort(); + await closeQuietly(connection); + await sendTask.catch(() => undefined); + } + + private async *runReceiveLoop( + invocationContext: InvocationContext, + connection: BaseLlmConnection, + llmRequest: LlmRequest, + sendAbort: AbortController, + ): AsyncGenerator { + for await (const llmResponse of connection.receive()) { + if (invocationContext.abortSignal?.aborted) { + return; + } + + // Capture the latest server-provided resumption handle on the + // invocation context so that any subsequent reconnect attempt can + // resume server-side state instead of replaying history. + if (llmResponse.liveSessionResumptionUpdate?.newHandle) { + invocationContext.liveSessionResumptionHandle = + llmResponse.liveSessionResumptionUpdate.newHandle; + } + + // GoAway is the server's "I'm about to close; reconnect with your + // resumption handle" signal. Throw a sentinel to break the outer + // reconnect loop in runLiveFlow. + if (llmResponse.goAway) { + logger.info('Received goAway from live server; triggering reconnect.'); + throw new LiveReconnectSignal('goAway'); + } + + const author = isUserAuthoredResponse(llmResponse) ? 'user' : this.name; + + const modelResponseEvent = createEvent({ + invocationId: invocationContext.invocationId, + author, + branch: invocationContext.branch, + }); + + for await (const event of this.postprocessLive( + invocationContext, + llmRequest, + llmResponse, + modelResponseEvent, + )) { + yield event; + + // Send function responses directly through the connection rather + // than via the live request queue. The TS LiveRequestQueue rejects + // sends after close (strict semantics), and callers commonly close + // the queue at end-of-input before the model finishes ferrying tool + // results back. Python's queue tolerates post-close sends, but + // porting that semantics is out of scope here. + if (event.content && getFunctionResponses(event).length > 0) { + await connection.sendContent(event.content); + } + + // Handle agent transfer triggered by a transfer_to_agent function + // response. The active connection is closed and the destination + // sub-agent's runLive is yielded into the same generator. + const transferTo = event.actions?.transferToAgent; + if (transferTo) { + // Brief delay lets the model finish flushing pending audio for + // the in-flight turn before we tear down the connection. + await sleep(TRANSFER_AGENT_DELAY_MS); + // Stop the parent send loop before the sub-agent starts its own, + // so the two never consume the shared liveRequestQueue + // concurrently (mirrors `send_task.cancel()` in the Python flow). + sendAbort.abort(); + await connection.close(); + const subAgent = + invocationContext.agent.rootAgent.findAgent(transferTo); + if (subAgent) { + const previousAgent = invocationContext.agent; + invocationContext.agent = subAgent; + // Child agent starts its own live session; do not carry over + // the parent's resumption handle. + const previousHandle = + invocationContext.liveSessionResumptionHandle; + invocationContext.liveSessionResumptionHandle = undefined; + try { + for await (const subEvent of subAgent.runLive( + invocationContext, + )) { + yield subEvent; + } + } finally { + invocationContext.agent = previousAgent; + invocationContext.liveSessionResumptionHandle = previousHandle; + } + } + return; + } + } + } + } + + private async *postprocessLive( + invocationContext: InvocationContext, + llmRequest: LlmRequest, + llmResponse: LlmResponse, + modelResponseEvent: Event, + ): AsyncGenerator { + for (const processor of this.responseProcessors) { + for await (const event of processor.runAsync( + invocationContext, + llmResponse, + )) { + yield event; + } + } + + // Skip purely empty responses, but allow control signals to surface. + if ( + !llmResponse.content && + !llmResponse.errorCode && + !llmResponse.interrupted && + !llmResponse.turnComplete && + !llmResponse.inputTranscription && + !llmResponse.outputTranscription && + !llmResponse.usageMetadata && + !llmResponse.liveSessionResumptionUpdate + ) { + return; + } + + // The connection layer (GeminiLlmConnection.receive) emits resumption + // updates and transcriptions as standalone, single-field responses -- + // never combined with `content`. Each is therefore handled with an early + // return; if that invariant changes, co-located fields would be dropped + // here and these branches would need to merge instead. + if (llmResponse.liveSessionResumptionUpdate) { + yield createEvent({ + ...modelResponseEvent, + liveSessionResumptionUpdate: llmResponse.liveSessionResumptionUpdate, + }); + return; + } + + if (llmResponse.inputTranscription) { + yield createEvent({ + ...modelResponseEvent, + inputTranscription: llmResponse.inputTranscription, + partial: llmResponse.partial, + }); + return; + } + if (llmResponse.outputTranscription) { + yield createEvent({ + ...modelResponseEvent, + outputTranscription: llmResponse.outputTranscription, + partial: llmResponse.partial, + }); + return; + } + + const mergedEvent = createEvent({ + ...modelResponseEvent, + ...llmResponse, + }); + + const functionCalls = getFunctionCalls(mergedEvent); + if (mergedEvent.content && functionCalls.length) { + populateClientFunctionCallId(mergedEvent); + mergedEvent.longRunningToolIds = Array.from( + getLongRunningFunctionCalls(functionCalls, llmRequest.toolsDict), + ); + } + + yield mergedEvent; + + // Execute any function calls returned in this event. + if (!functionCalls.length) { + return; + } + + const functionResponseEvent = await handleFunctionCallsAsync({ + invocationContext, + functionCallEvent: mergedEvent, + toolsDict: llmRequest.toolsDict, + beforeToolCallbacks: this.canonicalBeforeToolCallbacks, + afterToolCallbacks: this.canonicalAfterToolCallbacks, + }); + if (!functionResponseEvent) { + return; + } + const authEvent = generateAuthEvent( + invocationContext, + functionResponseEvent, + ); + if (authEvent) { + yield authEvent; + } + yield functionResponseEvent; } private async *runOneStepAsync( diff --git a/core/src/agents/run_config.ts b/core/src/agents/run_config.ts index 5312770a..190a6592 100644 --- a/core/src/agents/run_config.ts +++ b/core/src/agents/run_config.ts @@ -6,6 +6,7 @@ import { AudioTranscriptionConfig, + ContextWindowCompressionConfig, Modality, ProactivityConfig, RealtimeInputConfig, @@ -84,6 +85,12 @@ export interface RunConfig { */ realtimeInputConfig?: RealtimeInputConfig; + /** + * Context window compression config. When the running context exceeds + * `triggerTokens`, the server compresses older history to `targetTokens`. + */ + contextWindowCompression?: ContextWindowCompressionConfig; + /** * A limit on the total number of llm calls for a given run. * diff --git a/core/src/models/google_llm.ts b/core/src/models/google_llm.ts index 603834cf..4e14e4c7 100644 --- a/core/src/models/google_llm.ts +++ b/core/src/models/google_llm.ts @@ -218,6 +218,7 @@ export class Gemini extends BaseLlm { get liveApiVersion(): string { if (!this._liveApiVersion) { + // Vertex uses the beta API; the AI Studio backend uses v1alpha. this._liveApiVersion = this.apiBackend === GoogleLLMVariant.VERTEX_AI ? 'v1beta1' : 'v1alpha'; } @@ -283,6 +284,24 @@ export class Gemini extends BaseLlm { llmRequest.liveConnectConfig.tools = llmRequest.config?.tools; + // Gemini API (AI Studio) rejects `sessionResumption.transparent`; it is a + // Vertex-only flag. Strip it so callers can set a uniform resumption config + // regardless of backend. + if ( + this.apiBackend === GoogleLLMVariant.GEMINI_API && + llmRequest.liveConnectConfig.sessionResumption + ) { + const resumption = llmRequest.liveConnectConfig.sessionResumption as { + handle?: string; + transparent?: boolean; + }; + if (resumption.transparent !== undefined) { + llmRequest.liveConnectConfig.sessionResumption = { + handle: resumption.handle, + }; + } + } + const modelVersion = llmRequest.model ?? this.model; const messageQueue = new AsyncQueue(); diff --git a/core/src/runner/runner.ts b/core/src/runner/runner.ts index 7b9ef7b0..a7c6e4eb 100644 --- a/core/src/runner/runner.ts +++ b/core/src/runner/runner.ts @@ -4,7 +4,7 @@ * SPDX-License-Identifier: Apache-2.0 */ -import {Content, createPartFromText} from '@google/genai'; +import {Content, createPartFromText, Modality} from '@google/genai'; import {context, trace} from '@opentelemetry/api'; import {BaseAgent} from '../agents/base_agent.js'; @@ -12,6 +12,7 @@ import { InvocationContext, newInvocationContextId, } from '../agents/invocation_context.js'; +import {LiveRequestQueue} from '../agents/live_request_queue.js'; import {isLlmAgent} from '../agents/llm_agent.js'; import {createRunConfig, RunConfig} from '../agents/run_config.js'; import {BaseArtifactService} from '../artifacts/base_artifact_service.js'; @@ -527,7 +528,168 @@ export class Runner { } return true; } - // TODO - b/425992518: Implement runLive and related methods. + /** + * Runs the agent in the live (bidirectional streaming) mode. + * + * Model media events that carry raw inline bytes (audio, video, or image) + * are yielded but not appended to the session to avoid persisting large + * blobs; events with `fileData` references and most other live events + * (transcriptions, tool calls, usage) are persisted as in `runAsync`. + * + * This feature is **experimental** and its API may change. + * + * @param params.userId The user ID of the session. + * @param params.sessionId The session ID of the session. + * @param params.liveRequestQueue The queue used to feed the live model. + * @param params.runConfig The run config for the agent. + * @param params.abortSignal Optional signal to abort the live run. + * @param params.liveSessionResumptionHandle Optional session resumption + * handle observed from a prior `runLive` cycle on the same conversation. + * When set, the agent's live flow opens the connection with + * `liveConnectConfig.sessionResumption.handle` so the server restores its + * state instead of relying on client-side history replay. + * @yields The events generated by the agent. + */ + async *runLive(params: { + userId: string; + sessionId: string; + liveRequestQueue: LiveRequestQueue; + runConfig?: RunConfig; + abortSignal?: AbortSignal; + liveSessionResumptionHandle?: string; + }): AsyncGenerator { + if (!params.liveRequestQueue) { + throw new Error('liveRequestQueue is required for runLive.'); + } + + const runConfig = createRunConfig(params.runConfig); + if (!runConfig.responseModalities?.length) { + runConfig.responseModalities = [Modality.AUDIO]; + } + // For multi-agent live setups, the model's text transcription is needed + // as context for the transferred agent. + if (this.agent.subAgents?.length) { + if (runConfig.responseModalities.includes(Modality.AUDIO)) { + runConfig.outputAudioTranscription ??= {}; + } + runConfig.inputAudioTranscription ??= {}; + } + + const span = tracer.startSpan('invocation'); + const ctx = trace.setSpan(context.active(), span); + try { + yield* runAsyncGeneratorWithOtelContext( + ctx, + this, + async function* () { + const session = await this.sessionService.getOrCreateSession({ + appName: this.appName, + userId: params.userId, + sessionId: params.sessionId, + }); + + if (params.abortSignal?.aborted) { + return; + } + + const invocationContext = new InvocationContext({ + artifactService: this.artifactService, + sessionService: this.sessionService, + memoryService: this.memoryService, + credentialService: this.credentialService, + invocationId: newInvocationContextId(), + agent: this.agent, + session, + runConfig, + pluginManager: this.pluginManager, + liveRequestQueue: params.liveRequestQueue, + abortSignal: params.abortSignal, + liveSessionResumptionHandle: params.liveSessionResumptionHandle, + }); + + invocationContext.agent = this.determineAgentForResumption( + session, + this.agent, + ); + + // Step 1: before-run plugin hook (early exit if it returns content). + const beforeRunCallbackResponse = + await this.pluginManager.runBeforeRunCallback({ + invocationContext, + }); + if (params.abortSignal?.aborted) { + return; + } + if (beforeRunCallbackResponse) { + const earlyExitEvent = createEvent({ + invocationId: invocationContext.invocationId, + author: 'model', + content: beforeRunCallbackResponse, + }); + await this.sessionService.appendEvent({ + session, + event: earlyExitEvent, + }); + yield earlyExitEvent; + return; + } + + // Step 2: drive the agent's runLive and propagate events. + for await (const event of invocationContext.agent.runLive( + invocationContext, + )) { + if (params.abortSignal?.aborted) { + return; + } + + if (!event.partial && !isLiveModelMediaEventWithInlineData(event)) { + await this.sessionService.appendEvent({session, event}); + } + + const modifiedEvent = await this.pluginManager.runOnEventCallback({ + invocationContext, + event, + }); + if (params.abortSignal?.aborted) { + return; + } + + yield modifiedEvent ?? event; + } + + // Step 3: after-run plugin hook for cleanup/metrics. + await this.pluginManager.runAfterRunCallback({invocationContext}); + }, + ); + } finally { + span.end(); + } + } +} + +/** + * Whether a live event is a model media event carrying inline data (audio, + * video, or image). + * + * Such events are deliberately not persisted to the session to avoid storing + * large raw blobs. Media referenced via `fileData` (e.g. saved as artifacts) + * and all non-media events (transcriptions, tool calls, usage) are persisted + * as in `runAsync`. + */ +function isLiveModelMediaEventWithInlineData(event: Event): boolean { + const parts = event.content?.parts; + if (!parts?.length) { + return false; + } + return parts.some((part) => { + const mimeType = part.inlineData?.mimeType?.toLowerCase(); + return ( + mimeType !== undefined && + (mimeType.startsWith('audio/') || + mimeType.startsWith('video/') || + mimeType.startsWith('image/')) + ); + }); } /** diff --git a/core/test/models/google_llm_test.ts b/core/test/models/google_llm_test.ts index 1d745c40..4c5197f9 100644 --- a/core/test/models/google_llm_test.ts +++ b/core/test/models/google_llm_test.ts @@ -652,5 +652,51 @@ describe('GoogleLlm', () => { }), ); }); + + it('strips sessionResumption.transparent on the Gemini API backend', async () => { + const llm = new TestGemini({ + apiKey: 'test-key', + model: 'gemini-2.5-flash', + }); + const request: LlmRequest = { + model: 'gemini-2.5-flash', + contents: [], + liveConnectConfig: { + sessionResumption: {handle: 'h-1', transparent: true}, + }, + config: {}, + toolsDict: {}, + }; + + await llm.connect(request); + + expect(llm.liveApiClient.live.connect).toHaveBeenCalledWith( + expect.objectContaining({ + config: expect.objectContaining({ + sessionResumption: {handle: 'h-1'}, + }), + }), + ); + }); + }); + + describe('liveApiVersion', () => { + it('uses v1beta1 on the Vertex AI backend', () => { + const llm = new TestGemini({ + model: 'gemini-2.5-flash', + vertexai: true, + project: 'p', + location: 'us-central1', + }); + expect(llm.liveApiVersion).toBe('v1beta1'); + }); + + it('uses v1alpha on the Gemini API backend', () => { + const llm = new TestGemini({ + apiKey: 'test-key', + model: 'gemini-3.1-flash-live-preview', + }); + expect(llm.liveApiVersion).toBe('v1alpha'); + }); }); }); diff --git a/core/test/runner/run_live_test.ts b/core/test/runner/run_live_test.ts new file mode 100644 index 00000000..1dad9abe --- /dev/null +++ b/core/test/runner/run_live_test.ts @@ -0,0 +1,773 @@ +/** + * @license + * Copyright 2026 Google LLC + * SPDX-License-Identifier: Apache-2.0 + */ + +import { + BaseLlm, + BaseLlmConnection, + BaseTool, + Event, + InMemoryArtifactService, + InMemorySessionService, + LiveRequestQueue, + LlmAgent, + LlmRequest, + LlmResponse, + RunAsyncToolRequest, + Runner, +} from '@google/adk'; +import {Blob, Content, FunctionDeclaration, Modality} from '@google/genai'; +import {beforeEach, describe, expect, it} from 'vitest'; + +const TEST_APP_ID = 'test_app_id'; +const TEST_USER_ID = 'test_user_id'; +const TEST_SESSION_ID = 'test_session_id'; + +class RecordingConnection implements BaseLlmConnection { + readonly historyCalls: Content[][] = []; + readonly contentCalls: Content[] = []; + readonly realtimeCalls: Blob[] = []; + activityStartCalls = 0; + activityEndCalls = 0; + closed = false; + + constructor(private readonly responses: LlmResponse[]) {} + + async sendHistory(history: Content[]): Promise { + this.historyCalls.push(history); + } + async sendContent(content: Content): Promise { + this.contentCalls.push(content); + } + async sendRealtime(blob: Blob): Promise { + this.realtimeCalls.push(blob); + } + async sendActivityStart(): Promise { + this.activityStartCalls += 1; + } + async sendActivityEnd(): Promise { + this.activityEndCalls += 1; + } + async *receive(): AsyncGenerator { + for (const response of this.responses) { + yield response; + } + } + async close(): Promise { + this.closed = true; + } +} + +class FakeLiveLlm extends BaseLlm { + connection?: RecordingConnection; + llmRequestSeen?: LlmRequest; + readonly connections: RecordingConnection[] = []; + readonly llmRequestsSeen: LlmRequest[] = []; + + constructor( + private readonly responses: LlmResponse[] | LlmResponse[][], + model = 'fake-live-llm', + ) { + super({model}); + } + + // eslint-disable-next-line require-yield + override async *generateContentAsync(): AsyncGenerator< + LlmResponse, + void, + void + > { + throw new Error('generateContentAsync not used in live tests'); + } + + override async connect(llmRequest: LlmRequest): Promise { + // Snapshot the request as the caller may mutate `liveConnectConfig` + // across reconnect attempts (e.g. setting `sessionResumption.handle`). + this.llmRequestSeen = llmRequest; + this.llmRequestsSeen.push( + JSON.parse(JSON.stringify(llmRequest)) as LlmRequest, + ); + const isSequence = + Array.isArray(this.responses) && Array.isArray(this.responses[0]); + const responses = isSequence + ? ((this.responses as LlmResponse[][])[this.connections.length] ?? []) + : (this.responses as LlmResponse[]); + this.connection = new RecordingConnection(responses); + this.connections.push(this.connection); + return this.connection; + } +} + +class EchoTool extends BaseTool { + constructor() { + super({name: 'echo', description: 'Echoes back its input.'}); + } + override _getDeclaration(): FunctionDeclaration | undefined { + return {name: this.name, description: this.description}; + } + override async runAsync(request: RunAsyncToolRequest): Promise { + return {echoed: request.args}; + } +} + +describe('Runner.runLive', () => { + let sessionService: InMemorySessionService; + let artifactService: InMemoryArtifactService; + + beforeEach(async () => { + sessionService = new InMemorySessionService(); + artifactService = new InMemoryArtifactService(); + await sessionService.createSession({ + appName: TEST_APP_ID, + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + }); + }); + + it('throws when liveRequestQueue is missing', async () => { + const llm = new FakeLiveLlm([]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + await expect(async () => { + // @ts-expect-error - intentionally omit required argument + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + })) { + // no-op + } + }).rejects.toThrow('liveRequestQueue is required'); + }); + + it('creates the session when it does not exist', async () => { + const llm = new FakeLiveLlm([]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: 'missing', + liveRequestQueue: queue, + })) { + // no-op + } + + const session = await sessionService.getSession({ + appName: TEST_APP_ID, + userId: TEST_USER_ID, + sessionId: 'missing', + }); + expect(session).toBeDefined(); + }); + + it('forwards realtime blobs to the connection and yields model events', async () => { + const audioPart: Content = { + role: 'model', + parts: [{inlineData: {data: 'AAA=', mimeType: 'audio/pcm'}}], + }; + const textPart: Content = {role: 'model', parts: [{text: 'hello'}]}; + const llm = new FakeLiveLlm([ + {content: audioPart}, + {content: textPart}, + {turnComplete: true}, + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + const blob: Blob = {data: 'AAA=', mimeType: 'audio/pcm'}; + queue.sendRealtime(blob); + queue.close(); + + const events: Event[] = []; + for await (const event of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + events.push(event); + } + + expect(llm.connection).toBeDefined(); + expect(llm.connection!.realtimeCalls).toEqual([blob]); + expect(llm.connection!.closed).toBe(true); + + expect(events.some((e) => e.content === audioPart)).toBe(true); + expect(events.some((e) => e.content === textPart)).toBe(true); + expect(events.some((e) => e.turnComplete)).toBe(true); + }); + + it('defaults responseModalities to AUDIO and applies live config', async () => { + const llm = new FakeLiveLlm([{turnComplete: true}]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + + expect(llm.llmRequestSeen?.liveConnectConfig?.responseModalities).toEqual([ + Modality.AUDIO, + ]); + }); + + it('does not persist live audio events but persists transcription events', async () => { + const audioPart: Content = { + role: 'model', + parts: [{inlineData: {data: 'AAA=', mimeType: 'audio/pcm'}}], + }; + const llm = new FakeLiveLlm([ + {content: audioPart}, + { + outputTranscription: {text: 'hello world', finished: true}, + partial: false, + }, + {turnComplete: true}, + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + + const session = await sessionService.getSession({ + appName: TEST_APP_ID, + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + }); + const persisted = session!.events; + const hasAudioInline = persisted.some((event) => + event.content?.parts?.some((part) => + part.inlineData?.mimeType?.startsWith('audio/'), + ), + ); + expect(hasAudioInline).toBe(false); + const hasTranscription = persisted.some( + (event) => event.outputTranscription !== undefined, + ); + expect(hasTranscription).toBe(true); + }); + + it('skips inline video/image media and media in non-first parts', async () => { + const videoPart: Content = { + role: 'model', + parts: [{inlineData: {data: 'AAA=', mimeType: 'video/mp4'}}], + }; + const imagePart: Content = { + role: 'model', + parts: [{inlineData: {data: 'AAA=', mimeType: 'image/png'}}], + }; + const mixedPart: Content = { + role: 'model', + parts: [ + {text: 'ignored'}, + {inlineData: {data: 'AAA=', mimeType: 'audio/pcm'}}, + ], + }; + const textPart: Content = {role: 'model', parts: [{text: 'persist me'}]}; + const llm = new FakeLiveLlm([ + {content: videoPart}, + {content: imagePart}, + {content: mixedPart}, + {content: textPart, partial: false}, + {turnComplete: true}, + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + + const session = await sessionService.getSession({ + appName: TEST_APP_ID, + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + }); + const persisted = session!.events; + const hasInlineMedia = persisted.some((event) => + event.content?.parts?.some((part) => part.inlineData !== undefined), + ); + expect(hasInlineMedia).toBe(false); + const hasText = persisted.some((event) => + event.content?.parts?.some((part) => part.text === 'persist me'), + ); + expect(hasText).toBe(true); + }); + + it('runs tool calls and sends function responses back to the model', async () => { + const functionCall: Content = { + role: 'model', + parts: [{functionCall: {name: 'echo', args: {value: 1}}}], + }; + const llm = new FakeLiveLlm([ + {content: functionCall}, + {turnComplete: true}, + ]); + const agent = new LlmAgent({ + name: 'agent', + model: llm, + tools: [new EchoTool()], + }); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + const events: Event[] = []; + for await (const event of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + events.push(event); + } + + const responseEvents = events.filter((event) => + event.content?.parts?.some((part) => part.functionResponse), + ); + expect(responseEvents.length).toBe(1); + + expect(llm.connection!.contentCalls.length).toBe(1); + const sentBack = llm.connection!.contentCalls[0]; + expect(sentBack.parts?.[0]?.functionResponse?.name).toBe('echo'); + }); + + it('captures sessionResumptionUpdate handles into invocation context', async () => { + const llm = new FakeLiveLlm([ + {liveSessionResumptionUpdate: {newHandle: 'handle-1'}}, + {liveSessionResumptionUpdate: {newHandle: 'handle-2'}}, + {turnComplete: true}, + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + const events: Event[] = []; + for await (const event of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + events.push(event); + } + + const resumeEvents = events.filter((e) => e.liveSessionResumptionUpdate); + expect(resumeEvents.length).toBe(2); + expect(resumeEvents[1].liveSessionResumptionUpdate?.newHandle).toBe( + 'handle-2', + ); + }); + + it('reconnects with session handle on goAway and skips history replay', async () => { + const llm = new FakeLiveLlm([ + [ + {liveSessionResumptionUpdate: {newHandle: 'handle-1'}}, + {goAway: {timeLeft: '1s'}}, + ], + [{turnComplete: true}], + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + // Seed a content event so contents is non-empty on the first connect. + const session = (await sessionService.getSession({ + appName: TEST_APP_ID, + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + }))!; + await sessionService.appendEvent({ + session, + event: { + invocationId: 'seed', + author: 'user', + id: 'seed-evt', + actions: { + stateDelta: {}, + artifactDelta: {}, + requestedAuthConfigs: {}, + requestedToolConfirmations: {}, + }, + longRunningToolIds: [], + timestamp: Date.now(), + content: {role: 'user', parts: [{text: 'hello'}]}, + } as Event, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + + expect(llm.connections.length).toBe(2); + // First connection received history, second skipped it. + expect(llm.connections[0].historyCalls.length).toBe(1); + expect(llm.connections[1].historyCalls.length).toBe(0); + // Second connect carried the captured resumption handle. + expect( + llm.llmRequestsSeen[1].liveConnectConfig?.sessionResumption?.handle, + ).toBe('handle-1'); + expect( + llm.llmRequestsSeen[1].liveConnectConfig?.sessionResumption?.transparent, + ).toBe(true); + // First connect had no resumption handle set. + expect( + llm.llmRequestsSeen[0].liveConnectConfig?.sessionResumption?.handle, + ).toBeUndefined(); + }); + + it('uses an externally provided session resumption handle on first connect', async () => { + const llm = new FakeLiveLlm([{turnComplete: true}]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + // Seed contents so without a handle the runner would call sendHistory. + const session = (await sessionService.getSession({ + appName: TEST_APP_ID, + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + }))!; + await sessionService.appendEvent({ + session, + event: { + invocationId: 'seed', + author: 'user', + id: 'seed-evt', + actions: { + stateDelta: {}, + artifactDelta: {}, + requestedAuthConfigs: {}, + requestedToolConfirmations: {}, + }, + longRunningToolIds: [], + timestamp: Date.now(), + content: {role: 'user', parts: [{text: 'hello'}]}, + } as Event, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + liveSessionResumptionHandle: 'external-handle', + })) { + // drain + } + + // History was skipped because the caller supplied a handle. + expect(llm.connections[0].historyCalls.length).toBe(0); + expect( + llm.llmRequestsSeen[0].liveConnectConfig?.sessionResumption?.handle, + ).toBe('external-handle'); + }); + + it('does not reconnect when no resumption handle has been captured', async () => { + const llm = new FakeLiveLlm([ + [{goAway: {timeLeft: '1s'}}], + [{turnComplete: true}], + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + await expect(async () => { + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + }).rejects.toThrow(/live reconnect requested/); + + expect(llm.connections.length).toBe(1); + }); + + it('forwards activity-start and activity-end signals to the connection', async () => { + const llm = new FakeLiveLlm([{turnComplete: true}]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.sendActivityStart(); + queue.sendActivityEnd(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + + expect(llm.connection!.activityStartCalls).toBe(1); + expect(llm.connection!.activityEndCalls).toBe(1); + }); + + it('surfaces control-signal responses and attributes user-authored content', async () => { + const userContent: Content = { + role: 'user', + parts: [{text: 'echoed back'}], + }; + const llm = new FakeLiveLlm([ + {}, + {interrupted: true}, + {usageMetadata: {totalTokenCount: 5}}, + {content: userContent}, + {turnComplete: true}, + ]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + const events: Event[] = []; + for await (const event of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + events.push(event); + } + + expect(events.some((e) => e.interrupted)).toBe(true); + expect(events.some((e) => e.usageMetadata)).toBe(true); + const userEvent = events.find((e) => e.content === userContent); + expect(userEvent?.author).toBe('user'); + }); + + it('forwards turn-by-turn content to the connection', async () => { + const llm = new FakeLiveLlm([{turnComplete: true}]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + const content: Content = {role: 'user', parts: [{text: 'hi there'}]}; + queue.sendContent(content); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + // drain + } + + expect(llm.connection!.contentCalls).toEqual([content]); + }); + + it('stops early when the abort signal is already aborted', async () => { + const llm = new FakeLiveLlm([{turnComplete: true}]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + const controller = new AbortController(); + controller.abort(); + const events: Event[] = []; + for await (const event of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + abortSignal: controller.signal, + })) { + events.push(event); + } + + expect(events).toEqual([]); + }); + + it('transfers to a sub-agent on transfer_to_agent and yields its events', async () => { + const transferCall: Content = { + role: 'model', + parts: [ + {functionCall: {name: 'transfer_to_agent', args: {agentName: 'child'}}}, + ], + }; + const childText: Content = { + role: 'model', + parts: [{text: 'child speaking'}], + }; + const childLlm = new FakeLiveLlm([ + {content: childText}, + {turnComplete: true}, + ]); + const child = new LlmAgent({name: 'child', model: childLlm}); + const parentLlm = new FakeLiveLlm([ + {content: transferCall}, + {turnComplete: true}, + ]); + const parent = new LlmAgent({ + name: 'parent', + model: parentLlm, + subAgents: [child], + }); + const runner = new Runner({ + appName: TEST_APP_ID, + agent: parent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + const events: Event[] = []; + for await (const event of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + })) { + events.push(event); + } + + expect(parentLlm.connection!.closed).toBe(true); + expect(childLlm.connection).toBeDefined(); + expect(events.some((e) => e.content === childText)).toBe(true); + }); + + it('applies speech, transcription, and compression config from the run config', async () => { + const llm = new FakeLiveLlm([{turnComplete: true}]); + const agent = new LlmAgent({name: 'agent', model: llm}); + const runner = new Runner({ + appName: TEST_APP_ID, + agent, + sessionService, + artifactService, + }); + + const queue = new LiveRequestQueue(); + queue.close(); + for await (const _ of runner.runLive({ + userId: TEST_USER_ID, + sessionId: TEST_SESSION_ID, + liveRequestQueue: queue, + runConfig: { + responseModalities: [Modality.AUDIO], + speechConfig: {languageCode: 'en-US'}, + inputAudioTranscription: {}, + outputAudioTranscription: {}, + realtimeInputConfig: {}, + contextWindowCompression: {slidingWindow: {}}, + proactivity: {proactiveAudio: true}, + enableAffectiveDialog: true, + }, + })) { + // drain + } + + const liveConfig = llm.llmRequestSeen?.liveConnectConfig; + expect(liveConfig?.speechConfig).toEqual({languageCode: 'en-US'}); + expect(liveConfig?.inputAudioTranscription).toBeDefined(); + expect(liveConfig?.outputAudioTranscription).toBeDefined(); + expect(liveConfig?.contextWindowCompression).toEqual({slidingWindow: {}}); + expect(liveConfig?.proactivity).toEqual({proactiveAudio: true}); + expect(liveConfig?.enableAffectiveDialog).toBe(true); + }); +});