diff --git a/packages/client/src/VoiceConversation.test.ts b/packages/client/src/VoiceConversation.test.ts new file mode 100644 index 00000000..34499b04 --- /dev/null +++ b/packages/client/src/VoiceConversation.test.ts @@ -0,0 +1,239 @@ +import { describe, it, expect, vi, beforeEach, afterEach } from "vitest"; + +import { VoiceConversation } from "./VoiceConversation.js"; +import { + BaseConversation, + type Options, + type PartialOptions, +} from "./BaseConversation.js"; +import type { BaseConnection } from "./utils/BaseConnection.js"; +import type { InputController } from "./InputController.js"; +import type { OutputController } from "./OutputController.js"; +import type { AgentAudioEvent } from "./utils/events.js"; + +const noopInput = { + close: vi.fn(), + setDevice: vi.fn(), + setMuted: vi.fn(), + isMuted: () => false, + getAnalyser: () => undefined, + getVolume: () => 0, + getByteFrequencyData: () => {}, +} satisfies InputController; + +const noopOutput = { + close: vi.fn(), + setDevice: vi.fn(), + setVolume: vi.fn(), + interrupt: vi.fn(), + getAnalyser: () => undefined, + getVolume: () => 0, + getByteFrequencyData: () => {}, +} satisfies OutputController; + +export class TestVoiceConversation extends VoiceConversation { + public static create( + options: Partial = {}, + connection: BaseConnection = { + conversationId: "test-conversation-id", + inputFormat: { format: "pcm", sampleRate: 16000 }, + outputFormat: { format: "pcm", sampleRate: 16000 }, + onMessage: () => {}, + onDisconnect: () => {}, + onModeChange: () => {}, + close: () => {}, + sendMessage: () => {}, + } as unknown as BaseConnection + ) { + const fullOptions = BaseConversation.getFullOptions({ + agentId: "test-agent-id", + connectionType: "webrtc", + ...options, + } as PartialOptions); + + return new TestVoiceConversation( + fullOptions, + connection, + noopInput, + noopOutput, + null, + async () => {} + ); + } + + public handleAudioEvent(event: AgentAudioEvent) { + this.handleAudio(event); + } + + public simulateInterruption(eventId: number) { + this.handleInterruption({ + type: "interruption", + interruption_event: { event_id: eventId }, + }); + } +} + +describe("VoiceConversation", () => { + const alignment = { + chars: ["H", "e", "l", "l", "o"], + char_start_times_ms: [0, 80, 160, 240, 320], + char_durations_ms: [80, 80, 80, 80, 120], + }; + + it("fires onAudioAlignment when an audio event includes alignment data", () => { + const onAudioAlignment = vi.fn(); + const conversation = TestVoiceConversation.create({ onAudioAlignment }); + + conversation.handleAudioEvent({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 10, + alignment, + }, + }); + + expect(onAudioAlignment).toHaveBeenCalledWith(alignment); + }); + + it("does not fire onAudioAlignment for stale events after an interruption", () => { + const onAudioAlignment = vi.fn(); + const conversation = TestVoiceConversation.create({ onAudioAlignment }); + + conversation.simulateInterruption(20); + conversation.handleAudioEvent({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 5, + alignment, + }, + }); + + expect(onAudioAlignment).not.toHaveBeenCalled(); + }); + + it("fires onAudioAlignment without calling onAudio when audio_base_64 is omitted", () => { + const onAudioAlignment = vi.fn(); + const onAudio = vi.fn(); + const conversation = TestVoiceConversation.create({ + onAudioAlignment, + onAudio, + }); + + conversation.handleAudioEvent({ + type: "audio", + audio_event: { + audio_base_64: "", + event_id: 11, + alignment, + }, + }); + + expect(onAudioAlignment).toHaveBeenCalledWith(alignment); + expect(onAudio).not.toHaveBeenCalled(); + }); + + it("does not update mode or feedback for alignment-only events", () => { + const onModeChange = vi.fn(); + const onCanSendFeedbackChange = vi.fn(); + const conversation = TestVoiceConversation.create({ + onModeChange, + onCanSendFeedbackChange, + }); + + conversation.handleAudioEvent({ + type: "audio", + audio_event: { + audio_base_64: "", + event_id: 11, + alignment, + }, + }); + + expect(onModeChange).not.toHaveBeenCalled(); + expect(onCanSendFeedbackChange).not.toHaveBeenCalled(); + }); +}); + +describe("VoiceConversation WebSocket integration", () => { + const alignment = { + chars: ["A", "B"], + char_start_times_ms: [0, 100], + char_durations_ms: [100, 100], + }; + + let listeners: Map void)[]>; + let mockSocket: Record; + + beforeEach(() => { + listeners = new Map(); + mockSocket = { + addEventListener: vi.fn( + (type: string, handler: (event: { data: string }) => void) => { + if (!listeners.has(type)) listeners.set(type, []); + listeners.get(type)!.push(handler); + } + ), + removeEventListener: vi.fn(), + send: vi.fn(), + close: vi.fn(), + }; + vi.stubGlobal( + "WebSocket", + vi.fn(function WebSocket() { + return mockSocket; + }) + ); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it("delivers onAudioAlignment over WebSocket", async () => { + const { WebSocketConnection } = + await import("./utils/WebSocketConnection.js"); + const onAudioAlignment = vi.fn(); + + const promise = WebSocketConnection.create({ + agentId: "test-agent", + connectionType: "websocket", + }); + + for (const handler of listeners.get("open") ?? []) { + handler({ data: "" }); + } + for (const handler of listeners.get("message") ?? []) { + handler({ + data: JSON.stringify({ + type: "conversation_initiation_metadata", + conversation_initiation_metadata_event: { + conversation_id: "test-conv-id", + agent_output_audio_format: "pcm_16000", + user_input_audio_format: "pcm_16000", + }, + }), + }); + } + + const connection = await promise; + TestVoiceConversation.create({ onAudioAlignment }, connection); + + for (const handler of listeners.get("message") ?? []) { + handler({ + data: JSON.stringify({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 12, + alignment, + }, + }), + }); + } + + expect(onAudioAlignment).toHaveBeenCalledWith(alignment); + connection.close(); + }); +}); diff --git a/packages/client/src/VoiceConversation.ts b/packages/client/src/VoiceConversation.ts index caeadd4e..e9b7c81d 100644 --- a/packages/client/src/VoiceConversation.ts +++ b/packages/client/src/VoiceConversation.ts @@ -113,20 +113,19 @@ export class VoiceConversation extends BaseConversation { protected override handleAudio(event: AgentAudioEvent) { super.handleAudio(event); - if (event.audio_event.alignment && this.options.onAudioAlignment) { - this.options.onAudioAlignment(event.audio_event.alignment); - } - if (this.lastInterruptTimestamp <= event.audio_event.event_id) { + if (event.audio_event.alignment && this.options.onAudioAlignment) { + this.options.onAudioAlignment(event.audio_event.alignment); + } + if (event.audio_event.audio_base_64) { this.options.onAudio?.(event.audio_event.audio_base_64); // Audio routing is handled by attachConnectionToOutput for WebSocket // WebRTC handles audio playback directly through LiveKit tracks + this.currentEventId = event.audio_event.event_id; + this.updateCanSendFeedback(); + this.updateMode("speaking"); } - - this.currentEventId = event.audio_event.event_id; - this.updateCanSendFeedback(); - this.updateMode("speaking"); } } diff --git a/packages/client/src/utils/WebRTCConnection.test.ts b/packages/client/src/utils/WebRTCConnection.test.ts index fc59ae1d..f3ff90ad 100644 --- a/packages/client/src/utils/WebRTCConnection.test.ts +++ b/packages/client/src/utils/WebRTCConnection.test.ts @@ -45,6 +45,7 @@ vi.mock("livekit-client", () => { ConnectionStateChanged: "connectionStateChanged", DataReceived: "dataReceived", TrackSubscribed: "trackSubscribed", + TrackUnsubscribed: "trackUnsubscribed", ActiveSpeakersChanged: "activeSpeakersChanged", ParticipantDisconnected: "participantDisconnected", }, @@ -64,6 +65,7 @@ import { WebRTCConnection } from "./WebRTCConnection.js"; import { Room, createLocalAudioTrack } from "livekit-client"; import { setWebRTCAudioAdapterFactory } from "../WebRTCAudioAdapter.js"; import { WebAudioAdapter } from "../platform/web/webAudioAdapter.js"; +import { TestVoiceConversation } from "../VoiceConversation.test.js"; describe("WebRTCConnection", () => { beforeEach(() => { @@ -331,6 +333,141 @@ describe("WebRTCConnection", () => { }); }); + describe("WebRTC audio data channel", () => { + async function createWithHandlers() { + const mockRoom = new Room() as any; + const eventHandlers = new Map void>(); + + (mockRoom.on as ReturnType).mockImplementation( + (event: string, callback: (...args: unknown[]) => void) => { + eventHandlers.set(event, callback); + if (event === "connected") { + queueMicrotask(() => callback()); + } + } + ); + (mockRoom.once as ReturnType).mockImplementation( + (event: string, callback: (...args: unknown[]) => void) => { + if (event === "signalConnected") { + queueMicrotask(() => callback()); + } + } + ); + + const connection = await WebRTCConnection.create({ + conversationToken: "test-token", + connectionType: "webrtc", + }); + + return { connection, eventHandlers, mockRoom }; + } + + it("forwards alignment on data channel without playback audio", async () => { + const alignment = { + chars: ["H", "i"], + char_start_times_ms: [0, 100], + char_durations_ms: [100, 150], + }; + const { connection, eventHandlers } = await createWithHandlers(); + const onMessage = vi.fn(); + connection.onMessage(onMessage); + + eventHandlers.get("dataReceived")?.( + new TextEncoder().encode( + JSON.stringify({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 42, + alignment, + }, + }) + ) + ); + + expect(onMessage).toHaveBeenCalledWith({ + type: "audio", + audio_event: { + audio_base_64: "", + event_id: 42, + alignment, + }, + }); + connection.close(); + }); + + it("drops audio data channel messages without alignment", async () => { + const { connection, eventHandlers } = await createWithHandlers(); + const onMessage = vi.fn(); + connection.onMessage(onMessage); + + eventHandlers.get("dataReceived")?.( + new TextEncoder().encode( + JSON.stringify({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 44, + }, + }) + ) + ); + + expect(onMessage).not.toHaveBeenCalled(); + connection.close(); + }); + + it("forwards onAudioAlignment through VoiceConversation", async () => { + const alignment = { + chars: ["W", "o", "r", "d"], + char_start_times_ms: [0, 50, 100, 150], + char_durations_ms: [50, 50, 50, 80], + }; + const { connection, eventHandlers } = await createWithHandlers(); + const onAudioAlignment = vi.fn(); + const onAudio = vi.fn(); + + TestVoiceConversation.create({ onAudioAlignment, onAudio }, connection); + + eventHandlers.get("dataReceived")?.( + new TextEncoder().encode( + JSON.stringify({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 100, + alignment, + }, + }) + ) + ); + + expect(onAudioAlignment).toHaveBeenCalledWith(alignment); + expect(onAudio).not.toHaveBeenCalled(); + connection.close(); + }); + + it("setVolume uses LiveKit track.setVolume when no audio adapter is registered", async () => { + setWebRTCAudioAdapterFactory(undefined as never); + const { connection, eventHandlers } = await createWithHandlers(); + const mockSetVolume = vi.fn(); + + await eventHandlers.get("trackSubscribed")?.( + { + kind: "audio", + mediaStreamTrack: { id: "remote-track" }, + setVolume: mockSetVolume, + }, + {}, + { identity: "agent_123" } + ); + + connection.output.setVolume(0.75); + expect(mockSetVolume).toHaveBeenCalledWith(0.75); + connection.close(); + }); + }); + it.each([ { textOnly: true, shouldEnableMic: false }, { textOnly: false, shouldEnableMic: true }, diff --git a/packages/client/src/utils/WebRTCConnection.ts b/packages/client/src/utils/WebRTCConnection.ts index d48a6979..4a9e7aa0 100644 --- a/packages/client/src/utils/WebRTCConnection.ts +++ b/packages/client/src/utils/WebRTCConnection.ts @@ -62,6 +62,8 @@ export class WebRTCConnection extends BaseConnection { private audioAdapter: WebRTCAudioAdapter | null; + private remoteAudioTracks: RemoteAudioTrack[] = []; + private inputAnalyser: unknown = undefined; private inputVolumeProvider: VolumeProvider = NO_VOLUME; @@ -397,7 +399,15 @@ export class WebRTCConnection extends BaseConnection { const message = JSON.parse(new TextDecoder().decode(payload)); // Filter out audio messages for WebRTC - they're handled via audio tracks - if (message.type === "audio") { + if (isValidSocketEvent(message) && message.type === "audio") { + if (message.audio_event.alignment) { + const { audio_base_64: _audioBase64, ...audioEvent } = + message.audio_event; + this.handleMessage({ + type: "audio", + audio_event: { ...audioEvent, audio_base_64: "" }, + }); + } return; } @@ -425,6 +435,7 @@ export class WebRTCConnection extends BaseConnection { participant.identity.includes("agent") ) { const remoteAudioTrack = track as RemoteAudioTrack; + this.remoteAudioTracks.push(remoteAudioTrack); if (this.audioAdapter) { // Delegate playback to the platform-specific adapter @@ -442,6 +453,24 @@ export class WebRTCConnection extends BaseConnection { } ); + this.room.on( + RoomEvent.TrackUnsubscribed, + ( + track: Track, + _publication: TrackPublication, + participant: Participant + ) => { + if ( + track.kind === Track.Kind.Audio && + participant.identity.includes("agent") + ) { + this.remoteAudioTracks = this.remoteAudioTracks.filter( + remoteTrack => remoteTrack !== track + ); + } + } + ); + this.room.on( RoomEvent.ActiveSpeakersChanged, async (speakers: Participant[]) => { @@ -485,6 +514,7 @@ export class WebRTCConnection extends BaseConnection { // Delegate all audio cleanup to the adapter this.audioAdapter?.cleanup(); + this.remoteAudioTracks = []; this.inputAnalyser = undefined; this.outputAnalyser = undefined; this.inputVolumeProvider = NO_VOLUME; @@ -598,7 +628,14 @@ export class WebRTCConnection extends BaseConnection { } public setAudioVolume(volume: number) { - this.audioAdapter?.setVolume(volume); + if (this.audioAdapter) { + this.audioAdapter.setVolume(volume); + return; + } + + for (const track of this.remoteAudioTracks) { + track.setVolume?.(volume); + } } public async setAudioOutputDevice(deviceId: string): Promise { diff --git a/packages/client/src/utils/WebSocketConnection.test.ts b/packages/client/src/utils/WebSocketConnection.test.ts index d0c9953d..059732e7 100644 --- a/packages/client/src/utils/WebSocketConnection.test.ts +++ b/packages/client/src/utils/WebSocketConnection.test.ts @@ -168,4 +168,36 @@ describe("WebSocketConnection", () => { expect(listener).not.toHaveBeenCalled(); }); + + it("forwards alignment-bearing audio on the WebSocket", async () => { + const connection = await createConnection(); + const onMessage = vi.fn(); + const alignment = { + chars: ["H", "i"], + char_start_times_ms: [0, 100], + char_durations_ms: [100, 150], + }; + + connection.onMessage(onMessage); + + emit("message", { + data: JSON.stringify({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 2, + alignment, + }, + }), + }); + + expect(onMessage).toHaveBeenCalledWith({ + type: "audio", + audio_event: { + audio_base_64: "dGVzdA==", + event_id: 2, + alignment, + }, + }); + }); });