diff --git a/.changeset/update-agent-models-runtime.md b/.changeset/update-agent-models-runtime.md new file mode 100644 index 000000000..dcf4f85eb --- /dev/null +++ b/.changeset/update-agent-models-runtime.md @@ -0,0 +1,6 @@ +--- +'@livekit/agents': patch +--- + +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.ts b/agents/src/voice/agent.ts index 05536663b..6c412257a 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; @@ -255,19 +270,19 @@ export class Agent { } get vad(): VAD | undefined { - return this._vad; + return this._vad ?? undefined; } get stt(): STT | undefined { - return this._stt; + return this._stt ?? undefined; } get llm(): LLM | RealtimeModel | undefined { - return this._llm; + return this._llm ?? undefined; } get tts(): TTS | undefined { - return this._tts; + return this._tts ?? undefined; } get useTtsAlignedTranscript(): boolean | undefined { @@ -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.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 75cb22ac0..2351e1a87 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; } /** @@ -767,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 || 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,230 @@ 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); + } + + const unlock = await this.lock.lock(); + try { + if (this.closeAbort.signal.aborted || this.agent._agentActivity !== this) { + 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; + } + + 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; + + 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) { + 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, this.usingDefaultVad); + } + 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, '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._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 (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); + } + + if (rollbackErrors.length > 0) { + this.logger.error( + { error, rollbackErrors }, + 'failed to completely roll back model update', + ); + } + throw error; + } + } finally { + unlock(); + } + } + 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..d4381aaf8 --- /dev/null +++ b/agents/src/voice/agent_update_options.test.ts @@ -0,0 +1,940 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import type { AudioFrame } from '@livekit/rtc-node'; +import { describe, expect, it, vi } from 'vitest'; +import type { voice } from '../index.js'; +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 type { AudioRecognition } from './audio_recognition.js'; +import { AgentSessionEventTypes } from './events.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 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'; + } + + override get provider(): string { + return 'new-provider'; + } +} + +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(); + } + + 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'); + } +} + +function deferred(): { + promise: Promise; + resolve: () => void; +} { + let resolve!: () => void; + const promise = new Promise((res) => { + resolve = res; + }); + return { promise, resolve }; +} + +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(); + 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('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('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('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'); + 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(); + 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('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', + 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(agent.stt).toBeUndefined(); + 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('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(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(); + 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('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, + 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); + } 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 oldVad = new FakeVAD(); + const oldTts = new FakeTTS(); + const replacementStt = new ContextSTT('replacement-stt'); + 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, + 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, + vad: replacementVad, + 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' }); + 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..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; @@ -611,13 +612,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,16 +632,68 @@ 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, usingDefaultVad: boolean): Promise { + this.checkVadSilenceRequirement(undefined, 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; + + if (!this.closed && this.vad !== undefined) { + this.startVadTask(this.vad); + } + } finally { + unlock(); + } + } + async start(options?: { sttPipeline?: STTPipeline; turnDetectorStream?: BaseStreamingTurnDetectorStream; }) { 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), @@ -2144,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/audio_recognition_vad_reset.test.ts b/agents/src/voice/audio_recognition_vad_reset.test.ts index 8ec404fb0..9c32cb572 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 { 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); + }); }); 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'; 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