From fd5e8206a7698822f784cc96d6e65a06f823145f Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Wed, 21 Jan 2026 09:27:16 +0700 Subject: [PATCH] feat: relay transport (plaintext) --- .../server/src/client/daemon-client-v2.ts | 6 +- packages/server/src/server/bootstrap.ts | 16 ++- .../daemon-e2e/connection-offer.e2e.test.ts | 1 + .../daemon-e2e/relay-transport.e2e.test.ts | 107 ++++++++++++++ packages/server/src/server/relay-transport.ts | 135 ++++++++++++++++++ .../src/server/test-utils/paseo-daemon.ts | 6 +- .../server/src/server/websocket-server.ts | 10 +- 7 files changed, 274 insertions(+), 7 deletions(-) create mode 100644 packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts create mode 100644 packages/server/src/server/relay-transport.ts diff --git a/packages/server/src/client/daemon-client-v2.ts b/packages/server/src/client/daemon-client-v2.ts index ba2979825..3c3e336bc 100644 --- a/packages/server/src/client/daemon-client-v2.ts +++ b/packages/server/src/client/daemon-client-v2.ts @@ -274,9 +274,9 @@ export class DaemonClientV2 { } const targetUrl = this.conversationId - ? `${this.config.url}?conversationId=${encodeURIComponent( - this.conversationId - )}` + ? `${this.config.url}${ + this.config.url.includes("?") ? "&" : "?" + }conversationId=${encodeURIComponent(this.conversationId)}` : this.config.url; try { diff --git a/packages/server/src/server/bootstrap.ts b/packages/server/src/server/bootstrap.ts index 585bed458..f291fbbb7 100644 --- a/packages/server/src/server/bootstrap.ts +++ b/packages/server/src/server/bootstrap.ts @@ -54,6 +54,7 @@ import { createConnectionOfferV1, encodeOfferToFragmentUrl, } from "./connection-offer.js"; +import { startRelayTransport, type RelayTransportController } from "./relay-transport.js"; import type { AgentClient, AgentControlMcpConfig, @@ -100,6 +101,8 @@ export async function createPaseoDaemon( rootLogger: Logger ): Promise { const logger = rootLogger.child({ module: "bootstrap" }); + const connectionSessionId = randomUUID(); + let relayTransport: RelayTransportController | null = null; const agentMcpRoute = config.agentMcpRoute; const staticDir = config.staticDir; @@ -433,12 +436,22 @@ export async function createPaseoDaemon( }); const offer = await createConnectionOfferV1({ - sessionId: randomUUID(), + sessionId: connectionSessionId, endpoints, }); const url = encodeOfferToFragmentUrl({ offer, appBaseUrl }); logger.info({ url }, "pairing_offer"); + + if (relayEnabled) { + relayTransport?.stop().catch(() => undefined); + relayTransport = startRelayTransport({ + logger, + attachSocket: (ws) => wsServer.attachExternalSocket(ws), + relayEndpoint, + sessionId: connectionSessionId, + }); + } } else { logger.info({ path: listenTarget.path }, `Server listening on ${listenTarget.path}`); } @@ -465,6 +478,7 @@ export async function createPaseoDaemon( await closeAllAgents(logger, agentManager); await shutdownProviders(logger); terminalManager.killAll(); + await relayTransport?.stop().catch(() => undefined); await wsServer.close(); await new Promise((resolve) => { httpServer.close(() => resolve()); diff --git a/packages/server/src/server/daemon-e2e/connection-offer.e2e.test.ts b/packages/server/src/server/daemon-e2e/connection-offer.e2e.test.ts index 93f7a3720..82f9617c5 100644 --- a/packages/server/src/server/daemon-e2e/connection-offer.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/connection-offer.e2e.test.ts @@ -77,6 +77,7 @@ describe("ConnectionOfferV1 (daemon E2E)", () => { const daemon = await createTestPaseoDaemon({ listen: "0.0.0.0", logger, + relayEnabled: true, }); try { diff --git a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts new file mode 100644 index 000000000..7ca68913c --- /dev/null +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -0,0 +1,107 @@ +import { describe, expect, test } from "vitest"; +import WebSocket from "ws"; +import pino from "pino"; +import { Writable } from "node:stream"; + +import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; + +function createCapturingLogger() { + const lines: string[] = []; + const stream = new Writable({ + write(chunk, _enc, cb) { + lines.push(chunk.toString("utf8")); + cb(); + }, + }); + const logger = pino({ level: "info" }, stream); + return { logger, lines }; +} + +function parseOfferUrlFromLogs(lines: string[]): string { + for (const line of lines) { + if (!line.includes("pairing_offer")) continue; + try { + const obj = JSON.parse(line) as { msg?: string; url?: string }; + if (obj.msg === "pairing_offer" && typeof obj.url === "string") { + return obj.url; + } + } catch { + // ignore + } + } + throw new Error(`pairing_offer log not found. saw ${lines.length} lines`); +} + +function decodeOfferFromFragmentUrl(url: string): { sessionId: string } { + const marker = "#offer="; + const idx = url.indexOf(marker); + if (idx === -1) { + throw new Error(`missing ${marker} fragment: ${url}`); + } + const encoded = url.slice(idx + marker.length); + const json = Buffer.from(encoded, "base64url").toString("utf8"); + const offer = JSON.parse(json) as { sessionId?: string }; + if (!offer.sessionId) throw new Error("offer.sessionId missing"); + return { sessionId: offer.sessionId }; +} + +describe("Relay transport (plaintext) - daemon E2E", () => { + test( + "daemon connects to relay and client ping/pong works through relay", + async () => { + process.env.PASEO_PRIMARY_LAN_IP = "192.168.1.12"; + + const { logger, lines } = createCapturingLogger(); + const daemon = await createTestPaseoDaemon({ + listen: "127.0.0.1", + logger, + relayEnabled: true, + relayEndpoint: "relay.paseo.sh:443", + }); + + try { + const offerUrl = parseOfferUrlFromLogs(lines); + const { sessionId } = decodeOfferFromFragmentUrl(offerUrl); + + const ws = new WebSocket( + `wss://relay.paseo.sh/ws?session=${encodeURIComponent( + sessionId + )}&role=client` + ); + + const received = await new Promise((resolve, reject) => { + const timeout = setTimeout(() => { + ws.close(); + reject(new Error("timed out waiting for pong")); + }, 20000); + + ws.on("open", () => { + ws.send(JSON.stringify({ type: "ping" })); + }); + + ws.on("message", (data) => { + clearTimeout(timeout); + try { + resolve(JSON.parse(data.toString())); + } catch (err) { + reject(err); + } finally { + ws.close(); + } + }); + + ws.on("error", (err) => { + clearTimeout(timeout); + reject(err); + }); + }); + + expect(received).toEqual({ type: "pong" }); + } finally { + await daemon.close(); + } + }, + 60000 + ); +}); + diff --git a/packages/server/src/server/relay-transport.ts b/packages/server/src/server/relay-transport.ts new file mode 100644 index 000000000..eb3c684c7 --- /dev/null +++ b/packages/server/src/server/relay-transport.ts @@ -0,0 +1,135 @@ +import WebSocket from "ws"; +import type pino from "pino"; + +type RelayTransportOptions = { + logger: pino.Logger; + attachSocket: (ws: WebSocket) => Promise; + relayEndpoint: string; // "host:port" + sessionId: string; +}; + +export type RelayTransportController = { + stop: () => Promise; +}; + +export function startRelayTransport({ + logger, + attachSocket, + relayEndpoint, + sessionId, +}: RelayTransportOptions): RelayTransportController { + const relayLogger = logger.child({ module: "relay-transport" }); + + let stopped = false; + let ws: WebSocket | null = null; + let reconnectTimeout: ReturnType | null = null; + let reconnectAttempt = 0; + + const stop = async (): Promise => { + stopped = true; + if (reconnectTimeout) { + clearTimeout(reconnectTimeout); + reconnectTimeout = null; + } + if (ws) { + try { + ws.close(); + } catch { + // ignore + } + ws = null; + } + }; + + const connect = (): void => { + if (stopped) return; + + const url = buildRelayWebSocketUrl(relayEndpoint, sessionId, "server"); + const socket = new WebSocket(url); + ws = socket; + + let attached = false; + + socket.on("open", () => { + reconnectAttempt = 0; + relayLogger.info({ url }, "relay_connected"); + + if (attached) return; + attached = true; + void attachSocket(socket); + }); + + socket.on("close", (code, reason) => { + relayLogger.warn( + { code, reason: reason?.toString?.(), url }, + "relay_disconnected" + ); + scheduleReconnect(); + }); + + socket.on("error", (err) => { + relayLogger.warn({ err, url }, "relay_error"); + // close event will schedule reconnect + }); + }; + + const scheduleReconnect = (): void => { + if (stopped) return; + if (reconnectTimeout) return; + + reconnectAttempt += 1; + const delayMs = Math.min(30000, 1000 * reconnectAttempt); + reconnectTimeout = setTimeout(() => { + reconnectTimeout = null; + connect(); + }, delayMs); + }; + + connect(); + + return { stop }; +} + +function buildRelayWebSocketUrl( + relayEndpoint: string, + sessionId: string, + role: "server" | "client" +): string { + const { host, port } = parseHostPort(relayEndpoint); + const protocol = port === 443 ? "wss" : "ws"; + return `${protocol}://${host}:${port}/ws?session=${encodeURIComponent( + sessionId + )}&role=${role}`; +} + +function parseHostPort(input: string): { host: string; port: number } { + const trimmed = input.trim(); + + if (trimmed.startsWith("[")) { + const endIdx = trimmed.indexOf("]"); + if (endIdx === -1) { + throw new Error(`Invalid relay endpoint: ${input}`); + } + const host = trimmed.slice(1, endIdx); + const rest = trimmed.slice(endIdx + 1); + if (!rest.startsWith(":")) { + throw new Error(`Invalid relay endpoint: ${input}`); + } + const port = Number.parseInt(rest.slice(1), 10); + if (!Number.isFinite(port)) { + throw new Error(`Invalid relay port: ${input}`); + } + return { host: `[${host}]`, port }; + } + + const idx = trimmed.lastIndexOf(":"); + if (idx === -1) { + throw new Error(`Invalid relay endpoint (expected host:port): ${input}`); + } + const host = trimmed.slice(0, idx); + const port = Number.parseInt(trimmed.slice(idx + 1), 10); + if (!host || !Number.isFinite(port)) { + throw new Error(`Invalid relay endpoint: ${input}`); + } + return { host, port }; +} diff --git a/packages/server/src/server/test-utils/paseo-daemon.ts b/packages/server/src/server/test-utils/paseo-daemon.ts index c2a20a654..ea69f1384 100644 --- a/packages/server/src/server/test-utils/paseo-daemon.ts +++ b/packages/server/src/server/test-utils/paseo-daemon.ts @@ -11,6 +11,8 @@ type TestPaseoDaemonOptions = { corsAllowedOrigins?: string[]; listen?: string; logger?: Parameters[1]; + relayEnabled?: boolean; + relayEndpoint?: string; }; export type TestPaseoDaemon = { @@ -63,8 +65,8 @@ export async function createTestPaseoDaemon( agentControlMcp: { url: `http://127.0.0.1:${port}/mcp/agents`, }, - relayEnabled: true, - relayEndpoint: "relay.paseo.sh:443", + relayEnabled: options.relayEnabled ?? false, + relayEndpoint: options.relayEndpoint ?? "relay.paseo.sh:443", appBaseUrl: "https://app.paseo.sh", openai: openaiApiKey ? { apiKey: openaiApiKey } : undefined, downloadTokenTtlMs: options.downloadTokenTtlMs, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index ff5634bfb..12c0e2326 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -75,9 +75,17 @@ export class VoiceAssistantWebSocketServer { this.bridge.broadcast(message); } + public async attachExternalSocket(ws: Parameters[0]): Promise { + const fakeRequest = { + url: "/ws", + headers: {}, + method: "GET", + } as unknown as Parameters[1]; + await this.bridge.attach(ws, fakeRequest); + } + public async close(): Promise { await this.bridge.closeAll(); this.wss.close(); } } -