fix(voice): keep realtime turn after chat ctx timeout

.changeset/soft-realtime-turns.md

@@ -0,0 +1,6 @@
+---
+"@livekit/agents": patch
+"@livekit/agents-plugin-openai": patch
+---
+
+Do not drop realtime replies when the pre-reply chat context update times out.

agents/src/llm/index.ts

@@ -73,6 +73,7 @@ export {
 } from './llm.js';
 
 export {
+  RealtimeError,
   RealtimeModel,
   RealtimeSession,
   type GenerationCreatedEvent,

agents/src/llm/realtime.ts

@@ -71,6 +71,14 @@ export interface RealtimeCapabilities {
   nativeTranscriptSync?: boolean;
 }
 
+export class RealtimeError extends Error {
+  constructor(message: string, options?: ErrorOptions) {
+    super(message, options);
+    this.name = 'RealtimeError';
+    Error.captureStackTrace(this, RealtimeError);
+  }
+}
+
 export interface InputTranscriptionCompleted {
   itemId: string;
   transcript: string;

agents/src/voice/agent_activity.test.ts

@@ -23,6 +23,7 @@ import {
   FunctionCallOutput,
 } from '../llm/chat_context.js';
 import { LLM, type LLMStream } from '../llm/llm.js';
+import { type GenerationCreatedEvent, RealtimeError } from '../llm/realtime.js';
 import { type Tool, ToolContext, ToolFlag, Toolset, tool } from '../llm/tool_context.js';
 import { Future, Task } from '../utils.js';
 import { AgentTask, _getActivityTaskInfo } from './agent.js';
@@ -1096,6 +1097,103 @@ describe('AgentActivity - interrupted tool completion', () => {
   });
 });
 
