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.
2026-10-30 21:23:04 +00:00 · 2026-10-30 15:28:41 +01:00
parent fbe24fcdb6
commit 8c2a2a9a88
20 changed files with 608 additions and 110 deletions
@@ -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
@@ -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,
@@ -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) => {
@@ -1,5 +1,6 @@
 export interface DaemonTransport {
-  send: (data: string | Uint8Array | ArrayBuffer) => void;
+  maxFrameBytes?: number;
+  send: (data: string | Uint8Array | ArrayBuffer) => void | Promise<void>;
  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;
@@ -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<void>((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
@@ -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,
+  );
+});
@@ -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;
@@ -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`), {
@@ -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<void>((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<void>)[] = [];
  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<void>((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(
@@ -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) {
@@ -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;
@@ -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<void>((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"]);
+});
@@ -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<void> | undefined;
+
+  constructor(
+    private readonly io: {
+      frameBytes: number;
+      send: (frame: Frame) => void | Promise<void>;
+      deliver: (frame: Frame) => void;
+    },
+  ) {
+    if (io.frameBytes <= HEADER) throw new Error("Invalid transport frame budget");
+  }
+
+  send(frame: Frame): Promise<void> {
+    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<void> {
+    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;
+  }
+}
@@ -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",
@@ -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.
@@ -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();
+  },
+);
@@ -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<void>,
@@ -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<WebSocketLike, ChunkedTransport>();
+
+  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<void> {
    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, {
@@ -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<typeof ChunkedTransport>[] = [];
+  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<void>((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<void>((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);
@@ -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<void>;
@@ -43,6 +46,7 @@ export function createEncryptedRelaySocket(params: {
  });

  return {
+    maxFrameBytes: maxSecuredPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES),
    get readyState() {
      return readyState;
    },