puter/paseo
1
0
Fork 0
mirror of https://github.com/getpaseo/paseo.git synced 2026-07-29 12:01:31 +00:00
Code Issues Packages Projects Releases Wiki Activity

simplify: remove ws bridge; wrangler-based relay tests

Browse Source
This commit is contained in:
Mohamed Boudra
2026-02-05 10:45:51 +07:00
parent 36e105ed30
commit 574bdc16f6
16 changed files with 785 additions and 1276 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
.gitignore vendored
View File

@@ -57,3 +57,5 @@ CLAUDE.local.md
.debug.conversations/
.debug/
.paseo/
.wrangler/
**/.wrangler/

37
packages/app/e2e/global-setup.ts
View File

@@ -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<number> {
return new Promise((resolve, reject) => {
@@ -43,7 +42,7 @@ async function waitForServer(port: number, timeout = 15000): Promise<void> {
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 });

2
packages/app/e2e/pairing-offer-parsing.spec.ts
View File

@@ -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=...');

1
packages/relay/package.json
View File

@@ -6,7 +6,6 @@
"exports": {
".": "./src/index.ts",
"./e2ee": "./src/e2ee.ts",
"./node": "./src/node-adapter.ts",
"./cloudflare": "./src/cloudflare-adapter.ts"
},
"scripts": {

14
packages/relay/src/cloudflare-adapter.ts
View File

@@ -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 });
}

140
packages/relay/src/e2e.test.ts
View File

@@ -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<number> {
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<void> {
const start = Date.now();
while (Date.now() - start < timeout) {
try {
await new Promise<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((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<void>((r) => daemonWs.on("open", r));

4
packages/relay/src/index.ts
View File

@@ -1,9 +1,5 @@
export { Relay } from "./relay.js";
export type {
ConnectionRole,
RelayConnection,
RelayEvents,
RelaySession,
RelaySessionAttachment,
} from "./types.js";

387
packages/relay/src/node-adapter.ts
View File

@@ -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<void>;
stop(): Promise<void>;
}
type Session = {
serverId: string;
control: NodeWebSocket | null;
clients: Map<string, Set<NodeWebSocket>>; // clientId -> sockets
servers: Map<string, NodeWebSocket>; // clientId -> server-data socket
pending: Map<string, Array<string | ArrayBuffer>>; // 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<string, Session>();
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 "<binary>";
return bufferFromWsData(data).toString("utf8").slice(0, 200);
} catch {
return "<unavailable>";
}
})();
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();
}
});
},
};
}

152
packages/relay/src/relay.ts
View File

@@ -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<string, RelaySession>();
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);
}
}
}

24
packages/relay/src/types.ts
View File

@@ -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;
}

2
packages/server/src/client/daemon-client.ts
View File

@@ -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)) {

76
packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts
View File

@@ -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<number> {
});
}
async function waitForServer(port: number, timeout = 15000): Promise<void> {
const start = Date.now();
while (Date.now() - start < timeout) {
try {
await new Promise<void>((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

2
packages/server/src/server/messages.ts
View File

@@ -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<AgentStreamEventPayload, "shouldNotify">),

587
packages/server/src/server/websocket-server.ts
View File

@@ -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<Transport>;
@@ -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<WebSocketLike, Session> = 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<WebSocketSessionBridge["attach"]>[0]
ws: WebSocketLike
): Promise<void> {
const fakeRequest = {
url: "/ws",
headers: {},
method: "GET",
} as unknown as Parameters<WebSocketSessionBridge["attach"]>[1];
await this.bridge.attach(ws, fakeRequest);
await this.attachSocket(ws);
}
public async close(): Promise<void> {
await this.bridge.closeAll();
const cleanupPromises: Promise<void>[] = [];
for (const [ws, session] of this.sessions) {
cleanupPromises.push(session.cleanup());
cleanupPromises.push(
new Promise<void>((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<void> {
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<void> {
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<void> {
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 ?? "<unreadable>";
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;
}

602
packages/server/src/server/websocket-session-bridge.ts
View File

@@ -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<Transport>;
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<WebSocketLike, Session> = 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<void> {
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<void> {
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<void> {
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 ?? "<unreadable>";
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<void> {
const cleanupPromises: Promise<void>[] = [];
for (const [ws, session] of this.sessions) {
cleanupPromises.push(session.cleanup());
cleanupPromises.push(
new Promise<void>((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;
}

29
scripts/relay-local.ts
View File

@@ -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);
});