+describe('AgentActivity - realtime reply chat context push', () => {
+  function buildRealtimeReplyActivity(updateError?: unknown) {
+    const generationEvent = {} as GenerationCreatedEvent;
+    const realtimeSession = {
+      chatCtx: ChatContext.empty(),
+      tools: ToolContext.empty(),
+      updateChatCtx: vi.fn(async (chatCtx: ChatContext) => {
+        if (updateError) {
+          throw updateError;
+        }
+        realtimeSession.chatCtx = chatCtx.copy();
+      }),
+      updateTools: vi.fn(async () => {}),
+      updateOptions: vi.fn(),
+      generateReply: vi.fn(async () => generationEvent),
+    };
+    const activity = {
+      realtimeSession,
+      toolChoice: undefined,
+      _onEnterIgnoredTools: () => [],
+      agent: { _chatCtx: ChatContext.empty() },
+      agentSession: {
+        _conversationItemAdded: vi.fn(),
+      },
+      realtimeGenerationTask: vi.fn(async () => {}),
+      logger: { info() {}, debug() {}, warn: vi.fn(), error: vi.fn() },
+    };
+    Object.setPrototypeOf(activity, AgentActivity.prototype);
+
+    return { activity, realtimeSession };
+  }
+
+  async function runRealtimeReplyTask(activity: unknown, speechHandle: SpeechHandle) {
+    const realtimeReplyTask = (
+      AgentActivity.prototype as unknown as {
+        realtimeReplyTask(
+          this: unknown,
+          args: {
+            speechHandle: SpeechHandle;
+            modelSettings: { toolChoice?: never };
+            abortController: AbortController;
+            userInput: string;
+          },
+        ): Promise<void>;
+      }
+    ).realtimeReplyTask;
+
+    await realtimeReplyTask.call(activity, {
+      speechHandle,
+      modelSettings: {},
+      abortController: new AbortController(),
+      userInput: 'hello',
+    });
+  }
+
+  it('generates a reply when updateChatCtx succeeds', async () => {
+    const { activity, realtimeSession } = buildRealtimeReplyActivity();
+    const handle = SpeechHandle.create();
+    handle._authorizeGeneration();
+
+    await runRealtimeReplyTask(activity, handle);
+
+    expect(realtimeSession.generateReply).toHaveBeenCalledTimes(1);
+    expect(activity.realtimeGenerationTask).toHaveBeenCalledTimes(1);
+    expect(handle.done()).toBe(false);
+  });
+
+  it('still generates a reply when updateChatCtx raises RealtimeError', async () => {
+    const { activity, realtimeSession } = buildRealtimeReplyActivity(
+      new RealtimeError('update_chat_ctx timed out.'),
+    );
+    const handle = SpeechHandle.create();
+    handle._authorizeGeneration();
+
+    await runRealtimeReplyTask(activity, handle);
+
+    expect(realtimeSession.generateReply).toHaveBeenCalledTimes(1);
+    expect(activity.realtimeGenerationTask).toHaveBeenCalledTimes(1);
+    expect(handle.done()).toBe(false);
+    expect(activity.agent._chatCtx.items.some((item) => item.type === 'message')).toBe(true);
+  });
+
+  it('marks the speech handle failed for unexpected updateChatCtx errors', async () => {
+    const error = new Error('boom');
+    const { activity, realtimeSession } = buildRealtimeReplyActivity(error);
+    const handle = SpeechHandle.create();
+    handle._authorizeGeneration();
+
+    await runRealtimeReplyTask(activity, handle);
+
+    expect(realtimeSession.generateReply).not.toHaveBeenCalled();
+    expect(activity.realtimeGenerationTask).not.toHaveBeenCalled();
+    expect(handle.done()).toBe(true);
+    expect(handle.exception()).toBe(error);
+  });
+});
+
 describe('AgentActivity - interruption while waiting for tools', () => {
   function buildToolOutput() {
     const call = FunctionCall.create({

agents/src/voice/agent_activity.ts

@@ -38,6 +38,7 @@ import {
   type InputTranscriptionCompleted,
   LLM,
   type MessageGeneration,
+  RealtimeError,
   RealtimeModel,
   type RealtimeModelError,
   type RealtimeSession,
@@ -4077,7 +4078,23 @@ export class AgentActivity implements RecognitionHooks {
         role: 'user',
         content: userInput,
       });
-      await this.realtimeSession.updateChatCtx(chatCtx);
+      try {
+        await this.realtimeSession.updateChatCtx(chatCtx);
+      } catch (error) {
+        if (error instanceof RealtimeError) {
+          this.logger.warn(
+            { error: error.message },
+            'failed to update the chat context before generating the reply',
+          );
+        } else {
+          this.logger.error(
+            { error },
+            'failed to update the chat context before generating the reply',
+          );
+          speechHandle._markDone(error);
+          return;
+        }
+      }
       this.agent._chatCtx.insert(message);
       this.agentSession._conversationItemAdded(message);
     }

agents/src/voice/speech_handle.ts

@@ -80,6 +80,7 @@ export class SpeechHandle {
   private doneFut = new Future<void>();
   private generations: Future<void>[] = [];
   private _chatItems: ChatItem[] = [];
+  private _error: unknown;
 
   /** @internal */
   _tasks: Task<void>[] = [];
@@ -183,6 +184,14 @@ export class SpeechHandle {
     return this.doneFut.done;
   }
 
+  exception(): unknown {
+    if (!this.doneFut.done) {
+      throw new Error('SpeechHandle is not done yet');
+    }
+
+    return this._error;
+  }
+
   get chatItems(): ChatItem[] {
     return this._chatItems;
   }
@@ -357,8 +366,11 @@ export class SpeechHandle {
   }
 
   /** @internal */
-  _markDone(): void {
+  _markDone(error?: unknown): void {
     if (!this.doneFut.done) {
+      if (error !== undefined) {
+        this._error = error;
+      }
       this.doneFut.resolve();
     }
 

plugins/openai/src/realtime/realtime_model.ts

@@ -662,7 +662,7 @@ export class RealtimeSession extends llm.RealtimeSession {
       const timeoutController = new AbortController();
       const timeoutPromise = delay(5000, { signal: timeoutController.signal }).then(() => {
         cleanupTimedOutFutures();
-        throw new Error('Chat ctx update events timed out');
+        throw new llm.RealtimeError('update_chat_ctx timed out.');
       });
 
       try {