feat(server): built-in service proxy — absorb portless into daemon

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).
This commit is contained in:
Mohamed Boudra
2026-03-30 21:41:03 +07:00
parent 0ba2f6b194
commit 266d042528
9 changed files with 829 additions and 0 deletions
@@ -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,
});
}
@@ -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,
});
}
+38
View File
@@ -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<void>;
stop(): Promise<void>;
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,
@@ -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<void> {
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<void>((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<void>((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<void>((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<void>((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<void>((resolve, reject) => {
ws.on("open", resolve);
ws.on("error", reject);
});
// 4. Send a message and verify echo
const reply = await new Promise<string>((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<void>((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<void>((resolve) => server.close(() => resolve()));
});
});
+262
View File
@@ -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<string, number>();
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<string, string | string[]> {
const out: Record<string, string | string[]> = {};
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<number> {
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);
});
}
+11
View File
@@ -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<SpeechToTextProvider | null>;
tts: Resolvable<TextToSpeechProvider | null>;
terminalManager: TerminalManager | null;
serviceRouteStore?: ServiceRouteStore;
getDaemonTcpPort?: () => number | null;
voice?: {
voiceAgentMcpStdio?: VoiceMcpStdioConfig | null;
turnDetection?: Resolvable<TurnDetectionProvider | null>;
@@ -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<string>();
private unsubscribeTerminalsChanged: (() => void) | null = null;
private terminalExitSubscriptions: Map<string, () => 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,
});
}
@@ -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,
@@ -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<boolean>;
emitLiveTimelineItem?: (item: AgentTimelineItem) => Promise<boolean>;
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<WorktreeServiceResult[]> {
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<string, string> = {
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");
}
}
+40
View File
@@ -20,6 +20,7 @@ interface PaseoConfig {
teardown?: string[];
terminals?: WorktreeTerminalConfig[];
};
services?: Record<string, ServiceConfig>;
}
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<string, ServiceConfig> {
const config = readPaseoConfig(repoRoot);
const services = config?.services;
if (!services || typeof services !== "object") {
return new Map();
}
const result = new Map<string, ServiceConfig>();
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 },