From 266d04252817f3959e0d5eb59a62f04befa205c2 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Mon, 30 Mar 2026 21:41:03 +0700 Subject: [PATCH] =?UTF-8?q?feat(server):=20built-in=20service=20proxy=20?= =?UTF-8?q?=E2=80=94=20absorb=20portless=20into=20daemon?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Services defined in paseo.json get reverse-proxied through the daemon via hostname-based routing on *.localhost. Each service receives $PORT, $HOST, and $PASEO_SERVICE_URL env vars, and is accessible at {service}.localhost:6767 (main) or {branch}.{service}.localhost:6767 (worktrees). --- .../src/server/agent/agent-management-mcp.ts | 5 + .../server/src/server/agent/mcp-server.ts | 5 + packages/server/src/server/bootstrap.ts | 38 ++ .../server/src/server/service-proxy.test.ts | 347 ++++++++++++++++++ packages/server/src/server/service-proxy.ts | 262 +++++++++++++ packages/server/src/server/session.ts | 11 + .../server/src/server/websocket-server.ts | 9 + .../server/src/server/worktree-bootstrap.ts | 112 ++++++ packages/server/src/utils/worktree.ts | 40 ++ 9 files changed, 829 insertions(+) create mode 100644 packages/server/src/server/service-proxy.test.ts create mode 100644 packages/server/src/server/service-proxy.ts diff --git a/packages/server/src/server/agent/agent-management-mcp.ts b/packages/server/src/server/agent/agent-management-mcp.ts index 14086d973..36ccf9ab6 100644 --- a/packages/server/src/server/agent/agent-management-mcp.ts +++ b/packages/server/src/server/agent/agent-management-mcp.ts @@ -48,11 +48,14 @@ import { scheduleAgentMetadataGeneration } from "./agent-metadata-generator.js"; import { expandUserPath } from "../path-utils.js"; import type { TerminalManager } from "../../terminal/terminal-manager.js"; import { createAgentWorktree, runAsyncWorktreeBootstrap } from "../worktree-bootstrap.js"; +import type { ServiceRouteStore } from "../service-proxy.js"; export interface AgentManagementMcpOptions { agentManager: AgentManager; agentStorage: AgentStorage; terminalManager?: TerminalManager | null; + serviceRouteStore?: ServiceRouteStore; + getDaemonTcpPort?: () => number | null; paseoHome?: string; logger: Logger; } @@ -339,6 +342,8 @@ export async function createAgentManagementMcpServer( agentId: snapshot.id, item, }), + serviceRouteStore: options.serviceRouteStore, + daemonPort: options.getDaemonTcpPort?.() ?? null, logger: childLogger, }); } diff --git a/packages/server/src/server/agent/mcp-server.ts b/packages/server/src/server/agent/mcp-server.ts index 60f20cf5a..b1fb60104 100644 --- a/packages/server/src/server/agent/mcp-server.ts +++ b/packages/server/src/server/agent/mcp-server.ts @@ -28,11 +28,14 @@ import type { VoiceCallerContext, VoiceSpeakHandler } from "../voice-types.js"; import { expandUserPath, resolvePathFromBase } from "../path-utils.js"; import type { TerminalManager } from "../../terminal/terminal-manager.js"; import { createAgentWorktree, runAsyncWorktreeBootstrap } from "../worktree-bootstrap.js"; +import type { ServiceRouteStore } from "../service-proxy.js"; export interface AgentMcpServerOptions { agentManager: AgentManager; agentStorage: AgentStorage; terminalManager?: TerminalManager | null; + serviceRouteStore?: ServiceRouteStore; + getDaemonTcpPort?: () => number | null; paseoHome?: string; /** * ID of the agent that is connecting to this MCP server. @@ -510,6 +513,8 @@ export async function createAgentMcpServer(options: AgentMcpServerOptions): Prom agentId: snapshot.id, item, }), + serviceRouteStore: options.serviceRouteStore, + daemonPort: options.getDaemonTcpPort?.() ?? null, logger: childLogger, }); } diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index e73fe74a6..42db80269 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -113,6 +113,11 @@ 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 { + ServiceRouteStore, + createServiceProxyMiddleware, + createServiceProxyUpgradeHandler, +} from "./service-proxy.js"; import { createVoiceMcpSocketBridgeManager, type VoiceMcpSocketBridgeManager, @@ -189,6 +194,7 @@ export interface PaseoDaemon { agentManager: AgentManager; agentStorage: AgentStorage; terminalManager: TerminalManager; + serviceRouteStore: ServiceRouteStore; start(): Promise; stop(): Promise; getListenTarget(): ListenTarget | null; @@ -218,6 +224,8 @@ export async function createPaseoDaemon( const app = express(); let boundListenTarget: ListenTarget | null = null; + const serviceRouteStore = new ServiceRouteStore(); + // Host allowlist / DNS rebinding protection (vite-like semantics). // For non-TCP (unix sockets), skip host validation. if (listenTarget.type === "tcp") { @@ -231,6 +239,14 @@ export async function createPaseoDaemon( }); } + // Service proxy — intercepts requests for registered *.localhost hostnames + // and forwards them to the corresponding local service port. Placed after + // the host allowlist (*.localhost is already allowed) but before CORS and + // the rest of the routes so proxied requests skip unnecessary middleware. + app.use( + createServiceProxyMiddleware({ routeStore: serviceRouteStore, logger }), + ); + // CORS - allow same-origin + configured origins const allowedOrigins = new Set([ ...config.corsAllowedOrigins, @@ -352,6 +368,16 @@ export async function createPaseoDaemon( const httpServer = createHTTPServer(app); + // Service proxy WebSocket upgrade handler — must be registered before the + // VoiceAssistantWebSocketServer attaches its own "upgrade" listener so that + // service-bound upgrades are forwarded first. The handler is a no-op for + // requests that don't match a registered service route. + const serviceProxyUpgradeHandler = createServiceProxyUpgradeHandler({ + routeStore: serviceRouteStore, + logger, + }); + httpServer.on("upgrade", serviceProxyUpgradeHandler); + const agentStorage = new AgentStorage(config.agentStoragePath, logger); const projectRegistry = new FileBackedProjectRegistry( path.join(config.paseoHome, "projects", "projects.json"), @@ -433,6 +459,9 @@ export async function createPaseoDaemon( agentManager, agentStorage, terminalManager, + serviceRouteStore, + getDaemonTcpPort: () => + boundListenTarget?.type === "tcp" ? boundListenTarget.port : null, paseoHome: config.paseoHome, enableVoiceTools: false, resolveSpeakHandler: (callerAgentId) => @@ -459,6 +488,9 @@ export async function createPaseoDaemon( agentManager, agentStorage, terminalManager, + serviceRouteStore, + getDaemonTcpPort: () => + boundListenTarget?.type === "tcp" ? boundListenTarget.port : null, paseoHome: config.paseoHome, callerAgentId, enableVoiceTools: false, @@ -579,6 +611,9 @@ export async function createPaseoDaemon( agentManager, agentStorage, terminalManager, + serviceRouteStore, + getDaemonTcpPort: () => + boundListenTarget?.type === "tcp" ? boundListenTarget.port : null, paseoHome: config.paseoHome, callerAgentId, voiceOnly: true, @@ -639,6 +674,8 @@ export async function createPaseoDaemon( loopService, scheduleService, checkoutDiffManager, + serviceRouteStore, + () => (boundListenTarget?.type === "tcp" ? boundListenTarget.port : null), ); logger.info({ elapsed: elapsed() }, "Bootstrap complete, ready to start listening"); @@ -763,6 +800,7 @@ export async function createPaseoDaemon( agentManager, agentStorage, terminalManager, + serviceRouteStore, start, stop, getListenTarget: () => boundListenTarget, diff --git a/packages/server/src/server/service-proxy.test.ts b/packages/server/src/server/service-proxy.test.ts new file mode 100644 index 000000000..a39c04eb8 --- /dev/null +++ b/packages/server/src/server/service-proxy.test.ts @@ -0,0 +1,347 @@ +import { describe, it, expect, afterEach } from "vitest"; +import http from "node:http"; +import net from "node:net"; +import express from "express"; +import WebSocket, { WebSocketServer } from "ws"; +import pino from "pino"; +import { + ServiceRouteStore, + createServiceProxyMiddleware, + createServiceProxyUpgradeHandler, + findFreePort, +} from "./service-proxy.js"; + +const logger = pino({ level: "silent" }); + +// --------------------------------------------------------------------------- +// Helpers for cleanup +// --------------------------------------------------------------------------- + +function closeServer(server: http.Server): Promise { + return new Promise((resolve) => { + server.close(() => resolve()); + }); +} + +// --------------------------------------------------------------------------- +// ServiceRouteStore +// --------------------------------------------------------------------------- + +describe("ServiceRouteStore", () => { + it("addRoute and findRoute with exact match", () => { + const store = new ServiceRouteStore(); + store.addRoute("editor.localhost", 3000); + + const route = store.findRoute("editor.localhost"); + expect(route).toEqual({ hostname: "editor.localhost", port: 3000 }); + }); + + it("findRoute strips port from host header", () => { + const store = new ServiceRouteStore(); + store.addRoute("editor.localhost", 3000); + + const route = store.findRoute("editor.localhost:6767"); + expect(route).toEqual({ hostname: "editor.localhost", port: 3000 }); + }); + + it("findRoute subdomain match", () => { + const store = new ServiceRouteStore(); + store.addRoute("editor.localhost", 3000); + + const route = store.findRoute("fix-auth.editor.localhost"); + expect(route).toEqual({ hostname: "editor.localhost", port: 3000 }); + }); + + it("removeRoute works", () => { + const store = new ServiceRouteStore(); + store.addRoute("editor.localhost", 3000); + store.removeRoute("editor.localhost"); + + expect(store.findRoute("editor.localhost")).toBeNull(); + }); + + it("removeRoutesForPort works", () => { + const store = new ServiceRouteStore(); + store.addRoute("a.localhost", 3000); + store.addRoute("b.localhost", 3000); + store.addRoute("c.localhost", 4000); + + store.removeRoutesForPort(3000); + + expect(store.findRoute("a.localhost")).toBeNull(); + expect(store.findRoute("b.localhost")).toBeNull(); + expect(store.findRoute("c.localhost")).toEqual({ + hostname: "c.localhost", + port: 4000, + }); + }); + + it("findRoute returns null for unknown hosts", () => { + const store = new ServiceRouteStore(); + store.addRoute("editor.localhost", 3000); + + expect(store.findRoute("unknown.example.com")).toBeNull(); + }); + + it("listRoutes returns all routes", () => { + const store = new ServiceRouteStore(); + store.addRoute("a.localhost", 3000); + store.addRoute("b.localhost", 4000); + + const routes = store.listRoutes(); + expect(routes).toHaveLength(2); + expect(routes).toContainEqual({ hostname: "a.localhost", port: 3000 }); + expect(routes).toContainEqual({ hostname: "b.localhost", port: 4000 }); + }); +}); + +// --------------------------------------------------------------------------- +// HTTP proxy +// --------------------------------------------------------------------------- + +describe("HTTP proxy", () => { + const servers: http.Server[] = []; + + afterEach(async () => { + await Promise.all(servers.map(closeServer)); + servers.length = 0; + }); + + /** Start a real HTTP server that echoes back a known body and records received headers. */ + async function startUpstream(): Promise<{ + port: number; + server: http.Server; + receivedHeaders: () => http.IncomingHttpHeaders; + }> { + const port = await findFreePort(); + let lastHeaders: http.IncomingHttpHeaders = {}; + + const server = http.createServer((req, res) => { + lastHeaders = req.headers; + res.writeHead(200, { "content-type": "text/plain" }); + res.end("upstream-ok"); + }); + + await new Promise((resolve) => + server.listen(port, "127.0.0.1", resolve), + ); + servers.push(server); + + return { + port, + server, + receivedHeaders: () => lastHeaders, + }; + } + + /** Start an Express app with the service proxy middleware and an optional fallback. */ + async function startProxy( + routeStore: ServiceRouteStore, + opts?: { fallback?: boolean }, + ): Promise<{ port: number; server: http.Server }> { + const port = await findFreePort(); + const app = express(); + app.use(createServiceProxyMiddleware({ routeStore, logger })); + + if (opts?.fallback) { + app.use((_req, res) => { + res.status(404).send("no route"); + }); + } + + const server = http.createServer(app); + await new Promise((resolve) => + server.listen(port, "127.0.0.1", resolve), + ); + servers.push(server); + + return { port, server }; + } + + /** Simple HTTP GET helper that returns status code and body. */ + function httpGet( + port: number, + host: string, + path = "/", + ): Promise<{ status: number; body: string }> { + return new Promise((resolve, reject) => { + const req = http.get( + { hostname: "127.0.0.1", port, path, headers: { host } }, + (res) => { + let body = ""; + res.on("data", (chunk: Buffer) => (body += chunk.toString())); + res.on("end", () => + resolve({ status: res.statusCode ?? 0, body }), + ); + }, + ); + req.on("error", reject); + }); + } + + it("proxies requests to the correct upstream based on Host header", async () => { + const upstream = await startUpstream(); + const routeStore = new ServiceRouteStore(); + routeStore.addRoute("test-service.localhost", upstream.port); + + const proxy = await startProxy(routeStore); + const res = await httpGet( + proxy.port, + `test-service.localhost:${proxy.port}`, + ); + + expect(res.status).toBe(200); + expect(res.body).toBe("upstream-ok"); + + const headers = upstream.receivedHeaders(); + expect(headers["x-forwarded-for"]).toBeDefined(); + expect(headers["x-forwarded-host"]).toBe("test-service.localhost"); + }); + + it("falls through when no route matches", async () => { + const routeStore = new ServiceRouteStore(); + const proxy = await startProxy(routeStore, { fallback: true }); + + const res = await httpGet( + proxy.port, + `unknown.localhost:${proxy.port}`, + ); + + expect(res.status).toBe(404); + expect(res.body).toBe("no route"); + }); + + it("returns 502 when upstream is down", async () => { + // Get a port that nothing is listening on + const deadPort = await findFreePort(); + + const routeStore = new ServiceRouteStore(); + routeStore.addRoute("dead-service.localhost", deadPort); + + const proxy = await startProxy(routeStore); + const res = await httpGet( + proxy.port, + `dead-service.localhost:${proxy.port}`, + ); + + expect(res.status).toBe(502); + expect(res.body).toBe("502 Bad Gateway"); + }); +}); + +// --------------------------------------------------------------------------- +// WebSocket proxy +// --------------------------------------------------------------------------- + +describe("WebSocket proxy", () => { + const httpServers: http.Server[] = []; + const wsServers: WebSocketServer[] = []; + const wsClients: WebSocket[] = []; + + afterEach(async () => { + for (const ws of wsClients) { + if (ws.readyState === WebSocket.OPEN) ws.close(); + } + wsClients.length = 0; + + for (const wss of wsServers) { + wss.close(); + } + wsServers.length = 0; + + await Promise.all(httpServers.map(closeServer)); + httpServers.length = 0; + }); + + it("proxies WebSocket connections to the correct upstream", async () => { + // 1. Start a real WebSocket echo server + const upstreamPort = await findFreePort(); + const upstreamServer = http.createServer(); + const wss = new WebSocketServer({ server: upstreamServer }); + wsServers.push(wss); + + wss.on("connection", (ws) => { + ws.on("message", (data) => { + ws.send(`echo: ${data.toString()}`); + }); + }); + + await new Promise((resolve) => + upstreamServer.listen(upstreamPort, "127.0.0.1", resolve), + ); + httpServers.push(upstreamServer); + + // 2. Create the proxy server with the upgrade handler + const routeStore = new ServiceRouteStore(); + routeStore.addRoute("ws-service.localhost", upstreamPort); + + const proxyPort = await findFreePort(); + const proxyServer = http.createServer((_req, res) => { + res.writeHead(404); + res.end(); + }); + + const upgradeHandler = createServiceProxyUpgradeHandler({ + routeStore, + logger, + }); + proxyServer.on("upgrade", upgradeHandler); + + await new Promise((resolve) => + proxyServer.listen(proxyPort, "127.0.0.1", resolve), + ); + httpServers.push(proxyServer); + + // 3. Connect a WebSocket client through the proxy + const ws = new WebSocket(`ws://127.0.0.1:${proxyPort}`, { + headers: { host: `ws-service.localhost:${proxyPort}` }, + }); + wsClients.push(ws); + + await new Promise((resolve, reject) => { + ws.on("open", resolve); + ws.on("error", reject); + }); + + // 4. Send a message and verify echo + const reply = await new Promise((resolve, reject) => { + ws.on("message", (data) => resolve(data.toString())); + ws.on("error", reject); + ws.send("hello proxy"); + }); + + expect(reply).toBe("echo: hello proxy"); + }); +}); + +// --------------------------------------------------------------------------- +// findFreePort +// --------------------------------------------------------------------------- + +describe("findFreePort", () => { + it("returns a number", async () => { + const port = await findFreePort(); + expect(typeof port).toBe("number"); + expect(port).toBeGreaterThan(0); + expect(port).toBeLessThan(65536); + }); + + it("returns a port that is actually available", async () => { + const port = await findFreePort(); + + // Verify we can bind a server to it + const server = net.createServer(); + await new Promise((resolve, reject) => { + server.listen(port, "127.0.0.1", () => resolve()); + server.on("error", reject); + }); + + const addr = server.address(); + expect(addr).not.toBeNull(); + expect(typeof addr === "object" && addr !== null ? addr.port : -1).toBe( + port, + ); + + await new Promise((resolve) => server.close(() => resolve())); + }); +}); diff --git a/packages/server/src/server/service-proxy.ts b/packages/server/src/server/service-proxy.ts new file mode 100644 index 000000000..8839fadba --- /dev/null +++ b/packages/server/src/server/service-proxy.ts @@ -0,0 +1,262 @@ +import http from "node:http"; +import net from "node:net"; +import type { IncomingMessage } from "node:http"; +import type { Logger } from "pino"; +import type { RequestHandler } from "express"; + +// --------------------------------------------------------------------------- +// Hop-by-hop headers that must not be forwarded +// --------------------------------------------------------------------------- + +const HOP_BY_HOP_HEADERS = new Set([ + "connection", + "transfer-encoding", + "keep-alive", + "upgrade", + "proxy-connection", + "proxy-authenticate", + "proxy-authorization", + "te", + "trailer", +]); + +// --------------------------------------------------------------------------- +// ServiceRouteStore +// --------------------------------------------------------------------------- + +export interface ServiceRoute { + hostname: string; + port: number; +} + +export class ServiceRouteStore { + private routes = new Map(); + + addRoute(hostname: string, port: number): void { + this.routes.set(hostname, port); + } + + removeRoute(hostname: string): void { + this.routes.delete(hostname); + } + + removeRoutesForPort(port: number): void { + for (const [hostname, p] of this.routes) { + if (p === port) { + this.routes.delete(hostname); + } + } + } + + findRoute(host: string): ServiceRoute | null { + // Strip port suffix from the Host header value + const hostname = host.replace(/:\d+$/, ""); + + // 1. Exact match + const exactPort = this.routes.get(hostname); + if (exactPort !== undefined) { + return { hostname, port: exactPort }; + } + + // 2. Subdomain match — walk up the labels looking for a registered parent + const parts = hostname.split("."); + for (let i = 1; i < parts.length; i++) { + const candidate = parts.slice(i).join("."); + const candidatePort = this.routes.get(candidate); + if (candidatePort !== undefined) { + return { hostname: candidate, port: candidatePort }; + } + } + + return null; + } + + listRoutes(): ServiceRoute[] { + return Array.from(this.routes.entries()).map(([hostname, port]) => ({ + hostname, + port, + })); + } +} + +// --------------------------------------------------------------------------- +// Helpers +// --------------------------------------------------------------------------- + +function stripHopByHopHeaders( + rawHeaders: http.IncomingHttpHeaders, +): Record { + const out: Record = {}; + for (const [key, value] of Object.entries(rawHeaders)) { + if (value === undefined) continue; + if (HOP_BY_HOP_HEADERS.has(key.toLowerCase())) continue; + out[key] = value; + } + return out; +} + +// --------------------------------------------------------------------------- +// createServiceProxyMiddleware +// --------------------------------------------------------------------------- + +export function createServiceProxyMiddleware({ + routeStore, + logger, +}: { + routeStore: ServiceRouteStore; + logger: Logger; +}): RequestHandler { + return (req, res, next) => { + const hostHeader = req.headers.host; + if (!hostHeader) { + next(); + return; + } + + const route = routeStore.findRoute(hostHeader); + if (!route) { + next(); + return; + } + + const forwardedHeaders = stripHopByHopHeaders(req.headers); + forwardedHeaders["x-forwarded-for"] = + req.socket.remoteAddress ?? "127.0.0.1"; + forwardedHeaders["x-forwarded-host"] = hostHeader.replace(/:\d+$/, ""); + forwardedHeaders["x-forwarded-proto"] = req.protocol; + + const proxyReq = http.request( + { + hostname: "127.0.0.1", + port: route.port, + path: req.originalUrl, + method: req.method, + headers: forwardedHeaders, + }, + (proxyRes) => { + const responseHeaders = stripHopByHopHeaders(proxyRes.headers); + res.writeHead(proxyRes.statusCode ?? 502, responseHeaders); + proxyRes.pipe(res, { end: true }); + }, + ); + + proxyReq.on("error", (err) => { + logger.warn( + { err, hostname: route.hostname, port: route.port }, + "Service proxy: upstream unreachable", + ); + if (!res.headersSent) { + res.writeHead(502, { "content-type": "text/plain" }); + res.end("502 Bad Gateway"); + } + }); + + req.pipe(proxyReq, { end: true }); + }; +} + +// --------------------------------------------------------------------------- +// createServiceProxyUpgradeHandler +// --------------------------------------------------------------------------- + +export function createServiceProxyUpgradeHandler({ + routeStore, + logger, +}: { + routeStore: ServiceRouteStore; + logger: Logger; +}): (req: IncomingMessage, socket: net.Socket, head: Buffer) => void { + return (req, socket, head) => { + const hostHeader = req.headers.host; + if (!hostHeader) { + return; + } + + const route = routeStore.findRoute(hostHeader); + if (!route) { + return; + } + + const targetSocket = net.connect( + { host: "127.0.0.1", port: route.port }, + () => { + // Reconstruct the raw HTTP upgrade request to send to the target + const forwardedHeaders = stripHopByHopHeaders(req.headers); + forwardedHeaders["x-forwarded-for"] = + req.socket.remoteAddress ?? "127.0.0.1"; + forwardedHeaders["x-forwarded-host"] = hostHeader.replace(/:\d+$/, ""); + forwardedHeaders["x-forwarded-proto"] = "http"; + + // Re-include upgrade and connection headers — they are required for + // WebSocket handshake even though they are hop-by-hop. + forwardedHeaders["connection"] = "Upgrade"; + forwardedHeaders["upgrade"] = req.headers.upgrade ?? "websocket"; + + const headerLines: string[] = []; + headerLines.push( + `${req.method ?? "GET"} ${req.url ?? "/"} HTTP/${req.httpVersion}`, + ); + for (const [key, value] of Object.entries(forwardedHeaders)) { + if (Array.isArray(value)) { + for (const v of value) { + headerLines.push(`${key}: ${v}`); + } + } else { + headerLines.push(`${key}: ${value}`); + } + } + headerLines.push("\r\n"); + + targetSocket.write(headerLines.join("\r\n")); + + if (head.length > 0) { + targetSocket.write(head); + } + + // Pipe in both directions + targetSocket.pipe(socket); + socket.pipe(targetSocket); + }, + ); + + targetSocket.on("error", (err) => { + logger.warn( + { err, hostname: route.hostname, port: route.port }, + "Service proxy: WebSocket upstream unreachable", + ); + socket.end(); + }); + + socket.on("error", () => { + targetSocket.destroy(); + }); + }; +} + +// --------------------------------------------------------------------------- +// findFreePort +// --------------------------------------------------------------------------- + +export function findFreePort(): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.unref(); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (!address || typeof address === "string") { + server.close(); + reject(new Error("Failed to get assigned port")); + return; + } + const { port } = address; + server.close((err) => { + if (err) { + reject(err); + } else { + resolve(port); + } + }); + }); + server.on("error", reject); + }); +} diff --git a/packages/server/src/server/session.ts b/packages/server/src/server/session.ts index 45bb25437..05a22c5a6 100644 --- a/packages/server/src/server/session.ts +++ b/packages/server/src/server/session.ts @@ -139,6 +139,7 @@ import { type WorktreeConfig, } from "../utils/worktree.js"; import { runAsyncWorktreeBootstrap } from "./worktree-bootstrap.js"; +import type { ServiceRouteStore } from "./service-proxy.js"; import { getCheckoutDiff, getCheckoutShortstat, @@ -372,6 +373,8 @@ export type SessionOptions = { stt: Resolvable; tts: Resolvable; terminalManager: TerminalManager | null; + serviceRouteStore?: ServiceRouteStore; + getDaemonTcpPort?: () => number | null; voice?: { voiceAgentMcpStdio?: VoiceMcpStdioConfig | null; turnDetection?: Resolvable; @@ -567,6 +570,8 @@ export class Session { } | null = null; private readonly MOBILE_BACKGROUND_STREAM_GRACE_MS = 60_000; private readonly terminalManager: TerminalManager | null; + private readonly serviceRouteStore: ServiceRouteStore | null; + private readonly getDaemonTcpPort: (() => number | null) | null; private readonly subscribedTerminalDirectories = new Set(); private unsubscribeTerminalsChanged: (() => void) | null = null; private terminalExitSubscriptions: Map void> = new Map(); @@ -618,6 +623,8 @@ export class Session { stt, tts, terminalManager, + serviceRouteStore, + getDaemonTcpPort, voice, voiceBridge, dictation, @@ -642,6 +649,8 @@ export class Session { this.checkoutDiffManager = checkoutDiffManager; this.createAgentMcpTransport = createAgentMcpTransport; this.terminalManager = terminalManager; + this.serviceRouteStore = serviceRouteStore ?? null; + this.getDaemonTcpPort = getDaemonTcpPort ?? null; if (this.terminalManager) { this.unsubscribeTerminalsChanged = this.terminalManager.subscribeTerminalsChanged((event) => this.handleTerminalsChanged(event), @@ -2732,6 +2741,8 @@ export class Session { agentId: snapshot.id, item, }), + serviceRouteStore: this.serviceRouteStore ?? undefined, + daemonPort: this.getDaemonTcpPort?.() ?? null, logger: this.sessionLogger, }); } diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index ab9d0f787..00a88d219 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -33,6 +33,7 @@ import type { AgentProvider } from "./agent/agent-sdk-types.js"; import type { AgentProviderRuntimeSettingsMap } from "./agent/provider-launch-config.js"; import { PushTokenStore } from "./push/token-store.js"; import { PushService } from "./push/push-service.js"; +import type { ServiceRouteStore } from "./service-proxy.js"; import type { SpeechReadinessSnapshot, SpeechService } from "./speech/speech-runtime.js"; import type { VoiceCallerContext, VoiceMcpStdioConfig, VoiceSpeakHandler } from "./voice-types.js"; import { @@ -241,6 +242,8 @@ export class VoiceAssistantWebSocketServer { private readonly createAgentMcpTransport: AgentMcpTransportFactory; private readonly speech: SpeechService | null; private readonly terminalManager: TerminalManager | null; + private readonly serviceRouteStore: ServiceRouteStore | null; + private readonly getDaemonTcpPort: (() => number | null) | null; private readonly dictation: { finalTimeoutMs?: number; } | null; @@ -307,6 +310,8 @@ export class VoiceAssistantWebSocketServer { loopService?: LoopService, scheduleService?: ScheduleService, checkoutDiffManager?: CheckoutDiffManager, + serviceRouteStore?: ServiceRouteStore | null, + getDaemonTcpPort?: () => number | null, ) { this.logger = logger.child({ module: "websocket-server" }); this.serverId = serverId; @@ -343,6 +348,8 @@ export class VoiceAssistantWebSocketServer { this.dictation = dictation ?? null; this.agentProviderRuntimeSettings = agentProviderRuntimeSettings; this.onLifecycleIntent = onLifecycleIntent ?? null; + this.serviceRouteStore = serviceRouteStore ?? null; + this.getDaemonTcpPort = getDaemonTcpPort ?? null; this.serverCapabilities = buildServerCapabilities({ readiness: this.speech?.getReadiness() ?? null, }); @@ -643,6 +650,8 @@ export class VoiceAssistantWebSocketServer { stt: () => this.speech?.resolveStt() ?? null, tts: () => this.speech?.resolveTts() ?? null, terminalManager: this.terminalManager, + serviceRouteStore: this.serviceRouteStore ?? undefined, + getDaemonTcpPort: this.getDaemonTcpPort ?? undefined, voice: { ...(this.voice ?? {}), turnDetection: () => this.speech?.resolveTurnDetection() ?? null, diff --git a/packages/server/src/server/worktree-bootstrap.ts b/packages/server/src/server/worktree-bootstrap.ts index afeec0c29..849007779 100644 --- a/packages/server/src/server/worktree-bootstrap.ts +++ b/packages/server/src/server/worktree-bootstrap.ts @@ -7,15 +7,18 @@ import type { TerminalManager } from "../terminal/terminal-manager.js"; import type { TerminalSession } from "../terminal/terminal.js"; import { createWorktree, + getServiceConfigs, getWorktreeTerminalSpecs, listPaseoWorktrees, resolveWorktreeRuntimeEnv, runWorktreeSetupCommands, + slugify, WorktreeSetupError, type WorktreeConfig, type WorktreeSetupCommandResult, type WorktreeRuntimeEnv, } from "../utils/worktree.js"; +import { findFreePort, type ServiceRouteStore } from "./service-proxy.js"; import type { AgentTimelineItem } from "./agent/agent-sdk-types.js"; export interface WorktreeBootstrapTerminalResult { @@ -30,6 +33,8 @@ export interface RunAsyncWorktreeBootstrapOptions { agentId: string; worktree: WorktreeConfig; terminalManager: TerminalManager | null; + serviceRouteStore?: ServiceRouteStore; + daemonPort?: number | null; appendTimelineItem: (item: AgentTimelineItem) => Promise; emitLiveTimelineItem?: (item: AgentTimelineItem) => Promise; logger?: Logger; @@ -653,4 +658,111 @@ export async function runAsyncWorktreeBootstrap( } await runWorktreeTerminalBootstrap(options, runtimeEnv); + + if ( + !options.terminalManager || + !options.serviceRouteStore || + options.daemonPort === null || + options.daemonPort === undefined + ) { + return; + } + + try { + await spawnWorktreeServices({ + repoRoot: options.worktree.worktreePath, + branchName: options.worktree.branchName, + daemonPort: options.daemonPort, + routeStore: options.serviceRouteStore, + terminalManager: options.terminalManager, + logger: options.logger, + }); + } catch (error) { + options.logger?.warn( + { err: error, agentId: options.agentId, worktreePath: options.worktree.worktreePath }, + "Failed to spawn worktree services", + ); + } +} + +// --------------------------------------------------------------------------- +// Service lifecycle helpers +// --------------------------------------------------------------------------- + +export interface WorktreeServiceResult { + serviceName: string; + hostname: string; + port: number; + terminalId: string; +} + +export async function spawnWorktreeServices(options: { + repoRoot: string; + branchName: string | null; + daemonPort: number; + routeStore: ServiceRouteStore; + terminalManager: TerminalManager; + logger?: Logger; +}): Promise { + const { repoRoot, branchName, daemonPort, routeStore, terminalManager, logger } = options; + const serviceConfigs = getServiceConfigs(repoRoot); + if (serviceConfigs.size === 0) { + return []; + } + + const results: WorktreeServiceResult[] = []; + + for (const [serviceName, config] of serviceConfigs) { + const port = config.port ?? (await findFreePort()); + const branchHostnameLabel = branchName ? slugify(branchName) : null; + + const isDefaultBranch = + branchName === null || branchName === "main" || branchName === "master"; + const hostname = isDefaultBranch + ? `${serviceName}.localhost` + : `${branchHostnameLabel}.${serviceName}.localhost`; + + routeStore.addRoute(hostname, port); + + const env: Record = { + PORT: String(port), + HOST: "127.0.0.1", + PASEO_SERVICE_URL: `http://${hostname}:${daemonPort}`, + }; + + const terminal = await terminalManager.createTerminal({ + cwd: repoRoot, + name: serviceName, + env, + }); + + await waitForTerminalBootstrapReadiness(terminal); + terminal.send({ type: "input", data: `${config.command}\r` }); + + logger?.info( + { serviceName, hostname, port, terminalId: terminal.id }, + `Registered service proxy: ${hostname} -> 127.0.0.1:${port}`, + ); + + results.push({ + serviceName, + hostname, + port, + terminalId: terminal.id, + }); + } + + return results; +} + +export function teardownWorktreeServices(options: { + hostnames: string[]; + routeStore: ServiceRouteStore; + logger: Logger; +}): void { + const { hostnames, routeStore, logger } = options; + for (const hostname of hostnames) { + routeStore.removeRoute(hostname); + logger.info({ hostname }, "Removed service proxy route"); + } } diff --git a/packages/server/src/utils/worktree.ts b/packages/server/src/utils/worktree.ts index 193901553..3dcb38811 100644 --- a/packages/server/src/utils/worktree.ts +++ b/packages/server/src/utils/worktree.ts @@ -20,6 +20,7 @@ interface PaseoConfig { teardown?: string[]; terminals?: WorktreeTerminalConfig[]; }; + services?: Record; } const execAsync = promisify(exec); @@ -84,6 +85,11 @@ export interface WorktreeTerminalConfig { command: string; } +export interface ServiceConfig { + command: string; + port?: number; // explicit port override, otherwise auto-assigned +} + export class WorktreeSetupError extends Error { readonly results: WorktreeSetupCommandResult[]; @@ -194,6 +200,40 @@ export function getWorktreeTerminalSpecs(repoRoot: string): WorktreeTerminalConf return specs; } +export function getServiceConfigs(repoRoot: string): Map { + const config = readPaseoConfig(repoRoot); + const services = config?.services; + if (!services || typeof services !== "object") { + return new Map(); + } + + const result = new Map(); + for (const [name, entry] of Object.entries(services)) { + if (!entry || typeof entry !== "object") { + continue; + } + + const rawCommand = entry.command; + if (typeof rawCommand !== "string") { + continue; + } + const command = rawCommand.trim(); + if (!command) { + continue; + } + + const serviceConfig: ServiceConfig = { command }; + + if (typeof entry.port === "number" && Number.isFinite(entry.port)) { + serviceConfig.port = entry.port; + } + + result.set(name, serviceConfig); + } + + return result; +} + async function execSetupCommand( command: string, options: { cwd: string; env: NodeJS.ProcessEnv },