diff --git a/packages/cli/src/cli.ts b/packages/cli/src/cli.ts index b5f2d6c03..b3222c7c4 100644 --- a/packages/cli/src/cli.ts +++ b/packages/cli/src/cli.ts @@ -14,7 +14,6 @@ import { runInspectCommand } from './commands/agent/inspect.js' import { runWaitCommand } from './commands/agent/wait.js' import { runAttachCommand } from './commands/agent/attach.js' import { withOutput } from './output/index.js' -import { runVoiceMcpBridgeCommand } from './commands/voice-mcp-bridge.js' const VERSION = '0.1.0' @@ -142,13 +141,5 @@ export function createCli(): Command { // Worktree commands program.addCommand(createWorktreeCommand()) - // Internal voice MCP stdio bridge command (hidden). - program - .command('__paseo_voice_mcp_bridge', { hidden: true }) - .description('Internal voice MCP bridge command') - .requiredOption('--socket ') - .requiredOption('--caller-agent-id ') - .action(runVoiceMcpBridgeCommand) - return program } diff --git a/packages/cli/src/commands/voice-mcp-bridge.ts b/packages/cli/src/commands/voice-mcp-bridge.ts deleted file mode 100644 index e50887fa7..000000000 --- a/packages/cli/src/commands/voice-mcp-bridge.ts +++ /dev/null @@ -1,17 +0,0 @@ -import { runVoiceMcpBridgeCli } from "@getpaseo/server"; - -type VoiceBridgeOptions = { - socket: string; - callerAgentId: string; -}; - -export async function runVoiceMcpBridgeCommand( - options: VoiceBridgeOptions -): Promise { - await runVoiceMcpBridgeCli([ - "--socket", - options.socket, - "--caller-agent-id", - options.callerAgentId, - ]); -} diff --git a/packages/server/scripts/mcp-stdio-socket-bridge-cli.mjs b/packages/server/scripts/mcp-stdio-socket-bridge-cli.mjs new file mode 100644 index 000000000..bf8440ff3 --- /dev/null +++ b/packages/server/scripts/mcp-stdio-socket-bridge-cli.mjs @@ -0,0 +1,62 @@ +import net from "node:net"; + +function parseArgs(argv) { + let socketPath = null; + + for (let index = 0; index < argv.length; index += 1) { + const arg = argv[index]; + if (arg === "--socket") { + socketPath = argv[index + 1] ?? null; + index += 1; + } + } + + if (!socketPath || !socketPath.trim()) { + throw new Error("Missing required --socket "); + } + + return { socketPath: socketPath.trim() }; +} + +async function main() { + const { socketPath } = parseArgs(process.argv.slice(2)); + const socket = net.createConnection(socketPath); + + socket.on("error", (error) => { + process.stderr.write(`MCP stdio-socket bridge error: ${error.message}\n`); + process.exitCode = 1; + }); + + process.stdin.on("error", (error) => { + process.stderr.write(`MCP stdio-socket bridge stdin error: ${error.message}\n`); + socket.destroy(); + process.exitCode = 1; + }); + + process.stdout.on("error", (error) => { + process.stderr.write(`MCP stdio-socket bridge stdout error: ${error.message}\n`); + socket.destroy(); + process.exitCode = 1; + }); + + await new Promise((resolve, reject) => { + socket.once("connect", resolve); + socket.once("error", reject); + }); + + process.stdin.pipe(socket); + socket.pipe(process.stdout); + + await new Promise((resolve) => { + socket.once("close", resolve); + process.stdin.once("end", () => { + socket.end(); + resolve(); + }); + }); +} + +main().catch((error) => { + process.stderr.write(`${error instanceof Error ? error.message : String(error)}\n`); + process.exit(1); +}); diff --git a/packages/server/scripts/voice-main-daemon-e2e.ts b/packages/server/scripts/voice-main-daemon-e2e.ts new file mode 100644 index 000000000..a393dd0f6 --- /dev/null +++ b/packages/server/scripts/voice-main-daemon-e2e.ts @@ -0,0 +1,172 @@ +import { randomUUID } from "node:crypto"; +import pino from "pino"; + +import { DaemonClient } from "../src/server/test-utils/daemon-client.js"; +import { OpenAITTS } from "../src/server/speech/providers/openai/tts.js"; + +type RoundTripResult = { + voiceAgentId: string; + speakToolCalls: number; + audioChunks: number; +}; + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +async function streamToBuffer(stream: AsyncIterable): Promise { + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as ArrayBufferLike)); + } + return Buffer.concat(chunks); +} + +async function synthesizeInput(text: string, logger: pino.Logger): Promise { + const apiKey = process.env.OPENAI_API_KEY; + if (!apiKey) { + throw new Error("OPENAI_API_KEY is required for voice-main-daemon-e2e"); + } + + const tts = new OpenAITTS( + { + apiKey, + responseFormat: "pcm", + voice: "alloy", + }, + logger + ); + + const generated = await tts.synthesizeSpeech(text); + return streamToBuffer(generated.stream as AsyncIterable); +} + +async function runVoiceRoundTrip(params: { + daemonUrl: string; + voiceAgentId: string; + speechText: string; + timeoutMs: number; +}): Promise { + const client = new DaemonClient({ url: `${params.daemonUrl}/ws` }); + await client.connect(); + client.subscribeAgentUpdates({ subscriptionId: `voice-e2e-${randomUUID()}` }); + + const mode = await client.setVoiceMode(true, params.voiceAgentId); + if (!mode.accepted) { + throw new Error(`set_voice_mode rejected: ${mode.error ?? "unknown error"}`); + } + if (!mode.voiceAgentId) { + throw new Error("set_voice_mode returned null voiceAgentId"); + } + + const activeVoiceAgentId = mode.voiceAgentId; + const pcm = await synthesizeInput(params.speechText, pino({ level: "warn" })); + + let audioChunks = 0; + let speakToolCalls = 0; + let sawLastAudio = false; + + const done = new Promise((resolve) => { + const offAudio = client.on("audio_output", (msg) => { + if (msg.type !== "audio_output") return; + audioChunks += 1; + if (msg.payload.isLastChunk) { + sawLastAudio = true; + offAudio(); + resolve(); + } + }); + }); + + const offStream = client.on("agent_stream", (msg) => { + if (msg.type !== "agent_stream") return; + if (msg.payload.agentId !== activeVoiceAgentId) return; + if (msg.payload.event.type !== "timeline") return; + const item = msg.payload.event.item; + if (item.type !== "tool_call") return; + if (typeof item.name === "string" && item.name.toLowerCase().includes("speak")) { + speakToolCalls += 1; + } + }); + + const chunkBytes = 4800; + for (let offset = 0; offset < pcm.length; offset += chunkBytes) { + const chunk = pcm.subarray(offset, Math.min(pcm.length, offset + chunkBytes)); + const isLast = offset + chunkBytes >= pcm.length; + await client.sendVoiceAudioChunk( + chunk.toString("base64"), + "audio/pcm;rate=24000;bits=16", + isLast + ); + } + + try { + await Promise.race([ + done, + (async () => { + await sleep(params.timeoutMs); + throw new Error("Timed out waiting for audio_output"); + })(), + ]); + } finally { + offStream(); + await client.setVoiceMode(false).catch(() => undefined); + await client.close().catch(() => undefined); + } + + if (!sawLastAudio) { + throw new Error("No final audio chunk observed"); + } + if (audioChunks === 0) { + throw new Error("No audio chunks observed"); + } + if (speakToolCalls === 0) { + throw new Error("No speak tool call observed in agent stream"); + } + + return { + voiceAgentId: activeVoiceAgentId, + speakToolCalls, + audioChunks, + }; +} + +async function main(): Promise { + const daemonUrl = process.env.PASEO_DAEMON_URL ?? "ws://localhost:6767".replace(/^ws/, "http"); + const timeoutMs = Number(process.env.PASEO_VOICE_E2E_TIMEOUT_MS ?? "120000"); + const voiceAgentId = randomUUID(); + + const newAgentResult = await runVoiceRoundTrip({ + daemonUrl, + voiceAgentId, + speechText: + "Use the speak tool and say exactly: new voice agent round trip successful.", + timeoutMs, + }); + + const resumedResult = await runVoiceRoundTrip({ + daemonUrl, + voiceAgentId: newAgentResult.voiceAgentId, + speechText: + "Use the speak tool and say exactly: resumed voice agent round trip successful.", + timeoutMs, + }); + + console.log( + JSON.stringify( + { + ok: true, + daemonUrl, + newAgentResult, + resumedResult, + }, + null, + 2 + ) + ); +} + +main().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/packages/server/scripts/voice-mcp-agent-smoke.ts b/packages/server/scripts/voice-mcp-agent-smoke.ts new file mode 100644 index 000000000..41faa4eb4 --- /dev/null +++ b/packages/server/scripts/voice-mcp-agent-smoke.ts @@ -0,0 +1,207 @@ +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { mkdir, mkdtemp, rm } from "node:fs/promises"; +import { randomUUID } from "node:crypto"; +import pino from "pino"; + +import { AgentManager } from "../src/server/agent/agent-manager.js"; +import { AgentStorage } from "../src/server/agent/agent-storage.js"; +import { buildProviderRegistry, shutdownProviders } from "../src/server/agent/provider-registry.js"; +import type { + AgentProvider, + AgentSessionConfig, + AgentStreamEvent, +} from "../src/server/agent/agent-sdk-types.js"; +import { buildVoiceAgentMcpServerConfig } from "../src/server/session.js"; +import { createAgentMcpServer } from "../src/server/agent/mcp-server.js"; +import { createVoiceMcpSocketBridgeManager } from "../src/server/voice-mcp-bridge.js"; + +type CliOptions = { + provider: AgentProvider; + model?: string; + timeoutMs: number; +}; + +function parseArgs(argv: string[]): CliOptions { + let provider: AgentProvider = "claude"; + let model: string | undefined; + let timeoutMs = 120_000; + + for (let i = 0; i < argv.length; i += 1) { + const arg = argv[i]; + if (arg === "--provider" && argv[i + 1]) { + provider = argv[i + 1] as AgentProvider; + i += 1; + continue; + } + if (arg === "--model" && argv[i + 1]) { + model = argv[i + 1]; + i += 1; + continue; + } + if (arg === "--timeout-ms" && argv[i + 1]) { + timeoutMs = Number(argv[i + 1]) || timeoutMs; + i += 1; + } + } + + return { provider, model, timeoutMs }; +} + +async function streamWithTimeout( + iterator: AsyncGenerator, + timeoutMs: number, + onEvent: (event: AgentStreamEvent) => Promise | void +): Promise { + const timeout = new Promise((_, reject) => { + setTimeout(() => reject(new Error(`Timed out after ${timeoutMs}ms`)), timeoutMs); + }); + + const consume = (async () => { + for await (const event of iterator) { + await onEvent(event); + } + })(); + + await Promise.race([consume, timeout]); +} + +async function main(): Promise { + const opts = parseArgs(process.argv.slice(2)); + const logger = pino({ level: "info" }); + const scriptDir = path.dirname(fileURLToPath(import.meta.url)); + const serverDir = path.resolve(scriptDir, ".."); + const tmpRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-voice-mcp-smoke-")); + const paseoHome = path.join(tmpRoot, ".paseo"); + const storageDir = path.join(paseoHome, "agents"); + const voiceBridgeRuntimeDir = path.join(tmpRoot, "voice-mcp"); + const voiceWorkspace = path.join(tmpRoot, "voice-agent-workspace"); + await mkdir(voiceWorkspace, { recursive: true }); + + const providerRegistry = buildProviderRegistry(logger); + const providerDefinition = providerRegistry[opts.provider]; + if (!providerDefinition) { + throw new Error(`Unsupported provider: ${opts.provider}`); + } + + const agentStorage = new AgentStorage(storageDir, logger); + const agentManager = new AgentManager({ + logger, + registry: agentStorage, + clients: { + [opts.provider]: providerDefinition.createClient(logger), + }, + }); + + const speakCalls: string[] = []; + const resolveSpeakHandler = (callerAgentId: string) => { + return async ({ text }: { text: string; callerAgentId: string }) => { + speakCalls.push(text); + logger.info({ callerAgentId, text }, "Smoke speak handler invoked"); + }; + }; + + const voiceBridgeManager = createVoiceMcpSocketBridgeManager({ + runtimeDir: voiceBridgeRuntimeDir, + logger, + createAgentMcpServerForCaller: async (callerAgentId) => { + return createAgentMcpServer({ + agentManager, + agentStorage, + paseoHome, + callerAgentId, + enableVoiceTools: false, + resolveSpeakHandler, + resolveCallerContext: () => ({ + childAgentDefaultLabels: { ui: "true" }, + allowCustomCwd: true, + enableVoiceTools: true, + }), + logger, + }); + }, + }); + + const agentId = randomUUID(); + const socketPath = await voiceBridgeManager.ensureBridgeForCaller(agentId); + const bridgeScriptPath = path.resolve(serverDir, "scripts/mcp-stdio-socket-bridge-cli.mjs"); + const config: AgentSessionConfig = { + provider: opts.provider, + cwd: voiceWorkspace, + modeId: "default", + ...(opts.model ? { model: opts.model } : {}), + mcpServers: { + paseo: buildVoiceAgentMcpServerConfig({ + command: process.execPath, + baseArgs: [bridgeScriptPath], + socketPath, + env: { + PASEO_HOME: paseoHome, + }, + }), + }, + }; + + const prompt = [ + "You must call the MCP tool named speak.", + "Call it exactly once with this exact text: BRIDGE_SMOKE_OK", + "Do not call any other tools.", + "After the tool call, end your response.", + ].join("\n"); + + let sawSpeakTimeline = false; + let sawTurnFailed = false; + let failureError: string | null = null; + + try { + const created = await agentManager.createAgent(config, agentId, { + labels: { surface: "voice-smoke", ui: "true" }, + }); + logger.info({ provider: opts.provider, agentId: created.id }, "Created smoke agent"); + + const iterator = agentManager.streamAgent(created.id, prompt); + await streamWithTimeout(iterator, opts.timeoutMs, async (event) => { + if (event.type === "permission_requested") { + await agentManager.respondToPermission(created.id, event.request.id, { behavior: "allow" }); + } + if (event.type === "turn_failed") { + sawTurnFailed = true; + failureError = event.error; + } + if (event.type === "timeline" && event.item.type === "tool_call") { + logger.info({ name: event.item.name, callId: event.item.callId }, "Timeline tool call"); + if (typeof event.item.name === "string" && event.item.name.toLowerCase().includes("speak")) { + sawSpeakTimeline = true; + } + } + }); + + console.log( + JSON.stringify( + { + ok: !sawTurnFailed && sawSpeakTimeline && speakCalls.length > 0, + provider: opts.provider, + agentId, + sawSpeakTimeline, + speakCalls, + sawTurnFailed, + failureError, + }, + null, + 2 + ) + ); + } finally { + await agentManager.closeAgent(agentId).catch(() => undefined); + await agentManager.flush().catch(() => undefined); + await voiceBridgeManager.stop().catch(() => undefined); + await shutdownProviders(logger).catch(() => undefined); + await rm(tmpRoot, { recursive: true, force: true }); + } +} + +main().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/packages/server/scripts/voice-roundtrip-debug.ts b/packages/server/scripts/voice-roundtrip-debug.ts new file mode 100644 index 000000000..467ff1097 --- /dev/null +++ b/packages/server/scripts/voice-roundtrip-debug.ts @@ -0,0 +1,164 @@ +import { randomUUID } from "node:crypto"; +import pino from "pino"; + +import { createTestPaseoDaemon } from "../src/server/test-utils/paseo-daemon.js"; +import { DaemonClient } from "../src/server/test-utils/daemon-client.js"; +import { OpenAITTS } from "../src/server/speech/providers/openai/tts.js"; + +async function streamToBuffer(stream: AsyncIterable): Promise { + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk as ArrayBufferLike)); + } + return Buffer.concat(chunks); +} + +async function main(): Promise { + const apiKey = process.env.OPENAI_API_KEY; + if (!apiKey) { + throw new Error("OPENAI_API_KEY is required"); + } + + const logger = pino({ level: process.env.PASEO_LOG_LEVEL ?? "info" }); + const daemon = await createTestPaseoDaemon({ + logger, + agentClients: {}, + openai: { apiKey }, + speech: { + providers: { + dictationStt: { provider: "openai", explicit: true }, + voiceStt: { provider: "openai", explicit: true }, + voiceTts: { provider: "openai", explicit: true }, + }, + }, + voiceLlmProvider: "claude", + voiceLlmProviderExplicit: true, + }); + + const client = new DaemonClient({ + url: `ws://127.0.0.1:${daemon.port}/ws`, + }); + + const cleanup = async () => { + await client.close().catch(() => undefined); + await daemon.close().catch(() => undefined); + }; + + try { + await client.connect(); + client.subscribeAgentUpdates({ subscriptionId: "voice-debug" }); + + const voiceAgentId = randomUUID(); + const mode = await client.setVoiceMode(true, voiceAgentId); + console.log("set_voice_mode_response", mode); + + const offTranscript = client.on("transcription_result", (msg) => { + if (msg.type !== "transcription_result") return; + console.log("transcription_result", { + text: msg.payload.text, + isLowConfidence: msg.payload.isLowConfidence, + }); + }); + + const offActivity = client.on("activity_log", (msg) => { + if (msg.type !== "activity_log") return; + if (msg.payload.type === "transcript" || msg.payload.type === "error" || msg.payload.type === "assistant") { + console.log("activity_log", { + type: msg.payload.type, + content: msg.payload.content, + }); + } + }); + + const offStream = client.on("agent_stream", (msg) => { + if (msg.type !== "agent_stream") return; + if (msg.payload.event.type !== "timeline") return; + const item = msg.payload.event.item; + if (item.type !== "tool_call") return; + console.log("agent_stream:tool_call", { + agentId: msg.payload.agentId, + name: item.name, + status: item.status, + }); + }); + + let audioChunkCount = 0; + const firstAudio = new Promise((resolve) => { + const offAudio = client.on("audio_output", (msg) => { + if (msg.type !== "audio_output") return; + audioChunkCount += 1; + console.log("audio_output", { + id: msg.payload.id, + groupId: msg.payload.groupId, + chunkIndex: msg.payload.chunkIndex, + isLastChunk: msg.payload.isLastChunk, + format: msg.payload.format, + }); + if (audioChunkCount === 1) { + offAudio(); + resolve(); + } + }); + }); + const lastAudio = new Promise((resolve) => { + const offAudio = client.on("audio_output", (msg) => { + if (msg.type !== "audio_output") return; + if (msg.payload.isLastChunk) { + offAudio(); + resolve(); + } + }); + }); + + const tts = new OpenAITTS( + { + apiKey, + responseFormat: "pcm", + voice: "alloy", + }, + logger + ); + const generated = await tts.synthesizeSpeech( + "Use the speak tool and say exactly round trip successful." + ); + const pcm = await streamToBuffer(generated.stream as AsyncIterable); + + const chunkBytes = 4800; + for (let offset = 0; offset < pcm.length; offset += chunkBytes) { + const chunk = pcm.subarray(offset, Math.min(pcm.length, offset + chunkBytes)); + const isLast = offset + chunkBytes >= pcm.length; + await client.sendVoiceAudioChunk( + chunk.toString("base64"), + "audio/pcm;rate=24000;bits=16", + isLast + ); + } + + await Promise.race([ + firstAudio, + new Promise((_, reject) => + setTimeout(() => reject(new Error("Timed out waiting for first audio_output")), 120000) + ), + ]); + await Promise.race([ + lastAudio, + new Promise((_, reject) => + setTimeout(() => reject(new Error("Timed out waiting for final audio_output")), 120000) + ), + ]); + + console.log("success", { audioChunkCount }); + await client.setVoiceMode(false); + + offTranscript(); + offActivity(); + offStream(); + } finally { + await cleanup(); + } +} + +main().catch((error) => { + console.error(error); + process.exit(1); +}); diff --git a/packages/server/src/server/agent/agent-projections.ts b/packages/server/src/server/agent/agent-projections.ts index ae2bff296..e12f5c2c0 100644 --- a/packages/server/src/server/agent/agent-projections.ts +++ b/packages/server/src/server/agent/agent-projections.ts @@ -129,6 +129,9 @@ function buildSerializableConfig( if (extra !== undefined) { serializable.extra = extra; } + if (config.mcpServers) { + serializable.mcpServers = config.mcpServers; + } return Object.keys(serializable).length ? serializable : null; } diff --git a/packages/server/src/server/agent/agent-storage.test.ts b/packages/server/src/server/agent/agent-storage.test.ts index ba51f3dc5..4b0004d37 100644 --- a/packages/server/src/server/agent/agent-storage.test.ts +++ b/packages/server/src/server/agent/agent-storage.test.ts @@ -39,6 +39,7 @@ function createManagedAgent( modeId: configOverrides.modeId ?? "plan", model: configOverrides.model ?? "gpt-5.1", extra: configOverrides.extra ?? { claude: { maxThinkingTokens: 1024 } }, + mcpServers: configOverrides.mcpServers, }; const session = lifecycle === "closed" @@ -119,6 +120,13 @@ describe("AgentStorage", () => { modeId: "coding", model: "gpt-5.1", extra: { claude: { maxThinkingTokens: 1024 } }, + mcpServers: { + paseo: { + type: "stdio", + command: "node", + args: ["/tmp/mcp-stdio-socket-bridge-cli.mjs", "--socket", "/tmp/test.sock"], + }, + }, }, }) ); @@ -129,6 +137,13 @@ describe("AgentStorage", () => { expect(record.provider).toBe("claude"); expect(record.config?.modeId).toBe("coding"); expect(record.config?.model).toBe("gpt-5.1"); + expect(record.config?.mcpServers).toEqual({ + paseo: { + type: "stdio", + command: "node", + args: ["/tmp/mcp-stdio-socket-bridge-cli.mjs", "--socket", "/tmp/test.sock"], + }, + }); expect(record.lastModeId).toBe("coding"); expect(record.lastStatus).toBe("idle"); diff --git a/packages/server/src/server/agent/agent-storage.ts b/packages/server/src/server/agent/agent-storage.ts index c405b0885..0af305865 100644 --- a/packages/server/src/server/agent/agent-storage.ts +++ b/packages/server/src/server/agent/agent-storage.ts @@ -15,6 +15,7 @@ const SERIALIZABLE_CONFIG_SCHEMA = z model: z.string().nullable().optional(), thinkingOptionId: z.string().nullable().optional(), extra: z.record(z.any()).nullable().optional(), + mcpServers: z.record(z.any()).nullable().optional(), }) .nullable() .optional(); @@ -61,7 +62,7 @@ const STORED_AGENT_SCHEMA = z.object({ export type SerializableAgentConfig = Pick< AgentSessionConfig, - "modeId" | "model" | "thinkingOptionId" | "extra" + "modeId" | "model" | "thinkingOptionId" | "extra" | "mcpServers" >; export type StoredAgentRecord = z.infer; diff --git a/packages/server/src/server/agent/provider-manifest.ts b/packages/server/src/server/agent/provider-manifest.ts index 189ac78f7..f2ac24dc7 100644 --- a/packages/server/src/server/agent/provider-manifest.ts +++ b/packages/server/src/server/agent/provider-manifest.ts @@ -89,7 +89,7 @@ export const AGENT_PROVIDER_DEFINITIONS: AgentProviderDefinition[] = [ voice: { enabled: true, defaultModeId: "read-only", - defaultModel: "gpt-5.2-mini", + defaultModel: "gpt-5.1-codex-mini", }, }, { diff --git a/packages/server/src/server/agent/tts-manager.ts b/packages/server/src/server/agent/tts-manager.ts index 223b03707..0d6060b87 100644 --- a/packages/server/src/server/agent/tts-manager.ts +++ b/packages/server/src/server/agent/tts-manager.ts @@ -229,6 +229,17 @@ export class TTSManager { isVoiceMode, }, }); + this.logger.info( + { + audioId, + chunkId, + chunkIndex, + isLastChunk: next.done, + bytes: chunkBuffer.length, + isVoiceMode, + }, + "Emitted audio_output chunk to client" + ); chunkIndex += 1; @@ -287,6 +298,15 @@ export class TTSManager { } pending.pendingChunks = Math.max(0, pending.pendingChunks - 1); + this.logger.info( + { + chunkId, + audioId, + remainingPendingChunks: pending.pendingChunks, + streamEnded: pending.streamEnded, + }, + "Received audio playback confirmation from client" + ); if (pending.pendingChunks === 0 && pending.streamEnded) { pending.resolve(); diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 11b5f2705..d92db4a2b 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -67,39 +67,29 @@ import type { import { AGENT_PROVIDER_DEFINITIONS } from "./agent/provider-manifest.js"; import { acquirePidLock, releasePidLock } from "./pid-lock.js"; import { isHostAllowed, type AllowedHostsConfig } from "./allowed-hosts.js"; -import { createVoiceMcpBridgeSocketServer, type VoiceMcpBridgeSocketServer } from "./voice-mcp-bridge.js"; +import { + createVoiceMcpSocketBridgeManager, + type VoiceMcpSocketBridgeManager, +} from "./voice-mcp-bridge.js"; +import { resolveVoiceMcpBridgeFromRuntime } from "./voice-mcp-bridge-command.js"; type AgentMcpTransportMap = Map; function resolveVoiceMcpBridgeCommand(logger: Logger): { command: string; baseArgs: string[] } { - const explicit = process.env.PASEO_BIN_PATH?.trim(); - if (explicit) { - const resolved = { command: explicit, baseArgs: ["__paseo_voice_mcp_bridge"] }; - logger.info({ source: "PASEO_BIN_PATH", command: resolved.command, baseArgs: resolved.baseArgs }, "Resolved voice MCP bridge command"); - return resolved; - } - - const argv1 = process.argv[1]?.trim(); - if (!argv1) { - logger.warn("Could not resolve argv[1] for voice MCP bridge; falling back to 'paseo'"); - const resolved = { command: "paseo", baseArgs: ["__paseo_voice_mcp_bridge"] }; - logger.info({ source: "fallback", command: resolved.command, baseArgs: resolved.baseArgs }, "Resolved voice MCP bridge command"); - return resolved; - } - - const base = path.basename(argv1).toLowerCase(); - if (base.includes("tsx") && process.argv[2]) { - const resolved = { - command: process.execPath, - baseArgs: [argv1, process.argv[2], "__paseo_voice_mcp_bridge"], - }; - logger.info({ source: "tsx", command: resolved.command, baseArgs: resolved.baseArgs }, "Resolved voice MCP bridge command"); - return resolved; - } - - const resolved = { command: argv1, baseArgs: ["__paseo_voice_mcp_bridge"] }; - logger.info({ source: "argv", command: resolved.command, baseArgs: resolved.baseArgs }, "Resolved voice MCP bridge command"); - return resolved; + const decision = resolveVoiceMcpBridgeFromRuntime({ + bootstrapModuleUrl: import.meta.url, + execPath: process.execPath, + explicitScriptPath: process.env.PASEO_MCP_STDIO_SOCKET_BRIDGE_SCRIPT, + }); + logger.info( + { + source: decision.source, + command: decision.resolved.command, + baseArgs: decision.resolved.baseArgs, + }, + "Resolved voice MCP bridge command" + ); + return decision.resolved; } export type PaseoOpenAIConfig = OpenAiSpeechProviderConfig; @@ -382,7 +372,7 @@ export async function createPaseoDaemon( "Voice LLM provider reconciliation completed" ); let wsServer: VoiceAssistantWebSocketServer | null = null; - let voiceMcpBridgeServer: VoiceMcpBridgeSocketServer | null = null; + let voiceMcpBridgeManager: VoiceMcpSocketBridgeManager | null = null; // Create in-memory transport for Session's Agent MCP client (voice assistant tools) const createInMemoryAgentMcpTransport = async (): Promise => { @@ -522,10 +512,10 @@ export async function createPaseoDaemon( logger.info("Agent MCP HTTP endpoint disabled"); } - const voiceMcpSocketPath = path.join(config.paseoHome, "runtime", "voice-mcp.sock"); + const voiceMcpSocketDir = path.join(config.paseoHome, "runtime", "voice-mcp"); const voiceMcpBridgeCommand = resolveVoiceMcpBridgeCommand(logger); - voiceMcpBridgeServer = createVoiceMcpBridgeSocketServer({ - socketPath: voiceMcpSocketPath, + voiceMcpBridgeManager = createVoiceMcpSocketBridgeManager({ + runtimeDir: voiceMcpSocketDir, logger, createAgentMcpServerForCaller: async (callerAgentId) => { return createAgentMcpServer({ @@ -571,15 +561,16 @@ export async function createPaseoDaemon( voiceLlmModel: resolvedVoiceLlmModel, voiceAgentMcpStdio: { command: voiceMcpBridgeCommand.command, - baseArgs: [ - ...voiceMcpBridgeCommand.baseArgs, - "--socket", - voiceMcpSocketPath, - ], + baseArgs: [...voiceMcpBridgeCommand.baseArgs], env: { PASEO_HOME: config.paseoHome, }, }, + ensureVoiceMcpSocketForAgent: (agentId) => + voiceMcpBridgeManager?.ensureBridgeForCaller(agentId) ?? + Promise.reject(new Error("Voice MCP bridge manager is not initialized")), + removeVoiceMcpSocketForAgent: (agentId) => + voiceMcpBridgeManager?.removeBridgeForCaller(agentId) ?? Promise.resolve(), }, { finalTimeoutMs: config.dictationFinalTimeoutMs, @@ -675,9 +666,6 @@ export async function createPaseoDaemon( httpServer.listen(listenTarget.path); } }); - if (voiceMcpBridgeServer) { - await voiceMcpBridgeServer.start(); - } }; const stop = async () => { @@ -692,8 +680,8 @@ export async function createPaseoDaemon( if (wsServer) { await wsServer.close(); } - if (voiceMcpBridgeServer) { - await voiceMcpBridgeServer.stop().catch(() => undefined); + if (voiceMcpBridgeManager) { + await voiceMcpBridgeManager.stop().catch(() => undefined); } await new Promise((resolve) => { httpServer.close(() => resolve()); diff --git a/packages/server/src/server/exports.ts b/packages/server/src/server/exports.ts index c0779c865..31262a6a0 100644 --- a/packages/server/src/server/exports.ts +++ b/packages/server/src/server/exports.ts @@ -5,7 +5,6 @@ export { resolvePaseoHome } from "./paseo-home.js"; export { createRootLogger, type LogLevel, type LogFormat } from "./logger.js"; export { loadPersistedConfig, type PersistedConfig } from "./persisted-config.js"; export { DaemonClient, type DaemonClientConfig, type ConnectionState, type DaemonEvent } from "../client/daemon-client.js"; -export { runVoiceMcpBridgeCli } from "./voice-mcp-bridge.js"; // Agent SDK types for CLI commands export type { diff --git a/packages/server/src/server/index.ts b/packages/server/src/server/index.ts index 60c91a602..7737855f8 100644 --- a/packages/server/src/server/index.ts +++ b/packages/server/src/server/index.ts @@ -10,15 +10,8 @@ import { resolvePaseoHome } from "./paseo-home.js"; import { createRootLogger } from "./logger.js"; import { loadPersistedConfig } from "./persisted-config.js"; import { PidLockError } from "./pid-lock.js"; -import { runVoiceMcpBridgeCli } from "./voice-mcp-bridge.js"; async function main() { - const bridgeArgIndex = process.argv.findIndex((arg) => arg === "__paseo_voice_mcp_bridge"); - if (bridgeArgIndex >= 0) { - await runVoiceMcpBridgeCli(process.argv.slice(bridgeArgIndex + 1)); - return; - } - let paseoHome: string; let logger: ReturnType; let config: ReturnType; diff --git a/packages/server/src/server/persistence-hooks.ts b/packages/server/src/server/persistence-hooks.ts index 7552efc61..dc5a3b2ae 100644 --- a/packages/server/src/server/persistence-hooks.ts +++ b/packages/server/src/server/persistence-hooks.ts @@ -56,6 +56,7 @@ export function buildConfigOverrides( thinkingOptionId: record.config?.thinkingOptionId ?? undefined, title: record.title ?? undefined, extra: record.config?.extra ?? undefined, + mcpServers: record.config?.mcpServers ?? undefined, }; } @@ -74,6 +75,7 @@ export function buildSessionConfig( thinkingOptionId: overrides.thinkingOptionId, title: overrides.title, extra: overrides.extra, + mcpServers: overrides.mcpServers, }; } diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 8e0a8bc6e..bbb984027 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -128,16 +128,27 @@ const AUTO_GEN_MODEL = "haiku"; const VOICE_AGENT_SYSTEM_INSTRUCTION = [ "You are the Paseo voice assistant.", "The user cannot see your chat messages or tool calls.", - "Always narrate everything through the speak tool.", + "Always use the speak tool for all user-facing communication.", + "Before calling any non-speak tool, first call speak with a short acknowledgement of what you heard and what you will do next.", + "For long-running work, use speak to provide progress updates before and during execution.", + "Treat the user input as transcribed speech and confirm intent in your first speak response to catch transcription errors early.", + "If the transcription seems incomplete, cut off, or ambiguous, ask a clarifying question via speak before taking action.", "Use concise plain language suitable for speech output.", "Never use bash, file-edit, or web tools directly.", "Only use the paseo MCP tools.", ].join(" "); +function escapeXmlText(value: string): string { + return value + .replace(/&/g, "&") + .replace(//g, ">"); +} + export function buildVoiceAgentMcpServerConfig(params: { - callerAgentId: string; command: string; baseArgs: string[]; + socketPath: string; env?: Record; }): { type: "stdio"; @@ -148,7 +159,7 @@ export function buildVoiceAgentMcpServerConfig(params: { return { type: "stdio", command: params.command, - args: [...params.baseArgs, "--caller-agent-id", params.callerAgentId], + args: [...params.baseArgs, "--socket", params.socketPath], ...(params.env ? { env: params.env } : {}), }; } @@ -214,6 +225,8 @@ export type SessionOptions = { unregisterVoiceSpeakHandler?: (agentId: string) => void; registerVoiceCallerContext?: (agentId: string, context: VoiceCallerContext) => void; unregisterVoiceCallerContext?: (agentId: string) => void; + ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise; + removeVoiceMcpSocketForAgent?: (agentId: string) => Promise; }; dictation?: { finalTimeoutMs?: number; @@ -380,6 +393,8 @@ export class Session { context: VoiceCallerContext ) => void; private readonly unregisterVoiceCallerContext?: (agentId: string) => void; + private readonly ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise; + private readonly removeVoiceMcpSocketForAgent?: (agentId: string) => Promise; private voiceAssistantAgentId: string | null = null; constructor(options: SessionOptions) { @@ -428,6 +443,8 @@ export class Session { this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler; this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext; this.unregisterVoiceCallerContext = voiceBridge?.unregisterVoiceCallerContext; + this.ensureVoiceMcpSocketForAgent = voiceBridge?.ensureVoiceMcpSocketForAgent; + this.removeVoiceMcpSocketForAgent = voiceBridge?.removeVoiceMcpSocketForAgent; this.abortController = new AbortController(); this.sessionLogger = logger.child({ module: "session", @@ -1299,6 +1316,17 @@ export class Session { } this.isVoiceMode = false; + const disabledVoiceAgentId = this.voiceAssistantAgentId; + if (disabledVoiceAgentId && this.removeVoiceMcpSocketForAgent) { + try { + await this.removeVoiceMcpSocketForAgent(disabledVoiceAgentId); + } catch (error) { + this.sessionLogger.warn( + { err: error, voiceAssistantAgentId: disabledVoiceAgentId }, + "Failed to remove voice MCP socket bridge on disable" + ); + } + } this.sessionLogger.info( { voiceAssistantAgentId: this.voiceAssistantAgentId }, "Voice mode disabled (agent-backed)" @@ -1340,7 +1368,7 @@ export class Session { ...fallbackCandidates.filter( (candidate) => !scopedCandidates.some((scoped) => scoped.id === candidate.id) ), - ].filter((agent) => !agent.archivedAt); + ].filter((agent) => !agent.archivedAt && agent.labels?.ui !== "false"); if (available.length === 0) { return null; } @@ -4474,14 +4502,23 @@ export class Session { } private async ensureVoiceAssistantAgent(): Promise { + const ensureVoiceSocket = this.ensureVoiceMcpSocketForAgent; + if (!ensureVoiceSocket) { + throw new Error("Voice MCP socket bridge is not configured"); + } + if (this.voiceAssistantAgentId) { const existing = this.agentManager.getAgent(this.voiceAssistantAgentId); if (existing) { + await ensureVoiceSocket(existing.id); + this.registerVoiceBridgeForAgent(existing.id); return existing.id; } try { const hydrated = await this.ensureAgentLoaded(this.voiceAssistantAgentId); + await ensureVoiceSocket(hydrated.id); this.voiceAssistantAgentId = hydrated.id; + this.registerVoiceBridgeForAgent(hydrated.id); return hydrated.id; } catch (error) { this.sessionLogger.debug( @@ -4503,6 +4540,7 @@ export class Session { if (!mcpStdio) { throw new Error("Voice MCP stdio bridge is not configured"); } + const socketPath = await ensureVoiceSocket(voiceAgentId); const model = this.voiceLlmModel?.trim() || undefined; const config: AgentSessionConfig = { @@ -4512,9 +4550,9 @@ export class Session { ...(model ? { model } : {}), mcpServers: { paseo: buildVoiceAgentMcpServerConfig({ - callerAgentId: voiceAgentId, command: mcpStdio.command, baseArgs: mcpStdio.baseArgs, + socketPath, env: mcpStdio.env, }), }, @@ -4523,48 +4561,18 @@ export class Session { const created = await this.agentManager.createAgent(config, voiceAgentId, { labels: { surface: "voice", - ui: "false", + ui: "true", voiceClientId: this.clientId, }, }); this.voiceAssistantAgentId = created.id; - let speakHandlerRegistered = false; - let callerContextRegistered = false; try { - this.registerVoiceSpeakHandler?.(created.id, async ({ text, signal }) => { - const abortSignal = signal ?? this.abortController.signal; - await this.ttsManager.generateAndWaitForPlayback( - text, - (msg) => this.emit(msg), - abortSignal, - true - ); - this.emit({ - type: "activity_log", - payload: { - id: uuidv4(), - timestamp: new Date(), - type: "assistant", - content: text, - }, - }); - }); - speakHandlerRegistered = true; - - this.registerVoiceCallerContext?.(created.id, { - childAgentDefaultLabels: { ui: "true" }, - allowCustomCwd: true, - enableVoiceTools: true, - }); - callerContextRegistered = true; + this.registerVoiceBridgeForAgent(created.id); } catch (error) { - if (speakHandlerRegistered) { - this.unregisterVoiceSpeakHandler?.(created.id); - } - if (callerContextRegistered) { - this.unregisterVoiceCallerContext?.(created.id); - } + this.unregisterVoiceSpeakHandler?.(created.id); + this.unregisterVoiceCallerContext?.(created.id); + await this.removeVoiceMcpSocketForAgent?.(created.id).catch(() => undefined); await this.agentManager.closeAgent(created.id).catch(() => undefined); this.voiceAssistantAgentId = null; throw error; @@ -4582,11 +4590,53 @@ export class Session { return created.id; } + private registerVoiceBridgeForAgent(agentId: string): void { + this.registerVoiceSpeakHandler?.(agentId, async ({ text, signal }) => { + this.sessionLogger.info( + { + voiceAssistantAgentId: 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( + { voiceAssistantAgentId: 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: { ui: "true" }, + allowCustomCwd: true, + enableVoiceTools: true, + }); + } + private buildVoiceAgentPrompt(userText: string): string { + const transcription = escapeXmlText(userText.trim()); return [ VOICE_AGENT_SYSTEM_INSTRUCTION, "", - `User said: ${userText.trim()}`, + "", + transcription, + "", ].join("\n"); } @@ -4599,6 +4649,15 @@ export class Session { await this.interruptAgentIfRunning(agentId); + this.sessionLogger.info( + { + voiceAssistantAgentId: agentId, + userTextLength: userText.length, + userTextPreview: userText.slice(0, 160), + }, + "Starting voice agent turn" + ); + const prompt = this.buildVoiceAgentPrompt(userText); this.agentManager.recordUserMessage(agentId, userText); @@ -4610,6 +4669,14 @@ export class Session { } if (event.type === "timeline") { if (event.item.type === "tool_call" && typeof event.item.name === "string") { + this.sessionLogger.info( + { + voiceAssistantAgentId: agentId, + toolName: event.item.name, + callId: event.item.callId, + }, + "Voice agent timeline tool call" + ); if (event.item.name.toLowerCase().includes("speak")) { sawSpeakToolCall = true; } @@ -4638,6 +4705,11 @@ export class Session { { voiceAssistantAgentId: agentId }, "Voice agent turn completed without speak tool call; no audio playback emitted" ); + } else { + this.sessionLogger.info( + { voiceAssistantAgentId: agentId }, + "Voice agent turn included speak tool call" + ); } } @@ -4920,6 +4992,7 @@ export class Session { } finally { this.unregisterVoiceSpeakHandler?.(this.voiceAssistantAgentId); this.unregisterVoiceCallerContext?.(this.voiceAssistantAgentId); + await this.removeVoiceMcpSocketForAgent?.(this.voiceAssistantAgentId).catch(() => undefined); this.voiceAssistantAgentId = null; } } diff --git a/packages/server/src/server/session.voice-mcp-config.test.ts b/packages/server/src/server/session.voice-mcp-config.test.ts index 14f4c84e5..ae69dd65f 100644 --- a/packages/server/src/server/session.voice-mcp-config.test.ts +++ b/packages/server/src/server/session.voice-mcp-config.test.ts @@ -5,22 +5,20 @@ import { buildVoiceAgentMcpServerConfig } from "./session.js"; describe("voice MCP stdio config", () => { test("builds stdio MCP config for voice agent", () => { const config = buildVoiceAgentMcpServerConfig({ - callerAgentId: "voice-agent-123", - command: "/usr/local/bin/paseo", - baseArgs: ["__paseo_voice_mcp_bridge", "--socket", "/tmp/paseo-voice.sock"], + command: "/usr/local/bin/node", + baseArgs: ["/tmp/mcp-stdio-socket-bridge-cli.mjs"], + socketPath: "/tmp/paseo-voice.sock", env: { PASEO_HOME: "/tmp/paseo-home", }, }); expect(config.type).toBe("stdio"); - expect(config.command).toBe("/usr/local/bin/paseo"); + expect(config.command).toBe("/usr/local/bin/node"); expect(config.args).toEqual([ - "__paseo_voice_mcp_bridge", + "/tmp/mcp-stdio-socket-bridge-cli.mjs", "--socket", "/tmp/paseo-voice.sock", - "--caller-agent-id", - "voice-agent-123", ]); expect(config.env).toEqual({ PASEO_HOME: "/tmp/paseo-home", diff --git a/packages/server/src/server/voice-local-agent.e2e.test.ts b/packages/server/src/server/voice-local-agent.e2e.test.ts index 59e23ee8e..795c2afea 100644 --- a/packages/server/src/server/voice-local-agent.e2e.test.ts +++ b/packages/server/src/server/voice-local-agent.e2e.test.ts @@ -58,7 +58,7 @@ function waitForSignal( }, voiceLlmProvider: "codex", voiceLlmProviderExplicit: true, - voiceLlmModel: "gpt-5.2-mini", + voiceLlmModel: "gpt-5.1-codex-mini", }); }, 120000); diff --git a/packages/server/src/server/voice-mcp-bridge-command.test.ts b/packages/server/src/server/voice-mcp-bridge-command.test.ts new file mode 100644 index 000000000..ba5893ac1 --- /dev/null +++ b/packages/server/src/server/voice-mcp-bridge-command.test.ts @@ -0,0 +1,50 @@ +import path from "node:path"; +import { fileURLToPath } from "node:url"; +import { describe, expect, test } from "vitest"; + +import { + resolveVoiceMcpBridgeFromRuntime, + resolveVoiceMcpBridgeScriptPath, +} from "./voice-mcp-bridge-command.js"; + +describe("resolveVoiceMcpBridgeFromRuntime", () => { + const bootstrapModuleUrl = new URL("./bootstrap.ts", import.meta.url).toString(); + + test("resolves default JS bridge script with node execPath", () => { + const result = resolveVoiceMcpBridgeFromRuntime({ + bootstrapModuleUrl, + execPath: "/usr/local/bin/node", + }); + + const expectedScriptPath = fileURLToPath( + new URL("../../scripts/mcp-stdio-socket-bridge-cli.mjs", bootstrapModuleUrl) + ); + + expect(result.source).toBe("default-js-script"); + expect(result.resolved.command).toBe("/usr/local/bin/node"); + expect(result.resolved.baseArgs).toEqual([expectedScriptPath]); + }); + + test("uses explicit script override when provided", () => { + const explicitScriptPath = path.resolve(process.cwd(), "scripts/mcp-stdio-socket-bridge-cli.mjs"); + + const result = resolveVoiceMcpBridgeFromRuntime({ + bootstrapModuleUrl, + execPath: "/usr/local/bin/node", + explicitScriptPath, + }); + + expect(result.source).toBe("explicit-js-script"); + expect(result.resolved.command).toBe("/usr/local/bin/node"); + expect(result.resolved.baseArgs).toEqual([explicitScriptPath]); + }); + + test("throws when explicit script path is missing", () => { + expect(() => + resolveVoiceMcpBridgeScriptPath({ + bootstrapModuleUrl, + explicitScriptPath: "/tmp/does-not-exist-voice-bridge-script.mjs", + }) + ).toThrow("MCP stdio-socket bridge script not found"); + }); +}); diff --git a/packages/server/src/server/voice-mcp-bridge-command.ts b/packages/server/src/server/voice-mcp-bridge-command.ts new file mode 100644 index 000000000..63fbfe75b --- /dev/null +++ b/packages/server/src/server/voice-mcp-bridge-command.ts @@ -0,0 +1,50 @@ +import { existsSync } from "node:fs"; +import { fileURLToPath } from "node:url"; + +export type VoiceMcpBridgeCommand = { command: string; baseArgs: string[] }; + +const DEFAULT_BRIDGE_SCRIPT_RELATIVE_URL = "../../scripts/mcp-stdio-socket-bridge-cli.mjs"; + +export function resolveVoiceMcpBridgeScriptPath(params: { + bootstrapModuleUrl: string; + explicitScriptPath?: string | null; +}): string { + const explicitScriptPath = params.explicitScriptPath?.trim(); + if (explicitScriptPath) { + if (!existsSync(explicitScriptPath)) { + throw new Error( + `MCP stdio-socket bridge script not found at configured path: ${explicitScriptPath}` + ); + } + return explicitScriptPath; + } + + const scriptPath = fileURLToPath( + new URL(DEFAULT_BRIDGE_SCRIPT_RELATIVE_URL, params.bootstrapModuleUrl) + ); + if (!existsSync(scriptPath)) { + throw new Error(`MCP stdio-socket bridge script not found: ${scriptPath}`); + } + return scriptPath; +} + +export function resolveVoiceMcpBridgeFromRuntime(params: { + bootstrapModuleUrl: string; + execPath: string; + explicitScriptPath?: string | null; +}): { + resolved: VoiceMcpBridgeCommand; + source: string; +} { + const scriptPath = resolveVoiceMcpBridgeScriptPath({ + bootstrapModuleUrl: params.bootstrapModuleUrl, + explicitScriptPath: params.explicitScriptPath, + }); + return { + source: params.explicitScriptPath?.trim() ? "explicit-js-script" : "default-js-script", + resolved: { + command: params.execPath, + baseArgs: [scriptPath], + }, + }; +} diff --git a/packages/server/src/server/voice-mcp-bridge.test.ts b/packages/server/src/server/voice-mcp-bridge.test.ts index 942d55925..50a1285e7 100644 --- a/packages/server/src/server/voice-mcp-bridge.test.ts +++ b/packages/server/src/server/voice-mcp-bridge.test.ts @@ -8,16 +8,15 @@ import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js" import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import pino from "pino"; -import { createVoiceMcpBridgeSocketServer } from "./voice-mcp-bridge.js"; +import { createVoiceMcpSocketBridgeManager } from "./voice-mcp-bridge.js"; describe("voice MCP bridge", () => { - test("proxies stdio MCP messages through unix socket bridge", async () => { + test("proxies stdio MCP bytes through per-agent unix socket bridge", async () => { const tmpRoot = await mkdtemp(path.join(os.tmpdir(), "paseo-voice-mcp-bridge-")); - const socketPath = path.join(tmpRoot, "voice-mcp.sock"); const callerAgentId = "voice-agent-bridge-test"; - const bridge = createVoiceMcpBridgeSocketServer({ - socketPath, + const bridgeManager = createVoiceMcpSocketBridgeManager({ + runtimeDir: tmpRoot, logger: pino({ level: "silent" }), createAgentMcpServerForCaller: async (callerId) => { const server = new McpServer({ @@ -53,21 +52,16 @@ describe("voice MCP bridge", () => { }, }); - await bridge.start(); + const socketPath = await bridgeManager.ensureBridgeForCaller(callerAgentId); - const tsxBin = path.resolve(process.cwd(), "../../node_modules/.bin/tsx"); - const serverIndex = path.resolve(process.cwd(), "src/server/index.ts"); + const bridgeScript = path.resolve(process.cwd(), "scripts/mcp-stdio-socket-bridge-cli.mjs"); const transport = new StdioClientTransport({ command: process.execPath, args: [ - tsxBin, - serverIndex, - "__paseo_voice_mcp_bridge", + bridgeScript, "--socket", socketPath, - "--caller-agent-id", - callerAgentId, ], }); @@ -86,7 +80,7 @@ describe("voice MCP bridge", () => { expect(payload?.callerAgentId).toBe(callerAgentId); } finally { await client.close(); - await bridge.stop(); + await bridgeManager.stop(); await rm(tmpRoot, { recursive: true, force: true }); } }, 30_000); diff --git a/packages/server/src/server/voice-mcp-bridge.ts b/packages/server/src/server/voice-mcp-bridge.ts index ed6728fb6..1b46607dc 100644 --- a/packages/server/src/server/voice-mcp-bridge.ts +++ b/packages/server/src/server/voice-mcp-bridge.ts @@ -2,336 +2,147 @@ import net from "node:net"; import path from "node:path"; import { mkdir, rm } from "node:fs/promises"; import type { Logger } from "pino"; -import pino from "pino"; -import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; -import type { JSONRPCMessage } from "@modelcontextprotocol/sdk/types.js"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js"; +type BridgeServer = { + connect: (transport: StdioServerTransport) => Promise; + close?: () => Promise; +}; -type BridgeEnvelope = - | { type: "init"; callerAgentId: string } - | { type: "mcp"; message: JSONRPCMessage } - | { type: "ready" } - | { type: "error"; message: string }; - -function isRecord(value: unknown): value is Record { - return typeof value === "object" && value !== null; -} - -function parseEnvelope(raw: string): BridgeEnvelope { - const parsed = JSON.parse(raw); - if (!isRecord(parsed) || typeof parsed.type !== "string") { - throw new Error("Invalid bridge envelope"); - } - if (parsed.type === "init") { - const callerAgentId = typeof parsed.callerAgentId === "string" ? parsed.callerAgentId.trim() : ""; - if (!callerAgentId) { - throw new Error("Invalid init payload: callerAgentId is required"); - } - return { type: "init", callerAgentId }; - } - if (parsed.type === "mcp") { - if (!("message" in parsed)) { - throw new Error("Invalid mcp payload: message is required"); - } - return { type: "mcp", message: parsed.message as JSONRPCMessage }; - } - if (parsed.type === "ready") return { type: "ready" }; - if (parsed.type === "error") { - return { type: "error", message: typeof parsed.message === "string" ? parsed.message : "Unknown error" }; - } - throw new Error(`Unknown envelope type: ${parsed.type}`); -} - -function encodeEnvelope(envelope: BridgeEnvelope): string { - return `${JSON.stringify(envelope)}\n`; -} - -export type VoiceMcpBridgeSocketServer = { +type BridgeEntry = { socketPath: string; - start: () => Promise; + server: net.Server; + sockets: Set; +}; + +export type VoiceMcpSocketBridgeManager = { + ensureBridgeForCaller: (callerAgentId: string) => Promise; + removeBridgeForCaller: (callerAgentId: string) => Promise; stop: () => Promise; }; -export function createVoiceMcpBridgeSocketServer(params: { - socketPath: string; +function toSocketName(callerAgentId: string): string { + return `voice-mcp-${callerAgentId}.sock`; +} + +export function createVoiceMcpSocketBridgeManager(params: { + runtimeDir: string; logger: Logger; - createAgentMcpServerForCaller: (callerAgentId: string) => Promise<{ connect: (transport: InMemoryTransport) => Promise; close?: () => Promise }>; -}): VoiceMcpBridgeSocketServer { + createAgentMcpServerForCaller: (callerAgentId: string) => Promise; +}): VoiceMcpSocketBridgeManager { const logger = params.logger.child({ module: "voice-mcp-bridge" }); - const sockets = new Set(); + const entries = new Map(); + const pendingCreates = new Map>(); - const server = net.createServer((socket) => { - sockets.add(socket); - const connectionLogger = logger.child({ component: "connection" }); - let readBuffer = ""; - let initialized = false; - let clientTransport: InMemoryTransport | null = null; - let mcpServer: { close?: () => Promise } | null = null; + const ensureBridgeForCaller = async (callerAgentId: string): Promise => { + const existing = entries.get(callerAgentId); + if (existing) { + return existing.socketPath; + } - const send = (payload: BridgeEnvelope) => { - socket.write(encodeEnvelope(payload)); - }; + const pending = pendingCreates.get(callerAgentId); + if (pending) { + return pending; + } - const fail = (message: string) => { - send({ type: "error", message }); - socket.end(); - }; + const createPromise = (async () => { + const socketPath = path.join(params.runtimeDir, toSocketName(callerAgentId)); + const sockets = new Set(); + const server = net.createServer((socket) => { + sockets.add(socket); + const connectionLogger = logger.child({ callerAgentId, component: "connection" }); - const cleanup = async () => { - sockets.delete(socket); - const closeTasks: Promise[] = []; - if (clientTransport) { - closeTasks.push(clientTransport.close().catch(() => undefined)); - } - if (mcpServer?.close) { - closeTasks.push(mcpServer.close().catch(() => undefined)); - } - await Promise.all(closeTasks); - }; + let mcpServer: BridgeServer | null = null; + let transport: StdioServerTransport | null = null; - socket.on("data", (chunk) => { - readBuffer += chunk.toString("utf8"); - while (true) { - const newlineIndex = readBuffer.indexOf("\n"); - if (newlineIndex < 0) break; - const line = readBuffer.slice(0, newlineIndex).trim(); - readBuffer = readBuffer.slice(newlineIndex + 1); - if (!line) continue; + const cleanup = async () => { + sockets.delete(socket); + await Promise.all([ + transport?.close().catch(() => undefined), + mcpServer?.close?.().catch(() => undefined), + ]); + }; - let message: BridgeEnvelope; - try { - message = parseEnvelope(line); - } catch (error) { - fail(error instanceof Error ? error.message : String(error)); - return; - } + socket.on("error", (error) => { + connectionLogger.error({ err: error }, "Voice MCP bridge socket error"); + }); + socket.on("close", () => { + void cleanup(); + }); - if (message.type === "init") { - if (initialized) { - fail("Bridge already initialized"); - return; + void (async () => { + try { + mcpServer = await params.createAgentMcpServerForCaller(callerAgentId); + transport = new StdioServerTransport(socket, socket); + await mcpServer.connect(transport); + } catch (error) { + connectionLogger.error( + { err: error, callerAgentId }, + "Failed to initialize stream-level MCP bridge connection" + ); + socket.destroy(); } - initialized = true; - void (async () => { - try { - const [proxyClient, proxyServer] = InMemoryTransport.createLinkedPair(); - const serverInstance = await params.createAgentMcpServerForCaller(message.callerAgentId); - await serverInstance.connect(proxyServer); - await proxyClient.start(); - proxyClient.onmessage = (jsonrpcMessage) => { - send({ type: "mcp", message: jsonrpcMessage }); - }; - clientTransport = proxyClient; - mcpServer = serverInstance; - send({ type: "ready" }); - } catch (error) { - connectionLogger.error({ err: error }, "Failed to initialize voice MCP bridge connection"); - fail(error instanceof Error ? error.message : String(error)); - } - })(); - continue; - } + })(); + }); - if (message.type === "mcp") { - if (!clientTransport) { - fail("Bridge is not initialized"); - return; - } - void clientTransport.send(message.message).catch((error) => { - connectionLogger.error({ err: error }, "Failed to forward MCP message"); - fail(error instanceof Error ? error.message : String(error)); - }); - continue; - } - } - }); - - socket.on("error", (error) => { - connectionLogger.error({ err: error }, "Voice MCP bridge socket error"); - }); - socket.on("close", () => { - void cleanup(); - }); - }); - - return { - socketPath: params.socketPath, - async start() { - await mkdir(path.dirname(params.socketPath), { recursive: true }); - await rm(params.socketPath, { force: true }).catch(() => undefined); + await mkdir(params.runtimeDir, { recursive: true }); + await rm(socketPath, { force: true }).catch(() => undefined); await new Promise((resolve, reject) => { server.once("error", reject); - server.listen(params.socketPath, () => { + server.listen(socketPath, () => { server.off("error", reject); resolve(); }); }); - logger.info({ socketPath: params.socketPath }, "Voice MCP bridge socket server listening"); - }, - async stop() { - for (const socket of sockets) { - socket.destroy(); - } - await new Promise((resolve, reject) => { - server.close((error) => { - if (error) reject(error); - else resolve(); - }); - }); - await rm(params.socketPath, { force: true }).catch(() => undefined); - }, + + entries.set(callerAgentId, { socketPath, server, sockets }); + logger.info( + { callerAgentId, socketPath }, + "Voice MCP per-agent socket bridge listening" + ); + return socketPath; + })(); + + pendingCreates.set(callerAgentId, createPromise); + try { + return await createPromise; + } finally { + pendingCreates.delete(callerAgentId); + } }; -} -function parseBridgeCliArgs(argv: string[]): { socketPath: string; callerAgentId: string } { - let socketPath: string | null = null; - let callerAgentId: string | null = null; - - for (let index = 0; index < argv.length; index += 1) { - const arg = argv[index]; - if (arg === "--socket") { - socketPath = argv[index + 1] ?? null; - index += 1; - continue; + const removeBridgeForCaller = async (callerAgentId: string): Promise => { + const entry = entries.get(callerAgentId); + if (!entry) { + return; } - if (arg === "--caller-agent-id") { - callerAgentId = argv[index + 1] ?? null; - index += 1; - continue; - } - } + entries.delete(callerAgentId); - if (!socketPath?.trim()) { - throw new Error("Missing required --socket "); - } - if (!callerAgentId?.trim()) { - throw new Error("Missing required --caller-agent-id "); - } + for (const socket of entry.sockets) { + socket.destroy(); + } + await new Promise((resolve, reject) => { + entry.server.close((error) => { + if (error) reject(error); + else resolve(); + }); + }); + await rm(entry.socketPath, { force: true }).catch(() => undefined); + logger.info({ callerAgentId, socketPath: entry.socketPath }, "Voice MCP socket bridge removed"); + }; + + const stop = async (): Promise => { + const activeCallerIds = Array.from(entries.keys()); + for (const callerAgentId of activeCallerIds) { + await removeBridgeForCaller(callerAgentId).catch((error) => { + logger.warn({ err: error, callerAgentId }, "Failed to stop voice MCP socket bridge"); + }); + } + }; return { - socketPath: socketPath.trim(), - callerAgentId: callerAgentId.trim(), + ensureBridgeForCaller, + removeBridgeForCaller, + stop, }; } - -export async function runVoiceMcpBridgeCli(argv: string[], logger?: Logger): Promise { - const bridgeLogger = logger ?? pino({ level: "error" }); - const parsed = parseBridgeCliArgs(argv); - - const socket = net.createConnection(parsed.socketPath); - const stdioTransport = new StdioServerTransport(process.stdin, process.stdout); - let socketBuffer = ""; - let stdioStarted = false; - const pendingMcpMessages: JSONRPCMessage[] = []; - - let resolveReady: (() => void) | null = null; - let rejectReady: ((error: Error) => void) | null = null; - const readyPromise = new Promise((resolve, reject) => { - resolveReady = () => { - resolveReady = null; - rejectReady = null; - resolve(); - }; - rejectReady = (error: Error) => { - resolveReady = null; - rejectReady = null; - reject(error); - }; - }); - - const failReady = (message: string) => { - if (rejectReady) { - rejectReady(new Error(message)); - } - }; - - const sendEnvelope = (payload: BridgeEnvelope) => { - socket.write(encodeEnvelope(payload)); - }; - - socket.on("data", (chunk) => { - socketBuffer += chunk.toString("utf8"); - while (true) { - const newlineIndex = socketBuffer.indexOf("\n"); - if (newlineIndex < 0) break; - const line = socketBuffer.slice(0, newlineIndex).trim(); - socketBuffer = socketBuffer.slice(newlineIndex + 1); - if (!line) continue; - - let envelope: BridgeEnvelope; - try { - envelope = parseEnvelope(line); - } catch (error) { - const message = error instanceof Error ? error.message : String(error); - bridgeLogger.error({ err: error }, "Failed to parse voice MCP bridge envelope"); - failReady(`Failed to parse voice MCP bridge envelope: ${message}`); - socket.destroy(); - return; - } - if (envelope.type === "ready") { - resolveReady?.(); - continue; - } - if (envelope.type === "error") { - failReady(`Voice MCP bridge error: ${envelope.message}`); - socket.destroy(); - return; - } - if (envelope.type === "mcp") { - if (!stdioStarted) { - pendingMcpMessages.push(envelope.message); - continue; - } - void stdioTransport.send(envelope.message).catch((error) => { - bridgeLogger.error({ err: error }, "Failed to forward MCP message to stdio transport"); - socket.destroy(); - }); - } - } - }); - - socket.on("error", (error) => { - bridgeLogger.error({ err: error }, "Voice MCP bridge socket client error"); - failReady(`Voice MCP bridge socket client error: ${error.message}`); - }); - - await new Promise((resolve, reject) => { - socket.once("connect", () => resolve()); - socket.once("error", reject); - }); - - sendEnvelope({ type: "init", callerAgentId: parsed.callerAgentId }); - - const readyTimeoutPromise = new Promise((_, reject) => { - const timeout = setTimeout(() => { - reject(new Error("Timed out waiting for voice MCP bridge initialization")); - }, 10000); - readyPromise.finally(() => clearTimeout(timeout)).catch(() => undefined); - }); - await Promise.race([readyPromise, readyTimeoutPromise]); - - stdioTransport.onmessage = (message) => { - sendEnvelope({ type: "mcp", message }); - }; - stdioTransport.onerror = (error) => { - bridgeLogger.error({ err: error }, "Voice MCP stdio transport error"); - socket.destroy(); - }; - stdioTransport.onclose = () => { - socket.end(); - }; - - await stdioTransport.start(); - stdioStarted = true; - for (const message of pendingMcpMessages) { - await stdioTransport.send(message); - } - - await new Promise((resolve) => { - socket.once("close", () => resolve()); - process.stdin.once("end", () => resolve()); - }); - - await stdioTransport.close().catch(() => undefined); -} diff --git a/packages/server/src/server/voice-roundtrip.e2e.test.ts b/packages/server/src/server/voice-roundtrip.e2e.test.ts new file mode 100644 index 000000000..fdeaa3016 --- /dev/null +++ b/packages/server/src/server/voice-roundtrip.e2e.test.ts @@ -0,0 +1,287 @@ +import { describe, test, expect, beforeAll, afterAll } from "vitest"; +import { randomUUID } from "node:crypto"; +import { Readable } from "node:stream"; +import pino from "pino"; + +import { + createDaemonTestContext, + type DaemonTestContext, +} from "./test-utils/index.js"; +import { OpenAITTS } from "./speech/providers/openai/tts.js"; +import { OpenAISTT } from "./speech/providers/openai/stt.js"; +import { STTManager } from "./agent/stt-manager.js"; + +const openaiApiKey = process.env.OPENAI_API_KEY ?? null; +const shouldRun = process.env.PASEO_VOICE_ROUNDTRIP_E2E === "1" && Boolean(openaiApiKey); +const speechTest = shouldRun ? test : test.skip; + +function waitForSignal( + timeoutMs: number, + setup: ( + resolve: (value: T) => void, + reject: (error: Error) => void + ) => () => void +): Promise { + return new Promise((resolve, reject) => { + let cleanup: (() => void) | null = null; + const timeout = setTimeout(() => { + cleanup?.(); + reject(new Error(`Timeout waiting for event after ${timeoutMs}ms`)); + }, timeoutMs); + + cleanup = setup( + (value) => { + clearTimeout(timeout); + cleanup?.(); + resolve(value); + }, + (error) => { + clearTimeout(timeout); + cleanup?.(); + reject(error); + } + ); + }); +} + +async function withTimeout( + label: string, + timeoutMs: number, + task: Promise +): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout(() => { + reject(new Error(`Timed out during ${label} after ${timeoutMs}ms`)); + }, timeoutMs); + task.then( + (value) => { + clearTimeout(timer); + resolve(value); + }, + (error) => { + clearTimeout(timer); + reject(error); + } + ); + }); +} + +async function streamToBuffer(stream: Readable): Promise { + const chunks: Buffer[] = []; + for await (const chunk of stream) { + chunks.push(Buffer.isBuffer(chunk) ? chunk : Buffer.from(chunk)); + } + return Buffer.concat(chunks); +} + +describe("voice roundtrip e2e", () => { + let ctx: DaemonTestContext; + + beforeAll(async () => { + ctx = await createDaemonTestContext({ + agentClients: {}, + openai: { apiKey: openaiApiKey! }, + speech: { + providers: { + dictationStt: { provider: "openai", explicit: true }, + voiceStt: { provider: "openai", explicit: true }, + voiceTts: { provider: "openai", explicit: true }, + }, + }, + voiceLlmProvider: "claude", + voiceLlmProviderExplicit: true, + }); + }, 60000); + + afterAll(async () => { + await ctx.cleanup(); + }, 60000); + + speechTest( + "full roundtrip: voice input audio -> voice agent -> output audio -> transcribed output", + async () => { + const logger = pino({ level: "silent" }); + const ttsProvider = new OpenAITTS( + { + apiKey: openaiApiKey!, + responseFormat: "pcm", + voice: "alloy", + }, + logger + ); + const sttProvider = new OpenAISTT( + { + apiKey: openaiApiKey!, + model: "gpt-4o-mini-transcribe", + }, + logger + ); + const sttOutput = new STTManager("voice-roundtrip-e2e", logger, sttProvider); + + const voiceAgentId = randomUUID(); + const voiceMode = await withTimeout( + "setVoiceMode", + 15000, + ctx.client.setVoiceMode(true, voiceAgentId) + ); + expect(voiceMode.accepted).toBe(true); + expect(voiceMode.enabled).toBe(true); + const timelineTools: string[] = []; + const timelineToolAgentIds = new Set(); + const activityErrors: string[] = []; + + const offStream = ctx.client.on("agent_stream", (message) => { + if (message.type !== "agent_stream") { + return; + } + if (message.payload.event.type !== "timeline") { + return; + } + const item = message.payload.event.item; + if (item.type !== "tool_call") { + return; + } + timelineToolAgentIds.add(message.payload.agentId); + timelineTools.push(String(item.name ?? "")); + }); + const offErrors = ctx.client.on("activity_log", (message) => { + if (message.type !== "activity_log") { + return; + } + if (message.payload.type !== "error") { + return; + } + activityErrors.push(String(message.payload.content ?? "")); + }); + + const inputSpeech = await withTimeout( + "synthesizeInputAudio", + 30000, + ttsProvider.synthesizeSpeech( + "Use the speak tool and say exactly round trip successful." + ) + ); + const inputPcm = await withTimeout( + "collectInputAudio", + 15000, + streamToBuffer(inputSpeech.stream) + ); + const outputAudio = await (async () => { + try { + const transcriptPromise = waitForSignal<{ text: string; isLowConfidence: boolean }>( + 30000, + (resolve) => { + const offTranscript = ctx.client.on("transcription_result", (message) => { + if (message.type !== "transcription_result") { + return; + } + resolve({ + text: String(message.payload.text ?? ""), + isLowConfidence: Boolean(message.payload.isLowConfidence), + }); + }); + return () => { + offTranscript(); + }; + } + ); + + const outputAudioPromise = waitForSignal<{ + format: string; + chunks: Buffer[]; + }>(90000, (resolve, reject) => { + let targetGroupId: string | null = null; + const chunks: Array<{ index: number; bytes: Buffer }> = []; + let format = "pcm"; + + const offAudio = ctx.client.on("audio_output", (message) => { + if (message.type !== "audio_output") { + return; + } + const payload = message.payload; + if (!targetGroupId) { + targetGroupId = payload.groupId; + format = payload.format; + } + if (payload.groupId !== targetGroupId) { + return; + } + chunks.push({ + index: payload.chunkIndex, + bytes: Buffer.from(payload.audio, "base64"), + }); + if (payload.isLastChunk) { + chunks.sort((a, b) => a.index - b.index); + resolve({ + format, + chunks: chunks.map((entry) => entry.bytes), + }); + } + }); + + const offError = ctx.client.on("activity_log", (message) => { + if (message.type !== "activity_log") { + return; + } + if (message.payload.type !== "error") { + return; + } + reject(new Error(String(message.payload.content))); + }); + + return () => { + offAudio(); + offError(); + }; + }); + + const format = "audio/pcm;rate=24000;bits=16"; + const chunkBytes = 4800; // 100ms @ 24kHz mono PCM16 + for (let offset = 0; offset < inputPcm.length; offset += chunkBytes) { + const chunk = inputPcm.subarray(offset, Math.min(inputPcm.length, offset + chunkBytes)); + const isLast = offset + chunkBytes >= inputPcm.length; + await withTimeout( + "sendVoiceAudioChunk", + 5000, + ctx.client.sendVoiceAudioChunk(chunk.toString("base64"), format, isLast) + ); + } + const transcript = await withTimeout("waitForTranscription", 35000, transcriptPromise); + if (transcript.text.trim().length === 0) { + throw new Error( + `empty transcription (lowConfidence=${transcript.isLowConfidence})` + ); + } + return await withTimeout("waitForAudioOutput", 95000, outputAudioPromise); + } catch (error) { + throw new Error( + `${error instanceof Error ? error.message : String(error)} | requestedVoiceAgentId=${voiceAgentId} | timelineTools=${JSON.stringify(timelineTools)} | timelineToolAgentIds=${JSON.stringify(Array.from(timelineToolAgentIds))} | activityErrors=${JSON.stringify(activityErrors)}` + ); + } finally { + offStream(); + offErrors(); + } + })(); + + const outputRaw = Buffer.concat(outputAudio.chunks); + const outputFormat = + outputAudio.format === "pcm" + ? "audio/pcm;rate=24000;bits=16" + : outputAudio.format.includes("wav") + ? "audio/wav" + : `audio/${outputAudio.format}`; + const transcription = await withTimeout( + "transcribeOutputAudio", + 60000, + sttOutput.transcribe(outputRaw, outputFormat, { + label: "voice-roundtrip-output", + }) + ); + const normalized = transcription.text.trim().toLowerCase(); + + expect(normalized.length).toBeGreaterThan(0); + expect(normalized).toMatch(/round|trip|successful/); + }, + 180000 + ); +}); diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 4b63da744..8700a7526 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -88,6 +88,8 @@ export class VoiceAssistantWebSocketServer { voiceLlmProviderExplicit?: boolean; voiceLlmModel?: string | null; voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; + ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise; + removeVoiceMcpSocketForAgent?: (agentId: string) => Promise; } | null; private readonly voiceSpeakHandlers = new Map< string, @@ -113,6 +115,8 @@ export class VoiceAssistantWebSocketServer { voiceLlmProviderExplicit?: boolean; voiceLlmModel?: string | null; voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; + ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise; + removeVoiceMcpSocketForAgent?: (agentId: string) => Promise; }, dictation?: { finalTimeoutMs?: number; @@ -257,6 +261,8 @@ export class VoiceAssistantWebSocketServer { unregisterVoiceCallerContext: (agentId) => { this.voiceCallerContexts.delete(agentId); }, + ensureVoiceMcpSocketForAgent: this.voice?.ensureVoiceMcpSocketForAgent, + removeVoiceMcpSocketForAgent: this.voice?.removeVoiceMcpSocketForAgent, }, dictation: this.dictation ?? undefined, });