From 083d5a46101f689d102a71d78bc3cce3e76258e2 Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Wed, 22 Jul 2026 03:56:53 +0000 Subject: [PATCH 1/3] fix(voice): keep realtime turn after chat ctx timeout --- .changeset/soft-realtime-turns.md | 6 ++ agents/src/llm/index.ts | 1 + agents/src/llm/realtime.ts | 8 ++ agents/src/voice/agent_activity.test.ts | 98 +++++++++++++++++++ agents/src/voice/agent_activity.ts | 19 +++- agents/src/voice/speech_handle.ts | 14 ++- plugins/openai/src/realtime/realtime_model.ts | 2 +- 7 files changed, 145 insertions(+), 3 deletions(-) create mode 100644 .changeset/soft-realtime-turns.md diff --git a/.changeset/soft-realtime-turns.md b/.changeset/soft-realtime-turns.md new file mode 100644 index 000000000..def92bb8c --- /dev/null +++ b/.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. diff --git a/agents/src/llm/index.ts b/agents/src/llm/index.ts index 721fb0395..f74a0001d 100644 --- a/agents/src/llm/index.ts +++ b/agents/src/llm/index.ts @@ -73,6 +73,7 @@ export { } from './llm.js'; export { + RealtimeError, RealtimeModel, RealtimeSession, type GenerationCreatedEvent, diff --git a/agents/src/llm/realtime.ts b/agents/src/llm/realtime.ts index 10af53493..dfd3e6d78 100644 --- a/agents/src/llm/realtime.ts +++ b/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; diff --git a/agents/src/voice/agent_activity.test.ts b/agents/src/voice/agent_activity.test.ts index 05f732d89..3b3ed17e2 100644 --- a/agents/src/voice/agent_activity.test.ts +++ b/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; + } + ).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({ diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 6f7998979..6e65f40b4 100644 --- a/agents/src/voice/agent_activity.ts +++ b/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); } diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index 2c1fcb111..e4274513f 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -80,6 +80,7 @@ export class SpeechHandle { private doneFut = new Future(); private generations: Future[] = []; private _chatItems: ChatItem[] = []; + private _error: unknown; /** @internal */ _tasks: Task[] = []; @@ -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(); } diff --git a/plugins/openai/src/realtime/realtime_model.ts b/plugins/openai/src/realtime/realtime_model.ts index 88a76ef6f..74fe7a504 100644 --- a/plugins/openai/src/realtime/realtime_model.ts +++ b/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 { From f5c33b7268e25f9c5aa1e805c456799f7e825a4d Mon Sep 17 00:00:00 2001 From: Toubat Date: Wed, 22 Jul 2026 12:31:24 -0700 Subject: [PATCH 2/3] fix(voice): document realtime errors and propagate RunResult failures Add TypeDoc for RealtimeError and SpeechHandle.exception, and reject RunResult when the last SpeechHandle completed with an error so session.run() matches Python SpeechHandle._error propagation. Co-authored-by: Cursor --- agents/src/llm/realtime.ts | 4 +++ agents/src/voice/speech_handle.test.ts | 20 +++++++++++++++ agents/src/voice/speech_handle.ts | 5 ++++ agents/src/voice/testing/run_result.test.ts | 27 ++++++++++++++++++++- agents/src/voice/testing/run_result.ts | 10 ++++++++ 5 files changed, 65 insertions(+), 1 deletion(-) diff --git a/agents/src/llm/realtime.ts b/agents/src/llm/realtime.ts index dfd3e6d78..63db3a42e 100644 --- a/agents/src/llm/realtime.ts +++ b/agents/src/llm/realtime.ts @@ -71,6 +71,10 @@ export interface RealtimeCapabilities { nativeTranscriptSync?: boolean; } +/** + * Error raised by a realtime provider when an operation fails or times out + * (for example a chat-context update or reply generation). + */ export class RealtimeError extends Error { constructor(message: string, options?: ErrorOptions) { super(message, options); diff --git a/agents/src/voice/speech_handle.test.ts b/agents/src/voice/speech_handle.test.ts index 3ae783e1d..e75d0bc17 100644 --- a/agents/src/voice/speech_handle.test.ts +++ b/agents/src/voice/speech_handle.test.ts @@ -184,3 +184,23 @@ describe('SpeechHandle._markDone - generation completion', () => { expect(outcome).toBe('resolved'); }); }); + +describe('SpeechHandle.exception', () => { + it('throws when the handle is not done yet', () => { + const handle = SpeechHandle.create(); + expect(() => handle.exception()).toThrow(/not done yet/); + }); + + it('returns undefined when the handle completed without an error', () => { + const handle = SpeechHandle.create(); + handle._markDone(); + expect(handle.exception()).toBeUndefined(); + }); + + it('returns the error passed to _markDone', () => { + const handle = SpeechHandle.create(); + const error = new Error('realtime failed'); + handle._markDone(error); + expect(handle.exception()).toBe(error); + }); +}); diff --git a/agents/src/voice/speech_handle.ts b/agents/src/voice/speech_handle.ts index e4274513f..5fc4c74c0 100644 --- a/agents/src/voice/speech_handle.ts +++ b/agents/src/voice/speech_handle.ts @@ -184,6 +184,11 @@ export class SpeechHandle { return this.doneFut.done; } + /** + * Returns the error that caused this SpeechHandle to complete, if any. + * + * @throws Error if the SpeechHandle is not done yet. + */ exception(): unknown { if (!this.doneFut.done) { throw new Error('SpeechHandle is not done yet'); diff --git a/agents/src/voice/testing/run_result.test.ts b/agents/src/voice/testing/run_result.test.ts index 3aa01aba0..34fc15536 100644 --- a/agents/src/voice/testing/run_result.test.ts +++ b/agents/src/voice/testing/run_result.test.ts @@ -9,7 +9,7 @@ import { ToolContext, tool } from '../../llm/tool_context.js'; import { Agent } from '../agent.js'; import { performToolExecutions } from '../generation.js'; import { SpeechHandle } from '../speech_handle.js'; -import { activeMockTools, withMockTools } from './run_result.js'; +import { activeMockTools, RunResult, withMockTools } from './run_result.js'; class AgentA extends Agent { constructor() { @@ -180,3 +180,28 @@ describe('withMockTools', () => { expect(output.output[0]?.toolCallOutput?.isError).toBe(true); }); }); + +describe('RunResult speech handle error propagation', () => { + it('rejects when the last SpeechHandle completed with an error', async () => { + const run = new RunResult(); + const handle = SpeechHandle.create(); + run._watchHandle(handle); + + const error = new Error('update_chat_ctx failed unexpectedly'); + handle._markDone(error); + run._markDoneIfNeeded(handle); + + await expect(run.wait()).rejects.toThrow('update_chat_ctx failed unexpectedly'); + }); + + it('resolves when the last SpeechHandle completed without an error', async () => { + const run = new RunResult(); + const handle = SpeechHandle.create(); + run._watchHandle(handle); + + handle._markDone(); + run._markDoneIfNeeded(handle); + + await expect(run.wait()).resolves.toBe(run); + }); +}); diff --git a/agents/src/voice/testing/run_result.ts b/agents/src/voice/testing/run_result.ts index 1bc92e7fb..e43e63502 100644 --- a/agents/src/voice/testing/run_result.ts +++ b/agents/src/voice/testing/run_result.ts @@ -280,6 +280,16 @@ export class RunResult { return; } + // Propagate speech-handle errors (e.g. LLM or realtime failures), matching + // Python RunResult which rejects on SpeechHandle._error before final output. + const handleError = this.lastSpeechHandle.exception(); + if (handleError !== undefined && handleError !== null) { + this.doneFut.reject( + handleError instanceof Error ? handleError : new Error(String(handleError)), + ); + return; + } + const finalOutput = this.lastSpeechHandle._maybeRunFinalOutput; if (finalOutput instanceof Error) { this.doneFut.reject(finalOutput); From a909c1dc4698db2f0d66af813fb256657fa76c05 Mon Sep 17 00:00:00 2001 From: Toubat Date: Wed, 22 Jul 2026 12:34:04 -0700 Subject: [PATCH 3/3] style: sort RunResult test imports for prettier Co-authored-by: Cursor --- agents/src/voice/testing/run_result.test.ts | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/agents/src/voice/testing/run_result.test.ts b/agents/src/voice/testing/run_result.test.ts index 34fc15536..60e1f27f2 100644 --- a/agents/src/voice/testing/run_result.test.ts +++ b/agents/src/voice/testing/run_result.test.ts @@ -9,7 +9,7 @@ import { ToolContext, tool } from '../../llm/tool_context.js'; import { Agent } from '../agent.js'; import { performToolExecutions } from '../generation.js'; import { SpeechHandle } from '../speech_handle.js'; -import { activeMockTools, RunResult, withMockTools } from './run_result.js'; +import { RunResult, activeMockTools, withMockTools } from './run_result.js'; class AgentA extends Agent { constructor() {