feat(core): implement Runner.runLive and LlmAgent live flow

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<string, ActiveStreamingTool>;
   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;
   }
 
   /**

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<LiveRequest> {
+  async get(abortSignal?: AbortSignal): Promise<LiveRequest> {
     if (this.queue.length > 0) {
       return this.queue.shift()!;
     }
     if (this.isClosed) {
       return {close: true};
     }
-    return new Promise<LiveRequest>((resolve) => {
-      this.resolveFnFifoQueue.push(resolve);
+    if (abortSignal?.aborted) {
+      throw new Error('LiveRequestQueue.get() was aborted.');
+    }
+    return new Promise<LiveRequest>((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});
     });
   }