From 574bdc16f6e248fb40b0bddfd8cf077dffd18b54 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Thu, 5 Feb 2026 10:45:51 +0700 Subject: [PATCH] simplify: remove ws bridge; wrangler-based relay tests --- .gitignore | 2 + packages/app/e2e/global-setup.ts | 37 +- .../app/e2e/pairing-offer-parsing.spec.ts | 2 +- packages/relay/package.json | 1 - packages/relay/src/cloudflare-adapter.ts | 14 +- packages/relay/src/e2e.test.ts | 140 +++- packages/relay/src/index.ts | 4 - packages/relay/src/node-adapter.ts | 387 ----------- packages/relay/src/relay.ts | 152 ----- packages/relay/src/types.ts | 24 +- packages/server/src/client/daemon-client.ts | 2 +- .../daemon-e2e/relay-transport.e2e.test.ts | 76 ++- packages/server/src/server/messages.ts | 2 +- .../server/src/server/websocket-server.ts | 587 ++++++++++++++++- .../src/server/websocket-session-bridge.ts | 602 ------------------ scripts/relay-local.ts | 29 - 16 files changed, 785 insertions(+), 1276 deletions(-) delete mode 100644 packages/relay/src/node-adapter.ts delete mode 100644 packages/relay/src/relay.ts delete mode 100644 packages/server/src/server/websocket-session-bridge.ts delete mode 100644 scripts/relay-local.ts diff --git a/.gitignore b/.gitignore index 202f121ce..f789b9be2 100644 --- a/.gitignore +++ b/.gitignore @@ -57,3 +57,5 @@ CLAUDE.local.md .debug.conversations/ .debug/ .paseo/ +.wrangler/ +**/.wrangler/ diff --git a/packages/app/e2e/global-setup.ts b/packages/app/e2e/global-setup.ts index b4232bf9f..54f833b0d 100644 --- a/packages/app/e2e/global-setup.ts +++ b/packages/app/e2e/global-setup.ts @@ -4,7 +4,6 @@ import { tmpdir } from 'node:os'; import path from 'node:path'; import net from 'node:net'; import { Buffer } from 'node:buffer'; -import { createRelayServer, type RelayServer } from '@paseo/relay/node'; async function getAvailablePort(): Promise { return new Promise((resolve, reject) => { @@ -43,7 +42,7 @@ async function waitForServer(port: number, timeout = 15000): Promise { let daemonProcess: ChildProcess | null = null; let metroProcess: ChildProcess | null = null; let paseoHome: string | null = null; -let relayServer: RelayServer | null = null; +let relayProcess: ChildProcess | null = null; type OfferPayload = { v: 2; @@ -78,8 +77,32 @@ export default async function globalSetup() { const metroPort = await getAvailablePort(); paseoHome = await mkdtemp(path.join(tmpdir(), 'paseo-e2e-home-')); - relayServer = createRelayServer({ port: relayPort, host: '127.0.0.1' }); - await relayServer.start(); + const relayDir = path.resolve(__dirname, '..', '..', 'relay'); + relayProcess = spawn( + 'npx', + ['wrangler', 'dev', '--local', '--ip', '127.0.0.1', '--port', String(relayPort)], + { + cwd: relayDir, + env: { ...process.env }, + stdio: ['ignore', 'pipe', 'pipe'], + detached: false, + } + ); + + relayProcess.stdout?.on('data', (data: Buffer) => { + const lines = data.toString().split('\n').filter((l) => l.trim()); + for (const line of lines) { + console.log(`[relay] ${line}`); + } + }); + relayProcess.stderr?.on('data', (data: Buffer) => { + const lines = data.toString().split('\n').filter((l) => l.trim()); + for (const line of lines) { + console.error(`[relay] ${line}`); + } + }); + + await waitForServer(relayPort, 30000); // Start Metro bundler on dynamic port const appDir = path.resolve(__dirname, '..'); @@ -198,9 +221,9 @@ export default async function globalSetup() { metroProcess.kill('SIGTERM'); metroProcess = null; } - if (relayServer) { - await relayServer.stop(); - relayServer = null; + if (relayProcess) { + relayProcess.kill('SIGTERM'); + relayProcess = null; } if (paseoHome) { await rm(paseoHome, { recursive: true, force: true }); diff --git a/packages/app/e2e/pairing-offer-parsing.spec.ts b/packages/app/e2e/pairing-offer-parsing.spec.ts index 77f4a495e..6ecc5e8ec 100644 --- a/packages/app/e2e/pairing-offer-parsing.spec.ts +++ b/packages/app/e2e/pairing-offer-parsing.spec.ts @@ -29,7 +29,7 @@ test('pairing flow accepts #offer=ConnectionOfferV2 and stores relay-only host', const offerUrl = `https://app.paseo.sh/#offer=${encodeBase64Url(JSON.stringify(offer))}`; - await page.getByText('+ Add host', { exact: true }).click(); + await page.getByText('+ Add connection', { exact: true }).click(); await page.getByText('Paste pairing link', { exact: true }).click(); const input = page.getByPlaceholder('https://app.paseo.sh/#offer=...'); diff --git a/packages/relay/package.json b/packages/relay/package.json index 8827e7c2d..67c491451 100644 --- a/packages/relay/package.json +++ b/packages/relay/package.json @@ -6,7 +6,6 @@ "exports": { ".": "./src/index.ts", "./e2ee": "./src/e2ee.ts", - "./node": "./src/node-adapter.ts", "./cloudflare": "./src/cloudflare-adapter.ts" }, "scripts": { diff --git a/packages/relay/src/cloudflare-adapter.ts b/packages/relay/src/cloudflare-adapter.ts index 21b36ac7b..73fa82fdb 100644 --- a/packages/relay/src/cloudflare-adapter.ts +++ b/packages/relay/src/cloudflare-adapter.ts @@ -54,11 +54,13 @@ interface DurableObjectStub { /** * Durable Object that handles WebSocket relay for a single session. * - * Two WebSockets connect to this DO: - * - role=server: The Paseo daemon - * - role=client: The mobile/web app + * WebSockets connect to this DO in three shapes: + * - role=server (no clientId): daemon control socket (one per serverId) + * - role=server&clientId=...: daemon per-client data socket (one per clientId) + * - role=client&clientId=...: app/client socket (one per clientId) * - * Messages are forwarded between them. The DO hibernates when idle. + * Messages are forwarded between the per-client data sockets and their matching + * client sockets. The DO hibernates when idle. */ interface CFResponseInit extends ResponseInit { webSocket?: WebSocket; @@ -139,8 +141,8 @@ export class RelayDurableObject { return new Response("Missing serverId parameter", { status: 400 }); } - // v2 relay protocol: clients must provide a clientId so the daemon can create - // an independent E2EE channel per client connection. + // Clients must provide a clientId so the daemon can create an independent + // E2EE channel per client connection. if (role === "client" && !clientId) { return new Response("Missing clientId parameter", { status: 400 }); } diff --git a/packages/relay/src/e2e.test.ts b/packages/relay/src/e2e.test.ts index 601468929..1c8b932a5 100644 --- a/packages/relay/src/e2e.test.ts +++ b/packages/relay/src/e2e.test.ts @@ -1,6 +1,8 @@ import { describe, it, expect, beforeAll, afterAll } from "vitest"; import { WebSocket } from "ws"; -import { createRelayServer, type RelayServer } from "./node-adapter.js"; +import net from "node:net"; +import { spawn, type ChildProcess } from "node:child_process"; +import { Buffer } from "node:buffer"; import { generateKeyPair, exportPublicKey, @@ -10,21 +12,83 @@ import { decrypt, } from "./crypto.js"; -const TEST_PORT = 19999; +async function getAvailablePort(): Promise { + return new Promise((resolve, reject) => { + const server = net.createServer(); + server.once("error", reject); + server.listen(0, "127.0.0.1", () => { + const address = server.address(); + if (!address || typeof address === "string") { + server.close(() => reject(new Error("Failed to acquire port"))); + return; + } + server.close(() => resolve(address.port)); + }); + }); +} + +async function waitForServer(port: number, timeout = 15000): Promise { + const start = Date.now(); + while (Date.now() - start < timeout) { + try { + await new Promise((resolve, reject) => { + const socket = net.connect(port, "127.0.0.1", () => { + socket.end(); + resolve(); + }); + socket.on("error", reject); + }); + return; + } catch { + await new Promise((r) => setTimeout(r, 100)); + } + } + throw new Error(`Server did not start on port ${port} within ${timeout}ms`); +} describe("E2E Relay with E2EE", () => { - let relay: RelayServer; + let relayPort: number; + let relayProcess: ChildProcess | null = null; beforeAll(async () => { - relay = createRelayServer({ port: TEST_PORT, host: "127.0.0.1" }); - await relay.start(); + relayPort = await getAvailablePort(); + relayProcess = spawn( + "npx", + ["wrangler", "dev", "--local", "--ip", "127.0.0.1", "--port", String(relayPort)], + { + cwd: process.cwd(), + env: { ...process.env }, + stdio: ["ignore", "pipe", "pipe"], + detached: false, + } + ); + + relayProcess.stdout?.on("data", (data: Buffer) => { + const lines = data.toString().split("\n").filter((l) => l.trim()); + for (const line of lines) { + // eslint-disable-next-line no-console + console.log(`[relay] ${line}`); + } + }); + relayProcess.stderr?.on("data", (data: Buffer) => { + const lines = data.toString().split("\n").filter((l) => l.trim()); + for (const line of lines) { + // eslint-disable-next-line no-console + console.error(`[relay] ${line}`); + } + }); + + await waitForServer(relayPort, 30000); }); afterAll(async () => { - await relay.stop(); + if (relayProcess) { + relayProcess.kill("SIGTERM"); + relayProcess = null; + } }); - it("full flow: daemon and client exchange encrypted messages through relay", async () => { + it("full flow: daemon and client exchange encrypted messages through relay", { timeout: 20_000 }, async () => { const serverId = "test-session-" + Date.now(); const clientId = "clt_test_" + Date.now() + "_" + Math.random().toString(36).slice(2); @@ -37,7 +101,7 @@ describe("E2E Relay with E2EE", () => { // Daemon connects to relay as "server" control role const daemonControlWs = new WebSocket( - `ws://127.0.0.1:${TEST_PORT}/ws?serverId=${serverId}&role=server` + `ws://127.0.0.1:${relayPort}/ws?serverId=${serverId}&role=server` ); await new Promise((resolve, reject) => { @@ -58,9 +122,41 @@ describe("E2E Relay with E2EE", () => { daemonPubKeyOnClient ); - // Client connects to relay as "client" role (must include clientId in v2) + const waitForClientSeen = new Promise((resolve, reject) => { + const timeout = setTimeout( + () => reject(new Error("timed out waiting for client_connected")), + 5000 + ); + const onMessage = (raw: unknown) => { + try { + const text = + typeof raw === "string" + ? raw + : raw && typeof (raw as any).toString === "function" + ? (raw as any).toString() + : ""; + const msg = JSON.parse(text); + if (msg?.type === "client_connected" && msg.clientId === clientId) { + clearTimeout(timeout); + daemonControlWs.off("message", onMessage); + resolve(); + return; + } + if (msg?.type === "sync" && Array.isArray(msg.clientIds) && msg.clientIds.includes(clientId)) { + clearTimeout(timeout); + daemonControlWs.off("message", onMessage); + resolve(); + } + } catch { + // ignore + } + }; + daemonControlWs.on("message", onMessage); + }); + + // Client connects to relay as "client" role (must include clientId) const clientWs = new WebSocket( - `ws://127.0.0.1:${TEST_PORT}/ws?serverId=${serverId}&role=client&clientId=${clientId}` + `ws://127.0.0.1:${relayPort}/ws?serverId=${serverId}&role=client&clientId=${clientId}` ); await new Promise((resolve, reject) => { @@ -68,24 +164,10 @@ describe("E2E Relay with E2EE", () => { clientWs.on("error", reject); }); - // Wait for relay to notify daemon control socket, then open the per-client server-data socket. - await new Promise((resolve, reject) => { - const timeout = setTimeout(() => reject(new Error("timed out waiting for client_connected")), 5000); - daemonControlWs.on("message", (raw) => { - try { - const msg = JSON.parse(raw.toString()); - if (msg && msg.type === "client_connected" && msg.clientId === clientId) { - clearTimeout(timeout); - resolve(); - } - } catch { - // ignore - } - }); - }); + await waitForClientSeen; const daemonWs = new WebSocket( - `ws://127.0.0.1:${TEST_PORT}/ws?serverId=${serverId}&role=server&clientId=${clientId}` + `ws://127.0.0.1:${relayPort}/ws?serverId=${serverId}&role=server&clientId=${clientId}` ); await new Promise((resolve, reject) => { daemonWs.on("open", resolve); @@ -197,7 +279,7 @@ describe("E2E Relay with E2EE", () => { ); const daemonControlWs = new WebSocket( - `ws://127.0.0.1:${TEST_PORT}/ws?serverId=${serverId}&role=server` + `ws://127.0.0.1:${relayPort}/ws?serverId=${serverId}&role=server` ); await new Promise((r) => daemonControlWs.on("open", r)); @@ -234,13 +316,13 @@ describe("E2E Relay with E2EE", () => { }); const clientWs = new WebSocket( - `ws://127.0.0.1:${TEST_PORT}/ws?serverId=${serverId}&role=client&clientId=${clientId}` + `ws://127.0.0.1:${relayPort}/ws?serverId=${serverId}&role=client&clientId=${clientId}` ); await new Promise((r) => clientWs.on("open", r)); await waitForClientSeen; const daemonWs = new WebSocket( - `ws://127.0.0.1:${TEST_PORT}/ws?serverId=${serverId}&role=server&clientId=${clientId}` + `ws://127.0.0.1:${relayPort}/ws?serverId=${serverId}&role=server&clientId=${clientId}` ); await new Promise((r) => daemonWs.on("open", r)); diff --git a/packages/relay/src/index.ts b/packages/relay/src/index.ts index 8ae9211d7..c17bb678f 100644 --- a/packages/relay/src/index.ts +++ b/packages/relay/src/index.ts @@ -1,9 +1,5 @@ -export { Relay } from "./relay.js"; export type { ConnectionRole, - RelayConnection, - RelayEvents, - RelaySession, RelaySessionAttachment, } from "./types.js"; diff --git a/packages/relay/src/node-adapter.ts b/packages/relay/src/node-adapter.ts deleted file mode 100644 index cf97cd922..000000000 --- a/packages/relay/src/node-adapter.ts +++ /dev/null @@ -1,387 +0,0 @@ -import http from "http"; -import type { Socket } from "net"; -import { WebSocketServer, WebSocket as NodeWebSocket } from "ws"; -import type { ConnectionRole } from "./types.js"; - -export interface NodeRelayServerConfig { - port: number; - host?: string; -} - -/** - * Standalone Node.js relay server for self-hosting. - * - * v2 protocol: - * - Daemon connects a single control socket: - * ws://host:port/ws?serverId=abc&role=server - * - Each app client connects with a clientId: - * ws://host:port/ws?serverId=abc&role=client&clientId=clt_... - * - For every connected clientId, the daemon opens a dedicated server-data socket: - * ws://host:port/ws?serverId=abc&role=server&clientId=clt_... - * - * This allows multiple independent clients, each with its own E2EE handshake. - */ -export interface RelayServer { - start(): Promise; - stop(): Promise; -} - -type Session = { - serverId: string; - control: NodeWebSocket | null; - clients: Map>; // clientId -> sockets - servers: Map; // clientId -> server-data socket - pending: Map>; // clientId -> buffered frames -}; - -function bufferFromWsData(data: unknown): Buffer { - if (Buffer.isBuffer(data)) return data; - - if (Array.isArray(data)) { - return Buffer.concat(data.map(bufferFromWsData)); - } - - if (data instanceof ArrayBuffer) { - return Buffer.from(data); - } - - if (ArrayBuffer.isView(data)) { - return Buffer.from(data.buffer, data.byteOffset, data.byteLength); - } - - if (typeof data === "string") { - return Buffer.from(data, "utf8"); - } - - return Buffer.from(String(data), "utf8"); -} - -function normalizeWsMessage(data: unknown, isBinary: boolean): string | ArrayBuffer { - if (!isBinary) { - if (typeof data === "string") return data; - return bufferFromWsData(data).toString("utf8"); - } - - if (data instanceof ArrayBuffer) return data; - const buffer = bufferFromWsData(data); - const out = new Uint8Array(buffer.byteLength); - out.set(buffer); - return out.buffer; -} - -export function createRelayServer(config: NodeRelayServerConfig): RelayServer { - const { port, host = "0.0.0.0" } = config; - const logFrames = process.env.PASEO_RELAY_LOG_FRAMES === "1"; - const logUpgrades = process.env.PASEO_RELAY_LOG_UPGRADES === "1"; - - const sessions = new Map(); - - const getSession = (serverId: string): Session => { - let session = sessions.get(serverId); - if (!session) { - session = { - serverId, - control: null, - clients: new Map(), - servers: new Map(), - pending: new Map(), - }; - sessions.set(serverId, session); - console.log(`[Relay] Session created: ${serverId}`); - } - return session; - }; - - const notifyControl = (session: Session, msg: unknown) => { - if (!session.control || session.control.readyState !== NodeWebSocket.OPEN) return; - try { - session.control.send(JSON.stringify(msg)); - } catch { - // ignore - } - }; - - const listClientIds = (session: Session): string[] => Array.from(session.clients.keys()); - - const bufferFrame = (session: Session, clientId: string, frame: string | ArrayBuffer) => { - const existing = session.pending.get(clientId) ?? []; - existing.push(frame); - if (existing.length > 200) existing.splice(0, existing.length - 200); - session.pending.set(clientId, existing); - }; - - const flushFrames = (session: Session, clientId: string) => { - const server = session.servers.get(clientId); - if (!server || server.readyState !== NodeWebSocket.OPEN) return; - const frames = session.pending.get(clientId); - if (!frames || frames.length === 0) return; - session.pending.delete(clientId); - for (const frame of frames) { - try { - server.send(frame); - } catch { - bufferFrame(session, clientId, frame); - break; - } - } - }; - - const httpServer = http.createServer((req, res) => { - if (req.url === "/health") { - res.writeHead(200, { "Content-Type": "application/json" }); - res.end(JSON.stringify({ status: "ok" })); - return; - } - res.writeHead(404); - res.end("Not found"); - }); - - const wss = new WebSocketServer({ noServer: true }); - - httpServer.on("upgrade", (req, socket: Socket, head) => { - if (logUpgrades) { - try { - console.log( - `[Relay] upgrade url=${JSON.stringify(req.url)} host=${JSON.stringify(req.headers.host)}` - ); - } catch { - // ignore - } - } - try { - const url = new URL(req.url ?? "", `http://${req.headers.host}`); - if (url.pathname !== "/ws") { - socket.write("HTTP/1.1 404 Not Found\r\n\r\n"); - socket.destroy(); - return; - } - - const serverId = url.searchParams.get("serverId"); - const role = url.searchParams.get("role"); - const clientId = (url.searchParams.get("clientId") ?? "").trim(); - if (!serverId || !role || (role !== "server" && role !== "client")) { - socket.write("HTTP/1.1 400 Bad Request\r\n\r\n"); - socket.destroy(); - return; - } - if (role === "client" && !clientId) { - socket.write("HTTP/1.1 400 Bad Request\r\n\r\n"); - socket.destroy(); - return; - } - - wss.handleUpgrade(req, socket, head, (ws) => { - wss.emit("connection", ws, req); - }); - } catch { - socket.write("HTTP/1.1 400 Bad Request\r\n\r\n"); - socket.destroy(); - } - }); - - wss.on("connection", (ws, req) => { - const url = new URL(req.url ?? "", `http://${req.headers.host}`); - const serverId = url.searchParams.get("serverId")!; - const role = url.searchParams.get("role") as ConnectionRole; - const clientId = (url.searchParams.get("clientId") ?? "").trim(); - const session = getSession(serverId); - - const isControl = role === "server" && !clientId; - const isServerData = role === "server" && !!clientId; - - if (isControl) { - try { - session.control?.close(1008, "Replaced by new connection"); - } catch { - // ignore - } - session.control = ws; - try { - ws.send(JSON.stringify({ type: "sync", clientIds: listClientIds(session) })); - } catch { - // ignore - } - } - - if (role === "client") { - const set = session.clients.get(clientId) ?? new Set(); - set.add(ws); - session.clients.set(clientId, set); - notifyControl(session, { type: "client_connected", clientId }); - } - - if (isServerData) { - const prev = session.servers.get(clientId); - if (prev && prev !== ws) { - try { - prev.close(1008, "Replaced by new connection"); - } catch { - // ignore - } - } - session.servers.set(clientId, ws); - flushFrames(session, clientId); - } - - ws.on("message", (data, isBinary) => { - const normalized = normalizeWsMessage(data, isBinary); - - if (logFrames) { - const preview = (() => { - try { - if (isBinary) return ""; - return bufferFromWsData(data).toString("utf8").slice(0, 200); - } catch { - return ""; - } - })(); - const len = (() => { - try { - return bufferFromWsData(data).byteLength; - } catch { - return -1; - } - })(); - console.log( - `[Relay] frame ${serverId}/${role}${clientId ? `(${clientId})` : ""} binary=${isBinary} len=${len} preview=${JSON.stringify(preview)}` - ); - } - - if (role === "client") { - const server = session.servers.get(clientId); - if (!server || server.readyState !== NodeWebSocket.OPEN) { - bufferFrame(session, clientId, normalized); - return; - } - try { - server.send(normalized); - } catch { - // ignore - } - return; - } - - if (isServerData) { - const targets = session.clients.get(clientId); - if (!targets) return; - for (const target of targets) { - if (target.readyState !== NodeWebSocket.OPEN) continue; - try { - target.send(normalized); - } catch { - // ignore - } - } - } - }); - - ws.on("close", (code, reasonBuf) => { - const reason = reasonBuf?.toString?.() ?? ""; - - if (isControl) { - if (session.control === ws) session.control = null; - } else if (role === "client") { - const set = session.clients.get(clientId); - if (set) { - set.delete(ws); - if (set.size === 0) { - session.clients.delete(clientId); - session.pending.delete(clientId); - const server = session.servers.get(clientId); - if (server) { - try { - server.close(1001, "Client disconnected"); - } catch { - // ignore - } - session.servers.delete(clientId); - } - notifyControl(session, { type: "client_disconnected", clientId }); - } - } - } else if (isServerData) { - if (session.servers.get(clientId) === ws) { - session.servers.delete(clientId); - } - const targets = session.clients.get(clientId); - if (targets) { - for (const target of targets) { - try { - target.close(1012, "Server disconnected"); - } catch { - // ignore - } - } - } - } - - if (logUpgrades) { - console.log( - `[Relay] close ${serverId}/${role}${clientId ? `(${clientId})` : ""} code=${code} reason=${JSON.stringify(reason)}` - ); - } - - if (session.control === null && session.clients.size === 0 && session.servers.size === 0) { - sessions.delete(serverId); - console.log(`[Relay] Session closed: ${serverId}`); - } - }); - - ws.on("error", (error) => { - console.error(`[Relay] WebSocket error for ${serverId}/${role}${clientId ? `(${clientId})` : ""}:`, error); - }); - }); - - return { - start() { - return new Promise((resolve, reject) => { - httpServer.on("error", reject); - httpServer.listen(port, host, () => { - console.log(`[Relay] Listening on ${host}:${port}`); - resolve(); - }); - }); - }, - - stop() { - return new Promise((resolve) => { - try { - httpServer.close(() => undefined); - } catch { - // ignore - } - - for (const ws of wss.clients) { - try { - (ws as unknown as { terminate?: () => void }).terminate?.(); - } catch { - try { - ws.close(1001, "Server shutting down"); - } catch { - // ignore - } - } - } - - let finished = false; - const finish = () => { - if (finished) return; - finished = true; - resolve(); - }; - - const timeout = setTimeout(() => finish(), 1500); - try { - wss.close(() => { - clearTimeout(timeout); - finish(); - }); - } catch { - clearTimeout(timeout); - finish(); - } - }); - }, - }; -} - diff --git a/packages/relay/src/relay.ts b/packages/relay/src/relay.ts deleted file mode 100644 index 8c1f529ab..000000000 --- a/packages/relay/src/relay.ts +++ /dev/null @@ -1,152 +0,0 @@ -import type { - ConnectionRole, - RelayConnection, - RelayEvents, - RelaySession, -} from "./types.js"; - -/** - * Core relay logic for bridging server and client WebSocket connections. - * - * This class is platform-agnostic and works with any WebSocket implementation - * that conforms to the RelayConnection interface. - */ -export class Relay { - private sessions = new Map(); - private events: RelayEvents; - - constructor(events: RelayEvents = {}) { - this.events = events; - } - - /** - * Register a connection for a session. - * If both server and client are connected, messages are bridged. - */ - addConnection( - serverId: string, - role: ConnectionRole, - connection: RelayConnection - ): void { - let session = this.sessions.get(serverId); - - if (!session) { - session = { - serverId, - server: null, - client: null, - createdAt: Date.now(), - }; - this.sessions.set(serverId, session); - this.events.onSessionCreated?.(serverId); - } - - const previousServer = session.server; - const previousClient = session.client; - - const existingConnection = session[role]; - if (existingConnection) { - existingConnection.close(1008, "Replaced by new connection"); - } - - session[role] = connection; - - // Important: the E2EE handshake is tied to a specific server↔client socket pair. - // If the daemon reconnects (server socket changes) while the client socket stays - // open, the client will continue sending encrypted frames using its existing - // channel state — but the new daemon socket hasn't handshaked yet. This leads - // to immediate decrypt/handshake failures. - // - // To keep the protocol simple and robust, whenever a new *server* connection - // arrives while a client socket is still connected, force the client to reconnect. - if (role === "server" && previousClient) { - // Only close if the server socket actually changed (or was previously missing). - const serverChanged = !previousServer || previousServer !== connection; - if (serverChanged) { - session.client = null; - previousClient.close(1012, "Server reconnected"); - } - } - - if (session.server && session.client) { - this.events.onSessionBridged?.(serverId); - } - } - - /** - * Remove a connection from a session. - * If both connections are gone, the session is cleaned up. - */ - removeConnection(serverId: string, role: ConnectionRole): void { - const session = this.sessions.get(serverId); - if (!session) return; - - session[role] = null; - - if (!session.server && !session.client) { - this.sessions.delete(serverId); - this.events.onSessionClosed?.(serverId); - } - } - - /** - * Forward a message from one side to the other. - */ - forward( - serverId: string, - fromRole: ConnectionRole, - data: string | ArrayBuffer - ): void { - const session = this.sessions.get(serverId); - if (!session) return; - - const target = fromRole === "server" ? session.client : session.server; - if (target) { - try { - target.send(data); - } catch (error) { - this.events.onError?.( - serverId, - error instanceof Error ? error : new Error(String(error)) - ); - } - } - } - - /** - * Get session info for debugging/monitoring. - */ - getSession(serverId: string): RelaySession | undefined { - return this.sessions.get(serverId); - } - - /** - * List all active sessions. - */ - listSessions(): RelaySession[] { - return Array.from(this.sessions.values()); - } - - /** - * Close a session and both connections. - */ - closeSession(serverId: string, code = 1000, reason = "Session closed"): void { - const session = this.sessions.get(serverId); - if (!session) return; - - session.server?.close(code, reason); - session.client?.close(code, reason); - - this.sessions.delete(serverId); - this.events.onSessionClosed?.(serverId); - } - - /** - * Restore sessions from persisted state (for Durable Objects hibernation). - */ - restoreSessions(sessions: RelaySession[]): void { - for (const session of sessions) { - this.sessions.set(session.serverId, session); - } - } -} diff --git a/packages/relay/src/types.ts b/packages/relay/src/types.ts index aa59ea88d..3ddd00f21 100644 --- a/packages/relay/src/types.ts +++ b/packages/relay/src/types.ts @@ -10,33 +10,13 @@ export type ConnectionRole = "server" | "client"; -export interface RelaySession { - serverId: string; - server: RelayConnection | null; - client: RelayConnection | null; - createdAt: number; -} - -export interface RelayConnection { - role: ConnectionRole; - send(data: string | ArrayBuffer): void; - close(code?: number, reason?: string): void; -} - export interface RelaySessionAttachment { serverId: string; role: ConnectionRole; /** - * v2 relay: unique id for the client connection. Allows the daemon to create - * an independent socket + E2EE channel per connected client. + * Unique id for the client connection. Allows the daemon to create an + * independent socket + E2EE channel per connected client. */ clientId?: string | null; createdAt: number; } - -export interface RelayEvents { - onSessionCreated?(serverId: string): void; - onSessionBridged?(serverId: string): void; - onSessionClosed?(serverId: string): void; - onError?(serverId: string, error: Error): void; -} diff --git a/packages/server/src/client/daemon-client.ts b/packages/server/src/client/daemon-client.ts index 7610892ca..dc1c9a77f 100644 --- a/packages/server/src/client/daemon-client.ts +++ b/packages/server/src/client/daemon-client.ts @@ -298,7 +298,7 @@ export class DaemonClient { constructor(private config: DaemonClientConfig) { this.logger = config.logger ?? consoleLogger; - // Relay v2 requires a clientId so the daemon can create an independent + // Relay requires a clientId so the daemon can create an independent // socket + E2EE channel per connected client. Generate one per DaemonClient // instance (stable across reconnects in this tab/app session). if (isRelayClientWebSocketUrl(this.config.url)) { 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 index 0e5e79409..d8c0749bd 100644 --- a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -3,10 +3,12 @@ import WebSocket from "ws"; import pino from "pino"; import { Writable } from "node:stream"; import net from "node:net"; +import path from "node:path"; +import { spawn, type ChildProcess } from "node:child_process"; +import { Buffer } from "node:buffer"; import { createTestPaseoDaemon } from "../test-utils/paseo-daemon.js"; import { createClientChannel, type Transport } from "@paseo/relay/e2ee"; -import { createRelayServer } from "@paseo/relay/node"; import { buildRelayWebSocketUrl } from "../../shared/daemon-endpoints.js"; function createCapturingLogger() { @@ -69,16 +71,74 @@ async function getAvailablePort(): Promise { }); } +async function waitForServer(port: number, timeout = 15000): Promise { + const start = Date.now(); + while (Date.now() - start < timeout) { + try { + await new Promise((resolve, reject) => { + const socket = net.connect(port, "127.0.0.1", () => { + socket.end(); + resolve(); + }); + socket.on("error", reject); + }); + return; + } catch { + await new Promise((r) => setTimeout(r, 100)); + } + } + throw new Error(`Server did not start on port ${port} within ${timeout}ms`); +} + describe("Relay transport (E2EE) - daemon E2E", () => { + let relayPort: number; + let relayProcess: ChildProcess | null = null; + + const startRelay = async () => { + relayPort = await getAvailablePort(); + const relayDir = path.resolve(process.cwd(), "../relay"); + relayProcess = spawn( + "npx", + ["wrangler", "dev", "--local", "--ip", "127.0.0.1", "--port", String(relayPort)], + { + cwd: relayDir, + env: { ...process.env }, + stdio: ["ignore", "pipe", "pipe"], + detached: false, + } + ); + + relayProcess.stdout?.on("data", (data: Buffer) => { + const lines = data.toString().split("\n").filter((l) => l.trim()); + for (const line of lines) { + // eslint-disable-next-line no-console + console.log(`[relay] ${line}`); + } + }); + relayProcess.stderr?.on("data", (data: Buffer) => { + const lines = data.toString().split("\n").filter((l) => l.trim()); + for (const line of lines) { + // eslint-disable-next-line no-console + console.error(`[relay] ${line}`); + } + }); + + await waitForServer(relayPort, 30000); + }; + + const stopRelay = async () => { + if (!relayProcess) return; + relayProcess.kill("SIGTERM"); + relayProcess = null; + }; + 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 relayPort = await getAvailablePort(); - const relay = createRelayServer({ host: "127.0.0.1", port: relayPort }); - await relay.start(); + await startRelay(); const daemon = await createTestPaseoDaemon({ listen: "127.0.0.1", @@ -170,7 +230,7 @@ describe("Relay transport (E2EE) - daemon E2E", () => { throw err; } finally { await daemon.close(); - await relay.stop(); + await stopRelay(); } }, 30000 @@ -182,9 +242,7 @@ describe("Relay transport (E2EE) - daemon E2E", () => { process.env.PASEO_PRIMARY_LAN_IP = "192.168.1.12"; const { logger, lines } = createCapturingLogger(); - const relayPort = await getAvailablePort(); - const relay = createRelayServer({ host: "127.0.0.1", port: relayPort }); - await relay.start(); + await startRelay(); const daemon = await createTestPaseoDaemon({ listen: "127.0.0.1", @@ -271,7 +329,7 @@ describe("Relay transport (E2EE) - daemon E2E", () => { throw err; } finally { await daemon.close(); - await relay.stop(); + await stopRelay(); } }, 45000 diff --git a/packages/server/src/server/messages.ts b/packages/server/src/server/messages.ts index dec134ebb..0374a74ae 100644 --- a/packages/server/src/server/messages.ts +++ b/packages/server/src/server/messages.ts @@ -20,7 +20,7 @@ export function serializeAgentStreamEvent( ): AgentStreamEventPayload { if (event.type === "attention_required") { // Providers may emit attention_required without per-client notification context. - // The websocket bridge also emits attention_required with shouldNotify computed per client. + // The websocket server emits attention_required with shouldNotify computed per client. // Normalize provider events so they satisfy the shared schema. return { ...(event as Omit), diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 2a75cb2eb..3857dfb95 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,6 +1,7 @@ import { WebSocketServer } from "ws"; import type { Server as HTTPServer } from "http"; import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; +import { join } from "path"; import type { AgentManager } from "./agent/agent-manager.js"; import type { AgentStorage } from "./agent/agent-storage.js"; import type { DownloadTokenStore } from "./file-download/token-store.js"; @@ -8,10 +9,18 @@ import type { OpenAISTT } from "./agent/stt-openai.js"; import type { OpenAITTS } from "./agent/tts-openai.js"; import type { TerminalManager } from "../terminal/terminal-manager.js"; import type pino from "pino"; -import type { WSOutboundMessage } from "./messages.js"; -import { WebSocketSessionBridge } from "./websocket-session-bridge.js"; +import { + WSInboundMessageSchema, + type WSOutboundMessage, + wrapSessionMessage, +} from "./messages.js"; import type { AllowedHostsConfig } from "./allowed-hosts.js"; import { isHostAllowed } from "./allowed-hosts.js"; +import { Session } from "./session.js"; +import type { AgentProvider } from "./agent/agent-sdk-types.js"; +import { PushTokenStore } from "./push/token-store.js"; +import { PushService } from "./push/push-service.js"; +import { VoiceConversationStore } from "./voice-conversation-store.js"; export type AgentMcpTransportFactory = () => Promise; @@ -20,13 +29,55 @@ type WebSocketServerConfig = { allowedHosts?: AllowedHostsConfig; }; +function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffer { + if (typeof data === "string") return Buffer.from(data, "utf8"); + if (Array.isArray(data)) { + return Buffer.concat( + data.map((item) => + Buffer.isBuffer(item) ? item : Buffer.from(item as ArrayBuffer) + ) + ); + } + if (Buffer.isBuffer(data)) return data; + return Buffer.from(data as ArrayBuffer); +} + +type WebSocketLike = { + readyState: number; + send: (data: string) => void; + close: (code?: number, reason?: string) => void; + on: (event: "message" | "close" | "error", listener: (...args: any[]) => void) => void; + once: (event: "close" | "error", listener: (...args: any[]) => void) => void; +}; + /** * WebSocket server that only accepts sockets + parses/forwards messages to the session layer. */ export class VoiceAssistantWebSocketServer { private readonly logger: pino.Logger; private readonly wss: WebSocketServer; - private readonly bridge: WebSocketSessionBridge; + private readonly sessions: Map = new Map(); + private clientIdCounter = 0; + private readonly serverId: string; + private readonly agentManager: AgentManager; + private readonly agentStorage: AgentStorage; + private readonly downloadTokenStore: DownloadTokenStore; + private readonly paseoHome: string; + private readonly pushTokenStore: PushTokenStore; + private readonly pushService: PushService; + private readonly createAgentMcpTransport: AgentMcpTransportFactory; + private readonly stt: OpenAISTT | null; + private readonly tts: OpenAITTS | null; + private readonly terminalManager: TerminalManager | null; + private readonly voiceConversationStore: VoiceConversationStore; + private readonly dictation: { + openaiApiKey?: string | null; + finalTimeoutMs?: number; + } | null; + private readonly voice: { + openrouterApiKey?: string | null; + voiceLlmModel?: string | null; + } | null; constructor( server: HTTPServer, @@ -50,19 +101,31 @@ export class VoiceAssistantWebSocketServer { } ) { this.logger = logger.child({ module: "websocket-server" }); - this.bridge = new WebSocketSessionBridge( - this.logger, - serverId, - agentManager, - agentStorage, - downloadTokenStore, - paseoHome, - createAgentMcpTransport, - speech, - terminalManager, - voice, - dictation + this.serverId = serverId; + this.agentManager = agentManager; + this.agentStorage = agentStorage; + this.downloadTokenStore = downloadTokenStore; + this.paseoHome = paseoHome; + this.createAgentMcpTransport = createAgentMcpTransport; + this.stt = speech?.stt ?? null; + this.tts = speech?.tts ?? null; + this.terminalManager = terminalManager ?? null; + this.voiceConversationStore = new VoiceConversationStore( + join(paseoHome, "voice-conversations") ); + this.voice = voice ?? null; + this.dictation = dictation ?? null; + + const pushLogger = this.logger.child({ module: "push" }); + this.pushTokenStore = new PushTokenStore( + pushLogger, + join(paseoHome, "push-tokens.json") + ); + this.pushService = new PushService(pushLogger, this.pushTokenStore); + + this.agentManager.setAgentAttentionCallback((params) => { + this.broadcastAgentAttention(params); + }); const { allowedOrigins, allowedHosts } = wsConfig; this.wss = new WebSocketServer({ @@ -91,29 +154,503 @@ export class VoiceAssistantWebSocketServer { }); this.wss.on("connection", (ws, request) => { - void this.bridge.attach(ws, request); + void this.attachSocket(ws, request); }); this.logger.info("WebSocket server initialized on /ws"); } public broadcast(message: WSOutboundMessage): void { - this.bridge.broadcast(message); + const payload = JSON.stringify(message); + for (const ws of this.sessions.keys()) { + // WebSocket.OPEN = 1 + if (ws.readyState === 1) { + ws.send(payload); + } + } } public async attachExternalSocket( - ws: Parameters[0] + ws: WebSocketLike ): Promise { - const fakeRequest = { - url: "/ws", - headers: {}, - method: "GET", - } as unknown as Parameters[1]; - await this.bridge.attach(ws, fakeRequest); + await this.attachSocket(ws); } public async close(): Promise { - await this.bridge.closeAll(); + const cleanupPromises: Promise[] = []; + for (const [ws, session] of this.sessions) { + cleanupPromises.push(session.cleanup()); + cleanupPromises.push( + new Promise((resolve) => { + // WebSocket.CLOSED = 3 + if (ws.readyState === 3) { + resolve(); + return; + } + ws.once("close", () => resolve()); + ws.close(); + }) + ); + } + await Promise.all(cleanupPromises); + this.sessions.clear(); this.wss.close(); } + + private sendToClient(ws: WebSocketLike, message: WSOutboundMessage): void { + // WebSocket.OPEN = 1 + if (ws.readyState === 1) { + ws.send(JSON.stringify(message)); + } + } + + private async attachSocket(ws: WebSocketLike, _request?: unknown): Promise { + const clientId = `client-${++this.clientIdCounter}`; + const connectionLogger = this.logger.child({ clientId }); + + const session = new Session( + clientId, + (msg) => { + this.sendToClient(ws, wrapSessionMessage(msg)); + }, + connectionLogger.child({ module: "session" }), + this.downloadTokenStore, + this.pushTokenStore, + this.paseoHome, + this.agentManager, + this.agentStorage, + this.createAgentMcpTransport, + this.stt, + this.tts, + this.terminalManager, + this.voiceConversationStore, + this.voice ?? undefined, + this.dictation ?? undefined + ); + + this.sessions.set(ws, session); + + // Advertise stable server identity immediately on connect (used for URL/shareable IDs). + this.sendToClient( + ws, + wrapSessionMessage({ + type: "status", + payload: { + status: "server_info", + serverId: this.serverId, + }, + }) + ); + + connectionLogger.info( + { clientId, totalSessions: this.sessions.size }, + "Client connected" + ); + + ws.on("message", (data) => { + void this.handleRawMessage(ws, data); + }); + + ws.on("close", async () => { + await this.detachSocket(ws, connectionLogger, clientId); + }); + + ws.on("error", async (error) => { + const err = error instanceof Error ? error : new Error(String(error)); + connectionLogger.error({ err }, "Client error"); + await this.detachSocket(ws, connectionLogger, clientId); + }); + } + + private async detachSocket( + ws: WebSocketLike, + connectionLogger: pino.Logger, + clientId: string + ): Promise { + const session = this.sessions.get(ws); + if (!session) return; + + connectionLogger.info( + { clientId, totalSessions: this.sessions.size - 1 }, + "Client disconnected" + ); + + await session.cleanup(); + this.sessions.delete(ws); + } + + private async handleRawMessage( + ws: WebSocketLike, + data: Buffer | ArrayBuffer | Buffer[] | string + ): Promise { + try { + const buffer = bufferFromWsData(data); + const parsed = JSON.parse(buffer.toString()); + const parsedMessage = WSInboundMessageSchema.safeParse(parsed); + if (!parsedMessage.success) { + const requestInfo = extractRequestInfoFromUnknownWsInbound(parsed); + const isUnknownSchema = + requestInfo?.requestId != null && + typeof parsed === "object" && + parsed != null && + "type" in parsed && + (parsed as { type?: unknown }).type === "session"; + + this.logger.warn( + { + requestId: requestInfo?.requestId, + requestType: requestInfo?.requestType, + error: parsedMessage.error.message, + }, + "WS inbound message validation failed" + ); + + if (requestInfo) { + this.sendToClient( + ws, + wrapSessionMessage({ + type: "rpc_error", + payload: { + requestId: requestInfo.requestId, + requestType: requestInfo.requestType, + error: isUnknownSchema ? "Unknown request schema" : "Invalid message", + code: isUnknownSchema ? "unknown_schema" : "invalid_message", + }, + }) + ); + return; + } + + const errorMessage = `Invalid message: ${parsedMessage.error.message}`; + this.sendToClient( + ws, + wrapSessionMessage({ + type: "status", + payload: { + status: "error", + message: errorMessage, + }, + }) + ); + return; + } + + const message = parsedMessage.data; + + const messageSummary = { + type: message.type, + ...(message.type === "session" && message.message + ? { sessionMessageType: message.message.type } + : {}), + }; + this.logger.debug(messageSummary, "Received message"); + + if (message.type === "ping") { + this.sendToClient(ws, { type: "pong" }); + return; + } + + if (message.type === "recording_state") { + this.logger.debug({ isRecording: message.isRecording }, "Recording state"); + return; + } + + const session = this.sessions.get(ws); + if (!session) { + this.logger.error("No session found for client"); + return; + } + + if (message.type === "session") { + if (message.message.type === "create_agent_request") { + this.logger.debug( + { + cwd: message.message.config.cwd, + initialMode: message.message.config.modeId, + worktreeName: message.message.worktreeName, + requestId: message.message.requestId, + }, + "create_agent_request details" + ); + } + await session.handleMessage(message.message); + } + } catch (error) { + const err = error instanceof Error ? error : new Error(String(error)); + let rawPayload: string | null = null; + let parsedPayload: unknown = null; + + try { + const buffer = bufferFromWsData(data); + rawPayload = buffer.toString(); + parsedPayload = JSON.parse(rawPayload); + } catch (payloadError) { + rawPayload = rawPayload ?? ""; + parsedPayload = parsedPayload ?? rawPayload; + const payloadErr = + payloadError instanceof Error ? payloadError : new Error(String(payloadError)); + this.logger.error({ err: payloadErr }, "Failed to decode raw payload"); + } + + const trimmedRawPayload = + typeof rawPayload === "string" && rawPayload.length > 2000 + ? `${rawPayload.slice(0, 2000)}... (truncated)` + : rawPayload; + + this.logger.error( + { + err, + rawPayload: trimmedRawPayload, + parsedPayload, + }, + "Failed to parse/handle message" + ); + + const requestInfo = extractRequestInfoFromUnknownWsInbound(parsedPayload); + if (requestInfo) { + this.sendToClient( + ws, + wrapSessionMessage({ + type: "rpc_error", + payload: { + requestId: requestInfo.requestId, + requestType: requestInfo.requestType, + error: "Invalid message", + code: "invalid_message", + }, + }) + ); + return; + } + + this.sendToClient( + ws, + wrapSessionMessage({ + type: "status", + payload: { + status: "error", + message: `Invalid message: ${err.message}`, + }, + }) + ); + } + } + + private readonly ACTIVITY_THRESHOLD_MS = 120_000; + + private getClientActivityState(session: Session): { + deviceType: "web" | "mobile" | null; + focusedAgentId: string | null; + isStale: boolean; + appVisible: boolean; + } { + const activity = session.getClientActivity(); + if (!activity) { + this.logger.debug("getClientActivityState: no activity for session"); + return { deviceType: null, focusedAgentId: null, isStale: true, appVisible: false }; + } + const now = Date.now(); + const ageMs = now - activity.lastActivityAt.getTime(); + const isStale = ageMs >= this.ACTIVITY_THRESHOLD_MS; + this.logger.debug( + { + deviceType: activity.deviceType, + focusedAgentId: activity.focusedAgentId, + lastActivityAt: activity.lastActivityAt.toISOString(), + ageMs, + isStale, + appVisible: activity.appVisible, + }, + "getClientActivityState" + ); + return { + deviceType: activity.deviceType, + focusedAgentId: activity.focusedAgentId, + isStale, + appVisible: activity.appVisible, + }; + } + + private computeShouldNotifyForClient( + clientState: { + deviceType: "web" | "mobile" | null; + focusedAgentId: string | null; + isStale: boolean; + appVisible: boolean; + }, + allClientStates: Array<{ + deviceType: "web" | "mobile" | null; + focusedAgentId: string | null; + isStale: boolean; + appVisible: boolean; + }>, + agentId: string + ): boolean { + const isAnyoneActiveOnAgent = allClientStates.some( + (state) => state.focusedAgentId === agentId && state.appVisible && !state.isStale + ); + if (isAnyoneActiveOnAgent) { + return false; + } + + if (clientState.deviceType === null) { + return true; + } + + if (!clientState.isStale && clientState.appVisible && clientState.focusedAgentId !== null) { + return true; + } + + if (!clientState.isStale) { + return false; + } + + const hasActiveWebClient = allClientStates.some( + (state) => state.deviceType === "web" && !state.isStale + ); + + if (clientState.deviceType === "mobile") { + return !hasActiveWebClient; + } + + if (clientState.deviceType === "web") { + const hasOtherClient = allClientStates.some( + (state) => state !== clientState && (state.deviceType === "mobile" || state.deviceType === null) + ); + return !hasOtherClient; + } + + return true; + } + + private broadcastAgentAttention(params: { + agentId: string; + provider: AgentProvider; + reason: "finished" | "error" | "permission"; + }): void { + const clientEntries: Array<{ + ws: WebSocketLike; + state: { + deviceType: "web" | "mobile" | null; + focusedAgentId: string | null; + isStale: boolean; + appVisible: boolean; + }; + }> = []; + + for (const [ws, session] of this.sessions) { + clientEntries.push({ + ws, + state: this.getClientActivityState(session), + }); + } + + const allStates = clientEntries.map((e) => e.state); + + this.logger.debug( + { + agentId: params.agentId, + reason: params.reason, + clientCount: clientEntries.length, + allStates, + }, + "broadcastAgentAttention" + ); + + const hasActiveWebClient = allStates.some( + (state) => state.deviceType === "web" && !state.isStale + ); + const hasActiveMobileForegroundClient = allStates.some( + (state) => state.deviceType === "mobile" && state.appVisible && !state.isStale + ); + + // Push is only a fallback when the user is away from their desktop/web. + // Also suppress push if they're actively using the mobile app. + const shouldSendPush = + params.reason !== "error" && + !hasActiveWebClient && + !hasActiveMobileForegroundClient; + + this.logger.debug( + { hasActiveWebClient, hasActiveMobileForegroundClient, shouldSendPush }, + "Push gating check" + ); + + if (shouldSendPush) { + const tokens = this.pushTokenStore.getAllTokens(); + this.logger.info({ tokenCount: tokens.length }, "Sending push notification"); + if (tokens.length > 0) { + const agent = this.agentManager.getAgent(params.agentId); + const agentTitle = agent?.config?.title ?? agent?.cwd ?? params.agentId; + const title = + params.reason === "permission" ? "Agent needs permission" : "Agent finished"; + const body = + params.reason === "permission" + ? `Permission requested: ${agentTitle}` + : `Finished: ${agentTitle}`; + + void this.pushService.sendPush(tokens, { + title, + body, + data: { agentId: params.agentId, reason: params.reason }, + }); + } + } + + for (const { ws, state } of clientEntries) { + const shouldNotify = this.computeShouldNotifyForClient(state, allStates, params.agentId); + + const message = wrapSessionMessage({ + type: "agent_stream", + payload: { + agentId: params.agentId, + event: { + type: "attention_required", + provider: params.provider, + reason: params.reason, + timestamp: new Date().toISOString(), + shouldNotify, + }, + timestamp: new Date().toISOString(), + }, + }); + + this.sendToClient(ws, message); + } + } +} + +function extractRequestInfoFromUnknownWsInbound( + payload: unknown +): { requestId: string; requestType?: string } | null { + if (!payload || typeof payload !== "object") { + return null; + } + + const record = payload as { + type?: unknown; + requestId?: unknown; + message?: unknown; + }; + + // Session-wrapped messages + if (record.type === "session" && record.message && typeof record.message === "object") { + const msg = record.message as { requestId?: unknown; type?: unknown }; + if (typeof msg.requestId === "string") { + return { + requestId: msg.requestId, + ...(typeof msg.type === "string" ? { requestType: msg.type } : {}), + }; + } + } + + // Non-session messages (future-proof) + if (typeof record.requestId === "string") { + return { + requestId: record.requestId, + ...(typeof record.type === "string" ? { requestType: record.type } : {}), + }; + } + + return null; } diff --git a/packages/server/src/server/websocket-session-bridge.ts b/packages/server/src/server/websocket-session-bridge.ts deleted file mode 100644 index bd8f3226a..000000000 --- a/packages/server/src/server/websocket-session-bridge.ts +++ /dev/null @@ -1,602 +0,0 @@ -import type { IncomingMessage } from "http"; -import type { Transport } from "@modelcontextprotocol/sdk/shared/transport.js"; -import { join } from "path"; -import { - WSInboundMessageSchema, - type WSOutboundMessage, - wrapSessionMessage, -} from "./messages.js"; -import { Session } from "./session.js"; -import { AgentManager } from "./agent/agent-manager.js"; -import { AgentStorage } from "./agent/agent-storage.js"; -import type { AgentProvider } from "./agent/agent-sdk-types.js"; -import { DownloadTokenStore } from "./file-download/token-store.js"; -import { PushTokenStore } from "./push/token-store.js"; -import { PushService } from "./push/push-service.js"; -import { VoiceConversationStore } from "./voice-conversation-store.js"; -import type { OpenAISTT } from "./agent/stt-openai.js"; -import type { OpenAITTS } from "./agent/tts-openai.js"; -import type { TerminalManager } from "../terminal/terminal-manager.js"; -import type pino from "pino"; - -export type AgentMcpTransportFactory = () => Promise; - -function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffer { - if (typeof data === "string") return Buffer.from(data, "utf8"); - if (Array.isArray(data)) { - return Buffer.concat( - data.map((item) => - Buffer.isBuffer(item) ? item : Buffer.from(item as ArrayBuffer) - ) - ); - } - if (Buffer.isBuffer(data)) return data; - return Buffer.from(data as ArrayBuffer); -} - -type WebSocketLike = { - readyState: number; - send: (data: string) => void; - close: (code?: number, reason?: string) => void; - on: (event: "message" | "close" | "error", listener: (...args: any[]) => void) => void; - once: (event: "close" | "error", listener: (...args: any[]) => void) => void; -}; - -export class WebSocketSessionBridge { - private readonly logger: pino.Logger; - private readonly sessions: Map = new Map(); - private clientIdCounter = 0; - private readonly serverId: string; - private readonly agentManager: AgentManager; - private readonly agentStorage: AgentStorage; - private readonly downloadTokenStore: DownloadTokenStore; - private readonly paseoHome: string; - private readonly pushTokenStore: PushTokenStore; - private readonly pushService: PushService; - private readonly createAgentMcpTransport: AgentMcpTransportFactory; - private readonly stt: OpenAISTT | null; - private readonly tts: OpenAITTS | null; - private readonly terminalManager: TerminalManager | null; - private readonly voiceConversationStore: VoiceConversationStore; - private readonly dictation: { - openaiApiKey?: string | null; - finalTimeoutMs?: number; - } | null; - private readonly voice: { - openrouterApiKey?: string | null; - voiceLlmModel?: string | null; - } | null; - - constructor( - logger: pino.Logger, - serverId: string, - agentManager: AgentManager, - agentStorage: AgentStorage, - downloadTokenStore: DownloadTokenStore, - paseoHome: string, - createAgentMcpTransport: AgentMcpTransportFactory, - speech?: { stt: OpenAISTT | null; tts: OpenAITTS | null }, - terminalManager?: TerminalManager | null, - voice?: { - openrouterApiKey?: string | null; - voiceLlmModel?: string | null; - }, - dictation?: { - openaiApiKey?: string | null; - finalTimeoutMs?: number; - } - ) { - this.logger = logger.child({ module: "websocket-session-bridge" }); - this.serverId = serverId; - this.agentManager = agentManager; - this.agentStorage = agentStorage; - this.downloadTokenStore = downloadTokenStore; - this.paseoHome = paseoHome; - this.createAgentMcpTransport = createAgentMcpTransport; - this.stt = speech?.stt ?? null; - this.tts = speech?.tts ?? null; - this.terminalManager = terminalManager ?? null; - this.voiceConversationStore = new VoiceConversationStore( - join(paseoHome, "voice-conversations") - ); - this.voice = voice ?? null; - this.dictation = dictation ?? null; - - const pushLogger = this.logger.child({ module: "push" }); - this.pushTokenStore = new PushTokenStore( - pushLogger, - join(paseoHome, "push-tokens.json") - ); - this.pushService = new PushService(pushLogger, this.pushTokenStore); - - this.agentManager.setAgentAttentionCallback((params) => { - this.broadcastAgentAttention(params); - }); - } - - public getSessionCount(): number { - return this.sessions.size; - } - - public async attach(ws: WebSocketLike, _request: IncomingMessage): Promise { - const clientId = `client-${++this.clientIdCounter}`; - const connectionLogger = this.logger.child({ clientId }); - - const session = new Session( - clientId, - (msg) => { - this.sendToClient(ws, wrapSessionMessage(msg)); - }, - connectionLogger.child({ module: "session" }), - this.downloadTokenStore, - this.pushTokenStore, - this.paseoHome, - this.agentManager, - this.agentStorage, - this.createAgentMcpTransport, - this.stt, - this.tts, - this.terminalManager, - this.voiceConversationStore, - this.voice ?? undefined, - this.dictation ?? undefined - ); - - this.sessions.set(ws, session); - - // Advertise stable server identity immediately on connect (used for URL/shareable IDs). - this.sendToClient( - ws, - wrapSessionMessage({ - type: "status", - payload: { - status: "server_info", - serverId: this.serverId, - }, - }) - ); - - connectionLogger.info( - { clientId, totalSessions: this.sessions.size }, - "Client connected" - ); - - ws.on("message", (data) => { - void this.handleRawMessage(ws, data); - }); - - ws.on("close", async () => { - await this.detach(ws, connectionLogger, clientId); - }); - - ws.on("error", async (error) => { - const err = error instanceof Error ? error : new Error(String(error)); - connectionLogger.error({ err }, "Client error"); - await this.detach(ws, connectionLogger, clientId); - }); - } - - private async detach(ws: WebSocketLike, connectionLogger: pino.Logger, clientId: string): Promise { - const session = this.sessions.get(ws); - if (!session) return; - - connectionLogger.info( - { clientId, totalSessions: this.sessions.size - 1 }, - "Client disconnected" - ); - - await session.cleanup(); - this.sessions.delete(ws); - } - - private async handleRawMessage( - ws: WebSocketLike, - data: Buffer | ArrayBuffer | Buffer[] | string - ): Promise { - try { - const buffer = bufferFromWsData(data); - const parsed = JSON.parse(buffer.toString()); - const parsedMessage = WSInboundMessageSchema.safeParse(parsed); - if (!parsedMessage.success) { - const requestInfo = extractRequestInfoFromUnknownWsInbound(parsed); - const isUnknownSchema = - requestInfo?.requestId != null && - typeof parsed === "object" && - parsed != null && - "type" in parsed && - (parsed as { type?: unknown }).type === "session"; - - this.logger.warn( - { - requestId: requestInfo?.requestId, - requestType: requestInfo?.requestType, - error: parsedMessage.error.message, - }, - "WS inbound message validation failed" - ); - - if (requestInfo) { - this.sendToClient( - ws, - wrapSessionMessage({ - type: "rpc_error", - payload: { - requestId: requestInfo.requestId, - requestType: requestInfo.requestType, - error: isUnknownSchema ? "Unknown request schema" : "Invalid message", - code: isUnknownSchema ? "unknown_schema" : "invalid_message", - }, - }) - ); - return; - } - - const errorMessage = `Invalid message: ${parsedMessage.error.message}`; - this.sendToClient( - ws, - wrapSessionMessage({ - type: "status", - payload: { - status: "error", - message: errorMessage, - }, - }) - ); - return; - } - - const message = parsedMessage.data; - - const messageSummary = { - type: message.type, - ...(message.type === "session" && message.message - ? { sessionMessageType: message.message.type } - : {}), - }; - this.logger.debug(messageSummary, "Received message"); - - if (message.type === "ping") { - this.sendToClient(ws, { type: "pong" }); - return; - } - - if (message.type === "recording_state") { - this.logger.debug({ isRecording: message.isRecording }, "Recording state"); - return; - } - - const session = this.sessions.get(ws); - if (!session) { - this.logger.error("No session found for client"); - return; - } - - if (message.type === "session") { - if (message.message.type === "create_agent_request") { - this.logger.debug( - { - cwd: message.message.config.cwd, - initialMode: message.message.config.modeId, - worktreeName: message.message.worktreeName, - requestId: message.message.requestId, - }, - "create_agent_request details" - ); - } - await session.handleMessage(message.message); - } - } catch (error) { - const err = error instanceof Error ? error : new Error(String(error)); - let rawPayload: string | null = null; - let parsedPayload: unknown = null; - - try { - const buffer = bufferFromWsData(data); - rawPayload = buffer.toString(); - parsedPayload = JSON.parse(rawPayload); - } catch (payloadError) { - rawPayload = rawPayload ?? ""; - parsedPayload = parsedPayload ?? rawPayload; - const payloadErr = - payloadError instanceof Error ? payloadError : new Error(String(payloadError)); - this.logger.error({ err: payloadErr }, "Failed to decode raw payload"); - } - - const trimmedRawPayload = - typeof rawPayload === "string" && rawPayload.length > 2000 - ? `${rawPayload.slice(0, 2000)}... (truncated)` - : rawPayload; - - this.logger.error( - { - err, - rawPayload: trimmedRawPayload, - parsedPayload, - }, - "Failed to parse/handle message" - ); - - const requestInfo = extractRequestInfoFromUnknownWsInbound(parsedPayload); - if (requestInfo) { - this.sendToClient( - ws, - wrapSessionMessage({ - type: "rpc_error", - payload: { - requestId: requestInfo.requestId, - requestType: requestInfo.requestType, - error: "Invalid message", - code: "invalid_message", - }, - }) - ); - return; - } - - this.sendToClient( - ws, - wrapSessionMessage({ - type: "status", - payload: { - status: "error", - message: `Invalid message: ${err.message}`, - }, - }) - ); - } - } - - private sendToClient(ws: WebSocketLike, message: WSOutboundMessage): void { - // WebSocket.OPEN = 1 - if (ws.readyState === 1) { - ws.send(JSON.stringify(message)); - } - } - - public broadcast(message: WSOutboundMessage): void { - const payload = JSON.stringify(message); - for (const ws of this.sessions.keys()) { - if (ws.readyState === 1) { - ws.send(payload); - } - } - } - - public async closeAll(): Promise { - const cleanupPromises: Promise[] = []; - for (const [ws, session] of this.sessions) { - cleanupPromises.push(session.cleanup()); - cleanupPromises.push( - new Promise((resolve) => { - // WebSocket.CLOSED = 3 - if (ws.readyState === 3) { - resolve(); - return; - } - ws.once("close", () => resolve()); - ws.close(); - }) - ); - } - await Promise.all(cleanupPromises); - this.sessions.clear(); - } - - private readonly ACTIVITY_THRESHOLD_MS = 120_000; - - private getClientActivityState(session: Session): { - deviceType: "web" | "mobile" | null; - focusedAgentId: string | null; - isStale: boolean; - appVisible: boolean; - } { - const activity = session.getClientActivity(); - if (!activity) { - this.logger.debug("getClientActivityState: no activity for session"); - return { deviceType: null, focusedAgentId: null, isStale: true, appVisible: false }; - } - const now = Date.now(); - const ageMs = now - activity.lastActivityAt.getTime(); - const isStale = ageMs >= this.ACTIVITY_THRESHOLD_MS; - this.logger.debug( - { - deviceType: activity.deviceType, - focusedAgentId: activity.focusedAgentId, - lastActivityAt: activity.lastActivityAt.toISOString(), - ageMs, - isStale, - appVisible: activity.appVisible, - }, - "getClientActivityState" - ); - return { - deviceType: activity.deviceType, - focusedAgentId: activity.focusedAgentId, - isStale, - appVisible: activity.appVisible, - }; - } - - private computeShouldNotifyForClient( - clientState: { - deviceType: "web" | "mobile" | null; - focusedAgentId: string | null; - isStale: boolean; - appVisible: boolean; - }, - allClientStates: Array<{ - deviceType: "web" | "mobile" | null; - focusedAgentId: string | null; - isStale: boolean; - appVisible: boolean; - }>, - agentId: string - ): boolean { - const isAnyoneActiveOnAgent = allClientStates.some( - (state) => state.focusedAgentId === agentId && state.appVisible && !state.isStale - ); - if (isAnyoneActiveOnAgent) { - return false; - } - - if (clientState.deviceType === null) { - return true; - } - - if (!clientState.isStale && clientState.appVisible && clientState.focusedAgentId !== null) { - return true; - } - - if (!clientState.isStale) { - return false; - } - - const hasActiveWebClient = allClientStates.some( - (state) => state.deviceType === "web" && !state.isStale - ); - - if (clientState.deviceType === "mobile") { - return !hasActiveWebClient; - } - - if (clientState.deviceType === "web") { - const hasOtherClient = allClientStates.some( - (state) => state !== clientState && (state.deviceType === "mobile" || state.deviceType === null) - ); - return !hasOtherClient; - } - - return true; - } - - private broadcastAgentAttention(params: { - agentId: string; - provider: AgentProvider; - reason: "finished" | "error" | "permission"; - }): void { - const clientEntries: Array<{ - ws: WebSocketLike; - state: { - deviceType: "web" | "mobile" | null; - focusedAgentId: string | null; - isStale: boolean; - appVisible: boolean; - }; - }> = []; - - for (const [ws, session] of this.sessions) { - clientEntries.push({ - ws, - state: this.getClientActivityState(session), - }); - } - - const allStates = clientEntries.map((e) => e.state); - - this.logger.debug( - { - agentId: params.agentId, - reason: params.reason, - clientCount: clientEntries.length, - allStates, - }, - "broadcastAgentAttention" - ); - - const hasActiveWebClient = allStates.some( - (state) => state.deviceType === "web" && !state.isStale - ); - const hasActiveMobileForegroundClient = allStates.some( - (state) => state.deviceType === "mobile" && state.appVisible && !state.isStale - ); - - // Push is only a fallback when the user is away from their desktop/web. - // Also suppress push if they're actively using the mobile app. - const shouldSendPush = - params.reason !== "error" && - !hasActiveWebClient && - !hasActiveMobileForegroundClient; - - this.logger.debug( - { hasActiveWebClient, hasActiveMobileForegroundClient, shouldSendPush }, - "Push gating check" - ); - - if (shouldSendPush) { - const tokens = this.pushTokenStore.getAllTokens(); - this.logger.info({ tokenCount: tokens.length }, "Sending push notification"); - if (tokens.length > 0) { - const agent = this.agentManager.getAgent(params.agentId); - const agentTitle = agent?.config?.title ?? agent?.cwd ?? params.agentId; - const title = - params.reason === "permission" ? "Agent needs permission" : "Agent finished"; - const body = - params.reason === "permission" - ? `Permission requested: ${agentTitle}` - : `Finished: ${agentTitle}`; - - void this.pushService.sendPush(tokens, { - title, - body, - data: { agentId: params.agentId, reason: params.reason }, - }); - } - } - - for (const { ws, state } of clientEntries) { - const shouldNotify = this.computeShouldNotifyForClient(state, allStates, params.agentId); - - const message = wrapSessionMessage({ - type: "agent_stream", - payload: { - agentId: params.agentId, - event: { - type: "attention_required", - provider: params.provider, - reason: params.reason, - timestamp: new Date().toISOString(), - shouldNotify, - }, - timestamp: new Date().toISOString(), - }, - }); - - this.sendToClient(ws, message); - } - } -} - -function extractRequestInfoFromUnknownWsInbound( - payload: unknown -): { requestId: string; requestType?: string } | null { - if (!payload || typeof payload !== "object") { - return null; - } - - const record = payload as { - type?: unknown; - requestId?: unknown; - message?: unknown; - }; - - // Session-wrapped messages - if (record.type === "session" && record.message && typeof record.message === "object") { - const msg = record.message as { requestId?: unknown; type?: unknown }; - if (typeof msg.requestId === "string") { - return { - requestId: msg.requestId, - ...(typeof msg.type === "string" ? { requestType: msg.type } : {}), - }; - } - } - - // Non-session messages (future-proof) - if (typeof record.requestId === "string") { - return { - requestId: record.requestId, - ...(typeof record.type === "string" ? { requestType: record.type } : {}), - }; - } - - return null; -} diff --git a/scripts/relay-local.ts b/scripts/relay-local.ts deleted file mode 100644 index f0ad733c3..000000000 --- a/scripts/relay-local.ts +++ /dev/null @@ -1,29 +0,0 @@ -import { createRelayServer } from "@paseo/relay/node"; - -const port = Number(process.env.PORT ?? "7778"); -const host = process.env.HOST ?? "0.0.0.0"; - -async function main() { - const server = createRelayServer({ port, host }); - await server.start(); - console.log(`[relay-local] ready host=${host} port=${port}`); - - const stop = async () => { - try { - await server.stop(); - } finally { - process.exit(0); - } - }; - - process.on("SIGINT", stop); - process.on("SIGTERM", stop); - - setInterval(() => {}, 1e9); -} - -main().catch((e) => { - console.error(e); - process.exit(1); -}); -