diff --git a/packages/server/src/server/session.test.ts b/packages/server/src/server/session.test.ts index 490a7b474..eb8e57ca1 100644 --- a/packages/server/src/server/session.test.ts +++ b/packages/server/src/server/session.test.ts @@ -1,5 +1,4 @@ import { execSync } from "child_process"; -import { EventEmitter } from "events"; import { mkdtempSync, realpathSync, rmSync, symlinkSync, writeFileSync } from "fs"; import { homedir, tmpdir } from "os"; import { join, resolve as resolvePath } from "path"; @@ -17,16 +16,6 @@ import { StructuredAgentFallbackError } from "./agent/agent-response-loop.js"; import type { StoredAgentRecord } from "./agent/agent-storage.js"; import type { ProviderSnapshotManager } from "./agent/provider-snapshot-manager.js"; import type { SessionOptions } from "./session.js"; -import type { - SpeechToTextProvider, - StreamingTranscriptionCommittedEvent, - StreamingTranscriptionEvent, - StreamingTranscriptionSession, -} from "./speech/speech-provider.js"; -import type { - TurnDetectionProvider, - TurnDetectionSession, -} from "./speech/turn-detection-provider.js"; import { asSessionInternals as asSessionInternalsHelper, asAgentManager, @@ -50,8 +39,6 @@ import type { } from "../services/github-service.js"; interface SessionHandlerInternals { - startVoiceTurnController(): Promise; - stopVoiceTurnController(): Promise; handleSendAgentMessage( agentId: string, text: string, @@ -82,9 +69,6 @@ interface SessionHandlerInternals { handleStashPopRequest(params: unknown): Promise; createPaseoWorktree(params: unknown): Promise; handleStartWorkspaceScriptRequest(params: unknown): Promise; - sttManager: { - transcribe(audio: Buffer, format: string): Promise; - }; } function asSessionInternals(session: Session): SessionHandlerInternals { @@ -313,198 +297,6 @@ function createSessionForTest(options: SessionForTestOptions = {}): Session { }); } -class FakeVoiceTurnDetectionSession extends EventEmitter implements TurnDetectionSession { - public readonly requiredSampleRate = 16000; - - async connect(): Promise {} - - appendPcm16(_chunk: Buffer): void {} - - flush(): void {} - reset(): void {} - close(): void {} -} - -class FakeVoiceSttSession extends EventEmitter implements StreamingTranscriptionSession { - public readonly requiredSampleRate = 16000; - public commitCount = 0; - - async connect(): Promise {} - - appendPcm16(_pcm16le: Buffer): void {} - - commit(): void { - this.commitCount += 1; - } - - clear(): void {} - close(): void {} - - emitCommitted(event: StreamingTranscriptionCommittedEvent): void { - this.emit("committed", event); - } - - emitTranscript(event: StreamingTranscriptionEvent): void { - this.emit("transcript", event); - } -} - -function createVoiceSessionHarness() { - const messages: unknown[] = []; - const detector = new FakeVoiceTurnDetectionSession(); - const sttSession = new FakeVoiceSttSession(); - const sttProvider: SpeechToTextProvider = { - id: "local", - createSession: vi.fn(() => sttSession), - }; - const turnDetection: TurnDetectionProvider = { - id: "local", - createSession: vi.fn(() => detector), - }; - const session = createSessionForTest({ - messages, - stt: sttProvider, - voice: { turnDetection }, - }); - Object.assign(session, { - isVoiceMode: true, - voiceModeAgentId: "11111111-1111-4111-8111-111111111111", - }); - const internals = asSessionInternals(session); - const sendAgentMessage = vi - .spyOn(internals, "handleSendAgentMessage") - .mockResolvedValue({ ok: true }); - const transcribe = vi.spyOn(asSessionInternals(session).sttManager, "transcribe"); - - return { - session, - internals, - messages, - detector, - sttSession, - sendAgentMessage, - transcribe, - }; -} - -async function settleVoiceSession(): Promise { - await Promise.resolve(); - await Promise.resolve(); - await Promise.resolve(); -} - -describe("session voice mode streaming transcription", () => { - test("submits the streaming final transcript to the agent without batch transcribe", async () => { - const harness = createVoiceSessionHarness(); - - await harness.internals.startVoiceTurnController(); - harness.detector.emit("speech_started"); - await settleVoiceSession(); - harness.detector.emit("speech_stopped"); - await settleVoiceSession(); - harness.sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); - harness.sttSession.emitTranscript({ - segmentId: "segment-1", - transcript: "ship the streaming final", - isFinal: true, - language: "en", - avgLogprob: -0.1, - isLowConfidence: false, - }); - await settleVoiceSession(); - - expect(harness.sttSession.commitCount).toBe(1); - expect(harness.transcribe).not.toHaveBeenCalled(); - expect(harness.sendAgentMessage).toHaveBeenCalledWith( - "11111111-1111-4111-8111-111111111111", - "ship the streaming final", - undefined, - undefined, - undefined, - undefined, - { spokenInput: true }, - ); - expect(harness.messages).toContainEqual( - expect.objectContaining({ - type: "transcription_result", - payload: expect.objectContaining({ - text: "ship the streaming final", - language: "en", - avgLogprob: -0.1, - }), - }), - ); - - await harness.internals.stopVoiceTurnController(); - }); - - test("uses the finalization timeout empty transcript path without agent submission", async () => { - vi.useFakeTimers(); - try { - const harness = createVoiceSessionHarness(); - - await harness.internals.startVoiceTurnController(); - harness.detector.emit("speech_started"); - await settleVoiceSession(); - harness.detector.emit("speech_stopped"); - await settleVoiceSession(); - harness.sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); - - await vi.advanceTimersByTimeAsync(10_000); - await settleVoiceSession(); - - expect(harness.transcribe).not.toHaveBeenCalled(); - expect(harness.sendAgentMessage).not.toHaveBeenCalled(); - expect(harness.messages).toContainEqual( - expect.objectContaining({ - type: "transcription_result", - payload: expect.objectContaining({ - text: "", - }), - }), - ); - - await harness.internals.stopVoiceTurnController(); - } finally { - vi.useRealTimers(); - } - }); - - test("filters low-confidence streaming finals without agent submission", async () => { - const harness = createVoiceSessionHarness(); - - await harness.internals.startVoiceTurnController(); - harness.detector.emit("speech_started"); - await settleVoiceSession(); - harness.detector.emit("speech_stopped"); - await settleVoiceSession(); - harness.sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); - harness.sttSession.emitTranscript({ - segmentId: "segment-1", - transcript: "background noise", - isFinal: true, - avgLogprob: -2.5, - isLowConfidence: true, - }); - await settleVoiceSession(); - - expect(harness.transcribe).not.toHaveBeenCalled(); - expect(harness.sendAgentMessage).not.toHaveBeenCalled(); - expect(harness.messages).toContainEqual( - expect.objectContaining({ - type: "transcription_result", - payload: expect.objectContaining({ - text: "", - avgLogprob: -2.5, - isLowConfidence: true, - }), - }), - ); - - await harness.internals.stopVoiceTurnController(); - }); -}); - describe("file explorer binary responses", () => { const tempDirs: string[] = []; diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index b22b99a4f..8c7ca3003 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -44,22 +44,10 @@ import { import { FileUploadStore } from "./file-upload/index.js"; import { CursorError } from "./pagination/cursor.js"; import { SortablePager, type SortSpec } from "./pagination/sortable-pager.js"; -import { TTSManager } from "./agent/tts-manager.js"; -import { STTManager } from "./agent/stt-manager.js"; import type { SpeechToTextProvider, TextToSpeechProvider } from "./speech/speech-provider.js"; import type { TurnDetectionProvider } from "./speech/turn-detection-provider.js"; -import { maybePersistTtsDebugAudio } from "./agent/tts-debug.js"; -import { isPaseoDictationDebugEnabled } from "./agent/recordings-debug.js"; import { getPidLockInfo } from "./pid-lock.js"; import { generateLocalPairingOffer } from "./pairing-offer.js"; -import { - DictationStreamManager, - type DictationStreamOutboundMessage, -} from "./dictation/dictation-stream-manager.js"; -import { - createVoiceTurnController, - type VoiceTurnController, -} from "./voice/voice-turn-controller.js"; import { buildConfigOverrides, extractTimestamps, @@ -179,12 +167,9 @@ import { type ProjectRegistry, type WorkspaceRegistry, } from "./workspace-registry.js"; -import { - buildVoiceModeSystemPrompt, - stripVoiceModeSystemPrompt, - wrapSpokenInput, -} from "./voice-config.js"; +import { wrapSpokenInput } from "./voice-config.js"; import { isVoicePermissionAllowed } from "./voice-permission-policy.js"; +import { VoiceSession } from "./voice/voice-session.js"; import { listDirectoryEntries, readExplorerFile, @@ -226,9 +211,8 @@ import { buildCheckoutPrStatusPayloadFromSnapshot, buildCheckoutStatusPayloadFromSnapshot, } from "./checkout/status-projection.js"; -import type { LocalSpeechModelId } from "./speech/providers/local/models.js"; -import { toResolver, type Resolvable } from "./speech/provider-resolver.js"; -import type { SpeechReadinessSnapshot, SpeechReadinessState } from "./speech/speech-runtime.js"; +import type { Resolvable } from "./speech/provider-resolver.js"; +import type { SpeechReadinessSnapshot } from "./speech/speech-runtime.js"; import type pino from "pino"; import { ChatServiceError, @@ -472,8 +456,6 @@ export function resolveWaitForFinishError(options: { return typeof message === "string" && message.trim().length > 0 ? message : "Agent failed"; } -type ProcessingPhase = "idle" | "transcribing"; - interface WorkspaceGitWatchTarget { cwd: string; workspaceId: string; @@ -548,26 +530,6 @@ class SessionRequestError extends Error { } } -const PCM_SAMPLE_RATE = 16000; -const PCM_CHANNELS = 1; -const PCM_BITS_PER_SAMPLE = 16; -const PCM_BYTES_PER_MS = (PCM_SAMPLE_RATE * PCM_CHANNELS * (PCM_BITS_PER_SAMPLE / 8)) / 1000; -const MIN_STREAMING_SEGMENT_DURATION_MS = 1000; -const MIN_STREAMING_SEGMENT_BYTES = Math.round( - PCM_BYTES_PER_MS * MIN_STREAMING_SEGMENT_DURATION_MS, -); -const AgentIdSchema = z.guid(); -interface VoiceModeBaseConfig { - systemPrompt?: string; -} - -interface AudioBufferState { - chunks: Buffer[]; - format: string; - isPCM: boolean; - totalPCMBytes: number; -} - export interface SessionFileSystem { isDirectory(path: string): Promise; } @@ -582,18 +544,6 @@ const nodeSessionFileSystem: SessionFileSystem = { // Stub types for features under development (modules not yet available) type AgentMcpTransportFactory = () => Promise; -interface VoiceTranscriptionResultPayload { - text: string; - requestId: string; - language?: string; - duration?: number; - avgLogprob?: number; - isLowConfidence?: boolean; - byteLength?: number; - format?: string; - debugRecordingPath?: string; -} - export interface SessionOptions { clientId: string; appVersion?: string | null; @@ -694,62 +644,6 @@ type PullRequestTimelinePayload = Extract< >["payload"]; type PullRequestTimelinePayloadItem = PullRequestTimelinePayload["items"][number]; -interface VoiceFeatureUnavailableContext { - reasonCode: SpeechReadinessSnapshot["voiceFeature"]["reasonCode"]; - message: string; - retryable: boolean; - missingModelIds: LocalSpeechModelId[]; -} - -interface VoiceFeatureUnavailableResponseMetadata { - reasonCode?: SpeechReadinessSnapshot["voiceFeature"]["reasonCode"]; - retryable?: boolean; - missingModelIds?: LocalSpeechModelId[]; -} - -class VoiceFeatureUnavailableError extends Error { - readonly reasonCode: SpeechReadinessSnapshot["voiceFeature"]["reasonCode"]; - readonly retryable: boolean; - readonly missingModelIds: LocalSpeechModelId[]; - - constructor(context: VoiceFeatureUnavailableContext) { - super(context.message); - this.name = "VoiceFeatureUnavailableError"; - this.reasonCode = context.reasonCode; - this.retryable = context.retryable; - this.missingModelIds = [...context.missingModelIds]; - } -} - -function convertPCMToWavBuffer( - pcmBuffer: Buffer, - sampleRate: number, - channels: number, - bitsPerSample: number, -): Buffer { - const headerSize = 44; - const wavBuffer = Buffer.alloc(headerSize + pcmBuffer.length); - const byteRate = (sampleRate * channels * bitsPerSample) / 8; - const blockAlign = (channels * bitsPerSample) / 8; - - wavBuffer.write("RIFF", 0); - wavBuffer.writeUInt32LE(36 + pcmBuffer.length, 4); - wavBuffer.write("WAVE", 8); - wavBuffer.write("fmt ", 12); - wavBuffer.writeUInt32LE(16, 16); - wavBuffer.writeUInt16LE(1, 20); - wavBuffer.writeUInt16LE(channels, 22); - wavBuffer.writeUInt32LE(sampleRate, 24); - wavBuffer.writeUInt32LE(byteRate, 28); - wavBuffer.writeUInt16LE(blockAlign, 32); - wavBuffer.writeUInt16LE(bitsPerSample, 34); - wavBuffer.write("data", 36); - wavBuffer.writeUInt32LE(pcmBuffer.length, 40); - pcmBuffer.copy(wavBuffer, 44); - - return wavBuffer; -} - function parseClientCapabilities( capabilities: Record | null | undefined, ): ReadonlySet { @@ -806,33 +700,6 @@ export class Session { private readonly paseoHome: string; private readonly worktreesRoot: string | undefined; - // State machine - private abortController: AbortController; - private processingPhase: ProcessingPhase = "idle"; - - // Voice mode state - private isVoiceMode = false; - private speechInProgress = false; - - private dictationStreamManager!: DictationStreamManager; - private resolveVoiceTurnDetection!: () => TurnDetectionProvider | null; - private voiceTurnController: VoiceTurnController | null = null; - private voiceInputChunkCount = 0; - private voiceInputBytes = 0; - private voiceInputWindowStartedAt = Date.now(); - - // Audio buffering for interruption handling - private pendingAudioSegments: Array<{ audio: Buffer; format: string }> = []; - private bufferTimeout: ReturnType | null = null; - private audioBuffer: AudioBufferState | null = null; - - // Optional TTS debug capture (persisted per utterance) - private readonly ttsDebugStreams = new Map(); - - // Per-session managers - private ttsManager!: TTSManager; - private sttManager!: STTManager; - // Per-session MCP client and tools private agentMcpClient: Awaited> | null = null; private agentTools: ToolSet | null = null; @@ -890,18 +757,11 @@ export class Session { private readonly workspaceGitSubscriptions = new Map void>(); private readonly fileUploads: FileUploadStore; private readonly workspaceDirectory: WorkspaceDirectory; - private registerVoiceSpeakHandler?: (agentId: string, handler: VoiceSpeakHandler) => void; - private unregisterVoiceSpeakHandler?: (agentId: string) => void; - private registerVoiceCallerContext?: (agentId: string, context: VoiceCallerContext) => void; - private unregisterVoiceCallerContext?: (agentId: string) => void; - private getSpeechReadiness?: () => SpeechReadinessSnapshot; - private readonly sttLanguage: string; + private readonly voiceSession: VoiceSession; private readonly serverId: string | undefined; private readonly daemonVersion: string | undefined; private readonly daemonRuntimeConfig: SessionOptions["daemonRuntimeConfig"]; private readonly createAgentLifecycleDispatch: CreateAgentLifecycleDispatch; - private voiceModeAgentId: string | null = null; - private voiceModeBaseConfig: VoiceModeBaseConfig | null = null; constructor(options: SessionOptions) { const { @@ -1040,13 +900,10 @@ export class Session { this.getDaemonTcpHost = getDaemonTcpHost ?? null; this.serviceProxyPublicBaseUrl = serviceProxyPublicBaseUrl ?? null; this.resolveScriptHealth = resolveScriptHealth ?? null; - this.sttLanguage = sttLanguage ?? "en"; this.subscribeToOptionalManagers(); - this.bindVoiceBridges({ voice, voiceBridge, dictation }); this.serverId = serverId; this.daemonVersion = daemonVersion; this.daemonRuntimeConfig = daemonRuntimeConfig; - this.abortController = new AbortController(); this.workspaceDirectory = new WorkspaceDirectory({ logger: this.sessionLogger, projectRegistry: this.projectRegistry, @@ -1057,7 +914,40 @@ export class Session { buildWorkspaceDescriptor: (input) => this.buildWorkspaceDescriptor(input), }); - this.initializePerSessionManagers({ tts, stt, sttLanguage, dictation }); + this.voiceSession = new VoiceSession({ + host: { + emit: (msg) => this.emit(msg), + loadAgent: (agentId) => + ensureAgentLoaded(agentId, { + agentManager: this.agentManager, + agentStorage: this.agentStorage, + logger: this.sessionLogger, + }), + reloadAgentSession: (agentId, overrides) => + this.agentManager.reloadAgentSession(agentId, overrides), + sendSpokenInput: async (agentId, text) => { + await this.handleSendAgentMessage( + agentId, + text, + undefined, + undefined, + undefined, + undefined, + { spokenInput: true }, + ); + }, + interruptAgentIfRunning: (agentId) => this.interruptAgentIfRunning(agentId), + hasActiveAgentRun: (agentId) => this.hasActiveAgentRun(agentId), + }, + logger: this.sessionLogger, + sessionId: this.sessionId, + sttLanguage, + tts, + stt, + voice, + voiceBridge, + dictation, + }); // Initialize agent MCP client asynchronously void this.initializeAgentMcp(); @@ -1358,41 +1248,6 @@ export class Session { }; } - private bindVoiceBridges(params: { - voice: SessionOptions["voice"]; - voiceBridge: SessionOptions["voiceBridge"]; - dictation: SessionOptions["dictation"]; - }): void { - const { voice, voiceBridge, dictation } = params; - this.resolveVoiceTurnDetection = toResolver(voice?.turnDetection ?? null); - this.registerVoiceSpeakHandler = voiceBridge?.registerVoiceSpeakHandler; - this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler; - this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext; - this.unregisterVoiceCallerContext = voiceBridge?.unregisterVoiceCallerContext; - this.getSpeechReadiness = dictation?.getSpeechReadiness; - } - - private initializePerSessionManagers(params: { - tts: SessionOptions["tts"]; - stt: SessionOptions["stt"]; - sttLanguage: SessionOptions["sttLanguage"]; - dictation: SessionOptions["dictation"]; - }): void { - const { tts, stt, sttLanguage, dictation } = params; - this.ttsManager = new TTSManager(this.sessionId, this.sessionLogger, tts); - this.sttManager = new STTManager(this.sessionId, this.sessionLogger, stt, { - language: sttLanguage, - }); - this.dictationStreamManager = new DictationStreamManager({ - logger: this.sessionLogger, - sessionId: this.sessionId, - emit: (msg) => this.handleDictationManagerMessage(msg), - stt: dictation?.stt ?? null, - language: dictation?.sttLanguage, - finalTimeoutMs: dictation?.finalTimeoutMs, - }); - } - private subscribeToAgentEvents(): void { if (this.unsubscribeAgentEvents) { this.unsubscribeAgentEvents(); @@ -1416,8 +1271,7 @@ export class Session { } if ( - this.isVoiceMode && - this.voiceModeAgentId === event.agentId && + this.voiceSession.isActiveForAgent(event.agentId) && event.event.type === "permission_requested" && isVoicePermissionAllowed(event.event.request) ) { @@ -1877,27 +1731,27 @@ export class Session { private dispatchVoiceAndControlMessage(msg: SessionInboundMessage): Promise | undefined { switch (msg.type) { case "voice_audio_chunk": - return this.handleAudioChunk(msg); + return this.voiceSession.handleAudioChunk(msg); case "abort_request": - return this.handleAbort(); + return this.voiceSession.handleAbort(); case "audio_played": - this.handleAudioPlayed(msg.id); + this.voiceSession.handleAudioPlayed(msg.id); return undefined; case "set_voice_mode": - return this.handleSetVoiceMode(msg.enabled, msg.agentId, msg.requestId); + return this.voiceSession.handleSetVoiceMode(msg.enabled, msg.agentId, msg.requestId); case "dictation_stream_start": - return this.handleDictationStreamStart(msg); + return this.voiceSession.handleDictationStreamStart(msg); case "dictation_stream_chunk": - return this.dictationStreamManager.handleChunk({ + return this.voiceSession.handleDictationChunk({ dictationId: msg.dictationId, seq: msg.seq, audioBase64: msg.audio, format: msg.format, }); case "dictation_stream_finish": - return this.dictationStreamManager.handleFinish(msg.dictationId, msg.finalSeq); + return this.voiceSession.handleDictationFinish(msg.dictationId, msg.finalSeq); case "dictation_stream_cancel": - this.dictationStreamManager.handleCancel(msg.dictationId); + this.voiceSession.handleDictationCancel(msg.dictationId); return undefined; case "restart_server_request": return this.handleRestartServerRequest(msg.requestId, msg.reason); @@ -1942,26 +1796,6 @@ export class Session { } } - private async handleDictationStreamStart( - msg: Extract, - ): Promise { - const unavailable = this.resolveVoiceFeatureUnavailableContext("dictation"); - if (unavailable) { - this.emit({ - type: "dictation_stream_error", - payload: { - dictationId: msg.dictationId, - error: unavailable.message, - retryable: unavailable.retryable, - reasonCode: unavailable.reasonCode, - missingModelIds: unavailable.missingModelIds, - }, - }); - return; - } - await this.dictationStreamManager.handleStart(msg.dictationId, msg.format); - } - private dispatchAgentLifecycleMessage(msg: SessionInboundMessage): Promise | undefined { switch (msg.type) { case "fetch_agents_request": @@ -2948,397 +2782,6 @@ export class Session { } } - private toVoiceFeatureUnavailableContext( - state: SpeechReadinessState, - ): VoiceFeatureUnavailableContext { - return { - reasonCode: state.reasonCode, - message: state.message, - retryable: state.retryable, - missingModelIds: [...state.missingModelIds], - }; - } - - private resolveModeReadinessState( - readiness: SpeechReadinessSnapshot, - mode: "voice_mode" | "dictation", - ): SpeechReadinessState { - if (mode === "voice_mode") { - return readiness.realtimeVoice; - } - return readiness.dictation; - } - - private getVoiceFeatureUnavailableResponseMetadata( - error: unknown, - ): VoiceFeatureUnavailableResponseMetadata { - if (!(error instanceof VoiceFeatureUnavailableError)) { - return {}; - } - return { - reasonCode: error.reasonCode, - retryable: error.retryable, - missingModelIds: error.missingModelIds, - }; - } - - private resolveVoiceFeatureUnavailableContext( - mode: "voice_mode" | "dictation", - ): VoiceFeatureUnavailableContext | null { - const readiness = this.getSpeechReadiness?.(); - if (!readiness) { - return null; - } - - const modeReadiness = this.resolveModeReadinessState(readiness, mode); - if (!modeReadiness.enabled) { - return this.toVoiceFeatureUnavailableContext(modeReadiness); - } - if (!readiness.voiceFeature.available) { - return this.toVoiceFeatureUnavailableContext(readiness.voiceFeature); - } - if (!modeReadiness.available) { - return this.toVoiceFeatureUnavailableContext(modeReadiness); - } - return null; - } - - /** - * Handle voice mode toggle - */ - private async handleSetVoiceMode( - enabled: boolean, - agentId?: string, - requestId?: string, - ): Promise { - const startedAt = Date.now(); - try { - this.sessionLogger.info( - { enabled, requestedAgentId: agentId ?? null, requestId: requestId ?? null }, - "set_voice_mode started", - ); - if (enabled) { - const unavailable = this.resolveVoiceFeatureUnavailableContext("voice_mode"); - if (unavailable) { - throw new VoiceFeatureUnavailableError(unavailable); - } - - const normalizedAgentId = this.parseVoiceTargetAgentId(agentId ?? "", "set_voice_mode"); - - if ( - this.isVoiceMode && - this.voiceModeAgentId && - this.voiceModeAgentId !== normalizedAgentId - ) { - this.sessionLogger.info( - { - previousAgentId: this.voiceModeAgentId, - nextAgentId: normalizedAgentId, - elapsedMs: Date.now() - startedAt, - }, - "set_voice_mode disabling previous active voice agent", - ); - await this.disableVoiceModeForActiveAgent(true); - } - - if (!this.isVoiceMode || this.voiceModeAgentId !== normalizedAgentId) { - this.sessionLogger.info( - { agentId: normalizedAgentId, elapsedMs: Date.now() - startedAt }, - "set_voice_mode enabling voice for agent", - ); - const refreshedAgentId = await this.enableVoiceModeForAgent(normalizedAgentId); - this.voiceModeAgentId = refreshedAgentId; - this.sessionLogger.info( - { agentId: refreshedAgentId, elapsedMs: Date.now() - startedAt }, - "set_voice_mode agent enable complete", - ); - } - - this.sessionLogger.info( - { agentId: this.voiceModeAgentId, elapsedMs: Date.now() - startedAt }, - "set_voice_mode starting voice turn controller", - ); - await this.startVoiceTurnController(); - this.sessionLogger.info( - { agentId: this.voiceModeAgentId, elapsedMs: Date.now() - startedAt }, - "set_voice_mode voice turn controller started", - ); - this.isVoiceMode = true; - this.sessionLogger.info( - { - agentId: this.voiceModeAgentId, - elapsedMs: Date.now() - startedAt, - }, - "Voice mode enabled for existing agent", - ); - if (requestId) { - this.emit({ - type: "set_voice_mode_response", - payload: { - requestId, - enabled: true, - agentId: this.voiceModeAgentId, - accepted: true, - error: null, - }, - }); - } - return; - } - - this.sessionLogger.info( - { agentId: this.voiceModeAgentId, elapsedMs: Date.now() - startedAt }, - "set_voice_mode disabling active voice mode", - ); - await this.disableVoiceModeForActiveAgent(true); - this.isVoiceMode = false; - this.sessionLogger.info({ elapsedMs: Date.now() - startedAt }, "Voice mode disabled"); - if (requestId) { - this.emit({ - type: "set_voice_mode_response", - payload: { - requestId, - enabled: false, - agentId: null, - accepted: true, - error: null, - }, - }); - } - } catch (error) { - const errorMessage = error instanceof Error ? error.message : "Failed to set voice mode"; - const unavailable = this.getVoiceFeatureUnavailableResponseMetadata(error); - this.sessionLogger.error( - { - err: error, - enabled, - requestedAgentId: agentId ?? null, - elapsedMs: Date.now() - startedAt, - }, - "set_voice_mode failed", - ); - if (requestId) { - this.emit({ - type: "set_voice_mode_response", - payload: { - requestId, - enabled: this.isVoiceMode, - agentId: this.voiceModeAgentId, - accepted: false, - error: errorMessage, - ...unavailable, - }, - }); - return; - } - throw error; - } - } - - private parseVoiceTargetAgentId(rawId: string, source: string): string { - const parsed = AgentIdSchema.safeParse(rawId.trim()); - if (!parsed.success) { - throw new Error(`${source}: agentId must be a UUID`); - } - return parsed.data; - } - - private async enableVoiceModeForAgent(agentId: string): Promise { - const startedAt = Date.now(); - this.sessionLogger.info({ agentId }, "enableVoiceModeForAgent.ensureAgentLoaded.start"); - const existing = await ensureAgentLoaded(agentId, { - agentManager: this.agentManager, - agentStorage: this.agentStorage, - logger: this.sessionLogger, - }); - this.sessionLogger.info( - { agentId, elapsedMs: Date.now() - startedAt }, - "enableVoiceModeForAgent.ensureAgentLoaded.done", - ); - - this.registerVoiceBridgeForAgent(agentId); - - const baseConfig: VoiceModeBaseConfig = { - systemPrompt: stripVoiceModeSystemPrompt(existing.config.systemPrompt), - }; - this.voiceModeBaseConfig = baseConfig; - const refreshOverrides: Partial = { - systemPrompt: buildVoiceModeSystemPrompt(baseConfig.systemPrompt, true), - }; - - try { - this.sessionLogger.info( - { agentId, elapsedMs: Date.now() - startedAt }, - "enableVoiceModeForAgent.reloadAgentSession.start", - ); - const refreshed = await this.agentManager.reloadAgentSession(agentId, refreshOverrides); - this.sessionLogger.info( - { agentId, refreshedAgentId: refreshed.id, elapsedMs: Date.now() - startedAt }, - "enableVoiceModeForAgent.reloadAgentSession.done", - ); - return refreshed.id; - } catch (error) { - this.unregisterVoiceSpeakHandler?.(agentId); - this.unregisterVoiceCallerContext?.(agentId); - this.voiceModeBaseConfig = null; - throw error; - } - } - - private async disableVoiceModeForActiveAgent(restoreAgentConfig: boolean): Promise { - await this.stopVoiceTurnController(); - - const agentId = this.voiceModeAgentId; - if (!agentId) { - this.voiceModeBaseConfig = null; - return; - } - - this.unregisterVoiceSpeakHandler?.(agentId); - this.unregisterVoiceCallerContext?.(agentId); - - if (restoreAgentConfig && this.voiceModeBaseConfig) { - const baseConfig = this.voiceModeBaseConfig; - try { - await this.agentManager.reloadAgentSession(agentId, { - systemPrompt: buildVoiceModeSystemPrompt(baseConfig.systemPrompt, false), - }); - } catch (error) { - this.sessionLogger.warn( - { err: error, agentId }, - "Failed to restore agent config while disabling voice mode", - ); - } - } - - this.voiceModeBaseConfig = null; - this.voiceModeAgentId = null; - } - - private handleDictationManagerMessage(msg: DictationStreamOutboundMessage): void { - this.emit(msg as unknown as SessionOutboundMessage); - } - - private async startVoiceTurnController(): Promise { - if (this.voiceTurnController) { - this.sessionLogger.info("startVoiceTurnController skipped: already running"); - return; - } - - const turnDetection = this.resolveVoiceTurnDetection(); - if (!turnDetection) { - throw new Error("Voice turn detection is not configured"); - } - const stt = this.sttManager.getProvider(); - if (!stt) { - throw new Error("Voice speech-to-text is not configured"); - } - - this.sessionLogger.info( - { providerId: turnDetection.id }, - "startVoiceTurnController creating controller", - ); - - const controller = createVoiceTurnController({ - logger: this.sessionLogger.child({ component: "voice-turn-controller" }), - turnDetection, - stt, - sttLanguage: this.sttLanguage, - callbacks: { - onSpeechStarted: async () => { - this.sessionLogger.debug("Voice VAD speech_started"); - }, - onPartialTranscript: async ({ segmentId, transcript }) => { - this.sessionLogger.info( - { segmentId, transcriptLength: transcript.trim().length }, - "voice_input_state emitting isSpeaking=true", - ); - this.emit({ - type: "voice_input_state", - payload: { - isSpeaking: true, - }, - }); - await this.handleVoiceSpeechStart(); - }, - onSpeechStopped: async () => { - this.handleVoiceSpeechStopped(); - this.setPhase("transcribing"); - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "system", - content: "Transcribing audio...", - }, - }); - }, - onFinalTranscript: async ({ - transcript, - language, - durationMs, - avgLogprob, - isLowConfidence, - }) => { - const requestId = uuidv4(); - const transcriptText = isLowConfidence ? "" : transcript.trim(); - if (isLowConfidence) { - this.sessionLogger.debug( - { text: transcript, avgLogprob }, - "Filtered low-confidence transcription (likely non-speech)", - ); - } - this.sessionLogger.info( - { - requestId, - isVoiceMode: this.isVoiceMode, - transcriptLength: transcriptText.length, - transcript: transcriptText, - }, - "Transcription result", - ); - await this.handleTranscriptionResultPayload({ - text: transcriptText, - requestId, - ...(language ? { language } : {}), - duration: durationMs, - ...(avgLogprob !== undefined ? { avgLogprob } : {}), - ...(isLowConfidence !== undefined ? { isLowConfidence } : {}), - }); - }, - onError: (error) => { - this.sessionLogger.error({ err: error }, "Voice turn controller failed"); - }, - }, - }); - - this.sessionLogger.info("startVoiceTurnController connecting controller"); - await controller.start(); - this.voiceTurnController = controller; - this.sessionLogger.info("startVoiceTurnController connected"); - } - - private async stopVoiceTurnController(): Promise { - if (!this.voiceTurnController) { - return; - } - - const controller = this.voiceTurnController; - this.voiceTurnController = null; - await controller.stop(); - } - - private handleVoiceSpeechStopped(): void { - this.sessionLogger.info("voice_input_state emitting isSpeaking=false"); - this.emit({ - type: "voice_input_state", - payload: { - isSpeaking: false, - }, - }); - } - /** * Handle text message to agent (with optional image attachments) */ @@ -9176,581 +8619,6 @@ export class Session { } } - /** - * Handle audio chunk for buffering and transcription - */ - private async ensureAudioBufferForFormat( - chunkFormat: string, - isPCMChunk: boolean, - ): Promise { - if (!this.audioBuffer) { - this.audioBuffer = { - chunks: [], - format: chunkFormat, - isPCM: isPCMChunk, - totalPCMBytes: 0, - }; - return this.audioBuffer; - } - if (this.audioBuffer.isPCM !== isPCMChunk) { - this.sessionLogger.debug( - { - oldFormat: this.audioBuffer.isPCM ? "pcm" : this.audioBuffer.format, - newFormat: chunkFormat, - }, - `Audio format changed mid-stream, flushing current buffer`, - ); - const finalized = this.finalizeBufferedAudio(); - if (finalized) { - await this.processCompletedAudio(finalized.audio, finalized.format); - } - this.audioBuffer = { - chunks: [], - format: chunkFormat, - isPCM: isPCMChunk, - totalPCMBytes: 0, - }; - return this.audioBuffer; - } - if (!this.audioBuffer.isPCM) { - this.audioBuffer.format = chunkFormat; - } - return this.audioBuffer; - } - - private async forwardAudioChunkToVoiceTurn( - msg: Extract, - chunkFormat: string, - ): Promise { - if (!this.voiceTurnController) { - throw new Error("Voice mode is enabled but the voice turn controller is not running"); - } - const chunkBytes = Buffer.byteLength(msg.audio, "base64"); - this.voiceInputChunkCount += 1; - this.voiceInputBytes += chunkBytes; - const now = Date.now(); - if (this.voiceInputChunkCount % 50 === 0 || now - this.voiceInputWindowStartedAt >= 1000) { - this.sessionLogger.info( - { - chunkCount: this.voiceInputChunkCount, - audioBytes: this.voiceInputBytes, - windowMs: now - this.voiceInputWindowStartedAt, - format: chunkFormat, - }, - "Voice input chunk summary", - ); - this.voiceInputWindowStartedAt = now; - this.voiceInputChunkCount = 0; - this.voiceInputBytes = 0; - } - await this.voiceTurnController.appendClientChunk({ - audioBase64: msg.audio, - format: chunkFormat, - }); - } - - private async handleAudioChunk( - msg: Extract, - ): Promise { - if (!this.isVoiceMode) { - this.sessionLogger.warn( - "Received voice_audio_chunk while voice mode is disabled; transcript will be emitted but voice assistant turn is skipped", - ); - } - - const chunkFormat = msg.format || "audio/wav"; - - if (this.isVoiceMode) { - await this.forwardAudioChunkToVoiceTurn(msg, chunkFormat); - return; - } - - const chunkBuffer = Buffer.from(msg.audio, "base64"); - const isPCMChunk = chunkFormat.toLowerCase().includes("pcm"); - - const buffer = await this.ensureAudioBufferForFormat(chunkFormat, isPCMChunk); - - buffer.chunks.push(chunkBuffer); - if (buffer.isPCM) { - buffer.totalPCMBytes += chunkBuffer.length; - } - - // In non-voice mode, use streaming threshold to process chunks - const reachedStreamingThreshold = - !this.isVoiceMode && buffer.isPCM && buffer.totalPCMBytes >= MIN_STREAMING_SEGMENT_BYTES; - - if (!msg.isLast && reachedStreamingThreshold) { - return; - } - - const bufferedState = this.audioBuffer; - const finalized = this.finalizeBufferedAudio(); - if (!finalized) { - return; - } - - if (!msg.isLast && reachedStreamingThreshold) { - this.sessionLogger.debug( - { - minDuration: MIN_STREAMING_SEGMENT_DURATION_MS, - pcmBytes: bufferedState?.totalPCMBytes ?? 0, - }, - `Minimum chunk duration reached (~${MIN_STREAMING_SEGMENT_DURATION_MS}ms, ${ - bufferedState?.totalPCMBytes ?? 0 - } PCM bytes) – triggering STT`, - ); - } else { - this.sessionLogger.debug( - { audioBytes: finalized.audio.length, chunks: bufferedState?.chunks.length ?? 0 }, - `Complete audio segment (${finalized.audio.length} bytes, ${bufferedState?.chunks.length ?? 0} chunk(s))`, - ); - } - - await this.processCompletedAudio(finalized.audio, finalized.format); - } - - private finalizeBufferedAudio(): { audio: Buffer; format: string } | null { - if (!this.audioBuffer) { - return null; - } - - const bufferState = this.audioBuffer; - this.audioBuffer = null; - - if (bufferState.isPCM) { - const pcmBuffer = Buffer.concat(bufferState.chunks); - const wavBuffer = convertPCMToWavBuffer( - pcmBuffer, - PCM_SAMPLE_RATE, - PCM_CHANNELS, - PCM_BITS_PER_SAMPLE, - ); - return { - audio: wavBuffer, - format: "audio/wav", - }; - } - - return { - audio: Buffer.concat(bufferState.chunks), - format: bufferState.format, - }; - } - - private async processCompletedAudio(audio: Buffer, format: string): Promise { - if (this.processingPhase === "transcribing") { - this.sessionLogger.debug( - { phase: this.processingPhase, segmentCount: this.pendingAudioSegments.length + 1 }, - `Buffering audio segment (phase: ${this.processingPhase})`, - ); - this.pendingAudioSegments.push({ - audio, - format, - }); - this.setBufferTimeout(); - return; - } - - if (this.pendingAudioSegments.length > 0) { - this.pendingAudioSegments.push({ - audio, - format, - }); - this.sessionLogger.debug( - { segmentCount: this.pendingAudioSegments.length }, - `Processing ${this.pendingAudioSegments.length} buffered segments together`, - ); - - const pendingSegments = [...this.pendingAudioSegments]; - this.pendingAudioSegments = []; - this.clearBufferTimeout(); - - const combinedAudio = Buffer.concat(pendingSegments.map((segment) => segment.audio)); - const combinedFormat = pendingSegments[pendingSegments.length - 1].format; - - await this.processAudio(combinedAudio, combinedFormat); - return; - } - - await this.processAudio(audio, format); - } - - private async flushPendingAudioSegments(reason: string): Promise { - if (this.processingPhase === "transcribing" || this.pendingAudioSegments.length === 0) { - return; - } - - const pendingSegments = [...this.pendingAudioSegments]; - this.pendingAudioSegments = []; - this.clearBufferTimeout(); - - this.sessionLogger.debug( - { reason, segmentCount: pendingSegments.length }, - `Flushing ${pendingSegments.length} buffered audio segment(s)`, - ); - - const combinedAudio = Buffer.concat(pendingSegments.map((segment) => segment.audio)); - const combinedFormat = pendingSegments[pendingSegments.length - 1].format; - - await this.processAudio(combinedAudio, combinedFormat); - } - - /** - * Process audio through STT and then LLM - */ - private async processAudio(audio: Buffer, format: string): Promise { - this.setPhase("transcribing"); - - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "system", - content: "Transcribing audio...", - }, - }); - - try { - const requestId = uuidv4(); - const result = await this.sttManager.transcribe(audio, format, { - requestId, - label: this.isVoiceMode ? "voice" : "buffered", - }); - - const transcriptText = result.text.trim(); - this.sessionLogger.info( - { - requestId, - isVoiceMode: this.isVoiceMode, - transcriptLength: transcriptText.length, - transcript: transcriptText, - }, - "Transcription result", - ); - - await this.handleTranscriptionResultPayload({ - text: result.text, - language: result.language, - duration: result.duration, - requestId, - avgLogprob: result.avgLogprob, - isLowConfidence: result.isLowConfidence, - byteLength: result.byteLength, - format: result.format, - debugRecordingPath: result.debugRecordingPath, - }); - } catch (error) { - this.setPhase("idle"); - this.clearSpeechInProgress("transcription error"); - await this.flushPendingAudioSegments("transcription error"); - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "error", - content: `Transcription error: ${getErrorMessage(error)}`, - }, - }); - throw error; - } - } - - private async handleTranscriptionResultPayload( - result: VoiceTranscriptionResultPayload, - ): Promise { - const transcriptText = result.text.trim(); - - this.emit({ - type: "transcription_result", - payload: { - text: result.text, - ...(result.language ? { language: result.language } : {}), - ...(result.duration !== undefined ? { duration: result.duration } : {}), - requestId: result.requestId, - ...(result.avgLogprob !== undefined ? { avgLogprob: result.avgLogprob } : {}), - ...(result.isLowConfidence !== undefined - ? { isLowConfidence: result.isLowConfidence } - : {}), - ...(result.byteLength !== undefined ? { byteLength: result.byteLength } : {}), - ...(result.format ? { format: result.format } : {}), - ...(result.debugRecordingPath ? { debugRecordingPath: result.debugRecordingPath } : {}), - }, - }); - - if (!transcriptText) { - this.sessionLogger.debug("Empty transcription (false positive), not aborting"); - this.setPhase("idle"); - this.clearSpeechInProgress("empty transcription"); - await this.flushPendingAudioSegments("empty transcription"); - return; - } - - // Has content - abort any in-progress stream now - this.createAbortController(); - - if (result.debugRecordingPath) { - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "system", - content: `Saved input audio: ${result.debugRecordingPath}`, - metadata: { - recordingPath: result.debugRecordingPath, - ...(result.format ? { format: result.format } : {}), - requestId: result.requestId, - }, - }, - }); - } - - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "transcript", - content: result.text, - metadata: { - ...(result.language ? { language: result.language } : {}), - ...(result.duration !== undefined ? { duration: result.duration } : {}), - }, - }, - }); - - this.clearSpeechInProgress("transcription complete"); - this.setPhase("idle"); - if (!this.isVoiceMode) { - this.sessionLogger.debug( - { requestId: result.requestId }, - "Skipping voice agent processing because voice mode is disabled", - ); - await this.flushPendingAudioSegments("voice mode disabled"); - return; - } - - const agentId = this.voiceModeAgentId; - if (!agentId) { - this.sessionLogger.warn( - { requestId: result.requestId }, - "Skipping voice agent processing because no agent is currently voice-enabled", - ); - await this.flushPendingAudioSegments("no active voice agent"); - return; - } - - await this.handleSendAgentMessage( - agentId, - result.text, - undefined, - undefined, - undefined, - undefined, - { spokenInput: true }, - ); - await this.flushPendingAudioSegments("transcription complete"); - } - - private registerVoiceBridgeForAgent(agentId: string): void { - this.registerVoiceSpeakHandler?.(agentId, async ({ text, signal }) => { - this.sessionLogger.info( - { - agentId, - textLength: text.length, - preview: text.slice(0, 160), - }, - "Voice speak tool call received by session handler", - ); - const abortSignal = signal ?? this.abortController.signal; - await this.ttsManager.generateAndWaitForPlayback( - text, - (msg) => this.emit(msg), - abortSignal, - true, - ); - this.sessionLogger.info( - { agentId, textLength: text.length }, - "Voice speak tool call finished playback", - ); - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "assistant", - content: text, - }, - }); - }); - - this.registerVoiceCallerContext?.(agentId, { - childAgentDefaultLabels: {}, - allowCustomCwd: false, - enableVoiceTools: true, - }); - } - - /** - * Handle abort request from client - */ - private async handleAbort(): Promise { - this.sessionLogger.info( - { phase: this.processingPhase }, - `Abort request, phase: ${this.processingPhase}`, - ); - - this.abortController.abort(); - this.ttsManager.cancelPendingPlaybacks("abort request"); - - // Voice abort should always interrupt active agent output immediately. - if (this.isVoiceMode && this.voiceModeAgentId) { - try { - await this.interruptAgentIfRunning(this.voiceModeAgentId); - } catch (error) { - this.sessionLogger.warn( - { err: error, agentId: this.voiceModeAgentId }, - "Failed to interrupt active voice-mode agent on abort", - ); - } - } - - if (this.processingPhase === "transcribing") { - // Still in STT phase - we'll buffer the next audio - this.sessionLogger.debug("Will buffer next audio (currently transcribing)"); - // Phase stays as 'transcribing', handleAudioChunk will handle buffering - return; - } - - // Reset phase to idle and clear pending non-voice buffers. - this.setPhase("idle"); - this.pendingAudioSegments = []; - this.clearBufferTimeout(); - } - - /** - * Handle audio playback confirmation from client - */ - private handleAudioPlayed(id: string): void { - this.ttsManager.confirmAudioPlayed(id); - } - - /** - * Mark speech detection start and abort any active playback/agent run. - */ - private async handleVoiceSpeechStart(): Promise { - if (this.speechInProgress) { - return; - } - - const chunkReceivedAt = Date.now(); - const phaseBeforeAbort = this.processingPhase; - const hadActiveStream = this.hasActiveAgentRun(this.voiceModeAgentId); - - this.speechInProgress = true; - this.sessionLogger.debug("Voice speech detected – aborting playback and active agent run"); - - if (this.pendingAudioSegments.length > 0) { - this.sessionLogger.debug( - { segmentCount: this.pendingAudioSegments.length }, - `Dropping ${this.pendingAudioSegments.length} buffered audio segment(s) due to voice speech`, - ); - this.pendingAudioSegments = []; - } - - if (this.audioBuffer) { - this.sessionLogger.debug( - { chunks: this.audioBuffer.chunks.length, pcmBytes: this.audioBuffer.totalPCMBytes }, - `Clearing partial audio buffer (${this.audioBuffer.chunks.length} chunk(s)${ - this.audioBuffer.isPCM ? `, ${this.audioBuffer.totalPCMBytes} PCM bytes` : "" - })`, - ); - this.audioBuffer = null; - } - - this.clearBufferTimeout(); - - this.abortController.abort(); - await this.handleAbort(); - - const latencyMs = Date.now() - chunkReceivedAt; - this.sessionLogger.debug( - { latencyMs, phaseBeforeAbort, hadActiveStream }, - "[Telemetry] barge_in.llm_abort_latency", - ); - } - - /** - * Clear speech-in-progress flag once the user turn has completed - */ - private clearSpeechInProgress(reason: string): void { - if (!this.speechInProgress) { - return; - } - - this.speechInProgress = false; - this.sessionLogger.debug({ reason }, `Speech turn complete (${reason}) – resuming TTS`); - } - - /** - * Create new AbortController, aborting the previous one - */ - private createAbortController(): AbortController { - this.abortController.abort(); - this.abortController = new AbortController(); - this.ttsDebugStreams.clear(); - return this.abortController; - } - - /** - * Set the processing phase - */ - private setPhase(phase: ProcessingPhase): void { - this.processingPhase = phase; - this.sessionLogger.debug({ phase }, `Phase: ${phase}`); - } - - /** - * Set timeout to process buffered audio segments - */ - private setBufferTimeout(): void { - this.clearBufferTimeout(); - - this.bufferTimeout = setTimeout(async () => { - this.sessionLogger.debug("Buffer timeout reached, processing pending segments"); - - if (this.processingPhase === "transcribing") { - this.sessionLogger.debug( - { segmentCount: this.pendingAudioSegments.length }, - "Buffer timeout deferred because transcription is still in progress", - ); - this.setBufferTimeout(); - return; - } - - if (this.pendingAudioSegments.length > 0) { - const segments = [...this.pendingAudioSegments]; - this.pendingAudioSegments = []; - this.bufferTimeout = null; - - const combined = Buffer.concat(segments.map((s) => s.audio)); - await this.processAudio(combined, segments[0].format); - } - }, 10000); // 10 second timeout - } - - /** - * Clear buffer timeout - */ - private clearBufferTimeout(): void { - if (this.bufferTimeout) { - clearTimeout(this.bufferTimeout); - this.bufferTimeout = null; - } - } - /** * Emit a message to the client */ @@ -9767,54 +8635,6 @@ export class Session { "agent.session.outbound", ); } - if ( - msg.type === "audio_output" && - (process.env.TTS_DEBUG_AUDIO_DIR || isPaseoDictationDebugEnabled()) && - msg.payload.groupId && - typeof msg.payload.audio === "string" - ) { - const groupId = msg.payload.groupId; - const existing = - this.ttsDebugStreams.get(groupId) ?? - ({ format: msg.payload.format, chunks: [] } satisfies { - format: string; - chunks: Buffer[]; - }); - - try { - existing.chunks.push(Buffer.from(msg.payload.audio, "base64")); - existing.format = msg.payload.format; - this.ttsDebugStreams.set(groupId, existing); - } catch { - // ignore malformed base64 - } - - if (msg.payload.isLastChunk) { - const final = this.ttsDebugStreams.get(groupId); - this.ttsDebugStreams.delete(groupId); - if (final && final.chunks.length > 0) { - void (async () => { - const recordingPath = await maybePersistTtsDebugAudio( - Buffer.concat(final.chunks), - { sessionId: this.sessionId, groupId, format: final.format }, - this.sessionLogger, - ); - if (recordingPath) { - this.onMessage({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "system", - content: `Saved TTS audio: ${recordingPath}`, - metadata: { recordingPath, format: final.format, groupId }, - }, - }); - } - })(); - } - } - } this.onMessage(msg); } @@ -9848,21 +8668,7 @@ export class Session { this.unsubscribeProviderSnapshotEvents = null; } - // Abort any ongoing operations - this.abortController.abort(); - - // Clear timeouts - this.clearBufferTimeout(); - - // Clear buffers - this.pendingAudioSegments = []; - this.audioBuffer = null; - await this.stopVoiceTurnController(); - - // Cleanup managers - this.ttsManager.cleanup(); - this.sttManager.cleanup(); - this.dictationStreamManager.cleanupAll(); + await this.voiceSession.cleanup(); // Close MCP clients if (this.agentMcpClient) { @@ -9875,9 +8681,6 @@ export class Session { this.agentTools = null; } - await this.disableVoiceModeForActiveAgent(true); - this.isVoiceMode = false; - this.terminalController.dispose(); for (const unsubscribe of this.checkoutDiffSubscriptions.values()) { diff --git a/packages/server/src/server/voice/voice-session.test.ts b/packages/server/src/server/voice/voice-session.test.ts new file mode 100644 index 000000000..917ee62c2 --- /dev/null +++ b/packages/server/src/server/voice/voice-session.test.ts @@ -0,0 +1,211 @@ +import { EventEmitter } from "node:events"; +import pino from "pino"; +import { describe, expect, test, vi } from "vitest"; + +import { VoiceSession, type VoiceSessionHost } from "./voice-session.js"; +import type { ManagedAgent } from "../agent/agent-manager.js"; +import type { SessionOutboundMessage } from "../messages.js"; +import type { + SpeechToTextProvider, + StreamingTranscriptionCommittedEvent, + StreamingTranscriptionEvent, + StreamingTranscriptionSession, +} from "../speech/speech-provider.js"; +import type { + TurnDetectionProvider, + TurnDetectionSession, +} from "../speech/turn-detection-provider.js"; + +const VOICE_AGENT_ID = "11111111-1111-4111-8111-111111111111"; + +class FakeVoiceTurnDetectionSession extends EventEmitter implements TurnDetectionSession { + public readonly requiredSampleRate = 16000; + + async connect(): Promise {} + + appendPcm16(_chunk: Buffer): void {} + + flush(): void {} + reset(): void {} + close(): void {} +} + +class FakeVoiceSttSession extends EventEmitter implements StreamingTranscriptionSession { + public readonly requiredSampleRate = 16000; + public commitCount = 0; + + async connect(): Promise {} + + appendPcm16(_pcm16le: Buffer): void {} + + commit(): void { + this.commitCount += 1; + } + + clear(): void {} + close(): void {} + + emitCommitted(event: StreamingTranscriptionCommittedEvent): void { + this.emit("committed", event); + } + + emitTranscript(event: StreamingTranscriptionEvent): void { + this.emit("transcript", event); + } +} + +interface FakeVoiceHost extends VoiceSessionHost { + readonly emitted: SessionOutboundMessage[]; + readonly spokenInput: Array<{ agentId: string; text: string }>; +} + +function createFakeHost(): FakeVoiceHost { + const emitted: SessionOutboundMessage[] = []; + const spokenInput: Array<{ agentId: string; text: string }> = []; + return { + emitted, + spokenInput, + emit: (msg) => { + emitted.push(msg); + }, + loadAgent: async (agentId) => + ({ id: agentId, config: { systemPrompt: undefined } }) as unknown as ManagedAgent, + reloadAgentSession: async (agentId) => ({ id: agentId }) as unknown as ManagedAgent, + sendSpokenInput: async (agentId, text) => { + spokenInput.push({ agentId, text }); + }, + interruptAgentIfRunning: async () => {}, + hasActiveAgentRun: () => false, + }; +} + +function createVoiceSession() { + const detector = new FakeVoiceTurnDetectionSession(); + const sttSession = new FakeVoiceSttSession(); + const stt: SpeechToTextProvider = { + id: "local", + createSession: vi.fn(() => sttSession), + }; + const turnDetection: TurnDetectionProvider = { + id: "local", + createSession: vi.fn(() => detector), + }; + const host = createFakeHost(); + const voiceSession = new VoiceSession({ + host, + logger: pino({ level: "silent" }), + sessionId: "voice-session-test", + sttLanguage: "en", + tts: null, + stt, + voice: { turnDetection }, + }); + return { voiceSession, detector, sttSession, host }; +} + +async function settle(): Promise { + await Promise.resolve(); + await Promise.resolve(); + await Promise.resolve(); +} + +describe("VoiceSession streaming transcription", () => { + test("delivers the streaming final transcript to the agent exactly once", async () => { + const { voiceSession, detector, sttSession, host } = createVoiceSession(); + + await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID); + detector.emit("speech_started"); + await settle(); + detector.emit("speech_stopped"); + await settle(); + sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + sttSession.emitTranscript({ + segmentId: "segment-1", + transcript: "ship the streaming final", + isFinal: true, + language: "en", + avgLogprob: -0.1, + isLowConfidence: false, + }); + await settle(); + + expect(sttSession.commitCount).toBe(1); + expect(host.spokenInput).toEqual([ + { agentId: VOICE_AGENT_ID, text: "ship the streaming final" }, + ]); + expect(host.emitted).toContainEqual( + expect.objectContaining({ + type: "transcription_result", + payload: expect.objectContaining({ + text: "ship the streaming final", + language: "en", + avgLogprob: -0.1, + }), + }), + ); + + await voiceSession.cleanup(); + }); + + test("emits an empty transcript on finalization timeout without submitting to the agent", async () => { + vi.useFakeTimers(); + try { + const { voiceSession, detector, sttSession, host } = createVoiceSession(); + + await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID); + detector.emit("speech_started"); + await settle(); + detector.emit("speech_stopped"); + await settle(); + sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + + await vi.advanceTimersByTimeAsync(10_000); + await settle(); + + expect(host.spokenInput).toEqual([]); + expect(host.emitted).toContainEqual( + expect.objectContaining({ + type: "transcription_result", + payload: expect.objectContaining({ text: "" }), + }), + ); + + await voiceSession.cleanup(); + } finally { + vi.useRealTimers(); + } + }); + + test("filters a low-confidence streaming final without submitting to the agent", async () => { + const { voiceSession, detector, sttSession, host } = createVoiceSession(); + + await voiceSession.handleSetVoiceMode(true, VOICE_AGENT_ID); + detector.emit("speech_started"); + await settle(); + detector.emit("speech_stopped"); + await settle(); + sttSession.emitCommitted({ segmentId: "segment-1", previousSegmentId: null }); + sttSession.emitTranscript({ + segmentId: "segment-1", + transcript: "background noise", + isFinal: true, + avgLogprob: -2.5, + isLowConfidence: true, + }); + await settle(); + + expect(host.spokenInput).toEqual([]); + expect(host.emitted).toContainEqual( + expect.objectContaining({ + type: "transcription_result", + payload: expect.objectContaining({ + text: "", + avgLogprob: -2.5, + isLowConfidence: true, + }), + }), + ); + + await voiceSession.cleanup(); + }); +}); diff --git a/packages/server/src/server/voice/voice-session.ts b/packages/server/src/server/voice/voice-session.ts new file mode 100644 index 000000000..dcfe77baf --- /dev/null +++ b/packages/server/src/server/voice/voice-session.ts @@ -0,0 +1,1301 @@ +import { v4 as uuidv4 } from "uuid"; +import { z } from "zod"; +import type pino from "pino"; +import { getErrorMessage } from "@getpaseo/protocol/error-utils"; +import type { SessionInboundMessage, SessionOutboundMessage } from "../messages.js"; +import { TTSManager } from "../agent/tts-manager.js"; +import { STTManager } from "../agent/stt-manager.js"; +import type { SpeechToTextProvider, TextToSpeechProvider } from "../speech/speech-provider.js"; +import type { TurnDetectionProvider } from "../speech/turn-detection-provider.js"; +import { maybePersistTtsDebugAudio } from "../agent/tts-debug.js"; +import { isPaseoDictationDebugEnabled } from "../agent/recordings-debug.js"; +import { + DictationStreamManager, + type DictationStreamOutboundMessage, +} from "../dictation/dictation-stream-manager.js"; +import { createVoiceTurnController, type VoiceTurnController } from "./voice-turn-controller.js"; +import { buildVoiceModeSystemPrompt, stripVoiceModeSystemPrompt } from "../voice-config.js"; +import type { VoiceCallerContext, VoiceSpeakHandler } from "../voice-types.js"; +import type { ManagedAgent } from "../agent/agent-manager.js"; +import type { AgentSessionConfig } from "../agent/agent-sdk-types.js"; +import type { LocalSpeechModelId } from "../speech/providers/local/models.js"; +import { toResolver, type Resolvable } from "../speech/provider-resolver.js"; +import type { SpeechReadinessSnapshot, SpeechReadinessState } from "../speech/speech-runtime.js"; + +const PCM_SAMPLE_RATE = 16000; +const PCM_CHANNELS = 1; +const PCM_BITS_PER_SAMPLE = 16; +const PCM_BYTES_PER_MS = (PCM_SAMPLE_RATE * PCM_CHANNELS * (PCM_BITS_PER_SAMPLE / 8)) / 1000; +const MIN_STREAMING_SEGMENT_DURATION_MS = 1000; +const MIN_STREAMING_SEGMENT_BYTES = Math.round( + PCM_BYTES_PER_MS * MIN_STREAMING_SEGMENT_DURATION_MS, +); +const AgentIdSchema = z.guid(); + +type ProcessingPhase = "idle" | "transcribing"; + +interface VoiceModeBaseConfig { + systemPrompt?: string; +} + +interface AudioBufferState { + chunks: Buffer[]; + format: string; + isPCM: boolean; + totalPCMBytes: number; +} + +interface VoiceTranscriptionResultPayload { + text: string; + requestId: string; + language?: string; + duration?: number; + avgLogprob?: number; + isLowConfidence?: boolean; + byteLength?: number; + format?: string; + debugRecordingPath?: string; +} + +interface VoiceFeatureUnavailableContext { + reasonCode: SpeechReadinessSnapshot["voiceFeature"]["reasonCode"]; + message: string; + retryable: boolean; + missingModelIds: LocalSpeechModelId[]; +} + +interface VoiceFeatureUnavailableResponseMetadata { + reasonCode?: SpeechReadinessSnapshot["voiceFeature"]["reasonCode"]; + retryable?: boolean; + missingModelIds?: LocalSpeechModelId[]; +} + +class VoiceFeatureUnavailableError extends Error { + readonly reasonCode: SpeechReadinessSnapshot["voiceFeature"]["reasonCode"]; + readonly retryable: boolean; + readonly missingModelIds: LocalSpeechModelId[]; + + constructor(context: VoiceFeatureUnavailableContext) { + super(context.message); + this.name = "VoiceFeatureUnavailableError"; + this.reasonCode = context.reasonCode; + this.retryable = context.retryable; + this.missingModelIds = [...context.missingModelIds]; + } +} + +function convertPCMToWavBuffer( + pcmBuffer: Buffer, + sampleRate: number, + channels: number, + bitsPerSample: number, +): Buffer { + const headerSize = 44; + const wavBuffer = Buffer.alloc(headerSize + pcmBuffer.length); + const byteRate = (sampleRate * channels * bitsPerSample) / 8; + const blockAlign = (channels * bitsPerSample) / 8; + + wavBuffer.write("RIFF", 0); + wavBuffer.writeUInt32LE(36 + pcmBuffer.length, 4); + wavBuffer.write("WAVE", 8); + wavBuffer.write("fmt ", 12); + wavBuffer.writeUInt32LE(16, 16); + wavBuffer.writeUInt16LE(1, 20); + wavBuffer.writeUInt16LE(channels, 22); + wavBuffer.writeUInt32LE(sampleRate, 24); + wavBuffer.writeUInt32LE(byteRate, 28); + wavBuffer.writeUInt16LE(blockAlign, 32); + wavBuffer.writeUInt16LE(bitsPerSample, 34); + wavBuffer.write("data", 36); + wavBuffer.writeUInt32LE(pcmBuffer.length, 40); + pcmBuffer.copy(wavBuffer, 44); + + return wavBuffer; +} + +/** + * The agent-facing operations VoiceSession needs from the session that owns it. + * VoiceSession owns all voice/audio state; it reaches back through this narrow + * seam only to deliver transcripts to the agent, drive TTS playback, and + * abort/inspect the active agent run. + */ +export interface VoiceSessionHost { + emit(msg: SessionOutboundMessage): void; + loadAgent(agentId: string): Promise; + reloadAgentSession( + agentId: string, + overrides: Partial, + ): Promise; + sendSpokenInput(agentId: string, text: string): Promise; + interruptAgentIfRunning(agentId: string): Promise; + hasActiveAgentRun(agentId: string | null): boolean; +} + +export interface VoiceSessionOptions { + host: VoiceSessionHost; + logger: pino.Logger; + sessionId: string; + sttLanguage?: string; + tts: Resolvable; + stt: Resolvable; + voice?: { + turnDetection?: Resolvable; + }; + voiceBridge?: { + registerVoiceSpeakHandler?: (agentId: string, handler: VoiceSpeakHandler) => void; + unregisterVoiceSpeakHandler?: (agentId: string) => void; + registerVoiceCallerContext?: (agentId: string, context: VoiceCallerContext) => void; + unregisterVoiceCallerContext?: (agentId: string) => void; + }; + dictation?: { + finalTimeoutMs?: number; + stt?: Resolvable; + sttLanguage?: string; + getSpeechReadiness?: () => SpeechReadinessSnapshot; + }; +} + +/** + * Owns the voice half of a client session: speech-to-text/text-to-speech + * managers, dictation streaming, the barge-in audio-buffering state machine, + * voice-turn detection, and the MCP voice bridge. The session delegates the + * voice/dictation/abort message types here and otherwise knows nothing about + * audio buffering or processing phases. + */ +export class VoiceSession { + private readonly host: VoiceSessionHost; + private readonly sessionLogger: pino.Logger; + private readonly sessionId: string; + private readonly sttLanguage: string; + + private abortController: AbortController; + private processingPhase: ProcessingPhase = "idle"; + + private isVoiceMode = false; + private speechInProgress = false; + + private readonly dictationStreamManager: DictationStreamManager; + private readonly resolveVoiceTurnDetection: () => TurnDetectionProvider | null; + private voiceTurnController: VoiceTurnController | null = null; + private voiceInputChunkCount = 0; + private voiceInputBytes = 0; + private voiceInputWindowStartedAt = Date.now(); + + // Audio buffering for interruption handling + private pendingAudioSegments: Array<{ audio: Buffer; format: string }> = []; + private bufferTimeout: ReturnType | null = null; + private audioBuffer: AudioBufferState | null = null; + + // Optional TTS debug capture (persisted per utterance) + private readonly ttsDebugStreams = new Map(); + + private readonly ttsManager: TTSManager; + private readonly sttManager: STTManager; + + private readonly registerVoiceSpeakHandler?: ( + agentId: string, + handler: VoiceSpeakHandler, + ) => void; + private readonly unregisterVoiceSpeakHandler?: (agentId: string) => void; + private readonly registerVoiceCallerContext?: ( + agentId: string, + context: VoiceCallerContext, + ) => void; + private readonly unregisterVoiceCallerContext?: (agentId: string) => void; + private readonly getSpeechReadiness?: () => SpeechReadinessSnapshot; + + private voiceModeAgentId: string | null = null; + private voiceModeBaseConfig: VoiceModeBaseConfig | null = null; + + constructor(options: VoiceSessionOptions) { + const { host, logger, sessionId, sttLanguage, tts, stt, voice, voiceBridge, dictation } = + options; + this.host = host; + this.sessionLogger = logger; + this.sessionId = sessionId; + this.sttLanguage = sttLanguage ?? "en"; + this.abortController = new AbortController(); + + this.resolveVoiceTurnDetection = toResolver(voice?.turnDetection ?? null); + this.registerVoiceSpeakHandler = voiceBridge?.registerVoiceSpeakHandler; + this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler; + this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext; + this.unregisterVoiceCallerContext = voiceBridge?.unregisterVoiceCallerContext; + this.getSpeechReadiness = dictation?.getSpeechReadiness; + + this.ttsManager = new TTSManager(this.sessionId, this.sessionLogger, tts); + this.sttManager = new STTManager(this.sessionId, this.sessionLogger, stt, { + language: sttLanguage, + }); + this.dictationStreamManager = new DictationStreamManager({ + logger: this.sessionLogger, + sessionId: this.sessionId, + emit: (msg) => this.handleDictationManagerMessage(msg), + stt: dictation?.stt ?? null, + language: dictation?.sttLanguage, + finalTimeoutMs: dictation?.finalTimeoutMs, + }); + } + + isActiveForAgent(agentId: string): boolean { + return this.isVoiceMode && this.voiceModeAgentId === agentId; + } + + handleDictationChunk(params: { + dictationId: string; + seq: number; + audioBase64: string; + format: string; + }): Promise { + return this.dictationStreamManager.handleChunk(params); + } + + handleDictationFinish(dictationId: string, finalSeq: number): Promise { + return this.dictationStreamManager.handleFinish(dictationId, finalSeq); + } + + handleDictationCancel(dictationId: string): void { + this.dictationStreamManager.handleCancel(dictationId); + } + + async handleDictationStreamStart( + msg: Extract, + ): Promise { + const unavailable = this.resolveVoiceFeatureUnavailableContext("dictation"); + if (unavailable) { + this.emit({ + type: "dictation_stream_error", + payload: { + dictationId: msg.dictationId, + error: unavailable.message, + retryable: unavailable.retryable, + reasonCode: unavailable.reasonCode, + missingModelIds: unavailable.missingModelIds, + }, + }); + return; + } + await this.dictationStreamManager.handleStart(msg.dictationId, msg.format); + } + + private toVoiceFeatureUnavailableContext( + state: SpeechReadinessState, + ): VoiceFeatureUnavailableContext { + return { + reasonCode: state.reasonCode, + message: state.message, + retryable: state.retryable, + missingModelIds: [...state.missingModelIds], + }; + } + + private resolveModeReadinessState( + readiness: SpeechReadinessSnapshot, + mode: "voice_mode" | "dictation", + ): SpeechReadinessState { + if (mode === "voice_mode") { + return readiness.realtimeVoice; + } + return readiness.dictation; + } + + private getVoiceFeatureUnavailableResponseMetadata( + error: unknown, + ): VoiceFeatureUnavailableResponseMetadata { + if (!(error instanceof VoiceFeatureUnavailableError)) { + return {}; + } + return { + reasonCode: error.reasonCode, + retryable: error.retryable, + missingModelIds: error.missingModelIds, + }; + } + + private resolveVoiceFeatureUnavailableContext( + mode: "voice_mode" | "dictation", + ): VoiceFeatureUnavailableContext | null { + const readiness = this.getSpeechReadiness?.(); + if (!readiness) { + return null; + } + + const modeReadiness = this.resolveModeReadinessState(readiness, mode); + if (!modeReadiness.enabled) { + return this.toVoiceFeatureUnavailableContext(modeReadiness); + } + if (!readiness.voiceFeature.available) { + return this.toVoiceFeatureUnavailableContext(readiness.voiceFeature); + } + if (!modeReadiness.available) { + return this.toVoiceFeatureUnavailableContext(modeReadiness); + } + return null; + } + + /** + * Handle voice mode toggle + */ + async handleSetVoiceMode(enabled: boolean, agentId?: string, requestId?: string): Promise { + const startedAt = Date.now(); + try { + this.sessionLogger.info( + { enabled, requestedAgentId: agentId ?? null, requestId: requestId ?? null }, + "set_voice_mode started", + ); + if (enabled) { + const unavailable = this.resolveVoiceFeatureUnavailableContext("voice_mode"); + if (unavailable) { + throw new VoiceFeatureUnavailableError(unavailable); + } + + const normalizedAgentId = this.parseVoiceTargetAgentId(agentId ?? "", "set_voice_mode"); + + if ( + this.isVoiceMode && + this.voiceModeAgentId && + this.voiceModeAgentId !== normalizedAgentId + ) { + this.sessionLogger.info( + { + previousAgentId: this.voiceModeAgentId, + nextAgentId: normalizedAgentId, + elapsedMs: Date.now() - startedAt, + }, + "set_voice_mode disabling previous active voice agent", + ); + await this.disableVoiceModeForActiveAgent(true); + } + + if (!this.isVoiceMode || this.voiceModeAgentId !== normalizedAgentId) { + this.sessionLogger.info( + { agentId: normalizedAgentId, elapsedMs: Date.now() - startedAt }, + "set_voice_mode enabling voice for agent", + ); + const refreshedAgentId = await this.enableVoiceModeForAgent(normalizedAgentId); + this.voiceModeAgentId = refreshedAgentId; + this.sessionLogger.info( + { agentId: refreshedAgentId, elapsedMs: Date.now() - startedAt }, + "set_voice_mode agent enable complete", + ); + } + + this.sessionLogger.info( + { agentId: this.voiceModeAgentId, elapsedMs: Date.now() - startedAt }, + "set_voice_mode starting voice turn controller", + ); + await this.startVoiceTurnController(); + this.sessionLogger.info( + { agentId: this.voiceModeAgentId, elapsedMs: Date.now() - startedAt }, + "set_voice_mode voice turn controller started", + ); + this.isVoiceMode = true; + this.sessionLogger.info( + { + agentId: this.voiceModeAgentId, + elapsedMs: Date.now() - startedAt, + }, + "Voice mode enabled for existing agent", + ); + if (requestId) { + this.emit({ + type: "set_voice_mode_response", + payload: { + requestId, + enabled: true, + agentId: this.voiceModeAgentId, + accepted: true, + error: null, + }, + }); + } + return; + } + + this.sessionLogger.info( + { agentId: this.voiceModeAgentId, elapsedMs: Date.now() - startedAt }, + "set_voice_mode disabling active voice mode", + ); + await this.disableVoiceModeForActiveAgent(true); + this.isVoiceMode = false; + this.sessionLogger.info({ elapsedMs: Date.now() - startedAt }, "Voice mode disabled"); + if (requestId) { + this.emit({ + type: "set_voice_mode_response", + payload: { + requestId, + enabled: false, + agentId: null, + accepted: true, + error: null, + }, + }); + } + } catch (error) { + const errorMessage = error instanceof Error ? error.message : "Failed to set voice mode"; + const unavailable = this.getVoiceFeatureUnavailableResponseMetadata(error); + this.sessionLogger.error( + { + err: error, + enabled, + requestedAgentId: agentId ?? null, + elapsedMs: Date.now() - startedAt, + }, + "set_voice_mode failed", + ); + if (requestId) { + this.emit({ + type: "set_voice_mode_response", + payload: { + requestId, + enabled: this.isVoiceMode, + agentId: this.voiceModeAgentId, + accepted: false, + error: errorMessage, + ...unavailable, + }, + }); + return; + } + throw error; + } + } + + private parseVoiceTargetAgentId(rawId: string, source: string): string { + const parsed = AgentIdSchema.safeParse(rawId.trim()); + if (!parsed.success) { + throw new Error(`${source}: agentId must be a UUID`); + } + return parsed.data; + } + + private async enableVoiceModeForAgent(agentId: string): Promise { + const startedAt = Date.now(); + this.sessionLogger.info({ agentId }, "enableVoiceModeForAgent.ensureAgentLoaded.start"); + const existing = await this.host.loadAgent(agentId); + this.sessionLogger.info( + { agentId, elapsedMs: Date.now() - startedAt }, + "enableVoiceModeForAgent.ensureAgentLoaded.done", + ); + + this.registerVoiceBridgeForAgent(agentId); + + const baseConfig: VoiceModeBaseConfig = { + systemPrompt: stripVoiceModeSystemPrompt(existing.config.systemPrompt), + }; + this.voiceModeBaseConfig = baseConfig; + const refreshOverrides: Partial = { + systemPrompt: buildVoiceModeSystemPrompt(baseConfig.systemPrompt, true), + }; + + try { + this.sessionLogger.info( + { agentId, elapsedMs: Date.now() - startedAt }, + "enableVoiceModeForAgent.reloadAgentSession.start", + ); + const refreshed = await this.host.reloadAgentSession(agentId, refreshOverrides); + this.sessionLogger.info( + { agentId, refreshedAgentId: refreshed.id, elapsedMs: Date.now() - startedAt }, + "enableVoiceModeForAgent.reloadAgentSession.done", + ); + return refreshed.id; + } catch (error) { + this.unregisterVoiceSpeakHandler?.(agentId); + this.unregisterVoiceCallerContext?.(agentId); + this.voiceModeBaseConfig = null; + throw error; + } + } + + private async disableVoiceModeForActiveAgent(restoreAgentConfig: boolean): Promise { + await this.stopVoiceTurnController(); + + const agentId = this.voiceModeAgentId; + if (!agentId) { + this.voiceModeBaseConfig = null; + return; + } + + this.unregisterVoiceSpeakHandler?.(agentId); + this.unregisterVoiceCallerContext?.(agentId); + + if (restoreAgentConfig && this.voiceModeBaseConfig) { + const baseConfig = this.voiceModeBaseConfig; + try { + await this.host.reloadAgentSession(agentId, { + systemPrompt: buildVoiceModeSystemPrompt(baseConfig.systemPrompt, false), + }); + } catch (error) { + this.sessionLogger.warn( + { err: error, agentId }, + "Failed to restore agent config while disabling voice mode", + ); + } + } + + this.voiceModeBaseConfig = null; + this.voiceModeAgentId = null; + } + + private handleDictationManagerMessage(msg: DictationStreamOutboundMessage): void { + this.emit(msg as unknown as SessionOutboundMessage); + } + + private async startVoiceTurnController(): Promise { + if (this.voiceTurnController) { + this.sessionLogger.info("startVoiceTurnController skipped: already running"); + return; + } + + const turnDetection = this.resolveVoiceTurnDetection(); + if (!turnDetection) { + throw new Error("Voice turn detection is not configured"); + } + const stt = this.sttManager.getProvider(); + if (!stt) { + throw new Error("Voice speech-to-text is not configured"); + } + + this.sessionLogger.info( + { providerId: turnDetection.id }, + "startVoiceTurnController creating controller", + ); + + const controller = createVoiceTurnController({ + logger: this.sessionLogger.child({ component: "voice-turn-controller" }), + turnDetection, + stt, + sttLanguage: this.sttLanguage, + callbacks: { + onSpeechStarted: async () => { + this.sessionLogger.debug("Voice VAD speech_started"); + }, + onPartialTranscript: async ({ segmentId, transcript }) => { + this.sessionLogger.info( + { segmentId, transcriptLength: transcript.trim().length }, + "voice_input_state emitting isSpeaking=true", + ); + this.emit({ + type: "voice_input_state", + payload: { + isSpeaking: true, + }, + }); + await this.handleVoiceSpeechStart(); + }, + onSpeechStopped: async () => { + this.handleVoiceSpeechStopped(); + this.setPhase("transcribing"); + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "system", + content: "Transcribing audio...", + }, + }); + }, + onFinalTranscript: async ({ + transcript, + language, + durationMs, + avgLogprob, + isLowConfidence, + }) => { + const requestId = uuidv4(); + const transcriptText = isLowConfidence ? "" : transcript.trim(); + if (isLowConfidence) { + this.sessionLogger.debug( + { text: transcript, avgLogprob }, + "Filtered low-confidence transcription (likely non-speech)", + ); + } + this.sessionLogger.info( + { + requestId, + isVoiceMode: this.isVoiceMode, + transcriptLength: transcriptText.length, + transcript: transcriptText, + }, + "Transcription result", + ); + await this.handleTranscriptionResultPayload({ + text: transcriptText, + requestId, + ...(language ? { language } : {}), + duration: durationMs, + ...(avgLogprob !== undefined ? { avgLogprob } : {}), + ...(isLowConfidence !== undefined ? { isLowConfidence } : {}), + }); + }, + onError: (error) => { + this.sessionLogger.error({ err: error }, "Voice turn controller failed"); + }, + }, + }); + + this.sessionLogger.info("startVoiceTurnController connecting controller"); + await controller.start(); + this.voiceTurnController = controller; + this.sessionLogger.info("startVoiceTurnController connected"); + } + + private async stopVoiceTurnController(): Promise { + if (!this.voiceTurnController) { + return; + } + + const controller = this.voiceTurnController; + this.voiceTurnController = null; + await controller.stop(); + } + + private handleVoiceSpeechStopped(): void { + this.sessionLogger.info("voice_input_state emitting isSpeaking=false"); + this.emit({ + type: "voice_input_state", + payload: { + isSpeaking: false, + }, + }); + } + + private async ensureAudioBufferForFormat( + chunkFormat: string, + isPCMChunk: boolean, + ): Promise { + if (!this.audioBuffer) { + this.audioBuffer = { + chunks: [], + format: chunkFormat, + isPCM: isPCMChunk, + totalPCMBytes: 0, + }; + return this.audioBuffer; + } + if (this.audioBuffer.isPCM !== isPCMChunk) { + this.sessionLogger.debug( + { + oldFormat: this.audioBuffer.isPCM ? "pcm" : this.audioBuffer.format, + newFormat: chunkFormat, + }, + `Audio format changed mid-stream, flushing current buffer`, + ); + const finalized = this.finalizeBufferedAudio(); + if (finalized) { + await this.processCompletedAudio(finalized.audio, finalized.format); + } + this.audioBuffer = { + chunks: [], + format: chunkFormat, + isPCM: isPCMChunk, + totalPCMBytes: 0, + }; + return this.audioBuffer; + } + if (!this.audioBuffer.isPCM) { + this.audioBuffer.format = chunkFormat; + } + return this.audioBuffer; + } + + private async forwardAudioChunkToVoiceTurn( + msg: Extract, + chunkFormat: string, + ): Promise { + if (!this.voiceTurnController) { + throw new Error("Voice mode is enabled but the voice turn controller is not running"); + } + const chunkBytes = Buffer.byteLength(msg.audio, "base64"); + this.voiceInputChunkCount += 1; + this.voiceInputBytes += chunkBytes; + const now = Date.now(); + if (this.voiceInputChunkCount % 50 === 0 || now - this.voiceInputWindowStartedAt >= 1000) { + this.sessionLogger.info( + { + chunkCount: this.voiceInputChunkCount, + audioBytes: this.voiceInputBytes, + windowMs: now - this.voiceInputWindowStartedAt, + format: chunkFormat, + }, + "Voice input chunk summary", + ); + this.voiceInputWindowStartedAt = now; + this.voiceInputChunkCount = 0; + this.voiceInputBytes = 0; + } + await this.voiceTurnController.appendClientChunk({ + audioBase64: msg.audio, + format: chunkFormat, + }); + } + + async handleAudioChunk( + msg: Extract, + ): Promise { + if (!this.isVoiceMode) { + this.sessionLogger.warn( + "Received voice_audio_chunk while voice mode is disabled; transcript will be emitted but voice assistant turn is skipped", + ); + } + + const chunkFormat = msg.format || "audio/wav"; + + if (this.isVoiceMode) { + await this.forwardAudioChunkToVoiceTurn(msg, chunkFormat); + return; + } + + const chunkBuffer = Buffer.from(msg.audio, "base64"); + const isPCMChunk = chunkFormat.toLowerCase().includes("pcm"); + + const buffer = await this.ensureAudioBufferForFormat(chunkFormat, isPCMChunk); + + buffer.chunks.push(chunkBuffer); + if (buffer.isPCM) { + buffer.totalPCMBytes += chunkBuffer.length; + } + + // In non-voice mode, use streaming threshold to process chunks + const reachedStreamingThreshold = + !this.isVoiceMode && buffer.isPCM && buffer.totalPCMBytes >= MIN_STREAMING_SEGMENT_BYTES; + + if (!msg.isLast && reachedStreamingThreshold) { + return; + } + + const bufferedState = this.audioBuffer; + const finalized = this.finalizeBufferedAudio(); + if (!finalized) { + return; + } + + if (!msg.isLast && reachedStreamingThreshold) { + this.sessionLogger.debug( + { + minDuration: MIN_STREAMING_SEGMENT_DURATION_MS, + pcmBytes: bufferedState?.totalPCMBytes ?? 0, + }, + `Minimum chunk duration reached (~${MIN_STREAMING_SEGMENT_DURATION_MS}ms, ${ + bufferedState?.totalPCMBytes ?? 0 + } PCM bytes) – triggering STT`, + ); + } else { + this.sessionLogger.debug( + { audioBytes: finalized.audio.length, chunks: bufferedState?.chunks.length ?? 0 }, + `Complete audio segment (${finalized.audio.length} bytes, ${bufferedState?.chunks.length ?? 0} chunk(s))`, + ); + } + + await this.processCompletedAudio(finalized.audio, finalized.format); + } + + private finalizeBufferedAudio(): { audio: Buffer; format: string } | null { + if (!this.audioBuffer) { + return null; + } + + const bufferState = this.audioBuffer; + this.audioBuffer = null; + + if (bufferState.isPCM) { + const pcmBuffer = Buffer.concat(bufferState.chunks); + const wavBuffer = convertPCMToWavBuffer( + pcmBuffer, + PCM_SAMPLE_RATE, + PCM_CHANNELS, + PCM_BITS_PER_SAMPLE, + ); + return { + audio: wavBuffer, + format: "audio/wav", + }; + } + + return { + audio: Buffer.concat(bufferState.chunks), + format: bufferState.format, + }; + } + + private async processCompletedAudio(audio: Buffer, format: string): Promise { + if (this.processingPhase === "transcribing") { + this.sessionLogger.debug( + { phase: this.processingPhase, segmentCount: this.pendingAudioSegments.length + 1 }, + `Buffering audio segment (phase: ${this.processingPhase})`, + ); + this.pendingAudioSegments.push({ + audio, + format, + }); + this.setBufferTimeout(); + return; + } + + if (this.pendingAudioSegments.length > 0) { + this.pendingAudioSegments.push({ + audio, + format, + }); + this.sessionLogger.debug( + { segmentCount: this.pendingAudioSegments.length }, + `Processing ${this.pendingAudioSegments.length} buffered segments together`, + ); + + const pendingSegments = [...this.pendingAudioSegments]; + this.pendingAudioSegments = []; + this.clearBufferTimeout(); + + const combinedAudio = Buffer.concat(pendingSegments.map((segment) => segment.audio)); + const combinedFormat = pendingSegments[pendingSegments.length - 1].format; + + await this.processAudio(combinedAudio, combinedFormat); + return; + } + + await this.processAudio(audio, format); + } + + private async flushPendingAudioSegments(reason: string): Promise { + if (this.processingPhase === "transcribing" || this.pendingAudioSegments.length === 0) { + return; + } + + const pendingSegments = [...this.pendingAudioSegments]; + this.pendingAudioSegments = []; + this.clearBufferTimeout(); + + this.sessionLogger.debug( + { reason, segmentCount: pendingSegments.length }, + `Flushing ${pendingSegments.length} buffered audio segment(s)`, + ); + + const combinedAudio = Buffer.concat(pendingSegments.map((segment) => segment.audio)); + const combinedFormat = pendingSegments[pendingSegments.length - 1].format; + + await this.processAudio(combinedAudio, combinedFormat); + } + + /** + * Process audio through STT and then LLM + */ + private async processAudio(audio: Buffer, format: string): Promise { + this.setPhase("transcribing"); + + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "system", + content: "Transcribing audio...", + }, + }); + + try { + const requestId = uuidv4(); + const result = await this.sttManager.transcribe(audio, format, { + requestId, + label: this.isVoiceMode ? "voice" : "buffered", + }); + + const transcriptText = result.text.trim(); + this.sessionLogger.info( + { + requestId, + isVoiceMode: this.isVoiceMode, + transcriptLength: transcriptText.length, + transcript: transcriptText, + }, + "Transcription result", + ); + + await this.handleTranscriptionResultPayload({ + text: result.text, + language: result.language, + duration: result.duration, + requestId, + avgLogprob: result.avgLogprob, + isLowConfidence: result.isLowConfidence, + byteLength: result.byteLength, + format: result.format, + debugRecordingPath: result.debugRecordingPath, + }); + } catch (error) { + this.setPhase("idle"); + this.clearSpeechInProgress("transcription error"); + await this.flushPendingAudioSegments("transcription error"); + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "error", + content: `Transcription error: ${getErrorMessage(error)}`, + }, + }); + throw error; + } + } + + private async handleTranscriptionResultPayload( + result: VoiceTranscriptionResultPayload, + ): Promise { + const transcriptText = result.text.trim(); + + this.emit({ + type: "transcription_result", + payload: { + text: result.text, + ...(result.language ? { language: result.language } : {}), + ...(result.duration !== undefined ? { duration: result.duration } : {}), + requestId: result.requestId, + ...(result.avgLogprob !== undefined ? { avgLogprob: result.avgLogprob } : {}), + ...(result.isLowConfidence !== undefined + ? { isLowConfidence: result.isLowConfidence } + : {}), + ...(result.byteLength !== undefined ? { byteLength: result.byteLength } : {}), + ...(result.format ? { format: result.format } : {}), + ...(result.debugRecordingPath ? { debugRecordingPath: result.debugRecordingPath } : {}), + }, + }); + + if (!transcriptText) { + this.sessionLogger.debug("Empty transcription (false positive), not aborting"); + this.setPhase("idle"); + this.clearSpeechInProgress("empty transcription"); + await this.flushPendingAudioSegments("empty transcription"); + return; + } + + // Has content - abort any in-progress stream now + this.createAbortController(); + + if (result.debugRecordingPath) { + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "system", + content: `Saved input audio: ${result.debugRecordingPath}`, + metadata: { + recordingPath: result.debugRecordingPath, + ...(result.format ? { format: result.format } : {}), + requestId: result.requestId, + }, + }, + }); + } + + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "transcript", + content: result.text, + metadata: { + ...(result.language ? { language: result.language } : {}), + ...(result.duration !== undefined ? { duration: result.duration } : {}), + }, + }, + }); + + this.clearSpeechInProgress("transcription complete"); + this.setPhase("idle"); + if (!this.isVoiceMode) { + this.sessionLogger.debug( + { requestId: result.requestId }, + "Skipping voice agent processing because voice mode is disabled", + ); + await this.flushPendingAudioSegments("voice mode disabled"); + return; + } + + const agentId = this.voiceModeAgentId; + if (!agentId) { + this.sessionLogger.warn( + { requestId: result.requestId }, + "Skipping voice agent processing because no agent is currently voice-enabled", + ); + await this.flushPendingAudioSegments("no active voice agent"); + return; + } + + await this.host.sendSpokenInput(agentId, result.text); + await this.flushPendingAudioSegments("transcription complete"); + } + + private registerVoiceBridgeForAgent(agentId: string): void { + this.registerVoiceSpeakHandler?.(agentId, async ({ text, signal }) => { + this.sessionLogger.info( + { + agentId, + textLength: text.length, + preview: text.slice(0, 160), + }, + "Voice speak tool call received by session handler", + ); + const abortSignal = signal ?? this.abortController.signal; + await this.ttsManager.generateAndWaitForPlayback( + text, + (msg) => this.emit(msg), + abortSignal, + true, + ); + this.sessionLogger.info( + { agentId, textLength: text.length }, + "Voice speak tool call finished playback", + ); + this.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "assistant", + content: text, + }, + }); + }); + + this.registerVoiceCallerContext?.(agentId, { + childAgentDefaultLabels: {}, + allowCustomCwd: false, + enableVoiceTools: true, + }); + } + + /** + * Handle abort request from client + */ + async handleAbort(): Promise { + this.sessionLogger.info( + { phase: this.processingPhase }, + `Abort request, phase: ${this.processingPhase}`, + ); + + this.abortController.abort(); + this.ttsManager.cancelPendingPlaybacks("abort request"); + + // Voice abort should always interrupt active agent output immediately. + if (this.isVoiceMode && this.voiceModeAgentId) { + try { + await this.host.interruptAgentIfRunning(this.voiceModeAgentId); + } catch (error) { + this.sessionLogger.warn( + { err: error, agentId: this.voiceModeAgentId }, + "Failed to interrupt active voice-mode agent on abort", + ); + } + } + + if (this.processingPhase === "transcribing") { + // Still in STT phase - we'll buffer the next audio + this.sessionLogger.debug("Will buffer next audio (currently transcribing)"); + // Phase stays as 'transcribing', handleAudioChunk will handle buffering + return; + } + + // Reset phase to idle and clear pending non-voice buffers. + this.setPhase("idle"); + this.pendingAudioSegments = []; + this.clearBufferTimeout(); + } + + /** + * Handle audio playback confirmation from client + */ + handleAudioPlayed(id: string): void { + this.ttsManager.confirmAudioPlayed(id); + } + + /** + * Mark speech detection start and abort any active playback/agent run. + */ + private async handleVoiceSpeechStart(): Promise { + if (this.speechInProgress) { + return; + } + + const chunkReceivedAt = Date.now(); + const phaseBeforeAbort = this.processingPhase; + const hadActiveStream = this.host.hasActiveAgentRun(this.voiceModeAgentId); + + this.speechInProgress = true; + this.sessionLogger.debug("Voice speech detected – aborting playback and active agent run"); + + if (this.pendingAudioSegments.length > 0) { + this.sessionLogger.debug( + { segmentCount: this.pendingAudioSegments.length }, + `Dropping ${this.pendingAudioSegments.length} buffered audio segment(s) due to voice speech`, + ); + this.pendingAudioSegments = []; + } + + if (this.audioBuffer) { + this.sessionLogger.debug( + { chunks: this.audioBuffer.chunks.length, pcmBytes: this.audioBuffer.totalPCMBytes }, + `Clearing partial audio buffer (${this.audioBuffer.chunks.length} chunk(s)${ + this.audioBuffer.isPCM ? `, ${this.audioBuffer.totalPCMBytes} PCM bytes` : "" + })`, + ); + this.audioBuffer = null; + } + + this.clearBufferTimeout(); + + this.abortController.abort(); + await this.handleAbort(); + + const latencyMs = Date.now() - chunkReceivedAt; + this.sessionLogger.debug( + { latencyMs, phaseBeforeAbort, hadActiveStream }, + "[Telemetry] barge_in.llm_abort_latency", + ); + } + + /** + * Clear speech-in-progress flag once the user turn has completed + */ + private clearSpeechInProgress(reason: string): void { + if (!this.speechInProgress) { + return; + } + + this.speechInProgress = false; + this.sessionLogger.debug({ reason }, `Speech turn complete (${reason}) – resuming TTS`); + } + + /** + * Create new AbortController, aborting the previous one + */ + private createAbortController(): AbortController { + this.abortController.abort(); + this.abortController = new AbortController(); + this.ttsDebugStreams.clear(); + return this.abortController; + } + + /** + * Set the processing phase + */ + private setPhase(phase: ProcessingPhase): void { + this.processingPhase = phase; + this.sessionLogger.debug({ phase }, `Phase: ${phase}`); + } + + /** + * Set timeout to process buffered audio segments + */ + private setBufferTimeout(): void { + this.clearBufferTimeout(); + + this.bufferTimeout = setTimeout(async () => { + this.sessionLogger.debug("Buffer timeout reached, processing pending segments"); + + if (this.processingPhase === "transcribing") { + this.sessionLogger.debug( + { segmentCount: this.pendingAudioSegments.length }, + "Buffer timeout deferred because transcription is still in progress", + ); + this.setBufferTimeout(); + return; + } + + if (this.pendingAudioSegments.length > 0) { + const segments = [...this.pendingAudioSegments]; + this.pendingAudioSegments = []; + this.bufferTimeout = null; + + const combined = Buffer.concat(segments.map((s) => s.audio)); + await this.processAudio(combined, segments[0].format); + } + }, 10000); // 10 second timeout + } + + /** + * Clear buffer timeout + */ + private clearBufferTimeout(): void { + if (this.bufferTimeout) { + clearTimeout(this.bufferTimeout); + this.bufferTimeout = null; + } + } + + /** + * Emit a message to the client. Captures TTS audio_output frames for optional + * debug persistence before forwarding to the session emitter. + */ + private emit(msg: SessionOutboundMessage): void { + if ( + msg.type === "audio_output" && + (process.env.TTS_DEBUG_AUDIO_DIR || isPaseoDictationDebugEnabled()) && + msg.payload.groupId && + typeof msg.payload.audio === "string" + ) { + const groupId = msg.payload.groupId; + const existing = + this.ttsDebugStreams.get(groupId) ?? + ({ format: msg.payload.format, chunks: [] } satisfies { + format: string; + chunks: Buffer[]; + }); + + try { + existing.chunks.push(Buffer.from(msg.payload.audio, "base64")); + existing.format = msg.payload.format; + this.ttsDebugStreams.set(groupId, existing); + } catch { + // ignore malformed base64 + } + + if (msg.payload.isLastChunk) { + const final = this.ttsDebugStreams.get(groupId); + this.ttsDebugStreams.delete(groupId); + if (final && final.chunks.length > 0) { + void (async () => { + const recordingPath = await maybePersistTtsDebugAudio( + Buffer.concat(final.chunks), + { sessionId: this.sessionId, groupId, format: final.format }, + this.sessionLogger, + ); + if (recordingPath) { + this.host.emit({ + type: "activity_log", + payload: { + id: uuidv4(), + timestamp: new Date(), + type: "system", + content: `Saved TTS audio: ${recordingPath}`, + metadata: { recordingPath, format: final.format, groupId }, + }, + }); + } + })(); + } + } + } + this.host.emit(msg); + } + + /** + * Tear down all voice resources. + */ + async cleanup(): Promise { + this.abortController.abort(); + this.clearBufferTimeout(); + this.pendingAudioSegments = []; + this.audioBuffer = null; + await this.stopVoiceTurnController(); + + this.ttsManager.cleanup(); + this.sttManager.cleanup(); + this.dictationStreamManager.cleanupAll(); + + await this.disableVoiceModeForActiveAgent(true); + this.isVoiceMode = false; + } +}