feat: relay transport (plaintext)

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 {

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<PaseoDaemon> {
   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<void>((resolve) => {
       httpServer.close(() => resolve());

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 {

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<unknown>((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
+  );
+});
+

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<void>;
+  relayEndpoint: string; // "host:port"
+  sessionId: string;
+};
+
+export type RelayTransportController = {
+  stop: () => Promise<void>;
+};
+
+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<typeof setTimeout> | null = null;
+  let reconnectAttempt = 0;
+
+  const stop = async (): Promise<void> => {
+    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 };
+}

packages/server/src/server/test-utils/paseo-daemon.ts

@@ -11,6 +11,8 @@ type TestPaseoDaemonOptions = {
   corsAllowedOrigins?: string[];
   listen?: string;
   logger?: Parameters<typeof createPaseoDaemon>[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,

packages/server/src/server/websocket-server.ts

@@ -75,9 +75,17 @@ export class VoiceAssistantWebSocketServer {
     this.bridge.broadcast(message);
   }
 
+  public async attachExternalSocket(ws: Parameters<WebSocketSessionBridge["attach"]>[0]): Promise<void> {
+    const fakeRequest = {
+      url: "/ws",
+      headers: {},
+      method: "GET",
+    } as unknown as Parameters<WebSocketSessionBridge["attach"]>[1];
+    await this.bridge.attach(ws, fakeRequest);
+  }
+
   public async close(): Promise<void> {
     await this.bridge.closeAll();
     this.wss.close();
   }
 }
-