From 09ea051459628c7e089c65f691bb0a4751c1c437 Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Fri, 17 Jul 2026 07:14:46 +0000 Subject: [PATCH 1/8] feat(voice): support runtime model updates --- .changeset/update-agent-models-runtime.md | 5 + agents/src/voice/agent.ts | 64 ++- agents/src/voice/agent_activity.ts | 107 ++++- agents/src/voice/agent_update_options.test.ts | 380 ++++++++++++++++++ agents/src/voice/agent_v2.ts | 8 +- agents/src/voice/audio_recognition.ts | 57 ++- agents/src/voice/keyterm_detection.ts | 10 + 7 files changed, 608 insertions(+), 23 deletions(-) create mode 100644 .changeset/update-agent-models-runtime.md create mode 100644 agents/src/voice/agent_update_options.test.ts diff --git a/.changeset/update-agent-models-runtime.md b/.changeset/update-agent-models-runtime.md new file mode 100644 index 000000000..52ee7ccdc --- /dev/null +++ b/.changeset/update-agent-models-runtime.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': minor +--- + +Add `Agent.updateOptions()` for swapping STT, VAD, LLM, and TTS models at runtime. diff --git a/agents/src/voice/agent.ts b/agents/src/voice/agent.ts index 05536663b..a34e1ec75 100644 --- a/agents/src/voice/agent.ts +++ b/agents/src/voice/agent.ts @@ -132,15 +132,26 @@ export interface ModelSettings { toolChoice?: ToolChoice; } +export interface AgentUpdateOptions { + /** New STT model. Pass `null` to disable the agent STT and override any session STT. */ + stt?: STT | STTModelString | null; + /** New VAD model. Pass `null` to disable the agent VAD and override any session VAD. */ + vad?: VAD | null; + /** New LLM model. Pass `null` to disable the agent LLM and override any session LLM. */ + llm?: LLM | RealtimeModel | LLMModels | null; + /** New TTS model. Pass `null` to disable the agent TTS and override any session TTS. */ + tts?: TTS | TTSModelString | null; +} + export interface AgentOptions { id?: string; instructions: string | Instructions; chatCtx?: ChatContext; tools?: ToolContextLike; - stt?: STT | STTModelString; - vad?: VAD; - llm?: LLM | RealtimeModel | LLMModels; - tts?: TTS | TTSModelString; + stt?: STT | STTModelString | null; + vad?: VAD | null; + llm?: LLM | RealtimeModel | LLMModels | null; + tts?: TTS | TTSModelString | null; turnHandling?: TurnHandlingOptions; toolHandling?: ToolHandlingOptions; minConsecutiveSpeechDelay?: number; @@ -153,10 +164,14 @@ export interface AgentOptions { export class Agent { private _id: string; - private _stt?: STT; - private _vad?: VAD; - private _llm?: LLM | RealtimeModel; - private _tts?: TTS; + /** @internal */ + _stt?: STT | null; + /** @internal */ + _vad?: VAD | null; + /** @internal */ + _llm?: LLM | RealtimeModel | null; + /** @internal */ + _tts?: TTS | null; private _turnHandling?: Partial; private _minConsecutiveSpeechDelay?: number; @@ -254,19 +269,19 @@ export class Agent { this._agentActivity = undefined; } - get vad(): VAD | undefined { + get vad(): VAD | null | undefined { return this._vad; } - get stt(): STT | undefined { + get stt(): STT | null | undefined { return this._stt; } - get llm(): LLM | RealtimeModel | undefined { + get llm(): LLM | RealtimeModel | null | undefined { return this._llm; } - get tts(): TTS | undefined { + get tts(): TTS | null | undefined { return this._tts; } @@ -383,6 +398,31 @@ export class Agent { await this._agentActivity.updateInstructions(instructions); } + async updateOptions(options: AgentUpdateOptions): Promise { + const resolved: AgentUpdateOptions = { ...options }; + if (typeof resolved.stt === 'string') { + resolved.stt = InferenceSTT.fromModelString(resolved.stt); + } + if (typeof resolved.llm === 'string') { + resolved.llm = InferenceLLM.fromModelString(resolved.llm); + } + if (typeof resolved.tts === 'string') { + resolved.tts = InferenceTTS.fromModelString(resolved.tts); + } + + if (!this._agentActivity) { + if (Object.hasOwn(resolved, 'stt')) this._stt = resolved.stt as STT | null; + if (Object.hasOwn(resolved, 'vad')) this._vad = resolved.vad as VAD | null; + if (Object.hasOwn(resolved, 'llm')) { + this._llm = resolved.llm as LLM | RealtimeModel | null; + } + if (Object.hasOwn(resolved, 'tts')) this._tts = resolved.tts as TTS | null; + return; + } + + await this._agentActivity.updateModels(resolved); + } + // TODO(parity): Add when AgentConfigUpdate is ported to ChatContext. async updateTools(tools: ToolContextLike): Promise { if (!this._agentActivity) { diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 75cb22ac0..ec0762121 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -85,6 +85,7 @@ import { import { VAD, type VADEvent } from '../vad.js'; import { Agent, + type AgentUpdateOptions, type ModelSettings, StopResponse, _getActivityTaskInfo, @@ -758,7 +759,7 @@ export class AgentActivity implements RecognitionHooks { } get vad(): VAD | undefined { - return this.agent.vad || this.agentSession.vad; + return this.agent.vad !== undefined ? this.agent.vad ?? undefined : this.agentSession.vad; } /** @@ -774,7 +775,7 @@ export class AgentActivity implements RecognitionHooks { } get stt(): STT | undefined { - return this.agent.stt || this.agentSession.stt; + return this.agent.stt !== undefined ? this.agent.stt ?? undefined : this.agentSession.stt; } private getSttProvider(): string | undefined { @@ -789,11 +790,11 @@ export class AgentActivity implements RecognitionHooks { } get llm(): LLM | RealtimeModel | undefined { - return this.agent.llm || this.agentSession.llm; + return this.agent.llm !== undefined ? this.agent.llm ?? undefined : this.agentSession.llm; } get tts(): TTS | undefined { - return this.agent.tts || this.agentSession.tts; + return this.agent.tts !== undefined ? this.agent.tts ?? undefined : this.agentSession.tts; } get tools(): ToolContext { @@ -974,6 +975,104 @@ export class AgentActivity implements RecognitionHooks { } } + async updateModels(options: AgentUpdateOptions): Promise { + if ( + Object.hasOwn(options, 'llm') && + (options.llm instanceof RealtimeModel || this.llm instanceof RealtimeModel) + ) { + throw new Error( + 'cannot swap to or from a RealtimeModel while the agent is running, use AgentSession.updateAgent() instead', + ); + } + + if (Object.hasOwn(options, 'vad') && this.audioRecognition !== undefined) { + this.audioRecognition.checkVadSilenceRequirement(undefined, options.vad ?? undefined); + } + + if (Object.hasOwn(options, 'stt')) { + const oldStt = this.stt; + if (oldStt instanceof STT) { + oldStt.off('metrics_collected', this.onMetricsCollected); + oldStt.off('error', this.onModelError); + this.agentSession.off( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + + this.agent._stt = options.stt as STT | null; + const resolvedStt = this.stt; + if (this.audioRecognition !== undefined) { + await this.audioRecognition.updateStt( + this.stt ? (...args) => this.agent.sttNode(...args) : undefined, + { + model: resolvedStt?.model, + provider: resolvedStt?.provider, + resetContext: true, + }, + ); + } + this.agentSession._keytermDetector.swapStt(resolvedStt); + + if (resolvedStt instanceof STT) { + resolvedStt.on('metrics_collected', this.onMetricsCollected); + resolvedStt.on('error', this.onModelError); + if (resolvedStt.capabilities.chatContext) { + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + } + } + + if (Object.hasOwn(options, 'vad')) { + const oldVad = this.vad; + if (oldVad instanceof VAD) { + oldVad.off('metrics_collected', this.onMetricsCollected); + } + + this.agent._vad = options.vad as VAD | null; + if (this.audioRecognition !== undefined) { + await this.audioRecognition.updateVad(this.vad); + } + if (this.vad instanceof VAD) { + this.vad.on('metrics_collected', this.onMetricsCollected); + } + } + + if (Object.hasOwn(options, 'llm')) { + const oldLlm = this.llm; + if (oldLlm instanceof LLM) { + oldLlm.off('metrics_collected', this.onMetricsCollected); + oldLlm.off('error', this.onModelError); + } + + this.agent._llm = options.llm as LLM | null; + if (this.llm instanceof LLM) { + this.llm.prewarm(); + this.llm.on('metrics_collected', this.onMetricsCollected); + this.llm.on('error', this.onModelError); + } + } + + if (Object.hasOwn(options, 'tts')) { + const oldTts = this.tts; + if (oldTts instanceof TTS) { + oldTts.off('metrics_collected', this.onMetricsCollected); + oldTts.off('error', this.onModelError); + } + + this.agent._tts = options.tts as TTS | null; + if (this.tts instanceof TTS) { + const maybePrewarm = this.tts as TTS & { prewarm?: () => void }; + maybePrewarm.prewarm?.(); + this.tts.on('metrics_collected', this.onMetricsCollected); + this.tts.on('error', this.onModelError); + } + } + } + updateOptions(options: { endpointing?: EndpointingOptions; toolChoice?: ToolChoice | null; diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts new file mode 100644 index 000000000..f1fee1f61 --- /dev/null +++ b/agents/src/voice/agent_update_options.test.ts @@ -0,0 +1,380 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { AudioFrame } from '@livekit/rtc-node'; +import { describe, expect, it } from 'vitest'; +import { + BaseStreamingTurnDetector, + type BaseStreamingTurnDetectorStream, +} from '../inference/eot/base.js'; +import { ThresholdOptions, type TurnDetectorModel } from '../inference/eot/languages.js'; +import { ChatContext } from '../llm/chat_context.js'; +import { type RealtimeCapabilities, RealtimeModel, RealtimeSession } from '../llm/index.js'; +import type { GenerationCreatedEvent } from '../llm/realtime.js'; +import { ToolContext } from '../llm/tool_context.js'; +import { initializeLogger } from '../log.js'; +import { FakeSTT } from '../stt/testing/fake_stt.js'; +import { type ChunkedStream, SynthesizeStream, TTS } from '../tts/index.js'; +import type { APIConnectOptions } from '../types.js'; +import { DEFAULT_API_CONNECT_OPTIONS } from '../types.js'; +import { VAD, VADStream } from '../vad.js'; +import { Agent } from './agent.js'; +import { AgentSession } from './agent_session.js'; +import { FakeLLM } from './testing/fake_llm.js'; + +initializeLogger({ pretty: false, level: 'silent' }); + +class FakeVAD extends VAD { + label = 'fake-vad'; + + constructor() { + super({ updateInterval: 100 }); + } + + stream(): VADStream { + return new FakeVADStream(this); + } +} + +class FakeVADStream extends VADStream {} + +class LowSilenceVAD extends FakeVAD { + override get minSilenceDuration(): number { + return 0; + } +} + +class FakeSynthesizeStream extends SynthesizeStream { + label = 'fake-tts-stream'; + + protected async run(): Promise {} +} + +class FakeTTS extends TTS { + label = 'fake-tts'; + + constructor() { + super(24000, 1, { streaming: true }); + } + + synthesize(): ChunkedStream { + throw new Error('not implemented'); + } + + stream(options?: { connOptions?: APIConnectOptions }): SynthesizeStream { + return new FakeSynthesizeStream(this, options?.connOptions ?? DEFAULT_API_CONNECT_OPTIONS); + } +} + +class LabeledSTT extends FakeSTT { + override get model(): string { + return 'new-model'; + } + + override get provider(): string { + return 'new-provider'; + } +} + +class FakeRealtimeSession extends RealtimeSession { + get chatCtx(): ChatContext { + return ChatContext.empty(); + } + + get tools(): ToolContext { + return ToolContext.empty(); + } + + async updateInstructions(): Promise {} + + async updateChatCtx(): Promise {} + + async updateTools(): Promise {} + + updateOptions(): void {} + + pushAudio(_frame: AudioFrame): void {} + + async generateReply(): Promise { + throw new Error('not implemented'); + } + + async commitAudio(): Promise {} + + async clearAudio(): Promise {} + + async interrupt(): Promise {} + + async truncate(): Promise {} +} + +class FakeRealtimeModel extends RealtimeModel { + constructor() { + const capabilities: RealtimeCapabilities = { + messageTruncation: false, + turnDetection: false, + userTranscription: false, + autoToolReplyGeneration: false, + audioOutput: true, + manualFunctionCalls: false, + }; + super(capabilities); + } + + get model(): string { + return 'fake-realtime'; + } + + session(): RealtimeSession { + return new FakeRealtimeSession(this); + } + + async close(): Promise {} +} + +class FakeStreamingTurnDetector extends BaseStreamingTurnDetector { + constructor() { + super({ + sampleRate: 16000, + thresholds: new ThresholdOptions('turn-detector-v1-mini'), + }); + } + + get model(): TurnDetectorModel { + return 'turn-detector-v1-mini'; + } + + stream(): BaseStreamingTurnDetectorStream { + throw new Error('not implemented'); + } +} + +describe('Agent.updateOptions', () => { + it('replaces model fields when the agent is not running', async () => { + const stt1 = new FakeSTT(); + const stt2 = new FakeSTT(); + const vad1 = new FakeVAD(); + const vad2 = new FakeVAD(); + const llm1 = new FakeLLM(); + const llm2 = new FakeLLM(); + const tts1 = new FakeTTS(); + const tts2 = new FakeTTS(); + + const agent = new Agent({ instructions: 'test', stt: stt1, vad: vad1, llm: llm1, tts: tts1 }); + await agent.updateOptions({ stt: stt2, vad: vad2, llm: llm2, tts: tts2 }); + + expect(agent.stt).toBe(stt2); + expect(agent.vad).toBe(vad2); + expect(agent.llm).toBe(llm2); + expect(agent.tts).toBe(tts2); + }); + + it('only touches provided models when the agent is not running', async () => { + const stt = new FakeSTT(); + const llm = new FakeLLM(); + const tts = new FakeTTS(); + const agent = new Agent({ instructions: 'test', stt, llm }); + + await agent.updateOptions({ tts }); + + expect(agent.stt).toBe(stt); + expect(agent.llm).toBe(llm); + expect(agent.tts).toBe(tts); + }); + + it('swaps TTS while running', async () => { + const oldTts = new FakeTTS(); + const newTts = new FakeTTS(); + const agent = new Agent({ instructions: 'test', llm: new FakeLLM(), tts: oldTts }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + + await agent.updateOptions({ tts: newTts }); + + expect(agent.tts).toBe(newTts); + expect(activity.tts).toBe(newTts); + expect(oldTts.listenerCount('metrics_collected')).toBe(0); + expect(newTts.listenerCount('metrics_collected')).toBeGreaterThan(0); + } finally { + await session.close().catch(() => {}); + } + }); + + it('swaps LLM while running', async () => { + const oldLlm = new FakeLLM(); + const newLlm = new FakeLLM(); + const agent = new Agent({ instructions: 'test', llm: oldLlm }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + + await agent.updateOptions({ llm: newLlm }); + + expect(agent.llm).toBe(newLlm); + expect(activity.llm).toBe(newLlm); + expect(oldLlm.listenerCount('metrics_collected')).toBe(0); + expect(newLlm.listenerCount('metrics_collected')).toBeGreaterThan(0); + } finally { + await session.close().catch(() => {}); + } + }); + + it('swaps STT and rewires the live pipeline while running', async () => { + const oldStt = new FakeSTT(); + const newStt = new FakeSTT(); + const agent = new Agent({ + instructions: 'test', + stt: oldStt, + vad: new FakeVAD(), + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: unknown }).audioRecognition; + const oldPipeline = (recognition as { sttPipeline: unknown }).sttPipeline; + + await agent.updateOptions({ stt: newStt }); + + expect(agent.stt).toBe(newStt); + expect(activity.stt).toBe(newStt); + expect((recognition as { sttPipeline: unknown }).sttPipeline).not.toBe(oldPipeline); + expect(oldStt.listenerCount('metrics_collected')).toBe(0); + expect(newStt.listenerCount('metrics_collected')).toBeGreaterThan(0); + } finally { + await session.close().catch(() => {}); + } + }); + + it('swaps VAD while running', async () => { + const oldVad = new FakeVAD(); + const newVad = new FakeVAD(); + const agent = new Agent({ + instructions: 'test', + stt: new FakeSTT(), + vad: oldVad, + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + + await agent.updateOptions({ vad: newVad }); + + expect(agent.vad).toBe(newVad); + expect(activity.vad).toBe(newVad); + expect(oldVad.listenerCount('metrics_collected')).toBe(0); + expect(newVad.listenerCount('metrics_collected')).toBeGreaterThan(0); + } finally { + await session.close().catch(() => {}); + } + }); + + it('disables STT while running', async () => { + const agent = new Agent({ + instructions: 'test', + stt: new FakeSTT(), + vad: new FakeVAD(), + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + + await agent.updateOptions({ stt: null }); + + expect(agent.stt).toBeNull(); + expect(activity.stt).toBeUndefined(); + const recognition = (activity as unknown as { audioRecognition: unknown }).audioRecognition; + expect((recognition as { sttPipeline: unknown }).sttPipeline).toBeUndefined(); + } finally { + await session.close().catch(() => {}); + } + }); + + it('rejects swapping to a RealtimeModel while running', async () => { + const agent = new Agent({ instructions: 'test', llm: new FakeLLM() }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + await expect(agent.updateOptions({ llm: new FakeRealtimeModel() })).rejects.toThrow( + 'RealtimeModel', + ); + expect(agent.llm).toBeInstanceOf(FakeLLM); + } finally { + await session.close().catch(() => {}); + } + }); + + it('rejects swapping away from a RealtimeModel while running', async () => { + const agent = new Agent({ instructions: 'test', llm: new FakeRealtimeModel() }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + await expect(agent.updateOptions({ llm: new FakeLLM() })).rejects.toThrow('RealtimeModel'); + expect(agent.llm).toBeInstanceOf(FakeRealtimeModel); + } finally { + await session.close().catch(() => {}); + } + }); + + it('refreshes STT model and provider on swap', async () => { + const agent = new Agent({ + instructions: 'test', + stt: new FakeSTT(), + vad: new FakeVAD(), + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const recognition = (session._activity as unknown as { audioRecognition: unknown }) + .audioRecognition; + + await agent.updateOptions({ stt: new LabeledSTT() }); + + expect((recognition as { sttModel: unknown }).sttModel).toBe('new-model'); + expect((recognition as { sttProvider: unknown }).sttProvider).toBe('new-provider'); + } finally { + await session.close().catch(() => {}); + } + }); + + it('checks VAD silence requirements atomically', async () => { + const oldStt = new FakeSTT(); + const oldVad = new FakeVAD(); + const agent = new Agent({ + instructions: 'test', + stt: oldStt, + vad: oldVad, + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const recognition = (session._activity as unknown as { audioRecognition: unknown }) + .audioRecognition; + (recognition as { turnDetector: unknown }).turnDetector = new FakeStreamingTurnDetector(); + + await expect( + agent.updateOptions({ stt: new FakeSTT(), vad: new LowSilenceVAD() }), + ).rejects.toThrow('minSilenceDuration'); + + expect(agent.stt).toBe(oldStt); + expect(agent.vad).toBe(oldVad); + } finally { + await session.close().catch(() => {}); + } + }); +}); diff --git a/agents/src/voice/agent_v2.ts b/agents/src/voice/agent_v2.ts index e2ab0f8c8..3d23d08da 100644 --- a/agents/src/voice/agent_v2.ts +++ b/agents/src/voice/agent_v2.ts @@ -450,19 +450,19 @@ class AgentHookContext implements AgentContext { constructor(readonly agent: Agent) {} get vad(): VAD | undefined { - return this.agent.vad; + return this.agent.vad ?? undefined; } get stt(): STT | undefined { - return this.agent.stt; + return this.agent.stt ?? undefined; } get llm(): LLM | RealtimeModel | undefined { - return this.agent.llm; + return this.agent.llm ?? undefined; } get tts(): TTS | undefined { - return this.agent.tts; + return this.agent.tts ?? undefined; } get useTtsAlignedTranscript(): boolean | undefined { diff --git a/agents/src/voice/audio_recognition.ts b/agents/src/voice/audio_recognition.ts index d2005bcbc..c2b6b85f4 100644 --- a/agents/src/voice/audio_recognition.ts +++ b/agents/src/voice/audio_recognition.ts @@ -611,13 +611,14 @@ export class AudioRecognition { * configure it: raise if the bound VAD exposes `minSilenceDuration` and it * is below the floor. VADs that don't expose the knob are left untouched. */ - private checkVadSilenceRequirement( + checkVadSilenceRequirement( detector: _TurnDetector | BaseStreamingTurnDetector | undefined = this.turnDetector, + vad: VAD | undefined = this.vad, ): void { - if (!(detector instanceof BaseStreamingTurnDetector) || this.vad === undefined) { + if (!(detector instanceof BaseStreamingTurnDetector) || vad === undefined) { return; } - const current = this.vad.minSilenceDuration; + const current = vad.minSilenceDuration; if (current === null) { return; } @@ -630,6 +631,56 @@ export class AudioRecognition { } } + async updateStt( + stt: STTNode | undefined, + options: { model?: string; provider?: string; resetContext?: boolean } = {}, + ): Promise { + const unlock = await this.sttLifecycleLock.lock(); + try { + this.stt = stt; + if (Object.hasOwn(options, 'model')) { + this.sttModel = options.model; + } + if (Object.hasOwn(options, 'provider')) { + this.sttProvider = options.provider; + } + if (options.resetContext) { + this.lastLanguage = undefined; + this.sttRequestIds = []; + this.transcriptBuffer = []; + this.ignoreUserTranscriptUntil = undefined; + } + + await this.stopSttTasks(); + await this.sttPipeline?.close(); + this.sttPipeline = undefined; + this.sttOwnershipTransferred = false; + + if (!this.closed && this.stt !== undefined) { + this.startSttTasks(); + } + } finally { + unlock(); + } + } + + async updateVad(vad: VAD | undefined): Promise { + this.checkVadSilenceRequirement(undefined, vad); + this.vad = vad; + this.isInterruptionEnabled = !!(this.interruptionDetection && this.vad); + + await this.vadTask?.cancelAndWait(); + this.vadTask = undefined; + this.vadStream = undefined; + + if (!this.closed && this.vad !== undefined) { + this.vadTask = Task.from(({ signal }) => this.createVadTask(this.vad, signal)); + this.vadTask.result.catch((err) => { + this.logger.error(`Error running VAD task: ${err}`); + }); + } + } + async start(options?: { sttPipeline?: STTPipeline; turnDetectorStream?: BaseStreamingTurnDetectorStream; diff --git a/agents/src/voice/keyterm_detection.ts b/agents/src/voice/keyterm_detection.ts index 317a3ef66..7acc5867a 100644 --- a/agents/src/voice/keyterm_detection.ts +++ b/agents/src/voice/keyterm_detection.ts @@ -282,6 +282,16 @@ export class KeytermDetector extends (EventEmitter as new () => TypedEmitter 0) { + stt._updateSessionKeyterms(this.keyterms); + } + } + /** Bind this activity's STT (always) and start detection (if enabled). */ start(session: KeytermDetectorSession, stt: STT): void { // static keyterms must reach the recognizer even with detection disabled From 50db09ae47fa27658aa5618b8cd83281dba7064e Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 10:35:02 -0700 Subject: [PATCH 2/8] fix(voice): serialize runtime model swaps Keep live model replacement coherent with activity teardown and correctly propagate user VAD ownership after replacing the session default. Co-authored-by: Cursor --- .changeset/update-agent-models-runtime.md | 5 +- agents/src/voice/agent_activity.ts | 139 ++++++++++-------- agents/src/voice/agent_update_options.test.ts | 86 ++++++++++- agents/src/voice/audio_recognition.ts | 61 +++++--- agents/src/voice/index.ts | 1 + 5 files changed, 202 insertions(+), 90 deletions(-) diff --git a/.changeset/update-agent-models-runtime.md b/.changeset/update-agent-models-runtime.md index 52ee7ccdc..dcf4f85eb 100644 --- a/.changeset/update-agent-models-runtime.md +++ b/.changeset/update-agent-models-runtime.md @@ -1,5 +1,6 @@ --- -'@livekit/agents': minor +'@livekit/agents': patch --- -Add `Agent.updateOptions()` for swapping STT, VAD, LLM, and TTS models at runtime. +Add `Agent.updateOptions()` for swapping STT, VAD, LLM, and TTS models at runtime, including +explicit `null` values to disable session fallback. diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index ec0762121..b3d7d948d 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -989,87 +989,96 @@ export class AgentActivity implements RecognitionHooks { this.audioRecognition.checkVadSilenceRequirement(undefined, options.vad ?? undefined); } - if (Object.hasOwn(options, 'stt')) { - const oldStt = this.stt; - if (oldStt instanceof STT) { - oldStt.off('metrics_collected', this.onMetricsCollected); - oldStt.off('error', this.onModelError); - this.agentSession.off( - AgentSessionEventTypes.ConversationItemAdded, - this.pushConversationItemToStt, - ); - } - - this.agent._stt = options.stt as STT | null; - const resolvedStt = this.stt; - if (this.audioRecognition !== undefined) { - await this.audioRecognition.updateStt( - this.stt ? (...args) => this.agent.sttNode(...args) : undefined, - { - model: resolvedStt?.model, - provider: resolvedStt?.provider, - resetContext: true, - }, - ); + const unlock = await this.lock.lock(); + try { + if (this.closeAbort.signal.aborted || this.agent._agentActivity !== this) { + return; } - this.agentSession._keytermDetector.swapStt(resolvedStt); - if (resolvedStt instanceof STT) { - resolvedStt.on('metrics_collected', this.onMetricsCollected); - resolvedStt.on('error', this.onModelError); - if (resolvedStt.capabilities.chatContext) { - this.agentSession.on( + if (Object.hasOwn(options, 'stt')) { + const oldStt = this.stt; + if (oldStt instanceof STT) { + oldStt.off('metrics_collected', this.onMetricsCollected); + oldStt.off('error', this.onModelError); + this.agentSession.off( AgentSessionEventTypes.ConversationItemAdded, this.pushConversationItemToStt, ); } - } - } - if (Object.hasOwn(options, 'vad')) { - const oldVad = this.vad; - if (oldVad instanceof VAD) { - oldVad.off('metrics_collected', this.onMetricsCollected); + this.agent._stt = options.stt as STT | null; + const resolvedStt = this.stt; + if (this.audioRecognition !== undefined) { + await this.audioRecognition.updateStt( + this.stt ? (...args) => this.agent.sttNode(...args) : undefined, + { + model: resolvedStt?.model, + provider: resolvedStt?.provider, + resetContext: true, + }, + ); + } + this.agentSession._keytermDetector.swapStt(resolvedStt); + + if (resolvedStt instanceof STT) { + resolvedStt.on('metrics_collected', this.onMetricsCollected); + resolvedStt.on('error', this.onModelError); + if (resolvedStt.capabilities.chatContext) { + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + } } - this.agent._vad = options.vad as VAD | null; - if (this.audioRecognition !== undefined) { - await this.audioRecognition.updateVad(this.vad); - } - if (this.vad instanceof VAD) { - this.vad.on('metrics_collected', this.onMetricsCollected); - } - } + if (Object.hasOwn(options, 'vad')) { + const oldVad = this.vad; + if (oldVad instanceof VAD) { + oldVad.off('metrics_collected', this.onMetricsCollected); + } - if (Object.hasOwn(options, 'llm')) { - const oldLlm = this.llm; - if (oldLlm instanceof LLM) { - oldLlm.off('metrics_collected', this.onMetricsCollected); - oldLlm.off('error', this.onModelError); + this.agent._vad = options.vad as VAD | null; + if (this.audioRecognition !== undefined) { + await this.audioRecognition.updateVad(this.vad, this.usingDefaultVad); + } + if (this.vad instanceof VAD) { + this.vad.on('metrics_collected', this.onMetricsCollected); + } } - this.agent._llm = options.llm as LLM | null; - if (this.llm instanceof LLM) { - this.llm.prewarm(); - this.llm.on('metrics_collected', this.onMetricsCollected); - this.llm.on('error', this.onModelError); - } - } + if (Object.hasOwn(options, 'llm')) { + const oldLlm = this.llm; + if (oldLlm instanceof LLM) { + oldLlm.off('metrics_collected', this.onMetricsCollected); + oldLlm.off('error', this.onModelError); + } - if (Object.hasOwn(options, 'tts')) { - const oldTts = this.tts; - if (oldTts instanceof TTS) { - oldTts.off('metrics_collected', this.onMetricsCollected); - oldTts.off('error', this.onModelError); + this.agent._llm = options.llm as LLM | null; + if (this.llm instanceof LLM) { + this.llm.prewarm(); + this.llm.on('metrics_collected', this.onMetricsCollected); + this.llm.on('error', this.onModelError); + } } - this.agent._tts = options.tts as TTS | null; - if (this.tts instanceof TTS) { - const maybePrewarm = this.tts as TTS & { prewarm?: () => void }; - maybePrewarm.prewarm?.(); - this.tts.on('metrics_collected', this.onMetricsCollected); - this.tts.on('error', this.onModelError); + if (Object.hasOwn(options, 'tts')) { + const oldTts = this.tts; + if (oldTts instanceof TTS) { + oldTts.off('metrics_collected', this.onMetricsCollected); + oldTts.off('error', this.onModelError); + } + + this.agent._tts = options.tts as TTS | null; + if (this.tts instanceof TTS) { + const maybePrewarm = this.tts as TTS & { prewarm?: () => void }; + maybePrewarm.prewarm?.(); + this.tts.on('metrics_collected', this.onMetricsCollected); + this.tts.on('error', this.onModelError); + } } + } finally { + unlock(); } } diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index f1fee1f61..b6730a1cc 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -2,7 +2,7 @@ // // SPDX-License-Identifier: Apache-2.0 import type { AudioFrame } from '@livekit/rtc-node'; -import { describe, expect, it } from 'vitest'; +import { describe, expect, it, vi } from 'vitest'; import { BaseStreamingTurnDetector, type BaseStreamingTurnDetectorStream, @@ -20,6 +20,7 @@ import { DEFAULT_API_CONNECT_OPTIONS } from '../types.js'; import { VAD, VADStream } from '../vad.js'; import { Agent } from './agent.js'; import { AgentSession } from './agent_session.js'; +import type { AudioRecognition } from './audio_recognition.js'; import { FakeLLM } from './testing/fake_llm.js'; initializeLogger({ pretty: false, level: 'silent' }); @@ -149,6 +150,17 @@ class FakeStreamingTurnDetector extends BaseStreamingTurnDetector { } } +function deferred(): { + promise: Promise; + resolve: () => void; +} { + let resolve!: () => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + describe('Agent.updateOptions', () => { it('replaces model fields when the agent is not running', async () => { const stt1 = new FakeSTT(); @@ -251,6 +263,52 @@ describe('Agent.updateOptions', () => { } }); + it('serializes concurrent STT swaps so only the final model retains listeners', async () => { + const oldStt = new FakeSTT(); + const firstStt = new FakeSTT(); + const finalStt = new FakeSTT(); + const agent = new Agent({ + instructions: 'test', + stt: oldStt, + vad: new FakeVAD(), + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: AudioRecognition }) + .audioRecognition; + const originalUpdateStt = recognition.updateStt.bind(recognition); + const firstSwapBlocked = deferred(); + const releaseFirstSwap = deferred(); + let updateCount = 0; + + vi.spyOn(recognition, 'updateStt').mockImplementation(async (...args) => { + updateCount += 1; + if (updateCount === 1) { + firstSwapBlocked.resolve(); + await releaseFirstSwap.promise; + } + await originalUpdateStt(...args); + }); + + const firstSwap = agent.updateOptions({ stt: firstStt }); + await firstSwapBlocked.promise; + const finalSwap = agent.updateOptions({ stt: finalStt }); + releaseFirstSwap.resolve(); + await Promise.all([firstSwap, finalSwap]); + + expect(agent.stt).toBe(finalStt); + expect(oldStt.listenerCount('metrics_collected')).toBe(0); + expect(firstStt.listenerCount('metrics_collected')).toBe(0); + expect(finalStt.listenerCount('metrics_collected')).toBeGreaterThan(0); + } finally { + await session.close().catch(() => {}); + } + }); + it('swaps VAD while running', async () => { const oldVad = new FakeVAD(); const newVad = new FakeVAD(); @@ -277,6 +335,32 @@ describe('Agent.updateOptions', () => { } }); + it('treats a runtime VAD replacement as user-provided after inheriting the session default', async () => { + const replacementVad = new FakeVAD(); + const agent = new Agent({ + instructions: 'test', + stt: new FakeSTT(), + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: unknown }).audioRecognition; + + expect(activity.usingDefaultVad).toBe(true); + expect((recognition as { hasUserVad: boolean }).hasUserVad).toBe(false); + + await agent.updateOptions({ vad: replacementVad }); + + expect(activity.usingDefaultVad).toBe(false); + expect((recognition as { hasUserVad: boolean }).hasUserVad).toBe(true); + } finally { + await session.close().catch(() => {}); + } + }); + it('disables STT while running', async () => { const agent = new Agent({ instructions: 'test', diff --git a/agents/src/voice/audio_recognition.ts b/agents/src/voice/audio_recognition.ts index c2b6b85f4..2285c6884 100644 --- a/agents/src/voice/audio_recognition.ts +++ b/agents/src/voice/audio_recognition.ts @@ -355,6 +355,7 @@ export class AudioRecognition { private silenceAudioWriter: WritableStreamDefaultWriter; private sttOwnershipTransferred = false; private readonly sttLifecycleLock = new Mutex(); + private readonly vadLifecycleLock = new Mutex(); // all cancellable tasks private bounceEOUTask?: Task; @@ -664,20 +665,23 @@ export class AudioRecognition { } } - async updateVad(vad: VAD | undefined): Promise { + async updateVad(vad: VAD | undefined, usingDefaultVad: boolean): Promise { this.checkVadSilenceRequirement(undefined, vad); - this.vad = vad; - this.isInterruptionEnabled = !!(this.interruptionDetection && this.vad); + const unlock = await this.vadLifecycleLock.lock(); + try { + this.vad = vad; + this.usingDefaultVad = usingDefaultVad; + this.isInterruptionEnabled = !!(this.interruptionDetection && this.vad); - await this.vadTask?.cancelAndWait(); - this.vadTask = undefined; - this.vadStream = undefined; + await this.vadTask?.cancelAndWait(); + this.vadTask = undefined; + this.vadStream = undefined; - if (!this.closed && this.vad !== undefined) { - this.vadTask = Task.from(({ signal }) => this.createVadTask(this.vad, signal)); - this.vadTask.result.catch((err) => { - this.logger.error(`Error running VAD task: ${err}`); - }); + if (!this.closed && this.vad !== undefined) { + this.startVadTask(this.vad); + } + } finally { + unlock(); } } @@ -687,10 +691,9 @@ export class AudioRecognition { }) { this.startSttTasks(options?.sttPipeline); - this.vadTask = Task.from(({ signal }) => this.createVadTask(this.vad, signal)); - this.vadTask.result.catch((err) => { - this.logger.error(`Error running VAD task: ${err}`); - }); + if (this.vad !== undefined) { + this.startVadTask(this.vad); + } this.interruptionTask = Task.from(({ signal }) => this.createInterruptionTask(this.interruptionDetection, signal), @@ -2195,14 +2198,28 @@ export class AudioRecognition { * and trigger interruptions correctly. */ private resetVad() { - if (!this.vad) return; + void this.resetVadTask().catch((err) => { + this.logger.error(`Error resetting VAD task: ${err}`); + }); + } - this.vadTask?.cancelAndWait().finally(() => { - if (this.closed) return; - this.vadTask = Task.from(({ signal }) => this.createVadTask(this.vad, signal)); - this.vadTask.result.catch((err) => { - this.logger.error(`Error running VAD task: ${err}`); - }); + private async resetVadTask(): Promise { + const unlock = await this.vadLifecycleLock.lock(); + try { + if (!this.vad || this.closed) return; + await this.vadTask?.cancelAndWait(); + if (!this.closed && this.vad !== undefined) { + this.startVadTask(this.vad); + } + } finally { + unlock(); + } + } + + private startVadTask(vad: VAD): void { + this.vadTask = Task.from(({ signal }) => this.createVadTask(vad, signal)); + this.vadTask.result.catch((err) => { + this.logger.error(`Error running VAD task: ${err}`); }); } diff --git a/agents/src/voice/index.ts b/agents/src/voice/index.ts index 02872fc8a..48b3dba1f 100644 --- a/agents/src/voice/index.ts +++ b/agents/src/voice/index.ts @@ -12,6 +12,7 @@ export { type AgentOptions, type AgentTaskContext, type AgentTaskCreateOptions, + type AgentUpdateOptions, type ModelSettings, } from './agent.js'; export * from './amd.js'; From e3ded252946556763c5db2282e14f4e3ff14e279 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 10:41:42 -0700 Subject: [PATCH 3/8] test(voice): cover runtime model lifecycle races Lock in cleanup, fallback, listener, keyterm, and VAD task ownership semantics across concurrent runtime updates. Co-authored-by: Cursor --- agents/src/voice/agent_update_options.test.ts | 146 ++++++++++++++++++ .../voice/audio_recognition_vad_reset.test.ts | 44 ++++++ 2 files changed, 190 insertions(+) diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index b6730a1cc..56a967c3f 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -3,6 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 import type { AudioFrame } from '@livekit/rtc-node'; import { describe, expect, it, vi } from 'vitest'; +import { voice } from '../index.js'; import { BaseStreamingTurnDetector, type BaseStreamingTurnDetectorStream, @@ -21,6 +22,7 @@ import { VAD, VADStream } from '../vad.js'; import { Agent } from './agent.js'; import { AgentSession } from './agent_session.js'; import type { AudioRecognition } from './audio_recognition.js'; +import { AgentSessionEventTypes } from './events.js'; import { FakeLLM } from './testing/fake_llm.js'; initializeLogger({ pretty: false, level: 'silent' }); @@ -77,6 +79,24 @@ class LabeledSTT extends FakeSTT { } } +class ContextSTT extends FakeSTT { + keytermUpdates: string[][] = []; + contextUpdates: Array[0]> = []; + + constructor(label: string) { + super({ label }); + this.updateCapabilities({ chatContext: true, keyterms: true }); + } + + override _updateSessionKeyterms(keyterms: string[]): void { + this.keytermUpdates.push([...keyterms]); + } + + override _pushConversationItem(ev: Parameters[0]): void { + this.contextUpdates.push(ev); + } +} + class FakeRealtimeSession extends RealtimeSession { get chatCtx(): ChatContext { return ChatContext.empty(); @@ -162,6 +182,11 @@ function deferred(): { } describe('Agent.updateOptions', () => { + it('exports AgentUpdateOptions from the public voice namespace', () => { + const options: voice.AgentUpdateOptions = { stt: null, vad: null, llm: null, tts: null }; + expect(options).toEqual({ stt: null, vad: null, llm: null, tts: null }); + }); + it('replaces model fields when the agent is not running', async () => { const stt1 = new FakeSTT(); const stt2 = new FakeSTT(); @@ -309,6 +334,72 @@ describe('Agent.updateOptions', () => { } }); + it('serializes an STT swap with activity close and leaves no listeners attached', async () => { + const oldStt = new FakeSTT(); + const replacementStt = new FakeSTT(); + const agent = new Agent({ + instructions: 'test', + stt: oldStt, + vad: new FakeVAD(), + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: AudioRecognition }) + .audioRecognition; + const originalUpdateStt = recognition.updateStt.bind(recognition); + const swapBlocked = deferred(); + const releaseSwap = deferred(); + vi.spyOn(recognition, 'updateStt').mockImplementation(async (...args) => { + swapBlocked.resolve(); + await releaseSwap.promise; + await originalUpdateStt(...args); + }); + + const swap = agent.updateOptions({ stt: replacementStt }); + await swapBlocked.promise; + const close = activity.close(); + expect(agent._agentActivity).toBe(activity); + + releaseSwap.resolve(); + await Promise.all([swap, close]); + + expect(agent._agentActivity).toBeUndefined(); + expect(oldStt.listenerCount('metrics_collected')).toBe(0); + expect(oldStt.listenerCount('error')).toBe(0); + expect(replacementStt.listenerCount('metrics_collected')).toBe(0); + expect(replacementStt.listenerCount('error')).toBe(0); + await session.close().catch(() => {}); + }); + + it('moves STT context, keyterms, and error listeners to the replacement', async () => { + const oldStt = new ContextSTT('old-context-stt'); + const replacementStt = new ContextSTT('replacement-context-stt'); + const agent = new Agent({ instructions: 'test', stt: oldStt }); + const session = new AgentSession({ + turnDetection: 'manual', + keytermsOptions: { keyterms: ['LiveKit'] }, + }); + await session.start({ agent }); + try { + await agent.updateOptions({ stt: replacementStt }); + const itemCtx = ChatContext.empty(); + itemCtx.addMessage({ role: 'user', content: 'hello' }); + const event = { item: itemCtx.items[0]! }; + session.emit(AgentSessionEventTypes.ConversationItemAdded, event); + + expect(oldStt.contextUpdates).toHaveLength(0); + expect(replacementStt.contextUpdates).toEqual([event]); + expect(replacementStt.keytermUpdates).toContainEqual(['LiveKit']); + expect(oldStt.listenerCount('error')).toBe(0); + expect(replacementStt.listenerCount('error')).toBeGreaterThan(0); + } finally { + await session.close().catch(() => {}); + } + }); + it('swaps VAD while running', async () => { const oldVad = new FakeVAD(); const newVad = new FakeVAD(); @@ -385,6 +476,61 @@ describe('Agent.updateOptions', () => { } }); + it('uses session fallbacks when omitted and suppresses all four with null', async () => { + const sessionStt = new FakeSTT(); + const sessionVad = new FakeVAD(); + const sessionLlm = new FakeLLM(); + const sessionTts = new FakeTTS(); + const agent = new Agent({ instructions: 'test' }); + const session = new AgentSession({ + turnDetection: 'manual', + stt: sessionStt, + vad: sessionVad, + llm: sessionLlm, + tts: sessionTts, + }); + await session.start({ agent }); + try { + const activity = session._activity!; + expect(activity.stt).toBe(sessionStt); + expect(activity.vad).toBe(sessionVad); + expect(activity.llm).toBe(sessionLlm); + expect(activity.tts).toBe(sessionTts); + + await agent.updateOptions({ stt: null, vad: null, llm: null, tts: null }); + + expect(activity.stt).toBeUndefined(); + expect(activity.vad).toBeUndefined(); + expect(activity.llm).toBeUndefined(); + expect(activity.tts).toBeUndefined(); + } finally { + await session.close().catch(() => {}); + } + }); + + it('removes listeners from every model after sequential swaps and close', async () => { + const stts = [new FakeSTT(), new FakeSTT(), new FakeSTT()]; + const llms = [new FakeLLM(), new FakeLLM(), new FakeLLM()]; + const ttss = [new FakeTTS(), new FakeTTS(), new FakeTTS()]; + const agent = new Agent({ + instructions: 'test', + stt: stts[0], + llm: llms[0], + tts: ttss[0], + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + + await agent.updateOptions({ stt: stts[1], llm: llms[1], tts: ttss[1] }); + await agent.updateOptions({ stt: stts[2], llm: llms[2], tts: ttss[2] }); + await session.close(); + + for (const model of [...stts, ...llms, ...ttss]) { + expect(model.listenerCount('metrics_collected')).toBe(0); + expect(model.listenerCount('error')).toBe(0); + } + }); + it('rejects swapping to a RealtimeModel while running', async () => { const agent = new Agent({ instructions: 'test', llm: new FakeLLM() }); const session = new AgentSession({ turnDetection: 'manual' }); diff --git a/agents/src/voice/audio_recognition_vad_reset.test.ts b/agents/src/voice/audio_recognition_vad_reset.test.ts index 8ec404fb0..61f6be003 100644 --- a/agents/src/voice/audio_recognition_vad_reset.test.ts +++ b/agents/src/voice/audio_recognition_vad_reset.test.ts @@ -22,6 +22,10 @@ interface RecognitionInternals { lastSpeakingTime?: number; onSTTEvent: (ev: SpeechEvent) => Promise; bounceEOUTask?: { cancelAndWait: () => Promise }; + vadTask?: { cancelAndWait: () => Promise }; + updateVad: (vad: VAD | undefined, usingDefaultVad: boolean) => Promise; + resetVadTask: () => Promise; + startVadTask: (vad: VAD) => void; } function makeHooks(): RecognitionHooks { @@ -135,4 +139,44 @@ describe('AudioRecognition STT end-of-speech VAD reset', () => { await internals.bounceEOUTask?.cancelAndWait().catch(() => {}); } }); + + it('serializes VAD replacement with reset and keeps one final task owner', async () => { + const { recognition, internals } = makeRecognition(); + const oldVad = {} as VAD; + const replacementVad = {} as VAD; + internals.vad = oldVad; + + let releaseInitialCancel!: () => void; + const initialCancel = new Promise((resolve) => { + releaseInitialCancel = resolve; + }); + let activeTasks = 1; + let maxActiveTasks = 1; + internals.vadTask = { + cancelAndWait: async () => { + await initialCancel; + activeTasks -= 1; + }, + }; + + vi.spyOn(internals, 'startVadTask').mockImplementation((vad) => { + activeTasks += 1; + maxActiveTasks = Math.max(maxActiveTasks, activeTasks); + internals.vadTask = { + cancelAndWait: async () => { + activeTasks -= 1; + }, + }; + internals.vad = vad; + }); + + const update = internals.updateVad(replacementVad, false); + const reset = internals.resetVadTask(); + releaseInitialCancel(); + await Promise.all([update, reset]); + + expect(internals.vad).toBe(replacementVad); + expect(activeTasks).toBe(1); + expect(maxActiveTasks).toBe(1); + }); }); From 30ccaa96e013d80ffd25be99f4a88ee4f406e979 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 10:42:40 -0700 Subject: [PATCH 4/8] chore(voice): satisfy lifecycle test lint Co-authored-by: Cursor --- agents/src/voice/agent_update_options.test.ts | 2 +- agents/src/voice/audio_recognition_vad_reset.test.ts | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index 56a967c3f..6ae3273a1 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -3,7 +3,7 @@ // SPDX-License-Identifier: Apache-2.0 import type { AudioFrame } from '@livekit/rtc-node'; import { describe, expect, it, vi } from 'vitest'; -import { voice } from '../index.js'; +import type { voice } from '../index.js'; import { BaseStreamingTurnDetector, type BaseStreamingTurnDetectorStream, diff --git a/agents/src/voice/audio_recognition_vad_reset.test.ts b/agents/src/voice/audio_recognition_vad_reset.test.ts index 61f6be003..9c32cb572 100644 --- a/agents/src/voice/audio_recognition_vad_reset.test.ts +++ b/agents/src/voice/audio_recognition_vad_reset.test.ts @@ -141,7 +141,7 @@ describe('AudioRecognition STT end-of-speech VAD reset', () => { }); it('serializes VAD replacement with reset and keeps one final task owner', async () => { - const { recognition, internals } = makeRecognition(); + const { internals } = makeRecognition(); const oldVad = {} as VAD; const replacementVad = {} as VAD; internals.vad = oldVad; From 546a1932b261057a29b26d4a3f22ff4469310bd0 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 10:56:52 -0700 Subject: [PATCH 5/8] fix(voice): keep paused model updates state-only Prevent queued updates from reattaching resources to stale handoff activities while preserving model changes for a later resume. Co-authored-by: Cursor --- agents/src/voice/agent_activity.ts | 8 ++ agents/src/voice/agent_update_options.test.ts | 88 +++++++++++++++++++ 2 files changed, 96 insertions(+) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index b3d7d948d..b5f000ce3 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -995,6 +995,14 @@ export class AgentActivity implements RecognitionHooks { return; } + if (this._schedulingPaused || this.agentSession._activity !== this) { + if (Object.hasOwn(options, 'stt')) this.agent._stt = options.stt as STT | null; + if (Object.hasOwn(options, 'vad')) this.agent._vad = options.vad as VAD | null; + if (Object.hasOwn(options, 'llm')) this.agent._llm = options.llm as LLM | null; + if (Object.hasOwn(options, 'tts')) this.agent._tts = options.tts as TTS | null; + return; + } + if (Object.hasOwn(options, 'stt')) { const oldStt = this.stt; if (oldStt instanceof STT) { diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index 6ae3273a1..e69abbe95 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -374,6 +374,94 @@ describe('Agent.updateOptions', () => { await session.close().catch(() => {}); }); + it('keeps a queued update state-only when handoff pauses the old activity', async () => { + const oldStt = new ContextSTT('old-stt'); + const oldLlm = new FakeLLM(); + const oldTts = new FakeTTS(); + const oldAgent = new Agent({ + instructions: 'old', + stt: oldStt, + llm: oldLlm, + tts: oldTts, + }); + const newStt = new ContextSTT('new-stt'); + const newAgent = new Agent({ + instructions: 'new', + stt: newStt, + llm: new FakeLLM(), + tts: new FakeTTS(), + }); + const replacementStt = new ContextSTT('replacement-stt'); + const replacementLlm = new FakeLLM(); + const replacementTts = new FakeTTS(); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent: oldAgent }); + + try { + const oldActivity = session._activity!; + const pauseResourcesClosed = deferred(); + const releasePause = deferred(); + const internals = oldActivity as unknown as { + _closeSessionResources: () => Promise; + }; + const originalCloseResources = internals._closeSessionResources.bind(internals); + vi.spyOn(internals, '_closeSessionResources').mockImplementation(async () => { + await originalCloseResources(); + pauseResourcesClosed.resolve(); + await releasePause.promise; + }); + + const handoff = session._updateActivity(newAgent, { previousActivity: 'pause' }); + await pauseResourcesClosed.promise; + const queuedUpdate = oldAgent.updateOptions({ + stt: replacementStt, + llm: replacementLlm, + tts: replacementTts, + }); + releasePause.resolve(); + await Promise.all([queuedUpdate, handoff]); + + expect(oldAgent.stt).toBe(replacementStt); + expect(oldAgent.llm).toBe(replacementLlm); + expect(oldAgent.tts).toBe(replacementTts); + for (const model of [ + oldStt, + oldLlm, + oldTts, + replacementStt, + replacementLlm, + replacementTts, + ]) { + expect(model.listenerCount('metrics_collected')).toBe(0); + expect(model.listenerCount('error')).toBe(0); + } + + const handoffCtx = ChatContext.empty(); + handoffCtx.addMessage({ role: 'user', content: 'after handoff' }); + const handoffEvent = { item: handoffCtx.items[0]! }; + session.emit(AgentSessionEventTypes.ConversationItemAdded, handoffEvent); + expect(replacementStt.contextUpdates).toHaveLength(0); + expect(newStt.contextUpdates).toEqual([handoffEvent]); + + await session._updateActivity(oldAgent, { + previousActivity: 'pause', + newActivity: 'resume', + }); + expect(session._activity).toBe(oldActivity); + expect(replacementStt.listenerCount('metrics_collected')).toBeGreaterThan(0); + expect(replacementLlm.listenerCount('metrics_collected')).toBeGreaterThan(0); + expect(replacementTts.listenerCount('metrics_collected')).toBeGreaterThan(0); + + const resumedCtx = ChatContext.empty(); + resumedCtx.addMessage({ role: 'user', content: 'after resume' }); + const resumedEvent = { item: resumedCtx.items[0]! }; + session.emit(AgentSessionEventTypes.ConversationItemAdded, resumedEvent); + expect(replacementStt.contextUpdates).toEqual([resumedEvent]); + } finally { + await session.close().catch(() => {}); + } + }); + it('moves STT context, keyterms, and error listeners to the replacement', async () => { const oldStt = new ContextSTT('old-context-stt'); const replacementStt = new ContextSTT('replacement-context-stt'); From 3eca30e9521b7c81440f8cb7dba5289c2edaaae2 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 11:06:26 -0700 Subject: [PATCH 6/8] fix(voice): make runtime model updates atomic Preflight model warmup and roll back recognition, keyterm, listener, and model ownership when operational swaps fail. Co-authored-by: Cursor --- agents/src/voice/agent_activity.ts | 231 +++++++++++++----- agents/src/voice/agent_update_options.test.ts | 173 +++++++++++++ 2 files changed, 343 insertions(+), 61 deletions(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index b5f000ce3..c3480e95e 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -1003,87 +1003,196 @@ export class AgentActivity implements RecognitionHooks { return; } - if (Object.hasOwn(options, 'stt')) { - const oldStt = this.stt; - if (oldStt instanceof STT) { - oldStt.off('metrics_collected', this.onMetricsCollected); - oldStt.off('error', this.onModelError); - this.agentSession.off( - AgentSessionEventTypes.ConversationItemAdded, - this.pushConversationItemToStt, - ); - } + const previous = { + stt: this.agent._stt, + vad: this.agent._vad, + llm: this.agent._llm, + tts: this.agent._tts, + resolvedStt: this.stt, + resolvedVad: this.vad, + resolvedLlm: this.llm, + resolvedTts: this.tts, + usingDefaultVad: this.usingDefaultVad, + }; + const nextLlm = options.llm !== undefined ? options.llm ?? undefined : this.agentSession.llm; + const nextTts = options.tts !== undefined ? options.tts ?? undefined : this.agentSession.tts; - this.agent._stt = options.stt as STT | null; - const resolvedStt = this.stt; - if (this.audioRecognition !== undefined) { - await this.audioRecognition.updateStt( - this.stt ? (...args) => this.agent.sttNode(...args) : undefined, - { - model: resolvedStt?.model, - provider: resolvedStt?.provider, - resetContext: true, - }, - ); + try { + if (Object.hasOwn(options, 'llm') && nextLlm instanceof LLM) { + nextLlm.prewarm(); + } + if (Object.hasOwn(options, 'tts') && nextTts instanceof TTS) { + const maybePrewarm = nextTts as TTS & { prewarm?: () => void }; + maybePrewarm.prewarm?.(); } - this.agentSession._keytermDetector.swapStt(resolvedStt); - if (resolvedStt instanceof STT) { - resolvedStt.on('metrics_collected', this.onMetricsCollected); - resolvedStt.on('error', this.onModelError); - if (resolvedStt.capabilities.chatContext) { - this.agentSession.on( + if (Object.hasOwn(options, 'stt')) { + const oldStt = this.stt; + if (oldStt instanceof STT) { + oldStt.off('metrics_collected', this.onMetricsCollected); + oldStt.off('error', this.onModelError); + this.agentSession.off( AgentSessionEventTypes.ConversationItemAdded, this.pushConversationItemToStt, ); } - } - } - if (Object.hasOwn(options, 'vad')) { - const oldVad = this.vad; - if (oldVad instanceof VAD) { - oldVad.off('metrics_collected', this.onMetricsCollected); + this.agent._stt = options.stt as STT | null; + const resolvedStt = this.stt; + if (this.audioRecognition !== undefined) { + await this.audioRecognition.updateStt( + this.stt ? (...args) => this.agent.sttNode(...args) : undefined, + { + model: resolvedStt?.model, + provider: resolvedStt?.provider, + resetContext: true, + }, + ); + } + this.agentSession._keytermDetector.swapStt(resolvedStt); + + if (resolvedStt instanceof STT) { + resolvedStt.on('metrics_collected', this.onMetricsCollected); + resolvedStt.on('error', this.onModelError); + if (resolvedStt.capabilities.chatContext) { + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + } } - this.agent._vad = options.vad as VAD | null; - if (this.audioRecognition !== undefined) { - await this.audioRecognition.updateVad(this.vad, this.usingDefaultVad); + if (Object.hasOwn(options, 'vad')) { + const oldVad = this.vad; + if (oldVad instanceof VAD) { + oldVad.off('metrics_collected', this.onMetricsCollected); + } + + this.agent._vad = options.vad as VAD | null; + if (this.audioRecognition !== undefined) { + await this.audioRecognition.updateVad(this.vad, this.usingDefaultVad); + } + if (this.vad instanceof VAD) { + this.vad.on('metrics_collected', this.onMetricsCollected); + } } - if (this.vad instanceof VAD) { - this.vad.on('metrics_collected', this.onMetricsCollected); + + if (Object.hasOwn(options, 'llm')) { + const oldLlm = this.llm; + if (oldLlm instanceof LLM) { + oldLlm.off('metrics_collected', this.onMetricsCollected); + oldLlm.off('error', this.onModelError); + } + + this.agent._llm = options.llm as LLM | null; + if (this.llm instanceof LLM) { + this.llm.on('metrics_collected', this.onMetricsCollected); + this.llm.on('error', this.onModelError); + } } - } - if (Object.hasOwn(options, 'llm')) { - const oldLlm = this.llm; - if (oldLlm instanceof LLM) { - oldLlm.off('metrics_collected', this.onMetricsCollected); - oldLlm.off('error', this.onModelError); + if (Object.hasOwn(options, 'tts')) { + const oldTts = this.tts; + if (oldTts instanceof TTS) { + oldTts.off('metrics_collected', this.onMetricsCollected); + oldTts.off('error', this.onModelError); + } + + this.agent._tts = options.tts as TTS | null; + if (this.tts instanceof TTS) { + this.tts.on('metrics_collected', this.onMetricsCollected); + this.tts.on('error', this.onModelError); + } + } + } catch (error) { + const rollbackErrors: unknown[] = []; + const failedStt = this.stt; + const failedVad = this.vad; + const failedLlm = this.llm; + const failedTts = this.tts; + + if (failedStt instanceof STT) { + failedStt.off('metrics_collected', this.onMetricsCollected); + failedStt.off('error', this.onModelError); + } + if (failedVad instanceof VAD) { + failedVad.off('metrics_collected', this.onMetricsCollected); + } + if (failedLlm instanceof LLM) { + failedLlm.off('metrics_collected', this.onMetricsCollected); + failedLlm.off('error', this.onModelError); + } + if (failedTts instanceof TTS) { + failedTts.off('metrics_collected', this.onMetricsCollected); + failedTts.off('error', this.onModelError); } + this.agentSession.off( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); - this.agent._llm = options.llm as LLM | null; - if (this.llm instanceof LLM) { - this.llm.prewarm(); - this.llm.on('metrics_collected', this.onMetricsCollected); - this.llm.on('error', this.onModelError); + this.agent._stt = previous.stt; + this.agent._vad = previous.vad; + this.agent._llm = previous.llm; + this.agent._tts = previous.tts; + + if (Object.hasOwn(options, 'stt')) { + try { + await this.audioRecognition?.updateStt( + previous.resolvedStt ? (...args) => this.agent.sttNode(...args) : undefined, + { + model: previous.resolvedStt?.model, + provider: previous.resolvedStt?.provider, + resetContext: true, + }, + ); + } catch (rollbackError) { + rollbackErrors.push(rollbackError); + } + try { + this.agentSession._keytermDetector.swapStt(previous.resolvedStt); + } catch (rollbackError) { + rollbackErrors.push(rollbackError); + } + } + if (Object.hasOwn(options, 'vad')) { + try { + await this.audioRecognition?.updateVad(previous.resolvedVad, previous.usingDefaultVad); + } catch (rollbackError) { + rollbackErrors.push(rollbackError); + } } - } - if (Object.hasOwn(options, 'tts')) { - const oldTts = this.tts; - if (oldTts instanceof TTS) { - oldTts.off('metrics_collected', this.onMetricsCollected); - oldTts.off('error', this.onModelError); + if (previous.resolvedStt instanceof STT) { + previous.resolvedStt.on('metrics_collected', this.onMetricsCollected); + previous.resolvedStt.on('error', this.onModelError); + if (previous.resolvedStt.capabilities.chatContext) { + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + } + if (previous.resolvedVad instanceof VAD) { + previous.resolvedVad.on('metrics_collected', this.onMetricsCollected); + } + if (previous.resolvedLlm instanceof LLM) { + previous.resolvedLlm.on('metrics_collected', this.onMetricsCollected); + previous.resolvedLlm.on('error', this.onModelError); + } + if (previous.resolvedTts instanceof TTS) { + previous.resolvedTts.on('metrics_collected', this.onMetricsCollected); + previous.resolvedTts.on('error', this.onModelError); } - this.agent._tts = options.tts as TTS | null; - if (this.tts instanceof TTS) { - const maybePrewarm = this.tts as TTS & { prewarm?: () => void }; - maybePrewarm.prewarm?.(); - this.tts.on('metrics_collected', this.onMetricsCollected); - this.tts.on('error', this.onModelError); + if (rollbackErrors.length > 0) { + this.logger.error( + { error, rollbackErrors }, + 'failed to completely roll back model update', + ); } + throw error; } } finally { unlock(); diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index e69abbe95..6a23bb367 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -69,6 +69,18 @@ class FakeTTS extends TTS { } } +class ThrowingPrewarmLLM extends FakeLLM { + override prewarm(): void { + throw new Error('injected LLM prewarm failure'); + } +} + +class ThrowingPrewarmTTS extends FakeTTS { + prewarm(): void { + throw new Error('injected TTS prewarm failure'); + } +} + class LabeledSTT extends FakeSTT { override get model(): string { return 'new-model'; @@ -619,6 +631,167 @@ describe('Agent.updateOptions', () => { } }); + it('rolls back a multi-model update when LLM prewarm throws', async () => { + const oldStt = new ContextSTT('old-stt'); + const oldLlm = new FakeLLM(); + const replacementStt = new ContextSTT('replacement-stt'); + const replacementLlm = new ThrowingPrewarmLLM(); + const agent = new Agent({ + instructions: 'test', + stt: oldStt, + llm: oldLlm, + tts: new FakeTTS(), + }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + await expect( + agent.updateOptions({ stt: replacementStt, llm: replacementLlm }), + ).rejects.toThrow('injected LLM prewarm failure'); + expect(agent.stt).toBe(oldStt); + expect(agent.llm).toBe(oldLlm); + expect(oldStt.listenerCount('metrics_collected')).toBe(1); + expect(oldStt.listenerCount('error')).toBe(1); + expect(oldLlm.listenerCount('metrics_collected')).toBe(1); + expect(oldLlm.listenerCount('error')).toBe(1); + expect(replacementStt.listenerCount('metrics_collected')).toBe(0); + expect(replacementStt.listenerCount('error')).toBe(0); + expect(replacementLlm.listenerCount('metrics_collected')).toBe(0); + expect(replacementLlm.listenerCount('error')).toBe(0); + expect(session.listenerCount(AgentSessionEventTypes.ConversationItemAdded)).toBe(1); + } finally { + await session.close().catch(() => {}); + } + }); + + it('rolls back STT recognition and keyterm ownership when keyterm swap throws', async () => { + const oldStt = new ContextSTT('old-stt'); + const replacementStt = new ContextSTT('replacement-stt'); + const agent = new Agent({ instructions: 'test', stt: oldStt }); + const session = new AgentSession({ + turnDetection: 'manual', + keytermsOptions: { keyterms: ['LiveKit'] }, + }); + await session.start({ agent }); + try { + const detector = session._keytermDetector; + const originalSwap = detector.swapStt.bind(detector); + vi.spyOn(detector, 'swapStt').mockImplementation((stt) => { + originalSwap(stt); + throw new Error('injected keyterm swap failure'); + }); + await expect(agent.updateOptions({ stt: replacementStt })).rejects.toThrow( + 'injected keyterm swap failure', + ); + expect(agent.stt).toBe(oldStt); + expect(oldStt.listenerCount('metrics_collected')).toBe(1); + expect(oldStt.listenerCount('error')).toBe(1); + expect(replacementStt.listenerCount('metrics_collected')).toBe(0); + expect(replacementStt.listenerCount('error')).toBe(0); + expect(session.listenerCount(AgentSessionEventTypes.ConversationItemAdded)).toBe(1); + } finally { + await session.close().catch(() => {}); + } + }); + + it('preflights TTS prewarm before mutating another model', async () => { + const oldStt = new ContextSTT('old-stt'); + const oldTts = new FakeTTS(); + const replacementStt = new ContextSTT('replacement-stt'); + const agent = new Agent({ instructions: 'test', stt: oldStt, tts: oldTts }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + await expect( + agent.updateOptions({ stt: replacementStt, tts: new ThrowingPrewarmTTS() }), + ).rejects.toThrow('injected TTS prewarm failure'); + expect(agent.stt).toBe(oldStt); + expect(agent.tts).toBe(oldTts); + expect(oldStt.listenerCount('metrics_collected')).toBe(1); + expect(oldStt.listenerCount('error')).toBe(1); + expect(oldTts.listenerCount('metrics_collected')).toBe(1); + expect(oldTts.listenerCount('error')).toBe(1); + expect(replacementStt.listenerCount('metrics_collected')).toBe(0); + expect(replacementStt.listenerCount('error')).toBe(0); + expect(session.listenerCount(AgentSessionEventTypes.ConversationItemAdded)).toBe(1); + await agent.updateOptions({ stt: replacementStt, tts: new FakeTTS() }); + expect(agent.stt).toBe(replacementStt); + } finally { + await session.close().catch(() => {}); + } + }); + + it('rolls back after recognition STT replacement fails operationally', async () => { + const oldStt = new ContextSTT('old-stt'); + const replacementStt = new ContextSTT('replacement-stt'); + const agent = new Agent({ instructions: 'test', stt: oldStt }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: AudioRecognition }) + .audioRecognition; + const originalUpdateStt = recognition.updateStt.bind(recognition); + let fail = true; + const spy = vi.spyOn(recognition, 'updateStt').mockImplementation(async (...args) => { + await originalUpdateStt(...args); + if (fail) { + fail = false; + throw new Error('injected recognition STT failure'); + } + }); + await expect(agent.updateOptions({ stt: replacementStt })).rejects.toThrow( + 'injected recognition STT failure', + ); + expect(agent.stt).toBe(oldStt); + expect(oldStt.listenerCount('metrics_collected')).toBe(1); + expect(oldStt.listenerCount('error')).toBe(1); + expect(replacementStt.listenerCount('metrics_collected')).toBe(0); + expect(replacementStt.listenerCount('error')).toBe(0); + expect(session.listenerCount(AgentSessionEventTypes.ConversationItemAdded)).toBe(1); + spy.mockRestore(); + await agent.updateOptions({ stt: replacementStt }); + expect(agent.stt).toBe(replacementStt); + } finally { + await session.close().catch(() => {}); + } + }); + + it('rolls back VAD reference and task ownership after an operational failure', async () => { + const oldVad = new FakeVAD(); + const replacementVad = new FakeVAD(); + replacementVad.label = 'replacement-vad'; + const agent = new Agent({ instructions: 'test', vad: oldVad }); + const session = new AgentSession({ turnDetection: 'manual' }); + await session.start({ agent }); + try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: AudioRecognition }) + .audioRecognition; + const originalUpdateVad = recognition.updateVad.bind(recognition); + let fail = true; + const spy = vi.spyOn(recognition, 'updateVad').mockImplementation(async (...args) => { + await originalUpdateVad(...args); + if (fail) { + fail = false; + throw new Error('injected recognition VAD failure'); + } + }); + await expect(agent.updateOptions({ vad: replacementVad })).rejects.toThrow( + 'injected recognition VAD failure', + ); + expect(agent.vad).toBe(oldVad); + expect(oldVad.listenerCount('metrics_collected')).toBe(1); + expect(replacementVad.listenerCount('metrics_collected')).toBe(0); + expect((recognition as unknown as { vad: VAD }).vad).toBe(oldVad); + spy.mockRestore(); + await agent.updateOptions({ vad: replacementVad }); + expect(agent.vad).toBe(replacementVad); + } finally { + await session.close().catch(() => {}); + } + }); + it('rejects swapping to a RealtimeModel while running', async () => { const agent = new Agent({ instructions: 'test', llm: new FakeLLM() }); const session = new AgentSession({ turnDetection: 'manual' }); From d1d06d462efc2300df89996cac6c1316f0b383de Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 11:14:56 -0700 Subject: [PATCH 7/8] fix(voice): isolate model prewarm failures Keep synchronous preflight rejection outside operational rollback so unchanged recognition pipelines and turn context remain intact. Co-authored-by: Cursor --- agents/src/voice/agent_activity.ts | 16 ++--- agents/src/voice/agent_update_options.test.ts | 68 +++++++++++++++++-- 2 files changed, 72 insertions(+), 12 deletions(-) diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index c3480e95e..873537c51 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -1017,15 +1017,15 @@ export class AgentActivity implements RecognitionHooks { const nextLlm = options.llm !== undefined ? options.llm ?? undefined : this.agentSession.llm; const nextTts = options.tts !== undefined ? options.tts ?? undefined : this.agentSession.tts; - try { - if (Object.hasOwn(options, 'llm') && nextLlm instanceof LLM) { - nextLlm.prewarm(); - } - if (Object.hasOwn(options, 'tts') && nextTts instanceof TTS) { - const maybePrewarm = nextTts as TTS & { prewarm?: () => void }; - maybePrewarm.prewarm?.(); - } + if (Object.hasOwn(options, 'llm') && nextLlm instanceof LLM) { + nextLlm.prewarm(); + } + if (Object.hasOwn(options, 'tts') && nextTts instanceof TTS) { + const maybePrewarm = nextTts as TTS & { prewarm?: () => void }; + maybePrewarm.prewarm?.(); + } + try { if (Object.hasOwn(options, 'stt')) { const oldStt = this.stt; if (oldStt instanceof STT) { diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index 6a23bb367..a58f90cd2 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -633,29 +633,64 @@ describe('Agent.updateOptions', () => { it('rolls back a multi-model update when LLM prewarm throws', async () => { const oldStt = new ContextSTT('old-stt'); + const oldVad = new FakeVAD(); const oldLlm = new FakeLLM(); const replacementStt = new ContextSTT('replacement-stt'); + const replacementVad = new FakeVAD(); const replacementLlm = new ThrowingPrewarmLLM(); const agent = new Agent({ instructions: 'test', stt: oldStt, + vad: oldVad, llm: oldLlm, tts: new FakeTTS(), }); const session = new AgentSession({ turnDetection: 'manual' }); await session.start({ agent }); try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: AudioRecognition }) + .audioRecognition; + const internals = recognition as unknown as { + sttPipeline: unknown; + vadTask: unknown; + transcriptBuffer: string[]; + sttRequestIds: string[]; + lastLanguage: string | undefined; + }; + internals.transcriptBuffer = ['partial transcript']; + internals.sttRequestIds = ['request-before-prewarm']; + internals.lastLanguage = 'en'; + const previousSttPipeline = internals.sttPipeline; + const previousVadTask = internals.vadTask; + const previousTranscriptBuffer = internals.transcriptBuffer; + const previousSttRequestIds = internals.sttRequestIds; + await expect( - agent.updateOptions({ stt: replacementStt, llm: replacementLlm }), + agent.updateOptions({ + stt: replacementStt, + vad: replacementVad, + llm: replacementLlm, + }), ).rejects.toThrow('injected LLM prewarm failure'); expect(agent.stt).toBe(oldStt); + expect(agent.vad).toBe(oldVad); expect(agent.llm).toBe(oldLlm); + expect(internals.sttPipeline).toBe(previousSttPipeline); + expect(internals.vadTask).toBe(previousVadTask); + expect(internals.transcriptBuffer).toBe(previousTranscriptBuffer); + expect(internals.transcriptBuffer).toEqual(['partial transcript']); + expect(internals.sttRequestIds).toBe(previousSttRequestIds); + expect(internals.sttRequestIds).toEqual(['request-before-prewarm']); + expect(internals.lastLanguage).toBe('en'); expect(oldStt.listenerCount('metrics_collected')).toBe(1); expect(oldStt.listenerCount('error')).toBe(1); + expect(oldVad.listenerCount('metrics_collected')).toBe(1); expect(oldLlm.listenerCount('metrics_collected')).toBe(1); expect(oldLlm.listenerCount('error')).toBe(1); expect(replacementStt.listenerCount('metrics_collected')).toBe(0); expect(replacementStt.listenerCount('error')).toBe(0); + expect(replacementVad.listenerCount('metrics_collected')).toBe(0); expect(replacementLlm.listenerCount('metrics_collected')).toBe(0); expect(replacementLlm.listenerCount('error')).toBe(0); expect(session.listenerCount(AgentSessionEventTypes.ConversationItemAdded)).toBe(1); @@ -696,25 +731,50 @@ describe('Agent.updateOptions', () => { it('preflights TTS prewarm before mutating another model', async () => { const oldStt = new ContextSTT('old-stt'); + const oldVad = new FakeVAD(); const oldTts = new FakeTTS(); const replacementStt = new ContextSTT('replacement-stt'); - const agent = new Agent({ instructions: 'test', stt: oldStt, tts: oldTts }); + const replacementVad = new FakeVAD(); + const agent = new Agent({ + instructions: 'test', + stt: oldStt, + vad: oldVad, + tts: oldTts, + }); const session = new AgentSession({ turnDetection: 'manual' }); await session.start({ agent }); try { + const activity = session._activity!; + const recognition = (activity as unknown as { audioRecognition: AudioRecognition }) + .audioRecognition; + const internals = recognition as unknown as { vadTask: unknown }; + const previousVadTask = internals.vadTask; + await expect( - agent.updateOptions({ stt: replacementStt, tts: new ThrowingPrewarmTTS() }), + agent.updateOptions({ + stt: replacementStt, + vad: replacementVad, + tts: new ThrowingPrewarmTTS(), + }), ).rejects.toThrow('injected TTS prewarm failure'); expect(agent.stt).toBe(oldStt); + expect(agent.vad).toBe(oldVad); expect(agent.tts).toBe(oldTts); + expect(internals.vadTask).toBe(previousVadTask); expect(oldStt.listenerCount('metrics_collected')).toBe(1); expect(oldStt.listenerCount('error')).toBe(1); + expect(oldVad.listenerCount('metrics_collected')).toBe(1); expect(oldTts.listenerCount('metrics_collected')).toBe(1); expect(oldTts.listenerCount('error')).toBe(1); expect(replacementStt.listenerCount('metrics_collected')).toBe(0); expect(replacementStt.listenerCount('error')).toBe(0); + expect(replacementVad.listenerCount('metrics_collected')).toBe(0); expect(session.listenerCount(AgentSessionEventTypes.ConversationItemAdded)).toBe(1); - await agent.updateOptions({ stt: replacementStt, tts: new FakeTTS() }); + await agent.updateOptions({ + stt: replacementStt, + vad: replacementVad, + tts: new FakeTTS(), + }); expect(agent.stt).toBe(replacementStt); } finally { await session.close().catch(() => {}); From 8b2ea3547edf7f0a20ed00daf3aa6b6f8867e705 Mon Sep 17 00:00:00 2001 From: Toubat Date: Fri, 17 Jul 2026 12:09:19 -0700 Subject: [PATCH 8/8] fix(voice): preserve agent model getter types Normalize explicit-null disable state at public getters while resolving activity fallbacks from raw internal fields. Co-authored-by: Cursor --- agents/src/voice/agent.ts | 16 +++++----- agents/src/voice/agent.type.test.ts | 29 +++++++++++++++++++ agents/src/voice/agent_activity.ts | 10 +++---- agents/src/voice/agent_update_options.test.ts | 11 ++++++- 4 files changed, 52 insertions(+), 14 deletions(-) create mode 100644 agents/src/voice/agent.type.test.ts diff --git a/agents/src/voice/agent.ts b/agents/src/voice/agent.ts index a34e1ec75..6c412257a 100644 --- a/agents/src/voice/agent.ts +++ b/agents/src/voice/agent.ts @@ -269,20 +269,20 @@ export class Agent { this._agentActivity = undefined; } - get vad(): VAD | null | undefined { - return this._vad; + get vad(): VAD | undefined { + return this._vad ?? undefined; } - get stt(): STT | null | undefined { - return this._stt; + get stt(): STT | undefined { + return this._stt ?? undefined; } - get llm(): LLM | RealtimeModel | null | undefined { - return this._llm; + get llm(): LLM | RealtimeModel | undefined { + return this._llm ?? undefined; } - get tts(): TTS | null | undefined { - return this._tts; + get tts(): TTS | undefined { + return this._tts ?? undefined; } get useTtsAlignedTranscript(): boolean | undefined { diff --git a/agents/src/voice/agent.type.test.ts b/agents/src/voice/agent.type.test.ts new file mode 100644 index 000000000..a14e78e25 --- /dev/null +++ b/agents/src/voice/agent.type.test.ts @@ -0,0 +1,29 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { describe, expectTypeOf, it } from 'vitest'; +import { Agent } from './agent.js'; + +describe('Agent model getters', () => { + it('preserves undefined-based narrowing for existing callers', () => { + const agent = new Agent({ instructions: 'test' }); + + expectTypeOf(agent.stt).not.toEqualTypeOf(); + expectTypeOf(agent.vad).not.toEqualTypeOf(); + expectTypeOf(agent.llm).not.toEqualTypeOf(); + expectTypeOf(agent.tts).not.toEqualTypeOf(); + + if (agent.stt !== undefined) { + agent.stt.stream(); + } + if (agent.vad !== undefined) { + agent.vad.stream(); + } + if (agent.llm !== undefined) { + agent.llm.label(); + } + if (agent.tts !== undefined) { + agent.tts.stream(); + } + }); +}); diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 873537c51..2351e1a87 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -759,7 +759,7 @@ export class AgentActivity implements RecognitionHooks { } get vad(): VAD | undefined { - return this.agent.vad !== undefined ? this.agent.vad ?? undefined : this.agentSession.vad; + return this.agent._vad !== undefined ? this.agent._vad ?? undefined : this.agentSession.vad; } /** @@ -768,14 +768,14 @@ export class AgentActivity implements RecognitionHooks { * session, even if the value happens to be the same silero model. */ get usingDefaultVad(): boolean { - if (this.agent.vad !== undefined) { + if (this.agent._vad !== undefined) { return false; } return this.agentSession._usingDefaultVad; } get stt(): STT | undefined { - return this.agent.stt !== undefined ? this.agent.stt ?? undefined : this.agentSession.stt; + return this.agent._stt !== undefined ? this.agent._stt ?? undefined : this.agentSession.stt; } private getSttProvider(): string | undefined { @@ -790,11 +790,11 @@ export class AgentActivity implements RecognitionHooks { } get llm(): LLM | RealtimeModel | undefined { - return this.agent.llm !== undefined ? this.agent.llm ?? undefined : this.agentSession.llm; + return this.agent._llm !== undefined ? this.agent._llm ?? undefined : this.agentSession.llm; } get tts(): TTS | undefined { - return this.agent.tts !== undefined ? this.agent.tts ?? undefined : this.agentSession.tts; + return this.agent._tts !== undefined ? this.agent._tts ?? undefined : this.agentSession.tts; } get tools(): ToolContext { diff --git a/agents/src/voice/agent_update_options.test.ts b/agents/src/voice/agent_update_options.test.ts index a58f90cd2..d4381aaf8 100644 --- a/agents/src/voice/agent_update_options.test.ts +++ b/agents/src/voice/agent_update_options.test.ts @@ -567,7 +567,8 @@ describe('Agent.updateOptions', () => { await agent.updateOptions({ stt: null }); - expect(agent.stt).toBeNull(); + expect(agent._stt).toBeNull(); + expect(agent.stt).toBeUndefined(); expect(activity.stt).toBeUndefined(); const recognition = (activity as unknown as { audioRecognition: unknown }).audioRecognition; expect((recognition as { sttPipeline: unknown }).sttPipeline).toBeUndefined(); @@ -599,6 +600,14 @@ describe('Agent.updateOptions', () => { await agent.updateOptions({ stt: null, vad: null, llm: null, tts: null }); + expect(agent._stt).toBeNull(); + expect(agent._vad).toBeNull(); + expect(agent._llm).toBeNull(); + expect(agent._tts).toBeNull(); + expect(agent.stt).toBeUndefined(); + expect(agent.vad).toBeUndefined(); + expect(agent.llm).toBeUndefined(); + expect(agent.tts).toBeUndefined(); expect(activity.stt).toBeUndefined(); expect(activity.vad).toBeUndefined(); expect(activity.llm).toBeUndefined();