import express from "express"; import { createServer as createHTTPServer } from "http"; import { createReadStream, unlinkSync, existsSync } from "fs"; import { stat } from "fs/promises"; import { randomUUID } from "node:crypto"; import { hostname as getHostname } from "node:os"; import path from "node:path"; import { StreamableHTTPServerTransport } from "@modelcontextprotocol/sdk/server/streamableHttp.js"; import { InMemoryTransport } from "@modelcontextprotocol/sdk/inMemory.js"; import { isInitializeRequest } from "@modelcontextprotocol/sdk/types.js"; import type { Logger } from "pino"; export type ListenTarget = | { type: "tcp"; host: string; port: number } | { type: "socket"; path: string } | { type: "pipe"; path: string }; function resolveBoundListenTarget( listenTarget: ListenTarget, httpServer: ReturnType, ): ListenTarget { if (listenTarget.type !== "tcp") { return listenTarget; } const address = httpServer.address(); if (!address || typeof address === "string") { throw new Error("HTTP server did not expose a TCP address after listening"); } return { type: "tcp", host: listenTarget.host, port: address.port, }; } // Matches a Windows drive-letter path like C:\ or D:\ const WINDOWS_DRIVE_RE = /^[A-Za-z]:\\/; export function parseListenString(listen: string): ListenTarget { // 1. Windows named pipes: \\.\pipe\... or pipe://... if (listen.startsWith("\\\\.\\pipe\\") || listen.startsWith("pipe://")) { return { type: "pipe", path: listen.startsWith("pipe://") ? listen.slice("pipe://".length) : listen, }; } // 2. Explicit unix:// prefix if (listen.startsWith("unix://")) { return { type: "socket", path: listen.slice(7) }; } // 3. Reject Windows absolute drive paths — they are not Unix sockets if (WINDOWS_DRIVE_RE.test(listen)) { throw new Error(`Invalid listen string (Windows path is not a valid listen target): ${listen}`); } // 4. POSIX absolute path (/ or ~) — Unix socket if (listen.startsWith("/") || listen.startsWith("~")) { return { type: "socket", path: listen }; } // 5. Pure numeric — TCP port on 127.0.0.1 const trimmed = listen.trim(); if (/^\d+$/.test(trimmed)) { const port = parseInt(trimmed, 10); return { type: "tcp", host: "127.0.0.1", port }; } // 6. host:port — TCP if (listen.includes(":")) { const [host, portStr] = listen.split(":"); const parsedPort = parseInt(portStr, 10); if (!Number.isFinite(parsedPort)) { throw new Error(`Invalid port in listen string: ${listen}`); } return { type: "tcp", host: host || "127.0.0.1", port: parsedPort }; } throw new Error(`Invalid listen string: ${listen}`); } function formatListenTarget(listenTarget: ListenTarget | null): string | null { if (!listenTarget) { return null; } if (listenTarget.type === "tcp") { return `${listenTarget.host}:${listenTarget.port}`; } return listenTarget.path; } import { VoiceAssistantWebSocketServer } from "./websocket-server.js"; import { DownloadTokenStore } from "./file-download/token-store.js"; import type { OpenAiSpeechProviderConfig } from "./speech/providers/openai/config.js"; import type { LocalSpeechProviderConfig } from "./speech/providers/local/config.js"; import type { RequestedSpeechProviders } from "./speech/speech-types.js"; import { createSpeechService } from "./speech/speech-runtime.js"; import { AgentManager } from "./agent/agent-manager.js"; import { AgentStorage } from "./agent/agent-storage.js"; import { attachAgentStoragePersistence } from "./persistence-hooks.js"; import { createAgentMcpServer } from "./agent/mcp-server.js"; import { createAllClients, shutdownProviders } from "./agent/provider-registry.js"; import { bootstrapWorkspaceRegistries } from "./workspace-registry-bootstrap.js"; import { FileBackedProjectRegistry, FileBackedWorkspaceRegistry } from "./workspace-registry.js"; import { FileBackedChatService } from "./chat/chat-service.js"; import { CheckoutDiffManager } from "./checkout-diff-manager.js"; import { LoopService } from "./loop-service.js"; import { ScheduleService } from "./schedule/service.js"; import { createTerminalManager, type TerminalManager } from "../terminal/terminal-manager.js"; import { createConnectionOfferV2, encodeOfferToFragmentUrl } from "./connection-offer.js"; import { loadOrCreateDaemonKeyPair } from "./daemon-keypair.js"; import { startRelayTransport, type RelayTransportController } from "./relay-transport.js"; import { getOrCreateServerId } from "./server-id.js"; import { resolveDaemonVersion } from "./daemon-version.js"; import type { AgentClient, AgentProvider } from "./agent/agent-sdk-types.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; import { isHostAllowed, type AllowedHostsConfig } from "./allowed-hosts.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[] } | null { try { 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; } catch (err) { logger.warn({ err }, "Voice MCP bridge script not available — voice MCP via stdio disabled"); return null; } } export type PaseoOpenAIConfig = OpenAiSpeechProviderConfig; export type PaseoLocalSpeechConfig = LocalSpeechProviderConfig; export type PaseoSpeechConfig = { providers: RequestedSpeechProviders; local?: PaseoLocalSpeechConfig; }; export type DaemonLifecycleIntent = | { type: "shutdown"; clientId: string; requestId: string; } | { type: "restart"; clientId: string; requestId: string; reason?: string; }; export type PaseoDaemonConfig = { listen: string; paseoHome: string; corsAllowedOrigins: string[]; allowedHosts?: AllowedHostsConfig; mcpEnabled?: boolean; staticDir: string; mcpDebug: boolean; agentClients: Partial>; agentStoragePath: string; relayEnabled?: boolean; relayEndpoint?: string; relayPublicEndpoint?: string; appBaseUrl?: string; openai?: PaseoOpenAIConfig; speech?: PaseoSpeechConfig; voiceLlmProvider?: AgentProvider | null; voiceLlmProviderExplicit?: boolean; voiceLlmModel?: string | null; dictationFinalTimeoutMs?: number; downloadTokenTtlMs?: number; agentProviderSettings?: AgentProviderRuntimeSettingsMap; onLifecycleIntent?: (intent: DaemonLifecycleIntent) => void; }; export interface PaseoDaemon { config: PaseoDaemonConfig; agentManager: AgentManager; agentStorage: AgentStorage; terminalManager: TerminalManager; start(): Promise; stop(): Promise; getListenTarget(): ListenTarget | null; } export async function createPaseoDaemon( config: PaseoDaemonConfig, rootLogger: Logger, ): Promise { const logger = rootLogger.child({ module: "bootstrap" }); const bootstrapStart = performance.now(); const elapsed = () => `${(performance.now() - bootstrapStart).toFixed(0)}ms`; const daemonVersion = resolveDaemonVersion(import.meta.url); try { const serverId = getOrCreateServerId(config.paseoHome, { logger }); const daemonKeyPair = await loadOrCreateDaemonKeyPair(config.paseoHome, logger); let relayTransport: RelayTransportController | null = null; const staticDir = config.staticDir; const downloadTokenTtlMs = config.downloadTokenTtlMs ?? 60000; const downloadTokenStore = new DownloadTokenStore({ ttlMs: downloadTokenTtlMs }); const listenTarget = parseListenString(config.listen); const app = express(); let boundListenTarget: ListenTarget | null = null; // Host allowlist / DNS rebinding protection (vite-like semantics). // For non-TCP (unix sockets), skip host validation. if (listenTarget.type === "tcp") { app.use((req, res, next) => { const hostHeader = typeof req.headers.host === "string" ? req.headers.host : undefined; if (!isHostAllowed(hostHeader, config.allowedHosts)) { res.status(403).json({ error: "Invalid Host header" }); return; } next(); }); } // CORS - allow same-origin + configured origins const allowedOrigins = new Set([ ...config.corsAllowedOrigins, // Packaged desktop renderers use the custom paseo:// protocol scheme. "paseo://app", // For TCP, add localhost variants ...(listenTarget.type === "tcp" ? [ `http://${listenTarget.host}:${listenTarget.port}`, `http://localhost:${listenTarget.port}`, `http://127.0.0.1:${listenTarget.port}`, ] : []), ]); app.use((req, res, next) => { const origin = req.headers.origin; if (origin && allowedOrigins.has(origin)) { res.setHeader("Access-Control-Allow-Origin", origin); res.setHeader("Access-Control-Allow-Methods", "GET, POST, DELETE, OPTIONS"); res.setHeader("Access-Control-Allow-Headers", "Content-Type, Authorization"); res.setHeader("Access-Control-Allow-Credentials", "true"); } if (req.method === "OPTIONS") { res.status(204).end(); return; } next(); }); // Serve static files from public directory app.use("/public", express.static(staticDir)); // Middleware app.use(express.json()); // Health check endpoint app.get("/api/health", (_req, res) => { res.json({ status: "ok", timestamp: new Date().toISOString() }); }); app.get("/api/status", (_req, res) => { res.json({ status: "server_info", serverId, hostname: getHostname(), version: daemonVersion, listen: formatListenTarget(boundListenTarget ?? listenTarget), }); }); app.get("/api/files/download", async (req, res) => { const token = typeof req.query.token === "string" && req.query.token.trim().length > 0 ? req.query.token.trim() : null; if (!token) { res.status(400).json({ error: "Missing download token" }); return; } const entry = downloadTokenStore.consumeToken(token); if (!entry) { res.status(403).json({ error: "Invalid or expired token" }); return; } try { const fileStats = await stat(entry.absolutePath); if (!fileStats.isFile()) { res.status(404).json({ error: "File not found" }); return; } const safeFileName = entry.fileName.replace(/["\r\n]/g, "_"); res.setHeader("Content-Type", entry.mimeType); res.setHeader("Content-Disposition", `attachment; filename="${safeFileName}"`); res.setHeader("Content-Length", entry.size.toString()); const stream = createReadStream(entry.absolutePath); stream.on("error", (err) => { logger.error({ err }, "Failed to stream download"); if (!res.headersSent) { res.status(500).json({ error: "Failed to read file" }); } else { res.end(); } }); stream.pipe(res); } catch (err) { logger.error({ err }, "Failed to download file"); if (!res.headersSent) { res.status(404).json({ error: "File not found" }); } } }); const httpServer = createHTTPServer(app); const agentStorage = new AgentStorage(config.agentStoragePath, logger); const projectRegistry = new FileBackedProjectRegistry( path.join(config.paseoHome, "projects", "projects.json"), logger, ); const workspaceRegistry = new FileBackedWorkspaceRegistry( path.join(config.paseoHome, "projects", "workspaces.json"), logger, ); const chatService = new FileBackedChatService({ paseoHome: config.paseoHome, logger, }); const agentManager = new AgentManager({ clients: { ...createAllClients(logger, { runtimeSettings: config.agentProviderSettings, }), ...config.agentClients, }, registry: agentStorage, logger, }); const terminalManager = createTerminalManager(); const detachAgentStoragePersistence = attachAgentStoragePersistence( logger, agentManager, agentStorage, ); await agentStorage.initialize(); logger.info({ elapsed: elapsed() }, "Agent storage initialized"); await bootstrapWorkspaceRegistries({ paseoHome: config.paseoHome, agentStorage, projectRegistry, workspaceRegistry, logger, }); logger.info({ elapsed: elapsed() }, "Workspace registries bootstrapped"); await chatService.initialize(); logger.info({ elapsed: elapsed() }, "Chat service initialized"); const checkoutDiffManager = new CheckoutDiffManager({ logger, paseoHome: config.paseoHome, }); const loopService = new LoopService({ paseoHome: config.paseoHome, logger, agentManager, }); await loopService.initialize(); logger.info({ elapsed: elapsed() }, "Loop service initialized"); const scheduleService = new ScheduleService({ paseoHome: config.paseoHome, logger, agentManager, agentStorage, }); await scheduleService.start(); logger.info({ elapsed: elapsed() }, "Schedule service initialized"); logger.info({ elapsed: elapsed() }, "Loading persisted agent registry"); const persistedRecords = await agentStorage.list(); logger.info( { elapsed: elapsed() }, `Agent registry loaded (${persistedRecords.length} record${persistedRecords.length === 1 ? "" : "s"}); agents will initialize on demand`, ); logger.info( "Voice mode configured for agent-scoped resume flow (no dedicated voice assistant provider)", ); logger.info({ elapsed: elapsed() }, "Preparing voice and MCP runtime"); let wsServer: VoiceAssistantWebSocketServer | null = null; let voiceMcpBridgeManager: VoiceMcpSocketBridgeManager | null = null; // Create in-memory transport for Session's Agent MCP client (voice assistant tools) const createInMemoryAgentMcpTransport = async (): Promise => { const agentMcpServer = await createAgentMcpServer({ agentManager, agentStorage, terminalManager, paseoHome: config.paseoHome, enableVoiceTools: false, resolveSpeakHandler: (callerAgentId) => wsServer?.resolveVoiceSpeakHandler(callerAgentId) ?? null, resolveCallerContext: (callerAgentId) => wsServer?.resolveVoiceCallerContext(callerAgentId) ?? null, logger, }); const [clientTransport, serverTransport] = InMemoryTransport.createLinkedPair(); await agentMcpServer.connect(serverTransport); return clientTransport; }; const mcpEnabled = config.mcpEnabled ?? true; if (mcpEnabled) { const agentMcpRoute = "/mcp/agents"; const agentMcpTransports: AgentMcpTransportMap = new Map(); const createAgentMcpTransport = async (callerAgentId?: string) => { const agentMcpServer = await createAgentMcpServer({ agentManager, agentStorage, terminalManager, paseoHome: config.paseoHome, callerAgentId, enableVoiceTools: false, resolveSpeakHandler: (agentId) => wsServer?.resolveVoiceSpeakHandler(agentId) ?? null, resolveCallerContext: (agentId) => wsServer?.resolveVoiceCallerContext(agentId) ?? null, logger, }); const transport = new StreamableHTTPServerTransport({ sessionIdGenerator: () => randomUUID(), onsessioninitialized: (sessionId) => { agentMcpTransports.set(sessionId, transport); logger.debug({ sessionId }, "Agent MCP session initialized"); }, onsessionclosed: (sessionId) => { agentMcpTransports.delete(sessionId); logger.debug({ sessionId }, "Agent MCP session closed"); }, // NOTE: We enforce a Vite-like host allowlist at the app/websocket layer. // StreamableHTTPServerTransport's built-in check requires exact Host header matches. enableDnsRebindingProtection: false, }); transport.onclose = () => { if (transport.sessionId) { agentMcpTransports.delete(transport.sessionId); } }; transport.onerror = (err) => { logger.error({ err }, "Agent MCP transport error"); }; await agentMcpServer.connect(transport); return transport; }; const handleAgentMcpRequest: express.RequestHandler = async (req, res) => { if (config.mcpDebug) { logger.debug( { method: req.method, url: req.originalUrl, sessionId: req.header("mcp-session-id"), authorization: req.header("authorization"), body: req.body, }, "Agent MCP request", ); } try { const sessionId = req.header("mcp-session-id"); let transport = sessionId ? agentMcpTransports.get(sessionId) : undefined; if (!transport) { if (req.method !== "POST") { res.status(400).json({ jsonrpc: "2.0", error: { code: -32000, message: "Missing or invalid MCP session", }, id: null, }); return; } if (!isInitializeRequest(req.body)) { res.status(400).json({ jsonrpc: "2.0", error: { code: -32000, message: "Initialization request expected", }, id: null, }); return; } const callerAgentIdRaw = req.query.callerAgentId; const callerAgentId = typeof callerAgentIdRaw === "string" ? callerAgentIdRaw : Array.isArray(callerAgentIdRaw) && typeof callerAgentIdRaw[0] === "string" ? callerAgentIdRaw[0] : undefined; transport = await createAgentMcpTransport(callerAgentId); } await transport.handleRequest(req as any, res as any, req.body); } catch (err) { logger.error({ err }, "Failed to handle Agent MCP request"); if (!res.headersSent) { res.status(500).json({ jsonrpc: "2.0", error: { code: -32603, message: "Internal MCP server error", }, id: null, }); } } }; app.post(agentMcpRoute, handleAgentMcpRequest); app.get(agentMcpRoute, handleAgentMcpRequest); app.delete(agentMcpRoute, handleAgentMcpRequest); logger.info({ route: agentMcpRoute }, "Agent MCP server mounted on main app"); } else { logger.info("Agent MCP HTTP endpoint disabled"); } const voiceMcpSocketDir = path.join(config.paseoHome, "runtime", "voice-mcp"); const voiceMcpBridgeCommand = resolveVoiceMcpBridgeCommand(logger); voiceMcpBridgeManager = createVoiceMcpSocketBridgeManager({ runtimeDir: voiceMcpSocketDir, logger, createAgentMcpServerForCaller: async (callerAgentId) => { return createAgentMcpServer({ agentManager, agentStorage, terminalManager, paseoHome: config.paseoHome, callerAgentId, voiceOnly: true, resolveSpeakHandler: (agentId) => wsServer?.resolveVoiceSpeakHandler(agentId) ?? null, resolveCallerContext: (agentId) => wsServer?.resolveVoiceCallerContext(agentId) ?? null, logger, }); }, }); const speechService = createSpeechService({ logger, openaiConfig: config.openai, speechConfig: config.speech, }); logger.info({ elapsed: elapsed() }, "Speech service created"); wsServer = new VoiceAssistantWebSocketServer( httpServer, logger, serverId, agentManager, agentStorage, downloadTokenStore, config.paseoHome, createInMemoryAgentMcpTransport, { allowedOrigins, allowedHosts: config.allowedHosts }, speechService, terminalManager, { voiceAgentMcpStdio: voiceMcpBridgeCommand ? { command: voiceMcpBridgeCommand.command, baseArgs: [...voiceMcpBridgeCommand.baseArgs], env: { ELECTRON_RUN_AS_NODE: "1", PASEO_HOME: config.paseoHome, }, } : null, 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, }, config.agentProviderSettings, daemonVersion, (intent) => { try { config.onLifecycleIntent?.(intent); } catch (error) { logger.error({ err: error, intent }, "Failed to handle daemon lifecycle intent"); } }, projectRegistry, workspaceRegistry, chatService, loopService, scheduleService, checkoutDiffManager, ); logger.info({ elapsed: elapsed() }, "Bootstrap complete, ready to start listening"); const start = async () => { // Start main HTTP server await new Promise((resolve, reject) => { const onError = (err: Error) => { httpServer.off("listening", onListening); reject(err); }; const onListening = () => { httpServer.off("error", onError); const logAndResolve = async () => { boundListenTarget = resolveBoundListenTarget(listenTarget, httpServer); const relayEnabled = config.relayEnabled ?? true; const relayEndpoint = config.relayEndpoint ?? "relay.paseo.sh:443"; const relayPublicEndpoint = config.relayPublicEndpoint ?? relayEndpoint; const appBaseUrl = config.appBaseUrl ?? "https://app.paseo.sh"; if (boundListenTarget.type === "tcp") { logger.info( { host: boundListenTarget.host, port: boundListenTarget.port, elapsed: elapsed(), }, `Server listening on http://${boundListenTarget.host}:${boundListenTarget.port}`, ); } else { logger.info( { path: boundListenTarget.path, elapsed: elapsed() }, `Server listening on ${boundListenTarget.path}`, ); } if (typeof process.send === "function" && process.env.PASEO_SUPERVISED === "1") { process.send({ type: "paseo:ready", listen: boundListenTarget.type === "tcp" ? `${boundListenTarget.host}:${boundListenTarget.port}` : boundListenTarget.path, }); } if (relayEnabled) { const offer = await createConnectionOfferV2({ serverId, daemonPublicKeyB64: daemonKeyPair.publicKeyB64, relay: { endpoint: relayPublicEndpoint }, }); const url = encodeOfferToFragmentUrl({ offer, appBaseUrl }); logger.info({ url }, "pairing_offer"); relayTransport?.stop().catch(() => undefined); relayTransport = startRelayTransport({ logger, attachSocket: (ws, metadata) => { if (!wsServer) { throw new Error("WebSocket server not initialized"); } return wsServer.attachExternalSocket(ws, metadata); }, relayEndpoint, serverId, daemonKeyPair: daemonKeyPair.keyPair, }); } }; logAndResolve().then(resolve, reject); }; httpServer.once("error", onError); httpServer.once("listening", onListening); if (listenTarget.type === "tcp") { httpServer.listen(listenTarget.port, listenTarget.host); } else { if (listenTarget.type === "socket" && existsSync(listenTarget.path)) { unlinkSync(listenTarget.path); } httpServer.listen(listenTarget.path); } }); // Start speech service after listening so synchronous Sherpa native // model loading doesn't block the server from accepting connections. speechService.start(); }; const stop = async () => { await closeAllAgents(logger, agentManager); await agentManager.flush().catch(() => undefined); detachAgentStoragePersistence(); await agentStorage.flush().catch(() => undefined); await shutdownProviders(logger, { runtimeSettings: config.agentProviderSettings, }); terminalManager.killAll(); speechService.stop(); await scheduleService.stop().catch(() => undefined); await relayTransport?.stop().catch(() => undefined); if (wsServer) { await wsServer.close(); } if (voiceMcpBridgeManager) { await voiceMcpBridgeManager.stop().catch(() => undefined); } await new Promise((resolve) => { httpServer.close(() => resolve()); }); // Clean up socket files if (listenTarget.type === "socket" && existsSync(listenTarget.path)) { unlinkSync(listenTarget.path); } }; return { config, agentManager, agentStorage, terminalManager, start, stop, getListenTarget: () => boundListenTarget, }; } catch (err) { throw err; } } async function closeAllAgents(logger: Logger, agentManager: AgentManager): Promise { const agents = agentManager.listAgents(); for (const agent of agents) { try { await agentManager.closeAgent(agent.id); } catch (err) { logger.error({ err, agentId: agent.id }, "Failed to close agent"); } } }