From 8c2a2a9a8833f3c6236e807a3ce698a6e1223cf2 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 30 Oct 2026 15:28:41 +0100 Subject: [PATCH] feat(transport): deliver large messages across bounded connections Frame logical messages below serialization and routing so large history responses preserve their contents while fitting encrypted relay and host-channel limits. Negotiate support per connection and retain the existing logical size bound. --- docs/protocol-compatibility.md | 18 ++ packages/client/src/connection/index.ts | 1 + .../src/daemon-client-relay-e2ee-transport.ts | 16 +- .../src/daemon-client-transport-types.ts | 4 +- .../src/daemon-client-websocket-transport.ts | 16 ++ packages/client/src/daemon-client.test.ts | 173 +++++++++++------- packages/client/src/daemon-client.ts | 31 +++- packages/client/src/hosts/index.ts | 2 + packages/client/src/node/connector.test.ts | 10 +- packages/client/src/node/index.ts | 5 +- .../protocol/src/binary-frames/host-route.ts | 5 + .../protocol/src/chunked-transport.test.ts | 128 +++++++++++++ packages/protocol/src/chunked-transport.ts | 123 +++++++++++++ packages/protocol/src/client-capabilities.ts | 1 + packages/protocol/src/messages.ts | 1 + .../server/connected-hosts/hosts.e2e.test.ts | 31 ++++ .../server/src/server/connected-hosts/link.ts | 7 +- .../server/src/server/websocket-server.ts | 71 ++++--- .../websocket/encrypted-relay-socket.test.ts | 71 +++++++ .../websocket/encrypted-relay-socket.ts | 4 + 20 files changed, 608 insertions(+), 110 deletions(-) create mode 100644 packages/protocol/src/chunked-transport.test.ts create mode 100644 packages/protocol/src/chunked-transport.ts diff --git a/docs/protocol-compatibility.md b/docs/protocol-compatibility.md index f7244e67ba..46b9eacdff 100644 --- a/docs/protocol-compatibility.md +++ b/docs/protocol-compatibility.md @@ -52,6 +52,24 @@ On capable daemons, connecting creates no timeline or event demand. Client subsc release it on unsubscribe, and restore it after reconnect. Raw message observers inspect traffic without requesting streams. Application caches and which agents are visible remain caller-owned. +## Binary chunking + +`binary_chunking_v1` in hello and optional `server_info.features.binaryChunkingV1` +negotiate connection-scoped framing. Application serializers and routed envelopes use +the same boundary. Small messages and handshakes keep their existing representation. +Older peers retain existing frame limits. + +The reserved binary opcode `0x31` is followed by version `1`, original kind +(`0` text, `1` binary), and START/END flag bits `0`/`1`. Remaining bytes are an +unchanged slice. There is no total length or chunk count; a future receiver can +consume slices incrementally. Today the receiver buffers up to the existing +100 MiB logical limit and decodes UTF-8 only after END. Transfers cannot interleave. +Disconnect clears partial state and negotiation. + +The lower transport supplies its plaintext frame budget, including encryption +overhead. The sender waits for each physical write before sending the next chunk. +The framing module never selects an encryption construction. + ## Owned observations `owned_subscriptions` and `server_info.features.ownedSubscriptions` negotiate the source-owned diff --git a/packages/client/src/connection/index.ts b/packages/client/src/connection/index.ts index eda2fec26b..1ad4b34dee 100644 --- a/packages/client/src/connection/index.ts +++ b/packages/client/src/connection/index.ts @@ -125,6 +125,7 @@ export class ConnectionSubscriptions extends OwnedSubscriptions { // Protocol support belongs to the installed client. Only browser hosting needs // a resource supplied by the caller. Keep this exhaustive as the protocol evolves. export const DEFAULT_CLIENT_CAPABILITIES = { + [CLIENT_CAPS.binaryChunkingV1]: true, [CLIENT_CAPS.helloRejection]: true, [CLIENT_CAPS.ownedSubscriptions]: true, [CLIENT_CAPS.allProviders]: true, diff --git a/packages/client/src/daemon-client-relay-e2ee-transport.ts b/packages/client/src/daemon-client-relay-e2ee-transport.ts index 258c0aa9b5..27081046d0 100644 --- a/packages/client/src/daemon-client-relay-e2ee-transport.ts +++ b/packages/client/src/daemon-client-relay-e2ee-transport.ts @@ -1,3 +1,4 @@ +import { maxSecuredPlaintextByteLength } from "@getpaseo/relay"; import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import { openClientSecureChannel, @@ -155,18 +156,15 @@ export function createEncryptedTransport( const relayTransport: RelayTransport = { send: (data) => { if (typeof data === "string") { - base.send(data); - return; + return base.send(data); } if (ArrayBuffer.isView(data)) { - base.send(normalizeTransportPayload(data)); - return; + return base.send(normalizeTransportPayload(data)); } if (data instanceof ArrayBuffer) { - base.send(data); - return; + return base.send(data); } - base.send(String(data)); + return base.send(String(data)); }, // The channel decides the close code. Report it now: a socket close event // arrives asynchronously, and the consumer must not act on an earlier @@ -226,6 +224,7 @@ export function createEncryptedTransport( }); return { + maxFrameBytes: maxSecuredPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES), send: (data) => { if (!channel) { throw new Error("Encrypted channel not ready"); @@ -239,8 +238,9 @@ export function createEncryptedTransport( // Encryption and its transport handoff are synchronous up to send's first await. // Reject capacity failures here so a routed sender loses only its own channel. assertOutboundCapacity(wireBytes); - void channel.send(payload).catch((error) => { + return channel.send(payload).catch((error) => { emitError(error); + throw error; }); }, close: (code?: number, reason?: string) => { diff --git a/packages/client/src/daemon-client-transport-types.ts b/packages/client/src/daemon-client-transport-types.ts index ddcb57246b..76afb52094 100644 --- a/packages/client/src/daemon-client-transport-types.ts +++ b/packages/client/src/daemon-client-transport-types.ts @@ -1,5 +1,6 @@ export interface DaemonTransport { - send: (data: string | Uint8Array | ArrayBuffer) => void; + maxFrameBytes?: number; + send: (data: string | Uint8Array | ArrayBuffer) => void | Promise; close: (code?: number, reason?: string) => void; onMessage: (handler: (data: unknown, isBinary: boolean) => void) => () => void; onOpen: (handler: () => void) => () => void; @@ -19,6 +20,7 @@ export type WebSocketFactory = ( ) => WebSocketLike; export interface WebSocketLike { + bufferedAmount?: number; readyState: number; send: (data: string | Uint8Array | ArrayBuffer) => void; close: (code?: number, reason?: string) => void; diff --git a/packages/client/src/daemon-client-websocket-transport.ts b/packages/client/src/daemon-client-websocket-transport.ts index f5734b4b51..f0667720f0 100644 --- a/packages/client/src/daemon-client-websocket-transport.ts +++ b/packages/client/src/daemon-client-websocket-transport.ts @@ -55,6 +55,22 @@ export function createWebSocketTransportFactory(factory: WebSocketFactory): Daem throw new Error(`WebSocket not open (readyState=${ws.readyState})`); } ws.send(data); + if ((ws.bufferedAmount ?? 0) > 1024 * 1024) { + return new Promise((resolve, reject) => { + const drain = () => { + if (ws.readyState !== 1) { + reject(new Error("WebSocket closed while draining")); + return; + } + if ((ws.bufferedAmount ?? 0) <= 1024 * 1024) { + resolve(); + return; + } + setTimeout(drain, 10); + }; + drain(); + }); + } }, close: (code?: number, reason?: string) => { // Node's `ws` may emit an `error` when a connecting socket is closed before the diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 112113caf6..058c9ba77b 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -1094,6 +1094,7 @@ test("advertises client capabilities in hello", async () => { provider_snapshot_references: true, explicit_event_subscriptions: true, owned_subscriptions: true, + binary_chunking_v1: true, compact_provider_snapshots: true, custom_mode_icons: true, project_updates: true, @@ -6817,91 +6818,102 @@ test("drops legacy agent_stream tool_call payloads and logs validation warning", expect(logger.warn).toHaveBeenCalled(); }); -test("parses canonical fetch_agent_timeline_response payloads without crashing", async () => { - const logger = createMockLogger(); - const mock = createMockTransport(); +test.each([false, true])( + "parses canonical timeline responses with chunking=%s", + async (chunked) => { + const logger = createMockLogger(); + const mock = createMockTransport(); - const client = new DaemonClient({ - url: "ws://test", - clientId: "clsk_unit_test", - logger, - reconnect: { enabled: false }, - transportFactory: () => mock.transport, - }); - clients.push(client); + const client = new DaemonClient({ + url: "ws://test", + clientId: "clsk_unit_test", + logger, + reconnect: { enabled: false }, + transportFactory: () => mock.transport, + }); + clients.push(client); - const connectPromise = client.connect(); - mock.triggerOpen(); - await connectPromise; + const connectPromise = client.connect(); + mock.triggerOpen({ features: { ownedSubscriptions: true, binaryChunkingV1: chunked } }); + await connectPromise; - const received: unknown[] = []; - const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => { - received.push(msg); - }); + const received: unknown[] = []; + const unsubscribe = client.on("fetch_agent_timeline_response", (msg) => { + received.push(msg); + }); - mock.triggerMessage( - wrapSessionMessage({ - type: "fetch_agent_timeline_response", + const command = chunked ? "x".repeat(50 * 1024 * 1024) : "pwd"; + const { ChunkedTransport } = await import("@getpaseo/protocol/chunked-transport"); + const sender = new ChunkedTransport({ + frameBytes: 24 * 1024 * 1024, + send: (frame) => mock.triggerMessage(frame), + deliver: () => {}, + }); + sender.enabled = chunked; + await sender.send( + wrapSessionMessage({ + type: "fetch_agent_timeline_response", + payload: { + requestId: "req-1", + agentId: "agent_cli", + agent: null, + direction: "tail", + projection: "projected", + epoch: "epoch-1", + reset: false, + staleCursor: false, + gap: false, + window: { minSeq: 1, maxSeq: 1, nextSeq: 2 }, + startCursor: { epoch: "epoch-1", seq: 1 }, + endCursor: { epoch: "epoch-1", seq: 1 }, + hasOlder: false, + hasNewer: false, + entries: [ + { + timestamp: "2026-02-08T20:20:00.000Z", + provider: "codex", + seqStart: 1, + seqEnd: 1, + sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], + collapsed: [], + item: { + type: "tool_call", + callId: "call_cli_snapshot", + name: "shell", + status: "running", + detail: { + type: "shell", + command, + }, + error: null, + }, + }, + ], + error: null, + }, + }), + ); + + unsubscribe(); + + expect(received).toHaveLength(1); + expect(received[0]).toMatchObject({ payload: { - requestId: "req-1", - agentId: "agent_cli", - agent: null, - direction: "tail", - projection: "projected", - epoch: "epoch-1", - reset: false, - staleCursor: false, - gap: false, - window: { minSeq: 1, maxSeq: 1, nextSeq: 2 }, - startCursor: { epoch: "epoch-1", seq: 1 }, - endCursor: { epoch: "epoch-1", seq: 1 }, - hasOlder: false, - hasNewer: false, entries: [ { - timestamp: "2026-02-08T20:20:00.000Z", - provider: "codex", - seqStart: 1, - seqEnd: 1, - sourceSeqRanges: [{ startSeq: 1, endSeq: 1 }], - collapsed: [], item: { type: "tool_call", - callId: "call_cli_snapshot", - name: "shell", status: "running", - detail: { - type: "shell", - command: "pwd", - }, error: null, + detail: { type: "shell", command }, }, }, ], - error: null, }, - }), - ); - - unsubscribe(); - - expect(received).toHaveLength(1); - expect(received[0]).toMatchObject({ - payload: { - entries: [ - { - item: { - type: "tool_call", - status: "running", - error: null, - detail: { type: "shell" }, - }, - }, - ], - }, - }); - expect(logger.warn).not.toHaveBeenCalled(); -}); + }); + expect(logger.warn).not.toHaveBeenCalled(); + }, +); test("rejects and logs a correlated response that violates the protocol schema", async () => { const mock = createMockTransport(); @@ -8471,3 +8483,24 @@ test("closes before admission when a protected-capable daemon is reached in lega expect(daemonReceived).toHaveLength(1); expect(JSON.parse(daemonReceived[0] as string)).toMatchObject({ type: "hello" }); }); + +test("uses the frame budget selected by an asynchronously opened transport", async () => { + const mock = createMockTransport(); + mock.transport.maxFrameBytes = 1024; + const client = new DaemonClient({ + url: "ws://test", + clientId: "dynamic-budget", + transportFactory: () => mock.transport, + reconnect: { enabled: false }, + }); + clients.push(client); + const connecting = client.connect(); + mock.transport.maxFrameBytes = 16; + mock.triggerOpen({ features: { binaryChunkingV1: true } }); + await connecting; + client.sendHostFrame("route", new Uint8Array(100)); + await vi.waitFor(() => expect(mock.sent.length).toBeGreaterThan(1)); + expect(mock.sent.every((frame) => typeof frame !== "string" && frame.byteLength <= 16)).toBe( + true, + ); +}); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index d4b0f28a39..8f8ec74ed4 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1,3 +1,4 @@ +import { ChunkedTransport } from "@getpaseo/protocol/chunked-transport"; import { HOST_ROUTE_OPCODE, encodeHostRouteFrame, @@ -1366,6 +1367,7 @@ export class DaemonClient { failed: (error) => this.logger.error({ err: error }, "Subscription failed"), }); private transport: DaemonTransport | null = null; + private chunkedTransport: ChunkedTransport | null = null; private helloAuth: HelloAuth; private transportCleanup: Array<() => void> = []; private rawMessageListeners: Set<(message: SessionOutboundMessage) => void> = new Set(); @@ -1560,6 +1562,13 @@ export class DaemonClient { }); this.traceConnectionAttempt += 1; this.transport = transport; + this.chunkedTransport = new ChunkedTransport({ + get frameBytes() { + return transport.maxFrameBytes ?? 32 * 1024 * 1024; + }, + send: (frame) => transport.send(frame), + deliver: (frame) => this.handleTransportMessage(frame), + }); this.lastServerInfoMessage = null; this.authFailureReasonValue = null; @@ -1666,7 +1675,15 @@ export class DaemonClient { }); }), transport.onMessage((data) => { - if (this.transport === transport) this.handleTransportMessage(data); + if (this.transport !== transport) return; + const bytes = asUint8Array(data); + try { + if (typeof data === "string" || bytes) + this.chunkedTransport!.receive(typeof data === "string" ? data : bytes!); + else this.handleTransportMessage(data); + } catch { + transport.close(4003, "Invalid chunked transport"); + } }), ]; } catch (error) { @@ -2083,7 +2100,11 @@ export class DaemonClient { }); try { submitted?.(); - this.transport.send(frame); + const payload = typeof frame === "string" ? frame : asUint8Array(frame)!; + const transport = this.transport; + void this.chunkedTransport!.send(payload).catch(() => + transport.close(4003, "Transport send failed"), + ); } finally { this.endTraceSection(isOpen); } @@ -6763,6 +6784,8 @@ export class DaemonClient { } private disposeTransport(code = 1001, reason = "Reconnecting"): void { + this.chunkedTransport?.close(); + this.chunkedTransport = null; this.owned.disconnected(); this.providerSnapshotUpdates.clear(); this.stopLivenessHeartbeat(); @@ -7289,6 +7312,10 @@ export class DaemonClient { if (consumerMessage.type === "status") { const serverInfo = parseServerInfoStatusPayload(consumerMessage.payload); if (serverInfo) { + if (this.chunkedTransport) + this.chunkedTransport.enabled = + serverInfo.features?.binaryChunkingV1 === true && + this.config.capabilities?.[CLIENT_CAPS.binaryChunkingV1] !== false; this.lastServerInfoMessage = serverInfo; this.destinationServerId = serverInfo.serverId; this.connectionFailure = undefined; diff --git a/packages/client/src/hosts/index.ts b/packages/client/src/hosts/index.ts index 1baa9ebe43..97b25f06da 100644 --- a/packages/client/src/hosts/index.ts +++ b/packages/client/src/hosts/index.ts @@ -1,3 +1,4 @@ +import { MAX_HOST_CHANNEL_FRAME_BYTES } from "@getpaseo/protocol/binary-frames/index"; import { safeRandomId } from "../daemon-client-transport-utils.js"; import { DaemonClient } from "../daemon-client.js"; import type { DaemonTransportFactory } from "../daemon-client-transport-types.js"; @@ -84,6 +85,7 @@ export function createHostChannelTransport( }) .catch(close); return { + maxFrameBytes: MAX_HOST_CHANNEL_FRAME_BYTES, send(data) { if (closed || !routeId) throw Object.assign(new Error(`Host ${serverId} is offline`), { diff --git a/packages/client/src/node/connector.test.ts b/packages/client/src/node/connector.test.ts index 6e916d4bec..a1eba1ef15 100644 --- a/packages/client/src/node/connector.test.ts +++ b/packages/client/src/node/connector.test.ts @@ -44,12 +44,14 @@ test("a slow consumer cannot exceed the shared byte budget and close releases re socket.pause(); await ready; const frame = new Uint8Array(1024 * 1024); - for (let index = 0; index < 4; index++) transport.send(frame); + const writes = Array.from({ length: 4 }, () => transport.send(frame)); + const drained = Promise.allSettled(writes); expect(budget.bufferedBytes).toBe(4 * 1024 * 1024); expect(() => transport.send(frame)).toThrow("Connected host buffer limit reached"); expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit); transport.close(); await expect.poll(() => budget.bufferedBytes).toBe(0); + await drained; }); test("direct connector preserves 25 MiB binary frames in both directions", async () => { const server = await endpoint(); @@ -124,17 +126,20 @@ test("encrypted connector rejects overload synchronously and preserves the physi }); await new Promise((resolve) => transport.onOpen(resolve)); await ready; + expect(transport.maxFrameBytes).toBeLessThan(32 * 1024 * 1024); peer.pause(); const frame = new Uint8Array(1024 * 1024).fill(173); let rejected: unknown; + const writes: (void | Promise)[] = []; for (let index = 0; index < 8; index++) { try { - transport.send(frame); + writes.push(transport.send(frame)); } catch (error) { rejected = error; break; } } + const drained = Promise.allSettled(writes); expect(rejected).toMatchObject({ code: "resource_limit" }); expect(budget.bufferedBytes).toBeLessThanOrEqual(budget.limit); await new Promise((resolve) => setImmediate(resolve)); @@ -142,6 +147,7 @@ test("encrypted connector rejects overload synchronously and preserves the physi expect(closed).toBe(false); transport.send("sibling control"); peer.resume(); + await drained; await expect.poll(() => received.includes("sibling control")).toBe(true); expect(received.filter((data) => data instanceof ArrayBuffer).length).toBeGreaterThan(0); expect( diff --git a/packages/client/src/node/index.ts b/packages/client/src/node/index.ts index 46b5cff17a..2516946a07 100644 --- a/packages/client/src/node/index.ts +++ b/packages/client/src/node/index.ts @@ -172,9 +172,12 @@ export function createHostTransportFactory( close(); }); return { + get maxFrameBytes() { + return transport?.maxFrameBytes; + }, send(data) { if (closed || !transport) throw new Error("Host transport is unavailable"); - transport.send(data); + return transport.send(data); }, close, onOpen(handler) { diff --git a/packages/protocol/src/binary-frames/host-route.ts b/packages/protocol/src/binary-frames/host-route.ts index f45cd714f9..643dcb4e98 100644 --- a/packages/protocol/src/binary-frames/host-route.ts +++ b/packages/protocol/src/binary-frames/host-route.ts @@ -3,6 +3,11 @@ import { MAX_WEBSOCKET_MESSAGE_BYTES } from "../websocket-message-limit.js"; export const HOST_ROUTE_OPCODE = 0x30; export const MAX_HOST_ROUTE_ID_BYTES = 256; export const MAX_HOST_ROUTE_HEADER_BYTES = 3 + MAX_HOST_ROUTE_ID_BYTES; +// A logical host channel crosses the 32 MiB route admission boundary. Text can +// expand sixfold when enclosed in JSON; binary adds the bounded route header. +export const MAX_HOST_CHANNEL_FRAME_BYTES = Math.floor( + (32 * 1024 * 1024 - MAX_HOST_ROUTE_HEADER_BYTES) / 6, +); export const MAX_ROUTED_PAYLOAD_BYTES = MAX_WEBSOCKET_MESSAGE_BYTES - MAX_HOST_ROUTE_HEADER_BYTES; export interface HostRouteFrame { routeId: string; diff --git a/packages/protocol/src/chunked-transport.test.ts b/packages/protocol/src/chunked-transport.test.ts new file mode 100644 index 0000000000..9131de581f --- /dev/null +++ b/packages/protocol/src/chunked-transport.test.ts @@ -0,0 +1,128 @@ +import { describe, expect, it } from "vitest"; +import { ChunkedTransport } from "./chunked-transport.js"; + +describe("chunked transport", () => { + it("delivers a large UTF-8 logical message once, preserving order", async () => { + const delivered: (string | Uint8Array)[] = []; + const receiver = new ChunkedTransport({ + frameBytes: 17, + send: () => {}, + deliver: (v) => delivered.push(v), + }); + receiver.enabled = true; + const frames: (string | Uint8Array)[] = []; + const sender = new ChunkedTransport({ + frameBytes: 17, + send: (v) => { + frames.push(v); + receiver.receive(v); + }, + deliver: () => {}, + }); + sender.enabled = true; + const text = "🌍".repeat(100); + await sender.send(text); + expect(delivered).toEqual([text]); + expect(frames.length).toBeGreaterThan(1); + expect(frames.every((v) => v instanceof Uint8Array && v.byteLength <= 17)).toBe(true); + }); +}); + +it("preserves binary kind and leaves legacy frames unchanged", async () => { + const delivered: (string | Uint8Array)[] = []; + const receiver = new ChunkedTransport({ + frameBytes: 8, + send: () => {}, + deliver: (v) => delivered.push(v), + }); + const frames: (string | Uint8Array)[] = []; + const sender = new ChunkedTransport({ + frameBytes: 8, + send: (v) => { + frames.push(v); + receiver.receive(v); + }, + deliver: () => {}, + }); + const data = new Uint8Array(40).fill(5); + await sender.send(data); + expect(frames).toEqual([data]); + frames.length = 0; + sender.enabled = receiver.enabled = true; + await sender.send(data); + expect(frames).toHaveLength(10); + expect(delivered).toEqual([data, data]); +}); + +it("rejects malformed sequences and drops incomplete transfers on close", () => { + const delivered: unknown[] = []; + const receiver = new ChunkedTransport({ + frameBytes: 8, + send: () => {}, + deliver: (v) => delivered.push(v), + }); + receiver.enabled = true; + receiver.receive(new Uint8Array([0x31, 1, 0, 1, 65])); + expect(() => receiver.receive("interrupted")).toThrow("Interrupted"); + receiver.receive(new Uint8Array([0x31, 1, 0, 2, 66])); + expect(delivered).toEqual([]); +}); + +it("waits for drain before the next chunk and preserves logical ordering", async () => { + const frames: (string | Uint8Array)[] = []; + let drain!: () => void; + const sender = new ChunkedTransport({ + frameBytes: 8, + send: (v) => { + frames.push(v); + if (frames.length === 1) + return new Promise((resolve) => { + drain = resolve; + }); + }, + deliver: () => {}, + }); + sender.enabled = true; + const first = sender.send("abcdefghij"); + const second = sender.send("next"); + await Promise.resolve(); + expect(frames).toHaveLength(1); + drain(); + await Promise.all([first, second]); + expect(frames).toHaveLength(4); + expect(frames[3]).toBe("next"); +}); + +it.each([ + [0x31, 2, 0, 3, 65], + [0x31, 1, 2, 3, 65], + [0x31, 1, 0, 4, 65], + [0x31, 1, 0, 2, 65], + [0x31, 1, 0, 1], +])("rejects malformed chunk metadata %j", (...bytes) => { + const receiver = new ChunkedTransport({ + frameBytes: 8, + send: () => {}, + deliver: () => { + throw new Error("Unexpected delivery"); + }, + }); + receiver.enabled = true; + expect(() => receiver.receive(new Uint8Array(bytes))).toThrow(); +}); + +it("reconnect discards partial state and requires negotiation again", () => { + const delivered: unknown[] = []; + const create = () => + new ChunkedTransport({ frameBytes: 8, send: () => {}, deliver: (v) => delivered.push(v) }); + const old = create(); + old.enabled = true; + old.receive(new Uint8Array([0x31, 1, 0, 1, 65])); + old.close(); + old.receive(new Uint8Array([0x31, 1, 0, 2, 66])); + const fresh = create(); + expect(fresh.enabled).toBe(false); + fresh.enabled = true; + fresh.receive(new Uint8Array([0x31, 1, 0, 3, 67])); + expect(delivered).toEqual(["C"]); +}); diff --git a/packages/protocol/src/chunked-transport.ts b/packages/protocol/src/chunked-transport.ts new file mode 100644 index 0000000000..f040e28822 --- /dev/null +++ b/packages/protocol/src/chunked-transport.ts @@ -0,0 +1,123 @@ +import { MAX_WEBSOCKET_MESSAGE_BYTES } from "./websocket-message-limit.js"; + +type Frame = string | Uint8Array; +// Reserved outside the application binary opcode space. Version, kind, flags follow. +const OPCODE = 0x31; +const HEADER = 4; + +/** Connection-scoped ordered framing, below serialization and routing, above encryption. */ +export class ChunkedTransport { + enabled = false; + private parts: Uint8Array[] | undefined; + private size = 0; + private kind = 0; + private closed = false; + private pending: Promise | undefined; + + constructor( + private readonly io: { + frameBytes: number; + send: (frame: Frame) => void | Promise; + deliver: (frame: Frame) => void; + }, + ) { + if (io.frameBytes <= HEADER) throw new Error("Invalid transport frame budget"); + } + + send(frame: Frame): Promise { + const bytes = typeof frame === "string" ? new TextEncoder().encode(frame) : frame; + if (bytes.byteLength > MAX_WEBSOCKET_MESSAGE_BYTES) + throw new Error("Logical message too large"); + const chunk = this.enabled && bytes.byteLength > this.io.frameBytes; + if (this.closed) throw new Error("Transport closed"); + const write = () => { + if (this.closed) throw new Error("Transport closed"); + return chunk ? this.sendChunks(bytes, typeof frame === "string") : this.io.send(frame); + }; + const result = this.pending ? this.pending.then(write) : write(); + if (!result) return Promise.resolve(); + const pending = Promise.resolve(result).catch((error) => { + this.close(); + throw error; + }); + this.pending = pending; + void pending.then( + () => { + if (this.pending === pending) this.pending = undefined; + return undefined; + }, + () => {}, + ); + return pending; + } + + private async sendChunks(bytes: Uint8Array, text: boolean): Promise { + const capacity = this.io.frameBytes - HEADER; + for (let offset = 0; offset < bytes.byteLength; offset += capacity) { + if (this.closed) throw new Error("Transport closed"); + const end = Math.min(bytes.byteLength, offset + capacity); + const wire = new Uint8Array(HEADER + end - offset); + wire.set([ + OPCODE, + 1, + text ? 0 : 1, + (offset === 0 ? 1 : 0) | (end === bytes.byteLength ? 2 : 0), + ]); + wire.set(bytes.subarray(offset, end), HEADER); + await this.io.send(wire); + } + } + + receive(frame: Frame): void { + if (this.closed) return; + try { + if (typeof frame === "string" || frame[0] !== OPCODE) { + if (this.parts) throw new Error("Interrupted chunked message"); + this.io.deliver(frame); + return; + } + if ( + !this.enabled || + frame.length <= HEADER || + frame[1] !== 1 || + frame[2]! > 1 || + (frame[3]! & ~3) !== 0 + ) + throw new Error("Invalid chunk header"); + const start = (frame[3]! & 1) !== 0; + const end = (frame[3]! & 2) !== 0; + if (start === !!this.parts) throw new Error("Invalid chunk sequence"); + if (start) { + this.parts = []; + this.kind = frame[2]!; + this.size = 0; + } + if (frame[2] !== this.kind) throw new Error("Chunk kind changed"); + this.size += frame.length - HEADER; + if (this.size > MAX_WEBSOCKET_MESSAGE_BYTES) throw new Error("Logical message too large"); + this.parts!.push(frame.slice(HEADER)); + if (!end) return; + const data = new Uint8Array(this.size); + let offset = 0; + for (const part of this.parts!) { + data.set(part, offset); + offset += part.length; + } + this.parts = undefined; + this.size = 0; + this.io.deliver( + this.kind === 0 ? new TextDecoder("utf-8", { fatal: true }).decode(data) : data, + ); + } catch (error) { + this.close(); + throw error; + } + } + + close(): void { + this.closed = true; + this.enabled = false; + this.parts = undefined; + this.size = 0; + } +} diff --git a/packages/protocol/src/client-capabilities.ts b/packages/protocol/src/client-capabilities.ts index a35852cc06..28ec8456cb 100644 --- a/packages/protocol/src/client-capabilities.ts +++ b/packages/protocol/src/client-capabilities.ts @@ -1,4 +1,5 @@ export const CLIENT_CAPS = { + binaryChunkingV1: "binary_chunking_v1", helloRejection: "hello_rejection", // COMPAT(ownedSubscriptions): added in v0.8.0, remove legacy ownership after 2027-03-09. ownedSubscriptions: "owned_subscriptions", diff --git a/packages/protocol/src/messages.ts b/packages/protocol/src/messages.ts index 18c3da9b67..34272bce48 100644 --- a/packages/protocol/src/messages.ts +++ b/packages/protocol/src/messages.ts @@ -3832,6 +3832,7 @@ export const ServerInfoStatusPayloadSchema = z selectiveAgentTimeline: z.boolean().optional(), explicitEventSubscriptions: z.boolean().optional(), ownedSubscriptions: z.boolean().optional(), + binaryChunkingV1: z.boolean().optional(), workspaceSetupObservation: z.boolean().optional(), workspaceSortTimes: z.boolean().optional(), // COMPAT(canonicalSubmittedPrompts): added in v0.2.6, remove gate after 2027-01-30. diff --git a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts index b6765051f6..0b7bff0532 100644 --- a/packages/server/src/server/connected-hosts/hosts.e2e.test.ts +++ b/packages/server/src/server/connected-hosts/hosts.e2e.test.ts @@ -2091,3 +2091,34 @@ test.each(["removed", "disabled"] as const)( ).toEqual([]); }, ); + +test( + "a 50 MiB timeline crosses a routed host without dropping content or the route", + { timeout: 30000 }, + async () => { + const b = await daemon(); + const a = await daemon(); + const destination = await api(b); + const origin = await api(a); + const workspace = await destination.paseo.workspaces.create({ + source: { kind: "directory", path: b.staticDir }, + }); + const agent = await workspace.agents.create({ + config: { provider: "claude/claude-test", modeId: "full-access" }, + title: "Large history", + }); + const text = "x".repeat(50 * 1024 * 1024); + await b.daemon.agentManager.appendTimelineItem(agent.id, { type: "assistant_message", text }); + await origin.paseo.hosts.add({ connection: `127.0.0.1:${b.port}` }); + const serverId = (await online(origin.paseo)).hosts[0]!.serverId!; + const remote = await origin.paseo.hosts.client(serverId); + const response = await remote.agents.ref(agent.id).timeline.refetch({ limit: 20 }); + const messages = response.entries.flatMap((row) => + row.item.type === "assistant_message" ? [row.item.text] : [], + ); + expect(messages).toHaveLength(1); + expect(messages[0] === text).toBe(true); + expect((await remote.workspaces.list()).entries).toHaveLength(1); + await remote.dispose(); + }, +); diff --git a/packages/server/src/server/connected-hosts/link.ts b/packages/server/src/server/connected-hosts/link.ts index 85fe06ed4e..a18c89aeb0 100644 --- a/packages/server/src/server/connected-hosts/link.ts +++ b/packages/server/src/server/connected-hosts/link.ts @@ -1,7 +1,11 @@ import type { SessionDelivery } from "../session/owned-subscriptions/index.js"; import { EventEmitter } from "node:events"; import { createHash } from "node:crypto"; -import { encodeHostRouteFrame, type HostRouteFrame } from "@getpaseo/protocol/binary-frames/index"; +import { + MAX_HOST_CHANNEL_FRAME_BYTES, + encodeHostRouteFrame, + type HostRouteFrame, +} from "@getpaseo/protocol/binary-frames/index"; import type { DaemonPermission, SessionInboundMessage, @@ -10,6 +14,7 @@ import type { /** A logical socket has no network resources or authority of its own. */ export class VirtualHostSocket extends EventEmitter { + readonly maxFrameBytes = MAX_HOST_CHANNEL_FRAME_BYTES; readyState = 1; constructor( private readonly output: (data: string | Uint8Array) => Promise, diff --git a/packages/server/src/server/websocket-server.ts b/packages/server/src/server/websocket-server.ts index 96c81ab49a..bc86fa02a2 100644 --- a/packages/server/src/server/websocket-server.ts +++ b/packages/server/src/server/websocket-server.ts @@ -1,3 +1,4 @@ +import { ChunkedTransport } from "@getpaseo/protocol/chunked-transport"; import { type WorkspaceRuntimeStore } from "./workspace-runtime-store/index.js"; import { createAgentMessageDelivery, @@ -124,7 +125,6 @@ import { MAX_PHYSICAL_SOCKET_BUFFERED_BYTES, outboundFrameByteLength, physicalSocketHasCapacity, - sendBoundedPhysicalFrame, sendBoundedPhysicalFrameAndWait, } from "./websocket/physical-socket.js"; @@ -440,6 +440,7 @@ function bufferFromWsData(data: Buffer | ArrayBuffer | Buffer[] | string): Buffe } export interface WebSocketLike { + maxFrameBytes?: number; readyState: number; bufferedAmount?: number; send: ( @@ -1273,6 +1274,33 @@ export class VoiceAssistantWebSocketServer { return { sockets: writableSockets, payload, bytes: outboundFrameByteLength(payload) }; } + private negotiateChunking(ws: WebSocketLike, hello: WSHelloMessage): void { + this.chunkedSocket(ws).enabled = hello.capabilities?.[CLIENT_CAPS.binaryChunkingV1] === true; + } + + private readonly chunkedSockets = new WeakMap(); + + private chunkedSocket(ws: WebSocketLike): ChunkedTransport { + let transport = this.chunkedSockets.get(ws); + if (!transport) { + transport = new ChunkedTransport({ + frameBytes: ws.maxFrameBytes ?? 32 * 1024 * 1024, + send: async (frame) => { + const sent = await sendBoundedPhysicalFrameAndWait({ + socket: ws, + frame, + onHighWater: () => this.closeAtOutboundHighWater(ws), + }); + if (!sent) throw new Error("Source is closed"); + }, + deliver: (frame) => + this.handleRawMessage(ws, typeof frame === "string" ? frame : Buffer.from(frame)), + }); + this.chunkedSockets.set(ws, transport); + } + return transport; + } + private async sendMessageToClientAndWait( ws: WebSocketLike, message: WSOutboundMessage, @@ -1280,13 +1308,7 @@ export class VoiceAssistantWebSocketServer { const prepared = this.prepareJsonDelivery([ws], message, true); if (!prepared) throw new Error("Source message delivery unavailable"); try { - const sent = await sendBoundedPhysicalFrameAndWait({ - socket: ws, - frame: prepared.payload, - frameBytes: prepared.bytes, - onHighWater: () => this.closeAtOutboundHighWater(ws), - }); - if (!sent) throw new Error("Source is closed"); + await this.chunkedSocket(ws).send(prepared.payload); this.runtimeMetrics.recordOutboundMessage(message, ws.bufferedAmount); } catch (err) { this.logger.warn({ err }, "ws_send_failed"); @@ -1313,14 +1335,7 @@ export class VoiceAssistantWebSocketServer { return; } try { - const sent = await sendBoundedPhysicalFrameAndWait({ - socket: ws, - frame, - onHighWater: () => this.closeAtOutboundHighWater(ws), - }); - if (!sent) { - throw new Error("Physical WebSocket is not open"); - } + await this.chunkedSocket(ws).send(frame); this.runtimeMetrics.recordOutboundBinaryFrame(ws.bufferedAmount); } catch (err) { this.logger.warn({ err }, "ws_send_failed"); @@ -1331,17 +1346,16 @@ export class VoiceAssistantWebSocketServer { private sendFrameToClient( ws: WebSocketLike, frame: string | Uint8Array, - frameBytes: number, + _frameBytes: number, recordSent: () => void, ): void { try { - const sent = sendBoundedPhysicalFrame({ - socket: ws, - frame, - frameBytes, - onHighWater: () => this.closeAtOutboundHighWater(ws), - }); - if (sent) recordSent(); + void this.chunkedSocket(ws) + .send(frame) + .then(recordSent, (err) => { + this.logger.warn({ err }, "ws_send_failed"); + ws.close(1011, "Transport send failed"); + }); } catch (err) { this.logger.warn({ err }, "ws_send_failed"); } @@ -1688,6 +1702,7 @@ export class VoiceAssistantWebSocketServer { pending: PendingConnection; }): Promise { const { ws, message, pending } = params; + this.negotiateChunking(ws, message); if (message.protocolVersion < 1) { this.clearPendingConnection(ws); @@ -1952,6 +1967,7 @@ export class VoiceAssistantWebSocketServer { usageSetting: true, usageSources: true, ownedSubscriptions: true, + binaryChunkingV1: true, agentRequestReceipts: true, workspaceRequestReceipts: true, creationLifecycle: true, @@ -2165,10 +2181,15 @@ export class VoiceAssistantWebSocketServer { private bindSocketHandlers(ws: WebSocketLike): void { ws.on("message", (...args: unknown[]) => { const data = args[0] as Buffer | ArrayBuffer | Buffer[] | string; - this.handleRawMessage(ws, data); + try { + this.chunkedSocket(ws).receive(typeof data === "string" ? data : bufferFromWsData(data)); + } catch { + ws.close(1002, "Invalid chunked transport"); + } }); ws.on("close", async (...args: unknown[]) => { + this.chunkedSockets.get(ws)?.close(); const code = args[0]; const reason = args[1]; await this.detachSocket(ws, { diff --git a/packages/server/src/server/websocket/encrypted-relay-socket.test.ts b/packages/server/src/server/websocket/encrypted-relay-socket.test.ts index f2a880f15f..3e17f0cbb4 100644 --- a/packages/server/src/server/websocket/encrypted-relay-socket.test.ts +++ b/packages/server/src/server/websocket/encrypted-relay-socket.test.ts @@ -140,3 +140,74 @@ test("encrypted sockets do not double-count bytes already buffered by the transp expect(socket.bufferedAmount).toBe(payload.byteLength + 40); }); + +test("50 MiB logical text and routed binary cross encrypted relay frames in both directions", async () => { + const { ChunkedTransport } = await import("@getpaseo/protocol/chunked-transport"); + const { openClientSecureChannel, openDaemonSecureChannel } = await import("@getpaseo/relay/e2ee"); + const { generateKeyPair, exportPublicKey, maxSecuredPlaintextByteLength } = + await import("@getpaseo/relay"); + const { encodeHostRouteFrame, decodeHostRouteFrame } = + await import("@getpaseo/protocol/binary-frames/index"); + type Transport = import("@getpaseo/relay/e2ee").Transport; + const limit = 32 * 1024 * 1024; + const wireSizes: number[] = []; + const endpoint = (): Transport => ({ + send: () => {}, + close: () => {}, + onmessage: null, + onclose: null, + onerror: null, + }); + const left = endpoint(); + const right = endpoint(); + const wire = (destination: Transport) => (data: string | ArrayBuffer) => { + const size = typeof data === "string" ? Buffer.byteLength(data) : data.byteLength; + expect(size).toBeLessThanOrEqual(limit); + wireSizes.push(size); + queueMicrotask(() => destination.onmessage?.({ data, isBinary: typeof data !== "string" })); + }; + left.send = wire(right); + right.send = wire(left); + const key = generateKeyPair(); + const delivered: (string | Uint8Array)[][] = [[], []]; + let receivers: InstanceType[] = []; + const incoming = (index: number) => (data: string | ArrayBuffer) => + receivers[index]!.receive(typeof data === "string" ? data : new Uint8Array(data)); + const daemonPromise = openDaemonSecureChannel(right, { + daemonKeyPair: key, + events: { onmessage: incoming(1) }, + }); + const client = await openClientSecureChannel(left, { + daemonPublicKeyB64: exportPublicKey(key.publicKey), + events: { onmessage: incoming(0) }, + }); + const daemon = await daemonPromise; + receivers = [client, daemon].map( + (channel, index) => + new ChunkedTransport({ + frameBytes: maxSecuredPlaintextByteLength(limit), + send: (frame) => channel.send(typeof frame === "string" ? frame : frame.slice().buffer), + deliver: (frame) => { + delivered[index]!.push(frame); + }, + }), + ); + for (const receiver of receivers) receiver.enabled = true; + const text = "x".repeat(50 * 1024 * 1024); + wireSizes.length = 0; + await receivers[1]!.send(text); + await new Promise((resolve) => setImmediate(resolve)); + expect(delivered[0]).toEqual([text]); + expect(wireSizes.length).toBeGreaterThan(1); + const routed = encodeHostRouteFrame("route", new Uint8Array(50 * 1024 * 1024).fill(7)); + await receivers[0]!.send(routed); + await new Promise((resolve) => setImmediate(resolve)); + expect(delivered[1]).toHaveLength(1); + const received = decodeHostRouteFrame(delivered[1]![0] as Uint8Array)!; + expect(received.routeId).toBe("route"); + expect( + Buffer.from(received.payload).equals(Buffer.from(new Uint8Array(50 * 1024 * 1024).fill(7))), + ).toBe(true); + client.close(); + daemon.close(); +}, 30_000); diff --git a/packages/server/src/server/websocket/encrypted-relay-socket.ts b/packages/server/src/server/websocket/encrypted-relay-socket.ts index f11162761c..794789718c 100644 --- a/packages/server/src/server/websocket/encrypted-relay-socket.ts +++ b/packages/server/src/server/websocket/encrypted-relay-socket.ts @@ -1,3 +1,5 @@ +import { maxSecuredPlaintextByteLength } from "@getpaseo/relay"; +import { MAX_RELAY_WEBSOCKET_MESSAGE_BYTES } from "@getpaseo/protocol/websocket-message-limit"; import { EventEmitter } from "node:events"; import { MAX_PHYSICAL_SOCKET_BUFFERED_BYTES } from "./physical-socket.js"; @@ -8,6 +10,7 @@ export interface EncryptedRelayChannel { } export interface EncryptedRelaySocket { + readonly maxFrameBytes: number; readonly readyState: number; readonly bufferedAmount: number; send: (data: string | Uint8Array | ArrayBuffer) => void | Promise; @@ -43,6 +46,7 @@ export function createEncryptedRelaySocket(params: { }); return { + maxFrameBytes: maxSecuredPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES), get readyState() { return readyState; },