Update files

This commit is contained in:
Mohamed Boudra
2026-02-07 12:52:22 +07:00
parent 4a9cf9e90f
commit 6cfae77a71
24 changed files with 1312 additions and 443 deletions
-9
View File
@@ -14,7 +14,6 @@ import { runInspectCommand } from './commands/agent/inspect.js'
import { runWaitCommand } from './commands/agent/wait.js' import { runWaitCommand } from './commands/agent/wait.js'
import { runAttachCommand } from './commands/agent/attach.js' import { runAttachCommand } from './commands/agent/attach.js'
import { withOutput } from './output/index.js' import { withOutput } from './output/index.js'
import { runVoiceMcpBridgeCommand } from './commands/voice-mcp-bridge.js'
const VERSION = '0.1.0' const VERSION = '0.1.0'
@@ -142,13 +141,5 @@ export function createCli(): Command {
// Worktree commands // Worktree commands
program.addCommand(createWorktreeCommand()) 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 <path>')
.requiredOption('--caller-agent-id <id>')
.action(runVoiceMcpBridgeCommand)
return program return program
} }
@@ -1,17 +0,0 @@
import { runVoiceMcpBridgeCli } from "@getpaseo/server";
type VoiceBridgeOptions = {
socket: string;
callerAgentId: string;
};
export async function runVoiceMcpBridgeCommand(
options: VoiceBridgeOptions
): Promise<void> {
await runVoiceMcpBridgeCli([
"--socket",
options.socket,
"--caller-agent-id",
options.callerAgentId,
]);
}
@@ -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 <path>");
}
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);
});
@@ -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<void> {
return new Promise((resolve) => setTimeout(resolve, ms));
}
async function streamToBuffer(stream: AsyncIterable<unknown>): Promise<Buffer> {
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<Buffer> {
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<unknown>);
}
async function runVoiceRoundTrip(params: {
daemonUrl: string;
voiceAgentId: string;
speechText: string;
timeoutMs: number;
}): Promise<RoundTripResult> {
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<void>((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<void> {
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);
});
@@ -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<AgentStreamEvent>,
timeoutMs: number,
onEvent: (event: AgentStreamEvent) => Promise<void> | void
): Promise<void> {
const timeout = new Promise<never>((_, 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<void> {
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);
});
@@ -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<unknown>): Promise<Buffer> {
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<void> {
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<void>((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<void>((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<unknown>);
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);
});
@@ -129,6 +129,9 @@ function buildSerializableConfig(
if (extra !== undefined) { if (extra !== undefined) {
serializable.extra = extra; serializable.extra = extra;
} }
if (config.mcpServers) {
serializable.mcpServers = config.mcpServers;
}
return Object.keys(serializable).length ? serializable : null; return Object.keys(serializable).length ? serializable : null;
} }
@@ -39,6 +39,7 @@ function createManagedAgent(
modeId: configOverrides.modeId ?? "plan", modeId: configOverrides.modeId ?? "plan",
model: configOverrides.model ?? "gpt-5.1", model: configOverrides.model ?? "gpt-5.1",
extra: configOverrides.extra ?? { claude: { maxThinkingTokens: 1024 } }, extra: configOverrides.extra ?? { claude: { maxThinkingTokens: 1024 } },
mcpServers: configOverrides.mcpServers,
}; };
const session = const session =
lifecycle === "closed" lifecycle === "closed"
@@ -119,6 +120,13 @@ describe("AgentStorage", () => {
modeId: "coding", modeId: "coding",
model: "gpt-5.1", model: "gpt-5.1",
extra: { claude: { maxThinkingTokens: 1024 } }, 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.provider).toBe("claude");
expect(record.config?.modeId).toBe("coding"); expect(record.config?.modeId).toBe("coding");
expect(record.config?.model).toBe("gpt-5.1"); 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.lastModeId).toBe("coding");
expect(record.lastStatus).toBe("idle"); expect(record.lastStatus).toBe("idle");
@@ -15,6 +15,7 @@ const SERIALIZABLE_CONFIG_SCHEMA = z
model: z.string().nullable().optional(), model: z.string().nullable().optional(),
thinkingOptionId: z.string().nullable().optional(), thinkingOptionId: z.string().nullable().optional(),
extra: z.record(z.any()).nullable().optional(), extra: z.record(z.any()).nullable().optional(),
mcpServers: z.record(z.any()).nullable().optional(),
}) })
.nullable() .nullable()
.optional(); .optional();
@@ -61,7 +62,7 @@ const STORED_AGENT_SCHEMA = z.object({
export type SerializableAgentConfig = Pick< export type SerializableAgentConfig = Pick<
AgentSessionConfig, AgentSessionConfig,
"modeId" | "model" | "thinkingOptionId" | "extra" "modeId" | "model" | "thinkingOptionId" | "extra" | "mcpServers"
>; >;
export type StoredAgentRecord = z.infer<typeof STORED_AGENT_SCHEMA>; export type StoredAgentRecord = z.infer<typeof STORED_AGENT_SCHEMA>;
@@ -89,7 +89,7 @@ export const AGENT_PROVIDER_DEFINITIONS: AgentProviderDefinition[] = [
voice: { voice: {
enabled: true, enabled: true,
defaultModeId: "read-only", defaultModeId: "read-only",
defaultModel: "gpt-5.2-mini", defaultModel: "gpt-5.1-codex-mini",
}, },
}, },
{ {
@@ -229,6 +229,17 @@ export class TTSManager {
isVoiceMode, isVoiceMode,
}, },
}); });
this.logger.info(
{
audioId,
chunkId,
chunkIndex,
isLastChunk: next.done,
bytes: chunkBuffer.length,
isVoiceMode,
},
"Emitted audio_output chunk to client"
);
chunkIndex += 1; chunkIndex += 1;
@@ -287,6 +298,15 @@ export class TTSManager {
} }
pending.pendingChunks = Math.max(0, pending.pendingChunks - 1); 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) { if (pending.pendingChunks === 0 && pending.streamEnded) {
pending.resolve(); pending.resolve();
+31 -43
View File
@@ -67,39 +67,29 @@ import type {
import { AGENT_PROVIDER_DEFINITIONS } from "./agent/provider-manifest.js"; import { AGENT_PROVIDER_DEFINITIONS } from "./agent/provider-manifest.js";
import { acquirePidLock, releasePidLock } from "./pid-lock.js"; import { acquirePidLock, releasePidLock } from "./pid-lock.js";
import { isHostAllowed, type AllowedHostsConfig } from "./allowed-hosts.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<string, StreamableHTTPServerTransport>; type AgentMcpTransportMap = Map<string, StreamableHTTPServerTransport>;
function resolveVoiceMcpBridgeCommand(logger: Logger): { command: string; baseArgs: string[] } { function resolveVoiceMcpBridgeCommand(logger: Logger): { command: string; baseArgs: string[] } {
const explicit = process.env.PASEO_BIN_PATH?.trim(); const decision = resolveVoiceMcpBridgeFromRuntime({
if (explicit) { bootstrapModuleUrl: import.meta.url,
const resolved = { command: explicit, baseArgs: ["__paseo_voice_mcp_bridge"] }; execPath: process.execPath,
logger.info({ source: "PASEO_BIN_PATH", command: resolved.command, baseArgs: resolved.baseArgs }, "Resolved voice MCP bridge command"); explicitScriptPath: process.env.PASEO_MCP_STDIO_SOCKET_BRIDGE_SCRIPT,
return resolved; });
} logger.info(
{
const argv1 = process.argv[1]?.trim(); source: decision.source,
if (!argv1) { command: decision.resolved.command,
logger.warn("Could not resolve argv[1] for voice MCP bridge; falling back to 'paseo'"); baseArgs: decision.resolved.baseArgs,
const resolved = { command: "paseo", baseArgs: ["__paseo_voice_mcp_bridge"] }; },
logger.info({ source: "fallback", command: resolved.command, baseArgs: resolved.baseArgs }, "Resolved voice MCP bridge command"); "Resolved voice MCP bridge command"
return resolved; );
} return decision.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;
} }
export type PaseoOpenAIConfig = OpenAiSpeechProviderConfig; export type PaseoOpenAIConfig = OpenAiSpeechProviderConfig;
@@ -382,7 +372,7 @@ export async function createPaseoDaemon(
"Voice LLM provider reconciliation completed" "Voice LLM provider reconciliation completed"
); );
let wsServer: VoiceAssistantWebSocketServer | null = null; 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) // Create in-memory transport for Session's Agent MCP client (voice assistant tools)
const createInMemoryAgentMcpTransport = async (): Promise<InMemoryTransport> => { const createInMemoryAgentMcpTransport = async (): Promise<InMemoryTransport> => {
@@ -522,10 +512,10 @@ export async function createPaseoDaemon(
logger.info("Agent MCP HTTP endpoint disabled"); 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); const voiceMcpBridgeCommand = resolveVoiceMcpBridgeCommand(logger);
voiceMcpBridgeServer = createVoiceMcpBridgeSocketServer({ voiceMcpBridgeManager = createVoiceMcpSocketBridgeManager({
socketPath: voiceMcpSocketPath, runtimeDir: voiceMcpSocketDir,
logger, logger,
createAgentMcpServerForCaller: async (callerAgentId) => { createAgentMcpServerForCaller: async (callerAgentId) => {
return createAgentMcpServer({ return createAgentMcpServer({
@@ -571,15 +561,16 @@ export async function createPaseoDaemon(
voiceLlmModel: resolvedVoiceLlmModel, voiceLlmModel: resolvedVoiceLlmModel,
voiceAgentMcpStdio: { voiceAgentMcpStdio: {
command: voiceMcpBridgeCommand.command, command: voiceMcpBridgeCommand.command,
baseArgs: [ baseArgs: [...voiceMcpBridgeCommand.baseArgs],
...voiceMcpBridgeCommand.baseArgs,
"--socket",
voiceMcpSocketPath,
],
env: { env: {
PASEO_HOME: config.paseoHome, 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, finalTimeoutMs: config.dictationFinalTimeoutMs,
@@ -675,9 +666,6 @@ export async function createPaseoDaemon(
httpServer.listen(listenTarget.path); httpServer.listen(listenTarget.path);
} }
}); });
if (voiceMcpBridgeServer) {
await voiceMcpBridgeServer.start();
}
}; };
const stop = async () => { const stop = async () => {
@@ -692,8 +680,8 @@ export async function createPaseoDaemon(
if (wsServer) { if (wsServer) {
await wsServer.close(); await wsServer.close();
} }
if (voiceMcpBridgeServer) { if (voiceMcpBridgeManager) {
await voiceMcpBridgeServer.stop().catch(() => undefined); await voiceMcpBridgeManager.stop().catch(() => undefined);
} }
await new Promise<void>((resolve) => { await new Promise<void>((resolve) => {
httpServer.close(() => resolve()); httpServer.close(() => resolve());
-1
View File
@@ -5,7 +5,6 @@ export { resolvePaseoHome } from "./paseo-home.js";
export { createRootLogger, type LogLevel, type LogFormat } from "./logger.js"; export { createRootLogger, type LogLevel, type LogFormat } from "./logger.js";
export { loadPersistedConfig, type PersistedConfig } from "./persisted-config.js"; export { loadPersistedConfig, type PersistedConfig } from "./persisted-config.js";
export { DaemonClient, type DaemonClientConfig, type ConnectionState, type DaemonEvent } from "../client/daemon-client.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 // Agent SDK types for CLI commands
export type { export type {
-7
View File
@@ -10,15 +10,8 @@ import { resolvePaseoHome } from "./paseo-home.js";
import { createRootLogger } from "./logger.js"; import { createRootLogger } from "./logger.js";
import { loadPersistedConfig } from "./persisted-config.js"; import { loadPersistedConfig } from "./persisted-config.js";
import { PidLockError } from "./pid-lock.js"; import { PidLockError } from "./pid-lock.js";
import { runVoiceMcpBridgeCli } from "./voice-mcp-bridge.js";
async function main() { 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 paseoHome: string;
let logger: ReturnType<typeof createRootLogger>; let logger: ReturnType<typeof createRootLogger>;
let config: ReturnType<typeof loadConfig>; let config: ReturnType<typeof loadConfig>;
@@ -56,6 +56,7 @@ export function buildConfigOverrides(
thinkingOptionId: record.config?.thinkingOptionId ?? undefined, thinkingOptionId: record.config?.thinkingOptionId ?? undefined,
title: record.title ?? undefined, title: record.title ?? undefined,
extra: record.config?.extra ?? undefined, extra: record.config?.extra ?? undefined,
mcpServers: record.config?.mcpServers ?? undefined,
}; };
} }
@@ -74,6 +75,7 @@ export function buildSessionConfig(
thinkingOptionId: overrides.thinkingOptionId, thinkingOptionId: overrides.thinkingOptionId,
title: overrides.title, title: overrides.title,
extra: overrides.extra, extra: overrides.extra,
mcpServers: overrides.mcpServers,
}; };
} }
+114 -41
View File
@@ -128,16 +128,27 @@ const AUTO_GEN_MODEL = "haiku";
const VOICE_AGENT_SYSTEM_INSTRUCTION = [ const VOICE_AGENT_SYSTEM_INSTRUCTION = [
"You are the Paseo voice assistant.", "You are the Paseo voice assistant.",
"The user cannot see your chat messages or tool calls.", "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.", "Use concise plain language suitable for speech output.",
"Never use bash, file-edit, or web tools directly.", "Never use bash, file-edit, or web tools directly.",
"Only use the paseo MCP tools.", "Only use the paseo MCP tools.",
].join(" "); ].join(" ");
function escapeXmlText(value: string): string {
return value
.replace(/&/g, "&amp;")
.replace(/</g, "&lt;")
.replace(/>/g, "&gt;");
}
export function buildVoiceAgentMcpServerConfig(params: { export function buildVoiceAgentMcpServerConfig(params: {
callerAgentId: string;
command: string; command: string;
baseArgs: string[]; baseArgs: string[];
socketPath: string;
env?: Record<string, string>; env?: Record<string, string>;
}): { }): {
type: "stdio"; type: "stdio";
@@ -148,7 +159,7 @@ export function buildVoiceAgentMcpServerConfig(params: {
return { return {
type: "stdio", type: "stdio",
command: params.command, command: params.command,
args: [...params.baseArgs, "--caller-agent-id", params.callerAgentId], args: [...params.baseArgs, "--socket", params.socketPath],
...(params.env ? { env: params.env } : {}), ...(params.env ? { env: params.env } : {}),
}; };
} }
@@ -214,6 +225,8 @@ export type SessionOptions = {
unregisterVoiceSpeakHandler?: (agentId: string) => void; unregisterVoiceSpeakHandler?: (agentId: string) => void;
registerVoiceCallerContext?: (agentId: string, context: VoiceCallerContext) => void; registerVoiceCallerContext?: (agentId: string, context: VoiceCallerContext) => void;
unregisterVoiceCallerContext?: (agentId: string) => void; unregisterVoiceCallerContext?: (agentId: string) => void;
ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise<string>;
removeVoiceMcpSocketForAgent?: (agentId: string) => Promise<void>;
}; };
dictation?: { dictation?: {
finalTimeoutMs?: number; finalTimeoutMs?: number;
@@ -380,6 +393,8 @@ export class Session {
context: VoiceCallerContext context: VoiceCallerContext
) => void; ) => void;
private readonly unregisterVoiceCallerContext?: (agentId: string) => void; private readonly unregisterVoiceCallerContext?: (agentId: string) => void;
private readonly ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise<string>;
private readonly removeVoiceMcpSocketForAgent?: (agentId: string) => Promise<void>;
private voiceAssistantAgentId: string | null = null; private voiceAssistantAgentId: string | null = null;
constructor(options: SessionOptions) { constructor(options: SessionOptions) {
@@ -428,6 +443,8 @@ export class Session {
this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler; this.unregisterVoiceSpeakHandler = voiceBridge?.unregisterVoiceSpeakHandler;
this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext; this.registerVoiceCallerContext = voiceBridge?.registerVoiceCallerContext;
this.unregisterVoiceCallerContext = voiceBridge?.unregisterVoiceCallerContext; this.unregisterVoiceCallerContext = voiceBridge?.unregisterVoiceCallerContext;
this.ensureVoiceMcpSocketForAgent = voiceBridge?.ensureVoiceMcpSocketForAgent;
this.removeVoiceMcpSocketForAgent = voiceBridge?.removeVoiceMcpSocketForAgent;
this.abortController = new AbortController(); this.abortController = new AbortController();
this.sessionLogger = logger.child({ this.sessionLogger = logger.child({
module: "session", module: "session",
@@ -1299,6 +1316,17 @@ export class Session {
} }
this.isVoiceMode = false; 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( this.sessionLogger.info(
{ voiceAssistantAgentId: this.voiceAssistantAgentId }, { voiceAssistantAgentId: this.voiceAssistantAgentId },
"Voice mode disabled (agent-backed)" "Voice mode disabled (agent-backed)"
@@ -1340,7 +1368,7 @@ export class Session {
...fallbackCandidates.filter( ...fallbackCandidates.filter(
(candidate) => !scopedCandidates.some((scoped) => scoped.id === candidate.id) (candidate) => !scopedCandidates.some((scoped) => scoped.id === candidate.id)
), ),
].filter((agent) => !agent.archivedAt); ].filter((agent) => !agent.archivedAt && agent.labels?.ui !== "false");
if (available.length === 0) { if (available.length === 0) {
return null; return null;
} }
@@ -4474,14 +4502,23 @@ export class Session {
} }
private async ensureVoiceAssistantAgent(): Promise<string> { private async ensureVoiceAssistantAgent(): Promise<string> {
const ensureVoiceSocket = this.ensureVoiceMcpSocketForAgent;
if (!ensureVoiceSocket) {
throw new Error("Voice MCP socket bridge is not configured");
}
if (this.voiceAssistantAgentId) { if (this.voiceAssistantAgentId) {
const existing = this.agentManager.getAgent(this.voiceAssistantAgentId); const existing = this.agentManager.getAgent(this.voiceAssistantAgentId);
if (existing) { if (existing) {
await ensureVoiceSocket(existing.id);
this.registerVoiceBridgeForAgent(existing.id);
return existing.id; return existing.id;
} }
try { try {
const hydrated = await this.ensureAgentLoaded(this.voiceAssistantAgentId); const hydrated = await this.ensureAgentLoaded(this.voiceAssistantAgentId);
await ensureVoiceSocket(hydrated.id);
this.voiceAssistantAgentId = hydrated.id; this.voiceAssistantAgentId = hydrated.id;
this.registerVoiceBridgeForAgent(hydrated.id);
return hydrated.id; return hydrated.id;
} catch (error) { } catch (error) {
this.sessionLogger.debug( this.sessionLogger.debug(
@@ -4503,6 +4540,7 @@ export class Session {
if (!mcpStdio) { if (!mcpStdio) {
throw new Error("Voice MCP stdio bridge is not configured"); throw new Error("Voice MCP stdio bridge is not configured");
} }
const socketPath = await ensureVoiceSocket(voiceAgentId);
const model = this.voiceLlmModel?.trim() || undefined; const model = this.voiceLlmModel?.trim() || undefined;
const config: AgentSessionConfig = { const config: AgentSessionConfig = {
@@ -4512,9 +4550,9 @@ export class Session {
...(model ? { model } : {}), ...(model ? { model } : {}),
mcpServers: { mcpServers: {
paseo: buildVoiceAgentMcpServerConfig({ paseo: buildVoiceAgentMcpServerConfig({
callerAgentId: voiceAgentId,
command: mcpStdio.command, command: mcpStdio.command,
baseArgs: mcpStdio.baseArgs, baseArgs: mcpStdio.baseArgs,
socketPath,
env: mcpStdio.env, env: mcpStdio.env,
}), }),
}, },
@@ -4523,48 +4561,18 @@ export class Session {
const created = await this.agentManager.createAgent(config, voiceAgentId, { const created = await this.agentManager.createAgent(config, voiceAgentId, {
labels: { labels: {
surface: "voice", surface: "voice",
ui: "false", ui: "true",
voiceClientId: this.clientId, voiceClientId: this.clientId,
}, },
}); });
this.voiceAssistantAgentId = created.id; this.voiceAssistantAgentId = created.id;
let speakHandlerRegistered = false;
let callerContextRegistered = false;
try { try {
this.registerVoiceSpeakHandler?.(created.id, async ({ text, signal }) => { this.registerVoiceBridgeForAgent(created.id);
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;
} catch (error) { } catch (error) {
if (speakHandlerRegistered) { this.unregisterVoiceSpeakHandler?.(created.id);
this.unregisterVoiceSpeakHandler?.(created.id); this.unregisterVoiceCallerContext?.(created.id);
} await this.removeVoiceMcpSocketForAgent?.(created.id).catch(() => undefined);
if (callerContextRegistered) {
this.unregisterVoiceCallerContext?.(created.id);
}
await this.agentManager.closeAgent(created.id).catch(() => undefined); await this.agentManager.closeAgent(created.id).catch(() => undefined);
this.voiceAssistantAgentId = null; this.voiceAssistantAgentId = null;
throw error; throw error;
@@ -4582,11 +4590,53 @@ export class Session {
return created.id; 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 { private buildVoiceAgentPrompt(userText: string): string {
const transcription = escapeXmlText(userText.trim());
return [ return [
VOICE_AGENT_SYSTEM_INSTRUCTION, VOICE_AGENT_SYSTEM_INSTRUCTION,
"", "",
`User said: ${userText.trim()}`, "<voice_transcription>",
transcription,
"</voice_transcription>",
].join("\n"); ].join("\n");
} }
@@ -4599,6 +4649,15 @@ export class Session {
await this.interruptAgentIfRunning(agentId); 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); const prompt = this.buildVoiceAgentPrompt(userText);
this.agentManager.recordUserMessage(agentId, userText); this.agentManager.recordUserMessage(agentId, userText);
@@ -4610,6 +4669,14 @@ export class Session {
} }
if (event.type === "timeline") { if (event.type === "timeline") {
if (event.item.type === "tool_call" && typeof event.item.name === "string") { 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")) { if (event.item.name.toLowerCase().includes("speak")) {
sawSpeakToolCall = true; sawSpeakToolCall = true;
} }
@@ -4638,6 +4705,11 @@ export class Session {
{ voiceAssistantAgentId: agentId }, { voiceAssistantAgentId: agentId },
"Voice agent turn completed without speak tool call; no audio playback emitted" "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 { } finally {
this.unregisterVoiceSpeakHandler?.(this.voiceAssistantAgentId); this.unregisterVoiceSpeakHandler?.(this.voiceAssistantAgentId);
this.unregisterVoiceCallerContext?.(this.voiceAssistantAgentId); this.unregisterVoiceCallerContext?.(this.voiceAssistantAgentId);
await this.removeVoiceMcpSocketForAgent?.(this.voiceAssistantAgentId).catch(() => undefined);
this.voiceAssistantAgentId = null; this.voiceAssistantAgentId = null;
} }
} }
@@ -5,22 +5,20 @@ import { buildVoiceAgentMcpServerConfig } from "./session.js";
describe("voice MCP stdio config", () => { describe("voice MCP stdio config", () => {
test("builds stdio MCP config for voice agent", () => { test("builds stdio MCP config for voice agent", () => {
const config = buildVoiceAgentMcpServerConfig({ const config = buildVoiceAgentMcpServerConfig({
callerAgentId: "voice-agent-123", command: "/usr/local/bin/node",
command: "/usr/local/bin/paseo", baseArgs: ["/tmp/mcp-stdio-socket-bridge-cli.mjs"],
baseArgs: ["__paseo_voice_mcp_bridge", "--socket", "/tmp/paseo-voice.sock"], socketPath: "/tmp/paseo-voice.sock",
env: { env: {
PASEO_HOME: "/tmp/paseo-home", PASEO_HOME: "/tmp/paseo-home",
}, },
}); });
expect(config.type).toBe("stdio"); 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([ expect(config.args).toEqual([
"__paseo_voice_mcp_bridge", "/tmp/mcp-stdio-socket-bridge-cli.mjs",
"--socket", "--socket",
"/tmp/paseo-voice.sock", "/tmp/paseo-voice.sock",
"--caller-agent-id",
"voice-agent-123",
]); ]);
expect(config.env).toEqual({ expect(config.env).toEqual({
PASEO_HOME: "/tmp/paseo-home", PASEO_HOME: "/tmp/paseo-home",
@@ -58,7 +58,7 @@ function waitForSignal<T>(
}, },
voiceLlmProvider: "codex", voiceLlmProvider: "codex",
voiceLlmProviderExplicit: true, voiceLlmProviderExplicit: true,
voiceLlmModel: "gpt-5.2-mini", voiceLlmModel: "gpt-5.1-codex-mini",
}); });
}, 120000); }, 120000);
@@ -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");
});
});
@@ -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],
},
};
}
@@ -8,16 +8,15 @@ import { StdioClientTransport } from "@modelcontextprotocol/sdk/client/stdio.js"
import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js"; import { McpServer } from "@modelcontextprotocol/sdk/server/mcp.js";
import pino from "pino"; import pino from "pino";
import { createVoiceMcpBridgeSocketServer } from "./voice-mcp-bridge.js"; import { createVoiceMcpSocketBridgeManager } from "./voice-mcp-bridge.js";
describe("voice MCP bridge", () => { 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 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 callerAgentId = "voice-agent-bridge-test";
const bridge = createVoiceMcpBridgeSocketServer({ const bridgeManager = createVoiceMcpSocketBridgeManager({
socketPath, runtimeDir: tmpRoot,
logger: pino({ level: "silent" }), logger: pino({ level: "silent" }),
createAgentMcpServerForCaller: async (callerId) => { createAgentMcpServerForCaller: async (callerId) => {
const server = new McpServer({ 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 bridgeScript = path.resolve(process.cwd(), "scripts/mcp-stdio-socket-bridge-cli.mjs");
const serverIndex = path.resolve(process.cwd(), "src/server/index.ts");
const transport = new StdioClientTransport({ const transport = new StdioClientTransport({
command: process.execPath, command: process.execPath,
args: [ args: [
tsxBin, bridgeScript,
serverIndex,
"__paseo_voice_mcp_bridge",
"--socket", "--socket",
socketPath, socketPath,
"--caller-agent-id",
callerAgentId,
], ],
}); });
@@ -86,7 +80,7 @@ describe("voice MCP bridge", () => {
expect(payload?.callerAgentId).toBe(callerAgentId); expect(payload?.callerAgentId).toBe(callerAgentId);
} finally { } finally {
await client.close(); await client.close();
await bridge.stop(); await bridgeManager.stop();
await rm(tmpRoot, { recursive: true, force: true }); await rm(tmpRoot, { recursive: true, force: true });
} }
}, 30_000); }, 30_000);
+112 -301
View File
@@ -2,336 +2,147 @@ import net from "node:net";
import path from "node:path"; import path from "node:path";
import { mkdir, rm } from "node:fs/promises"; import { mkdir, rm } from "node:fs/promises";
import type { Logger } from "pino"; 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"; import { StdioServerTransport } from "@modelcontextprotocol/sdk/server/stdio.js";
type BridgeServer = {
connect: (transport: StdioServerTransport) => Promise<void>;
close?: () => Promise<void>;
};
type BridgeEnvelope = type BridgeEntry = {
| { type: "init"; callerAgentId: string }
| { type: "mcp"; message: JSONRPCMessage }
| { type: "ready" }
| { type: "error"; message: string };
function isRecord(value: unknown): value is Record<string, unknown> {
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 = {
socketPath: string; socketPath: string;
start: () => Promise<void>; server: net.Server;
sockets: Set<net.Socket>;
};
export type VoiceMcpSocketBridgeManager = {
ensureBridgeForCaller: (callerAgentId: string) => Promise<string>;
removeBridgeForCaller: (callerAgentId: string) => Promise<void>;
stop: () => Promise<void>; stop: () => Promise<void>;
}; };
export function createVoiceMcpBridgeSocketServer(params: { function toSocketName(callerAgentId: string): string {
socketPath: string; return `voice-mcp-${callerAgentId}.sock`;
}
export function createVoiceMcpSocketBridgeManager(params: {
runtimeDir: string;
logger: Logger; logger: Logger;
createAgentMcpServerForCaller: (callerAgentId: string) => Promise<{ connect: (transport: InMemoryTransport) => Promise<void>; close?: () => Promise<void> }>; createAgentMcpServerForCaller: (callerAgentId: string) => Promise<BridgeServer>;
}): VoiceMcpBridgeSocketServer { }): VoiceMcpSocketBridgeManager {
const logger = params.logger.child({ module: "voice-mcp-bridge" }); const logger = params.logger.child({ module: "voice-mcp-bridge" });
const sockets = new Set<net.Socket>(); const entries = new Map<string, BridgeEntry>();
const pendingCreates = new Map<string, Promise<string>>();
const server = net.createServer((socket) => { const ensureBridgeForCaller = async (callerAgentId: string): Promise<string> => {
sockets.add(socket); const existing = entries.get(callerAgentId);
const connectionLogger = logger.child({ component: "connection" }); if (existing) {
let readBuffer = ""; return existing.socketPath;
let initialized = false; }
let clientTransport: InMemoryTransport | null = null;
let mcpServer: { close?: () => Promise<void> } | null = null;
const send = (payload: BridgeEnvelope) => { const pending = pendingCreates.get(callerAgentId);
socket.write(encodeEnvelope(payload)); if (pending) {
}; return pending;
}
const fail = (message: string) => { const createPromise = (async () => {
send({ type: "error", message }); const socketPath = path.join(params.runtimeDir, toSocketName(callerAgentId));
socket.end(); const sockets = new Set<net.Socket>();
}; const server = net.createServer((socket) => {
sockets.add(socket);
const connectionLogger = logger.child({ callerAgentId, component: "connection" });
const cleanup = async () => { let mcpServer: BridgeServer | null = null;
sockets.delete(socket); let transport: StdioServerTransport | null = null;
const closeTasks: Promise<unknown>[] = [];
if (clientTransport) {
closeTasks.push(clientTransport.close().catch(() => undefined));
}
if (mcpServer?.close) {
closeTasks.push(mcpServer.close().catch(() => undefined));
}
await Promise.all(closeTasks);
};
socket.on("data", (chunk) => { const cleanup = async () => {
readBuffer += chunk.toString("utf8"); sockets.delete(socket);
while (true) { await Promise.all([
const newlineIndex = readBuffer.indexOf("\n"); transport?.close().catch(() => undefined),
if (newlineIndex < 0) break; mcpServer?.close?.().catch(() => undefined),
const line = readBuffer.slice(0, newlineIndex).trim(); ]);
readBuffer = readBuffer.slice(newlineIndex + 1); };
if (!line) continue;
let message: BridgeEnvelope; socket.on("error", (error) => {
try { connectionLogger.error({ err: error }, "Voice MCP bridge socket error");
message = parseEnvelope(line); });
} catch (error) { socket.on("close", () => {
fail(error instanceof Error ? error.message : String(error)); void cleanup();
return; });
}
if (message.type === "init") { void (async () => {
if (initialized) { try {
fail("Bridge already initialized"); mcpServer = await params.createAgentMcpServerForCaller(callerAgentId);
return; 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") { await mkdir(params.runtimeDir, { recursive: true });
if (!clientTransport) { await rm(socketPath, { force: true }).catch(() => undefined);
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 new Promise<void>((resolve, reject) => { await new Promise<void>((resolve, reject) => {
server.once("error", reject); server.once("error", reject);
server.listen(params.socketPath, () => { server.listen(socketPath, () => {
server.off("error", reject); server.off("error", reject);
resolve(); resolve();
}); });
}); });
logger.info({ socketPath: params.socketPath }, "Voice MCP bridge socket server listening");
}, entries.set(callerAgentId, { socketPath, server, sockets });
async stop() { logger.info(
for (const socket of sockets) { { callerAgentId, socketPath },
socket.destroy(); "Voice MCP per-agent socket bridge listening"
} );
await new Promise<void>((resolve, reject) => { return socketPath;
server.close((error) => { })();
if (error) reject(error);
else resolve(); pendingCreates.set(callerAgentId, createPromise);
}); try {
}); return await createPromise;
await rm(params.socketPath, { force: true }).catch(() => undefined); } finally {
}, pendingCreates.delete(callerAgentId);
}
}; };
}
function parseBridgeCliArgs(argv: string[]): { socketPath: string; callerAgentId: string } { const removeBridgeForCaller = async (callerAgentId: string): Promise<void> => {
let socketPath: string | null = null; const entry = entries.get(callerAgentId);
let callerAgentId: string | null = null; if (!entry) {
return;
for (let index = 0; index < argv.length; index += 1) {
const arg = argv[index];
if (arg === "--socket") {
socketPath = argv[index + 1] ?? null;
index += 1;
continue;
} }
if (arg === "--caller-agent-id") { entries.delete(callerAgentId);
callerAgentId = argv[index + 1] ?? null;
index += 1;
continue;
}
}
if (!socketPath?.trim()) { for (const socket of entry.sockets) {
throw new Error("Missing required --socket <path>"); socket.destroy();
} }
if (!callerAgentId?.trim()) { await new Promise<void>((resolve, reject) => {
throw new Error("Missing required --caller-agent-id <id>"); 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<void> => {
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 { return {
socketPath: socketPath.trim(), ensureBridgeForCaller,
callerAgentId: callerAgentId.trim(), removeBridgeForCaller,
stop,
}; };
} }
export async function runVoiceMcpBridgeCli(argv: string[], logger?: Logger): Promise<void> {
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<void>((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<void>((resolve, reject) => {
socket.once("connect", () => resolve());
socket.once("error", reject);
});
sendEnvelope({ type: "init", callerAgentId: parsed.callerAgentId });
const readyTimeoutPromise = new Promise<void>((_, 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<void>((resolve) => {
socket.once("close", () => resolve());
process.stdin.once("end", () => resolve());
});
await stdioTransport.close().catch(() => undefined);
}
@@ -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<T>(
timeoutMs: number,
setup: (
resolve: (value: T) => void,
reject: (error: Error) => void
) => () => void
): Promise<T> {
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<T>(
label: string,
timeoutMs: number,
task: Promise<T>
): Promise<T> {
return new Promise<T>((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<Buffer> {
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<string>();
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
);
});
@@ -88,6 +88,8 @@ export class VoiceAssistantWebSocketServer {
voiceLlmProviderExplicit?: boolean; voiceLlmProviderExplicit?: boolean;
voiceLlmModel?: string | null; voiceLlmModel?: string | null;
voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; voiceAgentMcpStdio?: VoiceMcpStdioConfig | null;
ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise<string>;
removeVoiceMcpSocketForAgent?: (agentId: string) => Promise<void>;
} | null; } | null;
private readonly voiceSpeakHandlers = new Map< private readonly voiceSpeakHandlers = new Map<
string, string,
@@ -113,6 +115,8 @@ export class VoiceAssistantWebSocketServer {
voiceLlmProviderExplicit?: boolean; voiceLlmProviderExplicit?: boolean;
voiceLlmModel?: string | null; voiceLlmModel?: string | null;
voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; voiceAgentMcpStdio?: VoiceMcpStdioConfig | null;
ensureVoiceMcpSocketForAgent?: (agentId: string) => Promise<string>;
removeVoiceMcpSocketForAgent?: (agentId: string) => Promise<void>;
}, },
dictation?: { dictation?: {
finalTimeoutMs?: number; finalTimeoutMs?: number;
@@ -257,6 +261,8 @@ export class VoiceAssistantWebSocketServer {
unregisterVoiceCallerContext: (agentId) => { unregisterVoiceCallerContext: (agentId) => {
this.voiceCallerContexts.delete(agentId); this.voiceCallerContexts.delete(agentId);
}, },
ensureVoiceMcpSocketForAgent: this.voice?.ensureVoiceMcpSocketForAgent,
removeVoiceMcpSocketForAgent: this.voice?.removeVoiceMcpSocketForAgent,
}, },
dictation: this.dictation ?? undefined, dictation: this.dictation ?? undefined,
}); });