From 466ba88d378eac7840f53e7f8db872b9488f5d53 Mon Sep 17 00:00:00 2001 From: "rosetta-livekit-bot[bot]" <282703043+rosetta-livekit-bot[bot]@users.noreply.github.com> Date: Sat, 18 Jul 2026 00:54:56 +0000 Subject: [PATCH 1/3] Support AssemblyAI inference context carryover --- .changeset/fresh-pandas-listen.md | 5 ++ agents/src/inference/stt.test.ts | 125 +++++++++++++++++++++++++++++- agents/src/inference/stt.ts | 110 +++++++++++++++++++++----- 3 files changed, 217 insertions(+), 23 deletions(-) create mode 100644 .changeset/fresh-pandas-listen.md diff --git a/.changeset/fresh-pandas-listen.md b/.changeset/fresh-pandas-listen.md new file mode 100644 index 000000000..51e0e558c --- /dev/null +++ b/.changeset/fresh-pandas-listen.md @@ -0,0 +1,5 @@ +--- +'@livekit/agents': patch +--- + +Support AssemblyAI inference STT agent context carryover. diff --git a/agents/src/inference/stt.test.ts b/agents/src/inference/stt.test.ts index 65df44750..d915bc4ee 100644 --- a/agents/src/inference/stt.test.ts +++ b/agents/src/inference/stt.test.ts @@ -1,12 +1,14 @@ // SPDX-FileCopyrightText: 2025 LiveKit, Inc. // // SPDX-License-Identifier: Apache-2.0 -import { beforeAll, describe, expect, it } from 'vitest'; +import { afterEach, beforeAll, describe, expect, it, vi } from 'vitest'; import * as agents from '../index.js'; import { normalizeLanguage } from '../language.js'; -import { initializeLogger } from '../log.js'; +import { AgentHandoffItem, ChatMessage } from '../llm/index.js'; +import { initializeLogger, log } from '../log.js'; import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js'; import { VAD, type VADStream } from '../vad.js'; +import { createConversationItemAddedEvent } from '../voice/events.js'; import { STT, type STTFallbackModel, @@ -21,6 +23,10 @@ beforeAll(() => { initializeLogger({ level: 'silent', pretty: false }); }); +afterEach(() => { + vi.restoreAllMocks(); +}); + /** Helper to create STT with required credentials. */ function makeStt(overrides: Record = {}) { const defaults = { @@ -32,6 +38,16 @@ function makeStt(overrides: Record = {}) { return new STT({ ...defaults, ...overrides }); } +function makeAssemblyStt(overrides: Record = {}) { + return makeStt({ model: 'assemblyai/universal-3-5-pro', ...overrides }); +} + +function assistantItemEvent(text: string) { + return createConversationItemAddedEvent( + ChatMessage.create({ role: 'assistant', content: [text] }), + ); +} + describe('parseSTTModelString', () => { it('simple model without language', () => { const [model, language] = parseSTTModelString('deepgram'); @@ -329,6 +345,111 @@ describe('STT diarization capabilities', () => { }); }); +describe('STT agent_context carryover', () => { + it('agentContextCarryover defaults to enabled on AssemblyAI U3 Pro family models', () => { + for (const model of ['assemblyai/u3-rt-pro', 'assemblyai/universal-3-5-pro'] as const) { + const stt = makeAssemblyStt({ model }); + expect(stt.capabilities.chatContext).toBe(true); + } + }); + + it('agentContextCarryover defaults off for unsupported models without warning', () => { + const warnSpy = vi.spyOn(log(), 'warn'); + const stt = makeAssemblyStt({ model: 'assemblyai/universal-streaming' }); + + expect(stt.capabilities.chatContext).toBe(false); + expect(warnSpy).not.toHaveBeenCalled(); + }); + + it('agentContextCarryover is off for non-AssemblyAI models', () => { + const stt = makeAssemblyStt({ model: 'deepgram/nova-3' }); + expect(stt.capabilities.chatContext).toBe(false); + }); + + it('explicit true on unsupported model warns and is ignored', () => { + const warnSpy = vi.spyOn(log(), 'warn'); + const stt = makeAssemblyStt({ + model: 'assemblyai/universal-streaming', + agentContextCarryover: true, + }); + + expect(stt.capabilities.chatContext).toBe(false); + expect(warnSpy).toHaveBeenCalledWith( + { model: 'assemblyai/universal-streaming' }, + 'agentContextCarryover is enabled but model does not support it; ignoring', + ); + }); + + it('explicit false disables carryover on a supported model', () => { + const stt = makeAssemblyStt({ agentContextCarryover: false }); + expect(stt.capabilities.chatContext).toBe(false); + }); + + it('previous_context_n_turns=0 disables the default carryover', () => { + const stt = makeAssemblyStt({ modelOptions: { previous_context_n_turns: 0 } }); + expect(stt.capabilities.chatContext).toBe(false); + }); + + it('explicit true wins over previous_context_n_turns=0', () => { + const stt = makeAssemblyStt({ + modelOptions: { previous_context_n_turns: 0 }, + agentContextCarryover: true, + }); + expect(stt.capabilities.chatContext).toBe(true); + }); + + it('forwards short assistant replies verbatim', () => { + const stt = makeAssemblyStt(); + stt._pushConversationItem(assistantItemEvent('Your room is booked for Tuesday.')); + expect(stt['opts'].modelOptions).toHaveProperty( + 'agent_context', + 'Your room is booked for Tuesday.', + ); + }); + + it('truncates oversize replies keeping the tail', () => { + const text = 'a'.repeat(2000) + 'b'.repeat(1750); + const stt = makeAssemblyStt(); + stt._pushConversationItem(assistantItemEvent(text)); + expect(stt['opts'].modelOptions).toHaveProperty('agent_context', 'b'.repeat(1750)); + }); + + it('ignores non-assistant items', () => { + const stt = makeAssemblyStt(); + + stt._pushConversationItem( + createConversationItemAddedEvent(ChatMessage.create({ role: 'user', content: ['hi there'] })), + ); + expect(stt['opts'].modelOptions).not.toHaveProperty('agent_context'); + + stt._pushConversationItem( + createConversationItemAddedEvent(ChatMessage.create({ role: 'assistant', content: [] })), + ); + expect(stt['opts'].modelOptions).not.toHaveProperty('agent_context'); + }); + + it('ignores agent handoff items', () => { + const stt = makeAssemblyStt(); + stt._pushConversationItem( + createConversationItemAddedEvent(AgentHandoffItem.create({ newAgentId: 'agent-2' })), + ); + expect(stt['opts'].modelOptions).not.toHaveProperty('agent_context'); + }); + + it('preserves explicit agent_context and later overwrites it with carryover', () => { + const stt = makeAssemblyStt({ + modelOptions: { agent_context: 'The agent asked for a booking date.' }, + }); + expect(stt['opts'].modelOptions).toHaveProperty( + 'agent_context', + 'The agent asked for a booking date.', + ); + + stt._pushConversationItem(assistantItemEvent('And your zip code?')); + expect(stt['opts'].modelOptions).toHaveProperty('agent_context', 'And your zip code?'); + }); +}); + describe('STT session keyterms', () => { it('updateOptions does not bake session keyterms into the user baseline', () => { const stt = makeStt({ model: 'deepgram/nova-3' }); diff --git a/agents/src/inference/stt.ts b/agents/src/inference/stt.ts index 3cfc8e852..c8e5a0e72 100644 --- a/agents/src/inference/stt.ts +++ b/agents/src/inference/stt.ts @@ -7,6 +7,7 @@ import type { WebSocket } from 'ws'; import { APIError, APIStatusError } from '../_exceptions.js'; import { AudioByteStream } from '../audio.js'; import { type LanguageCode, areLanguagesEquivalent, normalizeLanguage } from '../language.js'; +import { ChatMessage } from '../llm/index.js'; import { log } from '../log.js'; import { createStreamChannel } from '../stream/stream_channel.js'; import { @@ -19,6 +20,7 @@ import { import { type APIConnectOptions, DEFAULT_API_CONNECT_OPTIONS } from '../types.js'; import { type AudioBuffer, Event, Task, cancelAndWait, shortuuid, waitForAbort } from '../utils.js'; import { type VAD, VADEventType, type VADStream } from '../vad.js'; +import type { ConversationItemAddedEvent } from '../voice/events.js'; import { type TimedString, createTimedString } from '../voice/io.js'; import { type SttServerEvent, @@ -120,8 +122,10 @@ export interface AssemblyAIOptions { keyterms_prompt?: string[]; /** Enable speaker diarization. Default: false. */ speaker_labels?: boolean; - /** Context to bias recognition. Only supported with u3-rt-pro. Max 1500 chars. */ + /** Context to bias recognition. Only supported with u3-rt-pro. Max 1750 chars. */ agent_context?: string; + /** Prior turns carried as context; 0 disables carryover. Only supported with u3-rt-pro. */ + previous_context_n_turns?: number; /** Isolate the primary voice. Only supported with u3-rt-pro. */ voice_focus?: 'near-field' | 'far-field'; /** Background suppression strength. Only supported with u3-rt-pro. */ @@ -270,6 +274,28 @@ function keytermsExtraForModel( return { [key]: [...new Set([...existing, ...sessionKeyterms])] }; } +const ASSEMBLYAI_CARRYOVER_MODELS = [ + 'assemblyai/u3-rt-pro', + 'assemblyai/universal-3-5-pro', +] as const; + +const ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS = 1750; + +function supportsAgentContextCarryover(model: string | undefined): boolean { + return ASSEMBLYAI_CARRYOVER_MODELS.includes( + model as (typeof ASSEMBLYAI_CARRYOVER_MODELS)[number], + ); +} + +function agentContextCarryoverEnabled(opts: { + model: string | undefined; + modelOptions: Record; + agentContextCarryover?: boolean; +}): boolean { + if (!supportsAgentContextCarryover(opts.model)) return false; + return opts.agentContextCarryover ?? opts.modelOptions.previous_context_n_turns !== 0; +} + type _STTModels = | DeepgramModels | DeepgramFluxModels @@ -370,6 +396,7 @@ export interface InferenceSTTOptions { apiKey: string; apiSecret: string; modelOptions: STTOptions; + agentContextCarryover?: boolean; fallback?: STTFallbackModel[]; connOptions?: APIConnectOptions; } @@ -408,23 +435,49 @@ export class STT extends BaseSTT { apiSecret?: string; modelOptions?: STTOptions; fallback?: STTFallbackModelType | STTFallbackModelType[]; + agentContextCarryover?: boolean; connOptions?: APIConnectOptions; vad?: VAD; }) { const modelOptions = (opts?.modelOptions ?? {}) as STTOptions; + // Parse language from model string if provided: "provider/model:language". + let nextModel = opts?.model; + let nextLanguage = opts?.language; + let hasLanguageConflict = false; + if (typeof nextModel === 'string') { + const [parsedModel, parsedLanguage] = parseSTTModelString(nextModel); + if (parsedLanguage !== undefined) { + hasLanguageConflict = + !!nextLanguage && !areLanguagesEquivalent(nextLanguage, parsedLanguage); + if (!hasLanguageConflict) { + nextLanguage = parsedLanguage as STTLanguages; + } + nextModel = parsedModel as TModel; + } + } + const carryoverEnabled = agentContextCarryoverEnabled({ + model: typeof nextModel === 'string' ? nextModel : undefined, + modelOptions: modelOptions as Record, + agentContextCarryover: opts?.agentContextCarryover, + }); super({ streaming: true, interimResults: true, alignedTranscript: 'word', diarization: diarizationEnabled(modelOptions as Record), keyterms: - keytermsExtraForModel(typeof opts?.model === 'string' ? opts.model : undefined) !== - undefined, + keytermsExtraForModel(typeof nextModel === 'string' ? nextModel : undefined) !== undefined, + chatContext: carryoverEnabled, }); + if (opts?.agentContextCarryover && !supportsAgentContextCarryover(nextModel)) { + log().warn( + { model: nextModel }, + 'agentContextCarryover is enabled but model does not support it; ignoring', + ); + } + const { - model, - language, baseURL, encoding = DEFAULT_ENCODING, sampleRate = DEFAULT_SAMPLE_RATE, @@ -447,22 +500,11 @@ export class STT extends BaseSTT { throw new Error('apiSecret is required: pass apiSecret or set LIVEKIT_API_SECRET'); } - // Parse language from model string if provided: "provider/model:language" - let nextModel = model; - let nextLanguage = language; - if (typeof nextModel === 'string') { - const [parsedModel, parsedLanguage] = parseSTTModelString(nextModel); - if (parsedLanguage !== undefined) { - if (nextLanguage && !areLanguagesEquivalent(nextLanguage, parsedLanguage)) { - this.#logger.warn( - '`language` is provided via both argument and model, using the one from the argument', - { language: nextLanguage, model: nextModel }, - ); - } else { - nextLanguage = parsedLanguage as STTLanguages; - } - nextModel = parsedModel as TModel; - } + if (hasLanguageConflict) { + this.#logger.warn( + '`language` is provided via both argument and model, using the one from the argument', + { language: nextLanguage, model: opts?.model }, + ); } const normalizedFallback = fallback ? normalizeSTTFallback(fallback) : undefined; this.vad = resolveVADForModel(nextModel, vad); @@ -476,6 +518,7 @@ export class STT extends BaseSTT { apiKey: lkApiKey, apiSecret: lkApiSecret, modelOptions, + agentContextCarryover: opts?.agentContextCarryover, fallback: normalizedFallback, connOptions: connOptions ?? DEFAULT_API_CONNECT_OPTIONS, }; @@ -531,12 +574,22 @@ export class STT extends BaseSTT { this._vadPromise = undefined; this.updateCapabilities({ keyterms: keytermsExtraForModel(this.opts.model) !== undefined, + chatContext: agentContextCarryoverEnabled({ + model: this.opts.model, + modelOptions: this.opts.modelOptions as Record, + agentContextCarryover: this.opts.agentContextCarryover, + }), }); } if (nextOpts.modelOptions) { this.updateCapabilities({ diarization: diarizationEnabled(this.opts.modelOptions as Record), + chatContext: agentContextCarryoverEnabled({ + model: this.opts.model, + modelOptions: this.opts.modelOptions as Record, + agentContextCarryover: this.opts.agentContextCarryover, + }), }); // re-apply the active session keyterms on top of the update sent to live streams, // so a user extra update doesn't drop them. `this.opts.modelOptions` must stay a @@ -587,6 +640,21 @@ export class STT extends BaseSTT { } } + override _pushConversationItem(ev: ConversationItemAddedEvent): void { + const chatItem = ev.item; + if (chatItem instanceof ChatMessage && chatItem.role === 'assistant' && chatItem.textContent) { + let text = chatItem.textContent; + if (text.length > ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS) { + this.#logger.debug( + { fromChars: text.length, toChars: ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS }, + 'truncating agent_context carryover', + ); + text = text.slice(-ASSEMBLYAI_MAX_AGENT_CONTEXT_CHARS); + } + this.updateOptions({ modelOptions: { agent_context: text } as STTOptions }); + } + } + stream(options?: { language?: STTLanguages | string; connOptions?: APIConnectOptions; From ee9e759c3dfe931566c7b3ba3de57e3590693184 Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 20 Jul 2026 13:55:25 -0700 Subject: [PATCH 2/3] Fix STT conversation context lifecycle Guard provider updates by live capabilities and propagate model capability changes so sessions and fallback adapters forward context only while supported. Co-authored-by: Cursor --- agents/src/inference/stt.test.ts | 39 +++++++++++++- agents/src/inference/stt.ts | 4 ++ agents/src/stt/fallback_adapter.test.ts | 54 +++++++++++++++++++ agents/src/stt/fallback_adapter.ts | 16 +++++- agents/src/stt/stt.ts | 16 +++++- agents/src/voice/agent_activity.ts | 22 +++++--- .../voice/agent_activity_stt_context.test.ts | 45 ++++++++++++++++ 7 files changed, 186 insertions(+), 10 deletions(-) create mode 100644 agents/src/voice/agent_activity_stt_context.test.ts diff --git a/agents/src/inference/stt.test.ts b/agents/src/inference/stt.test.ts index d915bc4ee..7a17a32d6 100644 --- a/agents/src/inference/stt.test.ts +++ b/agents/src/inference/stt.test.ts @@ -381,8 +381,15 @@ describe('STT agent_context carryover', () => { }); it('explicit false disables carryover on a supported model', () => { - const stt = makeAssemblyStt({ agentContextCarryover: false }); + const stt = makeAssemblyStt({ + agentContextCarryover: false, + modelOptions: { agent_context: 'keep me' }, + }); expect(stt.capabilities.chatContext).toBe(false); + + stt._pushConversationItem(assistantItemEvent('do not forward me')); + + expect(stt['opts'].modelOptions).toHaveProperty('agent_context', 'keep me'); }); it('previous_context_n_turns=0 disables the default carryover', () => { @@ -400,11 +407,19 @@ describe('STT agent_context carryover', () => { it('forwards short assistant replies verbatim', () => { const stt = makeAssemblyStt(); + const stream = stt.stream(); + stt._pushConversationItem(assistantItemEvent('Your room is booked for Tuesday.')); + expect(stt['opts'].modelOptions).toHaveProperty( 'agent_context', 'Your room is booked for Tuesday.', ); + expect(stream['opts'].modelOptions).toHaveProperty( + 'agent_context', + 'Your room is booked for Tuesday.', + ); + stream.close(); }); it('truncates oversize replies keeping the tail', () => { @@ -448,6 +463,28 @@ describe('STT agent_context carryover', () => { stt._pushConversationItem(assistantItemEvent('And your zip code?')); expect(stt['opts'].modelOptions).toHaveProperty('agent_context', 'And your zip code?'); }); + + it('starts forwarding after changing from an unsupported to a supported model', () => { + const stt = makeAssemblyStt({ model: 'assemblyai/universal-streaming' }); + + stt._pushConversationItem(assistantItemEvent('ignored before transition')); + expect(stt['opts'].modelOptions).not.toHaveProperty('agent_context'); + + stt.updateOptions({ model: 'assemblyai/universal-3-5-pro' }); + stt._pushConversationItem(assistantItemEvent('forwarded after transition')); + + expect(stt['opts'].modelOptions).toHaveProperty('agent_context', 'forwarded after transition'); + }); + + it('stops forwarding after changing from a supported to an unsupported model', () => { + const stt = makeAssemblyStt(); + stt._pushConversationItem(assistantItemEvent('last supported context')); + + stt.updateOptions({ model: 'assemblyai/universal-streaming' }); + stt._pushConversationItem(assistantItemEvent('ignored after transition')); + + expect(stt['opts'].modelOptions).toHaveProperty('agent_context', 'last supported context'); + }); }); describe('STT session keyterms', () => { diff --git a/agents/src/inference/stt.ts b/agents/src/inference/stt.ts index c8e5a0e72..80b7cbbdd 100644 --- a/agents/src/inference/stt.ts +++ b/agents/src/inference/stt.ts @@ -641,6 +641,10 @@ export class STT extends BaseSTT { } override _pushConversationItem(ev: ConversationItemAddedEvent): void { + if (!this.capabilities.chatContext) { + return; + } + const chatItem = ev.item; if (chatItem instanceof ChatMessage && chatItem.role === 'assistant' && chatItem.textContent) { let text = chatItem.textContent; diff --git a/agents/src/stt/fallback_adapter.test.ts b/agents/src/stt/fallback_adapter.test.ts index 5773b96b6..6f8a76914 100644 --- a/agents/src/stt/fallback_adapter.test.ts +++ b/agents/src/stt/fallback_adapter.test.ts @@ -4,7 +4,12 @@ import type { EventEmitter } from 'node:events'; import { beforeAll, describe, expect, it, vi } from 'vitest'; import { APIConnectionError, APIError } from '../_exceptions.js'; +import { ChatMessage } from '../llm/index.js'; import { initializeLogger } from '../log.js'; +import { + type ConversationItemAddedEvent, + createConversationItemAddedEvent, +} from '../voice/events.js'; import { FallbackAdapter } from './fallback_adapter.js'; import type { STT, SpeechEvent } from './stt.js'; import { FakeSTT, RecognizeSentinel, emptyAudioFrame } from './testing/fake_stt.js'; @@ -50,6 +55,55 @@ describe('FallbackAdapter', () => { expect(adapter.capabilities.streaming).toBe(true); }); + it('forwards conversation context only to children that support it', () => { + class ContextRecordingSTT extends FakeSTT { + readonly conversationItems: ConversationItemAddedEvent[] = []; + + constructor(label: string, supportsChatContext: boolean) { + super({ label }); + this.updateCapabilities({ chatContext: supportsChatContext }); + } + + override _pushConversationItem(ev: ConversationItemAddedEvent): void { + this.conversationItems.push(ev); + } + } + + const supported = new ContextRecordingSTT('supported', true); + const unsupported = new ContextRecordingSTT('unsupported', false); + const adapter = new FallbackAdapter({ sttInstances: [supported, unsupported] }); + const event = createConversationItemAddedEvent( + ChatMessage.create({ role: 'assistant', content: ['hello'] }), + ); + + adapter._pushConversationItem(event); + + expect(supported.conversationItems).toEqual([event]); + expect(unsupported.conversationItems).toEqual([]); + }); + + it('tracks dynamic child conversation-context capabilities and removes listeners on close', async () => { + class DynamicContextSTT extends FakeSTT { + setChatContext(supported: boolean): void { + this.updateCapabilities({ chatContext: supported }); + } + } + + const child = new DynamicContextSTT(); + const adapter = new FallbackAdapter({ sttInstances: [child] }); + expect(adapter.capabilities.chatContext).toBe(false); + expect(child.listenerCount('capabilities_changed')).toBe(1); + + child.setChatContext(true); + expect(adapter.capabilities.chatContext).toBe(true); + + child.setChatContext(false); + expect(adapter.capabilities.chatContext).toBe(false); + + await adapter.close(); + expect(child.listenerCount('capabilities_changed')).toBe(0); + }); + it('_recognize falls through to the next instance on error', async () => { const primary = new FakeSTT({ label: 'primary', diff --git a/agents/src/stt/fallback_adapter.ts b/agents/src/stt/fallback_adapter.ts index 002a22470..c221cb257 100644 --- a/agents/src/stt/fallback_adapter.ts +++ b/agents/src/stt/fallback_adapter.ts @@ -98,6 +98,7 @@ export class FallbackAdapter extends STT { private _status: STTStatus[] = []; private _logger = log(); private _metricsForwarders = new Map void>(); + private _capabilitiesForwarders = new Map void>(); // Last child that produced output or returned a recognize result. Surfaced // via the dynamic label/model/provider getters so OTel attributes like // `gen_ai.request.model` on `user_turn` (refreshed on every STT event by @@ -197,9 +198,10 @@ export class FallbackAdapter extends STT { } override _pushConversationItem(ev: ConversationItemAddedEvent): void { - // forward to every underlying STT; unsupported ones warn-and-skip internally for (const sttInstance of this.sttInstances) { - sttInstance._pushConversationItem(ev); + if (sttInstance.capabilities.chatContext) { + sttInstance._pushConversationItem(ev); + } } } @@ -213,8 +215,15 @@ export class FallbackAdapter extends STT { // SpeechStream.mainTask emits that on this STT instance naturally. for (const s of this.sttInstances) { const metricsForwarder = (metrics: STTMetrics) => this.emit('metrics_collected', metrics); + const capabilitiesForwarder = () => { + this.updateCapabilities({ + chatContext: this.sttInstances.some((stt) => !!stt.capabilities.chatContext), + }); + }; this._metricsForwarders.set(s, metricsForwarder); + this._capabilitiesForwarders.set(s, capabilitiesForwarder); s.on('metrics_collected', metricsForwarder); + s.on('capabilities_changed', capabilitiesForwarder); } } @@ -325,8 +334,11 @@ export class FallbackAdapter extends STT { for (const s of this.sttInstances) { const m = this._metricsForwarders.get(s); if (m) s.off('metrics_collected' as keyof STTCallbacks, m); + const c = this._capabilitiesForwarders.get(s); + if (c) s.off('capabilities_changed', c); } this._metricsForwarders.clear(); + this._capabilitiesForwarders.clear(); } } diff --git a/agents/src/stt/stt.ts b/agents/src/stt/stt.ts index 923517964..b8d7b45a6 100644 --- a/agents/src/stt/stt.ts +++ b/agents/src/stt/stt.ts @@ -154,6 +154,7 @@ export interface STTError { export type STTCallbacks = { ['metrics_collected']: (metrics: STTMetrics) => void; ['error']: (error: STTError) => void; + ['capabilities_changed']: (capabilities: STTCapabilities) => void; }; /** @@ -180,7 +181,20 @@ export abstract class STT extends (EventEmitter as new () => TypedEmitter): void { - this.#capabilities = { ...this.#capabilities, ...caps }; + const next = { ...this.#capabilities, ...caps }; + const changed = + next.streaming !== this.#capabilities.streaming || + next.interimResults !== this.#capabilities.interimResults || + next.alignedTranscript !== this.#capabilities.alignedTranscript || + next.diarization !== this.#capabilities.diarization || + next.keyterms !== this.#capabilities.keyterms || + next.chatContext !== this.#capabilities.chatContext; + if (!changed) { + return; + } + + this.#capabilities = next; + this.emit('capabilities_changed', next); } /** diff --git a/agents/src/voice/agent_activity.ts b/agents/src/voice/agent_activity.ts index 13cfd5bb3..d92ac9943 100644 --- a/agents/src/voice/agent_activity.ts +++ b/agents/src/voice/agent_activity.ts @@ -621,12 +621,8 @@ export class AgentActivity implements RecognitionHooks { // forward conversation turns to STTs that consume context natively; gated by the // STT's own capability (toggled via the STT's args). stateless and activity-scoped, // so it lives here rather than in the detector. - if (this.stt.capabilities.chatContext) { - this.agentSession.on( - AgentSessionEventTypes.ConversationItemAdded, - this.pushConversationItemToStt, - ); - } + this.stt.on('capabilities_changed', this.syncConversationItemForwarding); + this.syncConversationItemForwarding(); } // Bundled-default VAD is treated as absent when the RealtimeModel does @@ -1216,6 +1212,19 @@ export class AgentActivity implements RecognitionHooks { } }; + private syncConversationItemForwarding = (): void => { + this.agentSession.off( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + if (this.stt instanceof STT && this.stt.capabilities.chatContext) { + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.pushConversationItemToStt, + ); + } + }; + private onError(ev: RealtimeModelError | STTError | TTSError | LLMError): void { if (ev.type === 'realtime_model_error') { const errorEvent = createErrorEvent(ev, this.llm); @@ -4545,6 +4554,7 @@ export class AgentActivity implements RecognitionHooks { if (this.stt instanceof STT) { this.stt.off('metrics_collected', this.onMetricsCollected); this.stt.off('error', this.onModelError); + this.stt.off('capabilities_changed', this.syncConversationItemForwarding); this.agentSession.off( AgentSessionEventTypes.ConversationItemAdded, this.pushConversationItemToStt, diff --git a/agents/src/voice/agent_activity_stt_context.test.ts b/agents/src/voice/agent_activity_stt_context.test.ts new file mode 100644 index 000000000..8ff780d8f --- /dev/null +++ b/agents/src/voice/agent_activity_stt_context.test.ts @@ -0,0 +1,45 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { describe, expect, it, vi } from 'vitest'; +import { STT as InferenceSTT } from '../inference/stt.js'; +import { ChatMessage } from '../llm/index.js'; +import { initializeLogger } from '../log.js'; +import { Agent } from './agent.js'; +import { AgentSession } from './agent_session.js'; +import { AgentSessionEventTypes, createConversationItemAddedEvent } from './events.js'; + +describe('AgentActivity STT conversation-context lifecycle', () => { + initializeLogger({ pretty: false, level: 'silent' }); + + it('tracks model capability transitions and removes its listener on close', async () => { + const stt = new InferenceSTT({ + model: 'assemblyai/universal-streaming', + apiKey: 'test-key', + apiSecret: 'test-secret', + baseURL: 'https://example.livekit.cloud', + }); + const pushSpy = vi.spyOn(stt, '_pushConversationItem'); + const session = new AgentSession({ stt }); + const event = createConversationItemAddedEvent( + ChatMessage.create({ role: 'assistant', content: ['hello'] }), + ); + + await session.start({ agent: new Agent({ instructions: 'test' }) }); + expect(stt.listenerCount('capabilities_changed')).toBe(1); + + session.emit(AgentSessionEventTypes.ConversationItemAdded, event); + expect(pushSpy).not.toHaveBeenCalled(); + + stt.updateOptions({ model: 'assemblyai/universal-3-5-pro' }); + session.emit(AgentSessionEventTypes.ConversationItemAdded, event); + expect(pushSpy).toHaveBeenCalledTimes(1); + + stt.updateOptions({ model: 'assemblyai/universal-streaming' }); + session.emit(AgentSessionEventTypes.ConversationItemAdded, event); + expect(pushSpy).toHaveBeenCalledTimes(1); + + await session.close(); + expect(stt.listenerCount('capabilities_changed')).toBe(0); + }); +}); From 0eebde463c7ab84705ff644327b77bf566de5758 Mon Sep 17 00:00:00 2001 From: Toubat Date: Mon, 20 Jul 2026 14:01:17 -0700 Subject: [PATCH 3/3] Test STT carryover wire lifecycle Cover the emitted session.update payload and the runtime previous_context_n_turns opt-out required by the Rosetta stack contract. Co-authored-by: Cursor --- agents/src/inference/stt.test.ts | 20 ++++++++++++++----- .../voice/agent_activity_stt_context.test.ts | 5 +++-- 2 files changed, 18 insertions(+), 7 deletions(-) diff --git a/agents/src/inference/stt.test.ts b/agents/src/inference/stt.test.ts index 7a17a32d6..554b3b02c 100644 --- a/agents/src/inference/stt.test.ts +++ b/agents/src/inference/stt.test.ts @@ -405,9 +405,14 @@ describe('STT agent_context carryover', () => { expect(stt.capabilities.chatContext).toBe(true); }); - it('forwards short assistant replies verbatim', () => { + it('emits active carryover shaping in the session.update wire payload', () => { const stt = makeAssemblyStt(); const stream = stt.stream(); + const sent: string[] = []; + Reflect.set(stream, 'activeWs', { + readyState: 1, + send: (payload: string) => sent.push(payload), + }); stt._pushConversationItem(assistantItemEvent('Your room is booked for Tuesday.')); @@ -415,10 +420,15 @@ describe('STT agent_context carryover', () => { 'agent_context', 'Your room is booked for Tuesday.', ); - expect(stream['opts'].modelOptions).toHaveProperty( - 'agent_context', - 'Your room is booked for Tuesday.', - ); + expect(sent).toHaveLength(1); + expect(JSON.parse(sent[0]!)).toMatchObject({ + type: 'session.update', + settings: { + extra: { + agent_context: 'Your room is booked for Tuesday.', + }, + }, + }); stream.close(); }); diff --git a/agents/src/voice/agent_activity_stt_context.test.ts b/agents/src/voice/agent_activity_stt_context.test.ts index 8ff780d8f..82c156d25 100644 --- a/agents/src/voice/agent_activity_stt_context.test.ts +++ b/agents/src/voice/agent_activity_stt_context.test.ts @@ -12,7 +12,7 @@ import { AgentSessionEventTypes, createConversationItemAddedEvent } from './even describe('AgentActivity STT conversation-context lifecycle', () => { initializeLogger({ pretty: false, level: 'silent' }); - it('tracks model capability transitions and removes its listener on close', async () => { + it('stops forwarding when previous_context_n_turns disables carryover', async () => { const stt = new InferenceSTT({ model: 'assemblyai/universal-streaming', apiKey: 'test-key', @@ -35,7 +35,8 @@ describe('AgentActivity STT conversation-context lifecycle', () => { session.emit(AgentSessionEventTypes.ConversationItemAdded, event); expect(pushSpy).toHaveBeenCalledTimes(1); - stt.updateOptions({ model: 'assemblyai/universal-streaming' }); + stt.updateOptions({ modelOptions: { previous_context_n_turns: 0 } }); + expect(stt.capabilities.chatContext).toBe(false); session.emit(AgentSessionEventTypes.ConversationItemAdded, event); expect(pushSpy).toHaveBeenCalledTimes(1);