From 14033b4bcd3fe82a7f66e53a8259601967e1d108 Mon Sep 17 00:00:00 2001 From: Tina Nguyen Date: Thu, 16 Jul 2026 01:45:29 -0400 Subject: [PATCH 1/2] rm --- .changeset/remove-native-transcript-sync.md | 6 ++ agents/src/llm/realtime.ts | 2 - agents/src/voice/room_io/room_io.ts | 29 +------- .../src/voice/transcription/synchronizer.ts | 67 +++++-------------- plugins/phonic/src/realtime/realtime_model.ts | 1 - 5 files changed, 22 insertions(+), 83 deletions(-) create mode 100644 .changeset/remove-native-transcript-sync.md diff --git a/.changeset/remove-native-transcript-sync.md b/.changeset/remove-native-transcript-sync.md new file mode 100644 index 000000000..b1671ca75 --- /dev/null +++ b/.changeset/remove-native-transcript-sync.md @@ -0,0 +1,6 @@ +--- +'@livekit/agents-plugin-phonic': patch +'@livekit/agents': patch +--- + +Remove the unused `nativeTranscriptSync` realtime model capability. No model relies on native transcript synchronization anymore (Phonic switched to `stream_ahead_of_real_time` mode), so the transcription synchronizer is always enabled. diff --git a/agents/src/llm/realtime.ts b/agents/src/llm/realtime.ts index 1aef68d2b..cfc6cc170 100644 --- a/agents/src/llm/realtime.ts +++ b/agents/src/llm/realtime.ts @@ -64,8 +64,6 @@ export interface RealtimeCapabilities { midSessionToolsUpdate?: boolean; /** Whether the tool and tool choice can be specified per response. */ perResponseToolChoice?: boolean; - /** Whether the model can synchronize generated transcript timing natively. */ - nativeTranscriptSync?: boolean; } export interface InputTranscriptionCompleted { diff --git a/agents/src/voice/room_io/room_io.ts b/agents/src/voice/room_io/room_io.ts index ed5315bf7..8d2e39055 100644 --- a/agents/src/voice/room_io/room_io.ts +++ b/agents/src/voice/room_io/room_io.ts @@ -19,7 +19,6 @@ import { import type { WritableStreamDefaultWriter } from 'node:stream/web'; import { ATTRIBUTE_PUBLISH_ON_BEHALF, TOPIC_CHAT } from '../../constants.js'; import { type JobContext, getJobContext } from '../../job.js'; -import { RealtimeModel } from '../../llm/index.js'; import { log } from '../../log.js'; import { IdentityTransform } from '../../stream/identity_transform.js'; import { DEFAULT_API_CONNECT_OPTIONS } from '../../types.js'; @@ -29,15 +28,11 @@ import { AgentSessionEventTypes, type AgentStateChangedEvent, CloseReason, - type ConversationItemAddedEvent, type UserInputTranscribedEvent, } from '../events.js'; import type { AudioOutput, TextOutput } from '../io.js'; import type { TextInputCallback } from '../remote_session.js'; -import { - TranscriptionSynchronizer, - defaultTextSyncOptions, -} from '../transcription/synchronizer.js'; +import { TranscriptionSynchronizer } from '../transcription/synchronizer.js'; import { ParticipantAudioInputStream } from './_input.js'; import { ParalellTextOutput, @@ -290,16 +285,6 @@ export class RoomIO { }); }; - private onConversationItemAdded = (ev: ConversationItemAddedEvent) => { - if (ev.item.type !== 'agent_handoff' || !this.transcriptionSynchronizer) { - return; - } - const sessionLlm = this.agentSession.currentAgent?.llm ?? this.agentSession.llm; - const nativeTranscriptSync = - sessionLlm instanceof RealtimeModel && !!sessionLlm.capabilities.nativeTranscriptSync; - this.transcriptionSynchronizer.enabled = !nativeTranscriptSync; - }; - private onAgentSessionClose = () => { if (!this.inputOptions.deleteRoomOnClose || this.deleteRoomTask) { return; @@ -550,13 +535,9 @@ export class RoomIO { // TODO(AJS-176): check for agent output const audioOutput = this.participantAudioOutput; if (this.outputOptions.syncTranscription && audioOutput) { - const sessionLlm = this.agentSession.currentAgent?.llm ?? this.agentSession.llm; - const nativeTranscriptSync = - sessionLlm instanceof RealtimeModel && !!sessionLlm.capabilities.nativeTranscriptSync; this.transcriptionSynchronizer = new TranscriptionSynchronizer( audioOutput, this.agentTranscriptOutput, - { ...defaultTextSyncOptions, enabled: !nativeTranscriptSync }, ); } } @@ -585,10 +566,6 @@ export class RoomIO { this.agentSession.on(AgentSessionEventTypes.AgentStateChanged, this.onAgentStateChanged); this.agentSession.on(AgentSessionEventTypes.UserInputTranscribed, this.onUserInputTranscribed); this.agentSession.on(AgentSessionEventTypes.Close, this.onAgentSessionClose); - this.agentSession.on( - AgentSessionEventTypes.ConversationItemAdded, - this.onConversationItemAdded, - ); } async close() { @@ -598,10 +575,6 @@ export class RoomIO { this.agentSession.off(AgentSessionEventTypes.UserInputTranscribed, this.onUserInputTranscribed); this.agentSession.off(AgentSessionEventTypes.AgentStateChanged, this.onAgentStateChanged); this.agentSession.off(AgentSessionEventTypes.Close, this.onAgentSessionClose); - this.agentSession.off( - AgentSessionEventTypes.ConversationItemAdded, - this.onConversationItemAdded, - ); if (this.textStreamHandlerRegistered) { this.room.unregisterTextStreamHandler(TOPIC_CHAT); diff --git a/agents/src/voice/transcription/synchronizer.ts b/agents/src/voice/transcription/synchronizer.ts index dc5c081f2..ea52184ff 100644 --- a/agents/src/voice/transcription/synchronizer.ts +++ b/agents/src/voice/transcription/synchronizer.ts @@ -24,7 +24,6 @@ interface TextSyncOptions { hyphenateWord: (word: string) => string[]; splitWords: (words: string) => [string, number, number][]; wordTokenizer: WordTokenizer; - enabled: boolean; } interface TextData { @@ -142,7 +141,6 @@ interface AudioData { } class SegmentSynchronizerImpl { - private enabled: boolean; private textData: TextData; private audioData: AudioData; private speed: number; @@ -181,7 +179,6 @@ class SegmentSynchronizerImpl { */ private readonly seedFromPushAudio: boolean = false, ) { - this.enabled = options.enabled; this.speed = options.speed * STANDARD_SPEECH_RATE; // hyphens per second this.textData = { wordStream: options.wordTokenizer.stream(), @@ -199,15 +196,13 @@ class SegmentSynchronizerImpl { this.outputStreamWriter = this.outputStream.writable.getWriter(); this.outputEnabledFuture.resolve(); - if (this.enabled) { - this.mainTask() - .then(() => { - this.outputStreamWriter.close(); - }) - .catch((error) => { - this.logger.error({ error }, 'mainTask SegmentSynchronizerImpl'); - }); - } + this.mainTask() + .then(() => { + this.outputStreamWriter.close(); + }) + .catch((error) => { + this.logger.error({ error }, 'mainTask SegmentSynchronizerImpl'); + }); this.captureTask = this.captureTaskImpl(); } @@ -308,11 +303,7 @@ class SegmentSynchronizerImpl { } this.textData.done = true; - if (!this.enabled) { - this.outputStreamWriter.close(); - } else { - this.textData.wordStream.endInput(); - } + this.textData.wordStream.endInput(); } pause(): void { @@ -546,10 +537,6 @@ class SegmentSynchronizerImpl { if (!this.outputEnabledFuture.done) { this.outputEnabledFuture.resolve(); } - // Close the writer if endTextInput hasn't already done so (e.g. on interruption) - if (!this.enabled && !this.textData.done) { - this.outputStreamWriter.close(); - } this.textData.wordStream.close(); await this.captureTask; } @@ -560,7 +547,6 @@ export interface TranscriptionSynchronizerOptions { hyphenateWord: (word: string) => string[]; splitWords: (words: string) => [string, number, number][]; wordTokenizer: WordTokenizer; - enabled: boolean; } export const defaultTextSyncOptions: TranscriptionSynchronizerOptions = { @@ -568,7 +554,6 @@ export const defaultTextSyncOptions: TranscriptionSynchronizerOptions = { hyphenateWord: basic.hyphenateWord, splitWords: basic.splitWords, wordTokenizer: new basic.WordTokenizer(false), - enabled: true, }; export class TranscriptionSynchronizer { @@ -587,8 +572,8 @@ export class TranscriptionSynchronizer { _audioAttached: boolean = true; /** @internal */ _textAttached: boolean = true; - // warn once per enabled cycle when only one of audio/text is detached; reset when - // the synchronizer transitions back to enabled + // warn once per detach cycle when only one of audio/text is detached; reset when + // both outputs are reattached /** @internal */ _warnedAsymmetricDetach: boolean = false; @@ -626,7 +611,6 @@ export class TranscriptionSynchronizer { hyphenateWord: options.hyphenateWord, splitWords: options.splitWords, wordTokenizer: options.wordTokenizer, - enabled: options.enabled, }; // initial segment/first segment, recreated for each new segment @@ -640,18 +624,6 @@ export class TranscriptionSynchronizer { return this._outputsAttached; } - get enabled(): boolean { - return this.options.enabled; - } - - set enabled(value: boolean) { - if (this.options.enabled === value) { - return; - } - this.options.enabled = value; - this.rotateSegment(); - } - /** @internal */ _onAttachmentChanged(args: { audioAttached?: boolean; textAttached?: boolean }): void { if (args.audioAttached !== undefined) { @@ -790,10 +762,6 @@ class SyncedAudioOutput extends AudioOutput { return; } - if (!this.synchronizer.enabled) { - return; - } - if (this.synchronizer._impl.audioInputEnded) { this.logger.warn( 'SegmentSynchronizerImpl audio marked as ended in capture audio, rotating segment', @@ -812,7 +780,7 @@ class SyncedAudioOutput extends AudioOutput { this.lastSegmentAccepted = this.segmentAccepted; } - if (!this.synchronizer.outputsAttached || !this.synchronizer.enabled) { + if (!this.synchronizer.outputsAttached) { return; } @@ -861,7 +829,7 @@ class SyncedAudioOutput extends AudioOutput { */ private settleDriftFinish(): void { const ev: PlaybackFinishedEvent = { playbackPosition: 0, interrupted: true }; - if (!this.synchronizer.outputsAttached || !this.synchronizer.enabled) { + if (!this.synchronizer.outputsAttached) { super.onPlaybackFinished(ev); return; } @@ -891,14 +859,14 @@ class SyncedAudioOutput extends AudioOutput { // this is going to be automatically called by the next_in_chain onPlaybackStarted(createdAt: number): void { super.onPlaybackStarted(createdAt); - if (this.synchronizer.outputsAttached && this.synchronizer.enabled) { + if (this.synchronizer.outputsAttached) { this.synchronizer._impl.onPlaybackStarted(createdAt); } } // this is going to be automatically called by the next_in_chain onPlaybackFinished(ev: PlaybackFinishedEvent) { - if (!this.synchronizer.outputsAttached || !this.synchronizer.enabled) { + if (!this.synchronizer.outputsAttached) { super.onPlaybackFinished(ev); return; } @@ -960,11 +928,6 @@ class SyncedTextOutput extends TextOutput { const textStr = isTimedString(text) ? text.text : text; - if (!this.synchronizer.enabled) { - await this.nextInChain.captureText(textStr); - return; - } - if (!this.synchronizer.outputsAttached) { if ( this.synchronizer._textAttached && @@ -1006,7 +969,7 @@ class SyncedTextOutput extends TextOutput { // Wait for any pending rotation to complete before accessing _impl await this.synchronizer.barrier(); - if (!this.synchronizer.enabled || !this.synchronizer.outputsAttached) { + if (!this.synchronizer.outputsAttached) { this.capturing = false; this.nextInChain.flush(); return; diff --git a/plugins/phonic/src/realtime/realtime_model.ts b/plugins/phonic/src/realtime/realtime_model.ts index b2c3d245b..06663f4f8 100644 --- a/plugins/phonic/src/realtime/realtime_model.ts +++ b/plugins/phonic/src/realtime/realtime_model.ts @@ -195,7 +195,6 @@ export class RealtimeModel extends llm.RealtimeModel { midSessionInstructionsUpdate: true, midSessionToolsUpdate: true, perResponseToolChoice: false, - nativeTranscriptSync: false, }); const apiKey = options.apiKey || process.env.PHONIC_API_KEY; From 9d3a6a4b52730cf0fcaa601339b8465db97948b2 Mon Sep 17 00:00:00 2001 From: Toubat Date: Thu, 16 Jul 2026 15:59:33 -0700 Subject: [PATCH 2/2] fix(agents): preserve native transcript sync capability Keep the released realtime capability functional for third-party models while allowing built-in providers to stop declaring it. Co-authored-by: Cursor --- .changeset/remove-native-transcript-sync.md | 2 +- agents/src/llm/realtime.ts | 5 + agents/src/llm/realtime.type.test.ts | 21 ++++ agents/src/voice/room_io/room_io.test.ts | 110 +++++++++++++++++- agents/src/voice/room_io/room_io.ts | 29 ++++- .../voice/transcription/synchronizer.test.ts | 21 +++- .../src/voice/transcription/synchronizer.ts | 63 +++++++--- 7 files changed, 233 insertions(+), 18 deletions(-) create mode 100644 agents/src/llm/realtime.type.test.ts diff --git a/.changeset/remove-native-transcript-sync.md b/.changeset/remove-native-transcript-sync.md index b1671ca75..a28e1e4f9 100644 --- a/.changeset/remove-native-transcript-sync.md +++ b/.changeset/remove-native-transcript-sync.md @@ -3,4 +3,4 @@ '@livekit/agents': patch --- -Remove the unused `nativeTranscriptSync` realtime model capability. No model relies on native transcript synchronization anymore (Phonic switched to `stream_ahead_of_real_time` mode), so the transcription synchronizer is always enabled. +Deprecate the `nativeTranscriptSync` realtime model capability while preserving its existing transcript synchronization behavior for third-party models. Remove Phonic's redundant explicit opt-out now that it uses `stream_ahead_of_real_time` mode. diff --git a/agents/src/llm/realtime.ts b/agents/src/llm/realtime.ts index cfc6cc170..10af53493 100644 --- a/agents/src/llm/realtime.ts +++ b/agents/src/llm/realtime.ts @@ -64,6 +64,11 @@ export interface RealtimeCapabilities { midSessionToolsUpdate?: boolean; /** Whether the tool and tool choice can be specified per response. */ perResponseToolChoice?: boolean; + /** + * Whether the model synchronizes generated transcript timing natively. + * @deprecated Native transcript synchronization is no longer used by built-in models. + */ + nativeTranscriptSync?: boolean; } export interface InputTranscriptionCompleted { diff --git a/agents/src/llm/realtime.type.test.ts b/agents/src/llm/realtime.type.test.ts new file mode 100644 index 000000000..6b4b041ff --- /dev/null +++ b/agents/src/llm/realtime.type.test.ts @@ -0,0 +1,21 @@ +// SPDX-FileCopyrightText: 2026 LiveKit, Inc. +// +// SPDX-License-Identifier: Apache-2.0 +import { describe, expectTypeOf, it } from 'vitest'; +import type { RealtimeCapabilities } from './realtime.js'; + +describe('RealtimeCapabilities', () => { + it('accepts the released native transcript synchronization capability', () => { + const capabilities: RealtimeCapabilities = { + messageTruncation: true, + turnDetection: true, + userTranscription: true, + autoToolReplyGeneration: true, + audioOutput: true, + manualFunctionCalls: true, + nativeTranscriptSync: true, + }; + + expectTypeOf(capabilities.nativeTranscriptSync).toEqualTypeOf(); + }); +}); diff --git a/agents/src/voice/room_io/room_io.test.ts b/agents/src/voice/room_io/room_io.test.ts index 594879972..d526392e5 100644 --- a/agents/src/voice/room_io/room_io.test.ts +++ b/agents/src/voice/room_io/room_io.test.ts @@ -4,6 +4,7 @@ import { EventEmitter } from 'node:events'; import { afterEach, describe, expect, it, vi } from 'vitest'; import * as jobModule from '../../job.js'; +import { RealtimeModel } from '../../llm/index.js'; import { IdentityTransform } from '../../stream/identity_transform.js'; import { DEFAULT_API_CONNECT_OPTIONS } from '../../types.js'; import { AgentSessionEventTypes, CloseReason, createCloseEvent } from '../events.js'; @@ -65,14 +66,25 @@ function createFakeRoom() { }; } -function createFakeSession() { +type FakeSession = { + input: { audio: null }; + output: { audio: null; transcription: null }; + currentAgent?: { llm?: RealtimeModel }; + llm?: RealtimeModel; + on: ReturnType; + off: ReturnType; + emit: (event: string | symbol, value: unknown) => boolean; + _closeSoon: ReturnType; +}; + +function createFakeSession(llm?: RealtimeModel): FakeSession { const emitter = new EventEmitter(); return { input: { audio: null }, output: { audio: null, transcription: null }, currentAgent: undefined, - llm: undefined, + llm, on: vi.fn((event: string | symbol, listener: (...args: unknown[]) => void) => { emitter.on(event, listener); return emitter; @@ -86,6 +98,100 @@ function createFakeSession() { }; } +class FakeRealtimeModel extends RealtimeModel { + constructor(nativeTranscriptSync?: boolean) { + super({ + messageTruncation: true, + turnDetection: true, + userTranscription: true, + autoToolReplyGeneration: true, + audioOutput: true, + manualFunctionCalls: true, + nativeTranscriptSync, + }); + } + + get model(): string { + return 'fake-realtime'; + } + + session(): never { + throw new Error('not used'); + } + + async close(): Promise {} +} + +describe('RoomIO native transcript synchronization', () => { + it('disables SDK synchronization when the initial realtime model synchronizes natively', async () => { + const room = createFakeRoom(); + const session = createFakeSession(new FakeRealtimeModel(true)); + const roomIO = new RoomIO({ + // @ts-expect-error This focused test uses the minimal AgentSession surface RoomIO consumes. + agentSession: session, + // @ts-expect-error This focused test uses the minimal Room surface RoomIO consumes. + room, + inputOptions: { audioEnabled: false, textEnabled: false }, + }); + + roomIO.start(); + + const synchronizer = Reflect.get(roomIO, 'transcriptionSynchronizer'); + expect(synchronizer.enabled).toBe(false); + await roomIO.close(); + }); + + it.each([false, undefined])( + 'keeps SDK synchronization enabled when native synchronization is %s', + async (nativeTranscriptSync) => { + const room = createFakeRoom(); + const session = createFakeSession(new FakeRealtimeModel(nativeTranscriptSync)); + const roomIO = new RoomIO({ + // @ts-expect-error This focused test uses the minimal AgentSession surface RoomIO consumes. + agentSession: session, + // @ts-expect-error This focused test uses the minimal Room surface RoomIO consumes. + room, + inputOptions: { audioEnabled: false, textEnabled: false }, + }); + + roomIO.start(); + + const synchronizer = Reflect.get(roomIO, 'transcriptionSynchronizer'); + expect(synchronizer.enabled).toBe(true); + await roomIO.close(); + }, + ); + + it.each([ + { initial: true, handoff: false, expected: true }, + { initial: false, handoff: true, expected: false }, + { initial: true, handoff: undefined, expected: true }, + ])( + 'updates SDK synchronization after handoff from $initial to $handoff', + async ({ initial, handoff, expected }) => { + const room = createFakeRoom(); + const session = createFakeSession(new FakeRealtimeModel(initial)); + const roomIO = new RoomIO({ + // @ts-expect-error This focused test uses the minimal AgentSession surface RoomIO consumes. + agentSession: session, + // @ts-expect-error This focused test uses the minimal Room surface RoomIO consumes. + room, + inputOptions: { audioEnabled: false, textEnabled: false }, + }); + roomIO.start(); + + session.currentAgent = { llm: new FakeRealtimeModel(handoff) }; + session.emit(AgentSessionEventTypes.ConversationItemAdded, { + item: { type: 'agent_handoff' }, + }); + + const synchronizer = Reflect.get(roomIO, 'transcriptionSynchronizer'); + expect(synchronizer.enabled).toBe(expected); + await roomIO.close(); + }, + ); +}); + describe('RoomIO deleteRoomOnClose', () => { afterEach(() => { vi.restoreAllMocks(); diff --git a/agents/src/voice/room_io/room_io.ts b/agents/src/voice/room_io/room_io.ts index 8d2e39055..ed5315bf7 100644 --- a/agents/src/voice/room_io/room_io.ts +++ b/agents/src/voice/room_io/room_io.ts @@ -19,6 +19,7 @@ import { import type { WritableStreamDefaultWriter } from 'node:stream/web'; import { ATTRIBUTE_PUBLISH_ON_BEHALF, TOPIC_CHAT } from '../../constants.js'; import { type JobContext, getJobContext } from '../../job.js'; +import { RealtimeModel } from '../../llm/index.js'; import { log } from '../../log.js'; import { IdentityTransform } from '../../stream/identity_transform.js'; import { DEFAULT_API_CONNECT_OPTIONS } from '../../types.js'; @@ -28,11 +29,15 @@ import { AgentSessionEventTypes, type AgentStateChangedEvent, CloseReason, + type ConversationItemAddedEvent, type UserInputTranscribedEvent, } from '../events.js'; import type { AudioOutput, TextOutput } from '../io.js'; import type { TextInputCallback } from '../remote_session.js'; -import { TranscriptionSynchronizer } from '../transcription/synchronizer.js'; +import { + TranscriptionSynchronizer, + defaultTextSyncOptions, +} from '../transcription/synchronizer.js'; import { ParticipantAudioInputStream } from './_input.js'; import { ParalellTextOutput, @@ -285,6 +290,16 @@ export class RoomIO { }); }; + private onConversationItemAdded = (ev: ConversationItemAddedEvent) => { + if (ev.item.type !== 'agent_handoff' || !this.transcriptionSynchronizer) { + return; + } + const sessionLlm = this.agentSession.currentAgent?.llm ?? this.agentSession.llm; + const nativeTranscriptSync = + sessionLlm instanceof RealtimeModel && !!sessionLlm.capabilities.nativeTranscriptSync; + this.transcriptionSynchronizer.enabled = !nativeTranscriptSync; + }; + private onAgentSessionClose = () => { if (!this.inputOptions.deleteRoomOnClose || this.deleteRoomTask) { return; @@ -535,9 +550,13 @@ export class RoomIO { // TODO(AJS-176): check for agent output const audioOutput = this.participantAudioOutput; if (this.outputOptions.syncTranscription && audioOutput) { + const sessionLlm = this.agentSession.currentAgent?.llm ?? this.agentSession.llm; + const nativeTranscriptSync = + sessionLlm instanceof RealtimeModel && !!sessionLlm.capabilities.nativeTranscriptSync; this.transcriptionSynchronizer = new TranscriptionSynchronizer( audioOutput, this.agentTranscriptOutput, + { ...defaultTextSyncOptions, enabled: !nativeTranscriptSync }, ); } } @@ -566,6 +585,10 @@ export class RoomIO { this.agentSession.on(AgentSessionEventTypes.AgentStateChanged, this.onAgentStateChanged); this.agentSession.on(AgentSessionEventTypes.UserInputTranscribed, this.onUserInputTranscribed); this.agentSession.on(AgentSessionEventTypes.Close, this.onAgentSessionClose); + this.agentSession.on( + AgentSessionEventTypes.ConversationItemAdded, + this.onConversationItemAdded, + ); } async close() { @@ -575,6 +598,10 @@ export class RoomIO { this.agentSession.off(AgentSessionEventTypes.UserInputTranscribed, this.onUserInputTranscribed); this.agentSession.off(AgentSessionEventTypes.AgentStateChanged, this.onAgentStateChanged); this.agentSession.off(AgentSessionEventTypes.Close, this.onAgentSessionClose); + this.agentSession.off( + AgentSessionEventTypes.ConversationItemAdded, + this.onConversationItemAdded, + ); if (this.textStreamHandlerRegistered) { this.room.unregisterTextStreamHandler(TOPIC_CHAT); diff --git a/agents/src/voice/transcription/synchronizer.test.ts b/agents/src/voice/transcription/synchronizer.test.ts index d3b100d3f..a0ea88621 100644 --- a/agents/src/voice/transcription/synchronizer.test.ts +++ b/agents/src/voice/transcription/synchronizer.test.ts @@ -5,7 +5,11 @@ import { AudioFrame } from '@livekit/rtc-node'; import { afterEach, describe, expect, it, vi } from 'vitest'; import * as logModule from '../../log.js'; import { AudioOutput, TextOutput } from '../io.js'; -import { SpeakingRateData, TranscriptionSynchronizer } from './synchronizer.js'; +import { + SpeakingRateData, + TranscriptionSynchronizer, + defaultTextSyncOptions, +} from './synchronizer.js'; describe('SpeakingRateData', () => { describe('constructor', () => { @@ -230,6 +234,21 @@ class MockTextOutput extends TextOutput { flush(): void {} } +describe('TranscriptionSynchronizer enabled behavior', () => { + it('directly forwards text when synchronization is disabled', async () => { + const downstream = new MockTextOutput(); + const synchronizer = new TranscriptionSynchronizer(new MockAudioOutput(), downstream, { + ...defaultTextSyncOptions, + enabled: false, + }); + + await synchronizer.textOutput.captureText('hello'); + + expect(downstream.captured).toEqual(['hello']); + await synchronizer.close(); + }); +}); + describe('TranscriptionSynchronizer attachment warnings', () => { const textDetachedWarning = 'TranscriptSynchronizer text output was detached while audio output is still active; ' + diff --git a/agents/src/voice/transcription/synchronizer.ts b/agents/src/voice/transcription/synchronizer.ts index ea52184ff..005b88727 100644 --- a/agents/src/voice/transcription/synchronizer.ts +++ b/agents/src/voice/transcription/synchronizer.ts @@ -24,6 +24,7 @@ interface TextSyncOptions { hyphenateWord: (word: string) => string[]; splitWords: (words: string) => [string, number, number][]; wordTokenizer: WordTokenizer; + enabled: boolean; } interface TextData { @@ -141,6 +142,7 @@ interface AudioData { } class SegmentSynchronizerImpl { + private enabled: boolean; private textData: TextData; private audioData: AudioData; private speed: number; @@ -179,6 +181,7 @@ class SegmentSynchronizerImpl { */ private readonly seedFromPushAudio: boolean = false, ) { + this.enabled = options.enabled; this.speed = options.speed * STANDARD_SPEECH_RATE; // hyphens per second this.textData = { wordStream: options.wordTokenizer.stream(), @@ -196,13 +199,15 @@ class SegmentSynchronizerImpl { this.outputStreamWriter = this.outputStream.writable.getWriter(); this.outputEnabledFuture.resolve(); - this.mainTask() - .then(() => { - this.outputStreamWriter.close(); - }) - .catch((error) => { - this.logger.error({ error }, 'mainTask SegmentSynchronizerImpl'); - }); + if (this.enabled) { + this.mainTask() + .then(() => { + this.outputStreamWriter.close(); + }) + .catch((error) => { + this.logger.error({ error }, 'mainTask SegmentSynchronizerImpl'); + }); + } this.captureTask = this.captureTaskImpl(); } @@ -303,7 +308,11 @@ class SegmentSynchronizerImpl { } this.textData.done = true; - this.textData.wordStream.endInput(); + if (!this.enabled) { + this.outputStreamWriter.close(); + } else { + this.textData.wordStream.endInput(); + } } pause(): void { @@ -537,6 +546,10 @@ class SegmentSynchronizerImpl { if (!this.outputEnabledFuture.done) { this.outputEnabledFuture.resolve(); } + // Close the writer if endTextInput hasn't already done so (e.g. on interruption) + if (!this.enabled && !this.textData.done) { + this.outputStreamWriter.close(); + } this.textData.wordStream.close(); await this.captureTask; } @@ -547,6 +560,7 @@ export interface TranscriptionSynchronizerOptions { hyphenateWord: (word: string) => string[]; splitWords: (words: string) => [string, number, number][]; wordTokenizer: WordTokenizer; + enabled: boolean; } export const defaultTextSyncOptions: TranscriptionSynchronizerOptions = { @@ -554,6 +568,7 @@ export const defaultTextSyncOptions: TranscriptionSynchronizerOptions = { hyphenateWord: basic.hyphenateWord, splitWords: basic.splitWords, wordTokenizer: new basic.WordTokenizer(false), + enabled: true, }; export class TranscriptionSynchronizer { @@ -611,6 +626,7 @@ export class TranscriptionSynchronizer { hyphenateWord: options.hyphenateWord, splitWords: options.splitWords, wordTokenizer: options.wordTokenizer, + enabled: options.enabled, }; // initial segment/first segment, recreated for each new segment @@ -624,6 +640,18 @@ export class TranscriptionSynchronizer { return this._outputsAttached; } + get enabled(): boolean { + return this.options.enabled; + } + + set enabled(value: boolean) { + if (this.options.enabled === value) { + return; + } + this.options.enabled = value; + this.rotateSegment(); + } + /** @internal */ _onAttachmentChanged(args: { audioAttached?: boolean; textAttached?: boolean }): void { if (args.audioAttached !== undefined) { @@ -762,6 +790,10 @@ class SyncedAudioOutput extends AudioOutput { return; } + if (!this.synchronizer.enabled) { + return; + } + if (this.synchronizer._impl.audioInputEnded) { this.logger.warn( 'SegmentSynchronizerImpl audio marked as ended in capture audio, rotating segment', @@ -780,7 +812,7 @@ class SyncedAudioOutput extends AudioOutput { this.lastSegmentAccepted = this.segmentAccepted; } - if (!this.synchronizer.outputsAttached) { + if (!this.synchronizer.outputsAttached || !this.synchronizer.enabled) { return; } @@ -829,7 +861,7 @@ class SyncedAudioOutput extends AudioOutput { */ private settleDriftFinish(): void { const ev: PlaybackFinishedEvent = { playbackPosition: 0, interrupted: true }; - if (!this.synchronizer.outputsAttached) { + if (!this.synchronizer.outputsAttached || !this.synchronizer.enabled) { super.onPlaybackFinished(ev); return; } @@ -859,14 +891,14 @@ class SyncedAudioOutput extends AudioOutput { // this is going to be automatically called by the next_in_chain onPlaybackStarted(createdAt: number): void { super.onPlaybackStarted(createdAt); - if (this.synchronizer.outputsAttached) { + if (this.synchronizer.outputsAttached && this.synchronizer.enabled) { this.synchronizer._impl.onPlaybackStarted(createdAt); } } // this is going to be automatically called by the next_in_chain onPlaybackFinished(ev: PlaybackFinishedEvent) { - if (!this.synchronizer.outputsAttached) { + if (!this.synchronizer.outputsAttached || !this.synchronizer.enabled) { super.onPlaybackFinished(ev); return; } @@ -928,6 +960,11 @@ class SyncedTextOutput extends TextOutput { const textStr = isTimedString(text) ? text.text : text; + if (!this.synchronizer.enabled) { + await this.nextInChain.captureText(textStr); + return; + } + if (!this.synchronizer.outputsAttached) { if ( this.synchronizer._textAttached && @@ -969,7 +1006,7 @@ class SyncedTextOutput extends TextOutput { // Wait for any pending rotation to complete before accessing _impl await this.synchronizer.barrier(); - if (!this.synchronizer.outputsAttached) { + if (!this.synchronizer.enabled || !this.synchronizer.outputsAttached) { this.capturing = false; this.nextInChain.flush(); return;