fix(transport): preserve native admission and pace file writes

Sockets without a drain signal retain single-frame outbound admission while accepting chunked responses. Uploads and response-bearing control sends must observe queued admission failures instead of waiting for a response that cannot arrive.

docs/protocol-compatibility.md

@@ -75,8 +75,9 @@ The framing module never selects an encryption construction.
 Native uploads retain the platform's existing queue limits. React Native's
 WebSocket does not report its queued bytes, so the client cannot wait for native
 drain. Android's installed OkHttp transport closes when a send would put more
-than 16 MiB in its queue. Negotiating chunking does not establish large native
-upload support; daemon-to-native responses still use the daemon's paced writes.
+than 16 MiB in its queue. The native transport disables outbound chunking, including
+on routed channels, and retains its existing single-frame admission errors.
+Inbound chunking remains enabled; daemon-to-native responses use the daemon's paced writes.
 
 ## Owned observations
 

packages/client/src/daemon-client-relay-e2ee-transport.ts

@@ -224,6 +224,7 @@ export function createEncryptedTransport(
   });
 
   return {
+    supportsDrain: base.supportsDrain,
     maxFrameBytes: maxSecuredPlaintextByteLength(MAX_RELAY_WEBSOCKET_MESSAGE_BYTES),
     send: (data) => {
       if (!channel) {

packages/client/src/daemon-client-transport-types.ts

@@ -1,4 +1,6 @@
 export interface DaemonTransport {
+  /** False when the underlying socket cannot report write drain. */
+  supportsDrain?: boolean;
   maxFrameBytes?: number;
   send: (data: string | Uint8Array | ArrayBuffer) => void | Promise<void>;
   close: (code?: number, reason?: string) => void;
@@ -14,10 +16,13 @@ export type DaemonTransportFactory = (options: {
   protocols?: string[];
 }) => DaemonTransport;
 
-export type WebSocketFactory = (
-  url: string,
-  options?: { headers?: Record<string, string>; protocols?: string[] },
-) => WebSocketLike;
+export interface WebSocketFactory {
+  (
+    url: string,
+    options?: { headers?: Record<string, string>; protocols?: string[] },
+  ): WebSocketLike;
+  supportsDrain?: boolean;
+}
 
 export interface WebSocketLike {
   bufferedAmount?: number;

packages/client/src/daemon-client-transport.test.ts

@@ -175,3 +175,18 @@ describe("daemon-client transport helpers", () => {
     expect(decodeMessageData(bytes)).toBe("hello");
   });
 });
+
+test("native drain capability survives WebSocket and encryption wrappers", async () => {
+  const { nativeWebSocketFactory } = await import("./daemon-client-websocket-transport.js");
+  expect(nativeWebSocketFactory.supportsDrain).toBe(false);
+  const factory = Object.assign(() => ({ readyState: 1, send() {}, close() {} }), {
+    supportsDrain: false,
+  });
+  const base = createWebSocketTransportFactory(factory)({ url: "ws://test" });
+  expect(base.supportsDrain).toBe(false);
+  const encrypted = createEncryptedTransport(base, {
+    daemonPublicKeyB64: "unused-key",
+    logger: { warn() {} },
+  });
+  expect(encrypted.supportsDrain).toBe(false);
+});

packages/client/src/daemon-client-websocket-transport.ts

@@ -39,6 +39,9 @@ export function nativeWebSocketFactory(
   return new (getGlobalWebSocket())(url, options?.protocols, { headers: options?.headers });
 }
 
+// React Native leaves bufferedAmount at zero, including while native output is queued.
+nativeWebSocketFactory.supportsDrain = false;
+
 export function createWebSocketTransportFactory(factory: WebSocketFactory): DaemonTransportFactory {
   return ({ url, headers, protocols }) => {
     const ws = factory(url, { headers, protocols });
@@ -50,6 +53,7 @@ export function createWebSocketTransportFactory(factory: WebSocketFactory): Daem
       }
     }
     return {
+      supportsDrain: factory.supportsDrain !== false,
       send: (data) => {
         if (typeof ws.readyState === "number" && ws.readyState !== 1) {
           throw new Error(`WebSocket not open (readyState=${ws.readyState})`);

packages/client/src/daemon-client.test.ts

@@ -8664,3 +8664,121 @@ test("malformed chunk metadata still closes the client transport", async () => {
   mock.triggerMessage(new Uint8Array([0x31, 2, 0, 3, 65]));
   expect(closed).toHaveBeenCalledWith(4003, "Invalid chunked transport");
 });
+
+test.each([FileTransferOpcode.FileBegin, FileTransferOpcode.FileChunk, FileTransferOpcode.FileEnd])(
+  "upload stops at a queued pre-write rejection (opcode=%s)",
+  async (opcode) => {
+    const mock = createMockTransport();
+    const send = mock.transport.send;
+    let drain!: () => void;
+    let blocked = false;
+    const attempts: number[] = [];
+    mock.transport.send = (frame) => {
+      if (typeof frame !== "string") {
+        const decoded = decodeFileTransferFrame(assertUint8Array(frame));
+        if (decoded) {
+          attempts.push(decoded.opcode);
+          if (decoded.opcode === opcode) throw new Error("upload capacity");
+        } else if (blocked)
+          return new Promise<void>((resolve) => {
+            drain = resolve;
+          });
+      }
+      return send(frame);
+    };
+    const client = new DaemonClient({
+      url: "ws://test",
+      clientId: "upload-admission",
+      transportFactory: () => mock.transport,
+      reconnect: { enabled: false },
+    });
+    clients.push(client);
+    const connecting = client.connect();
+    mock.triggerOpen();
+    await connecting;
+    blocked = true;
+    const first = client.sendHostFrame("route", new Uint8Array([1]));
+    const upload = client.uploadFile({
+      fileName: "file.bin",
+      mimeType: "application/octet-stream",
+      bytes: new Uint8Array([1, 2]),
+      chunkSize: 1,
+    });
+    let rejected: unknown;
+    void upload.catch((error) => {
+      rejected = error;
+    });
+    await new Promise<void>((resolve) => setTimeout(resolve, 0));
+    drain();
+    await first;
+    await vi.waitFor(() => expect(rejected).toMatchObject({ message: "upload capacity" }), {
+      timeout: 500,
+    });
+    expect(attempts.at(-1)).toBe(opcode);
+    expect(client.isConnected).toBe(true);
+  },
+);
+
+test.each(["latency", "dictation start", "dictation finish"])(
+  "queued control send rejection reaches its caller (%s)",
+  async (operation) => {
+    const mock = createMockTransport();
+    const send = mock.transport.send;
+    let blocked = false;
+    let drain!: () => void;
+    mock.transport.send = (frame) => {
+      if (!blocked) return send(frame);
+      if (typeof frame !== "string")
+        return new Promise<void>((resolve) => {
+          drain = resolve;
+        });
+      throw new Error("control capacity");
+    };
+    const client = new DaemonClient({
+      url: "ws://test",
+      clientId: "control-admission",
+      transportFactory: () => mock.transport,
+      reconnect: { enabled: false },
+    });
+    clients.push(client);
+    const connecting = client.connect();
+    mock.triggerOpen();
+    await connecting;
+    blocked = true;
+    const first = client.sendHostFrame("route", new Uint8Array([1]));
+    let result: Promise<unknown>;
+    if (operation === "latency") result = client.measureLatency({ timeoutMs: 1000 });
+    else if (operation === "dictation start")
+      result = client.startDictationStream("dictation", "pcm");
+    else result = client.finishDictationStream("dictation", 0);
+    const rejected = expect(result).rejects.toThrow("control capacity");
+    drain();
+    await first;
+    await rejected;
+    expect(client.isConnected).toBe(true);
+  },
+);
+
+test("routed channels inherit an origin without drain support", async () => {
+  const { createHostChannelTransport } = await import("./hosts/index.js");
+  const mock = createMockTransport();
+  mock.transport.supportsDrain = false;
+  const client = new DaemonClient({
+    url: "ws://test",
+    clientId: "unpaced-route",
+    transportFactory: () => mock.transport,
+    reconnect: { enabled: false },
+  });
+  clients.push(client);
+  const connecting = client.connect();
+  mock.triggerOpen();
+  await connecting;
+  const route = createHostChannelTransport(
+    client,
+    "srv_destination",
+    "scope",
+    "route",
+  )({ url: "ws://route" });
+  expect(route.supportsDrain).toBe(false);
+  route.close();
+});

packages/client/src/daemon-client.ts

@@ -1566,6 +1566,9 @@ export class DaemonClient {
       this.traceConnectionAttempt += 1;
       this.transport = transport;
       this.chunkedTransport = new ChunkedTransport({
+        get supportsDrain() {
+          return transport.supportsDrain;
+        },
         get frameBytes() {
           return transport.maxFrameBytes ?? 32 * 1024 * 1024;
         },
@@ -1957,6 +1960,11 @@ export class DaemonClient {
   // Message Subscription
   // ============================================================================
 
+  /** Drain capability inherited by transports routed through this connection. */
+  get supportsDrain(): boolean {
+    return this.transport?.supportsDrain !== false;
+  }
+
   subscribe(handler: DaemonEventHandler): () => void {
     this.eventListeners.add(handler);
     return () => {
@@ -2140,10 +2148,10 @@ export class DaemonClient {
     }
   }
 
-  private sendBinaryFrame(frame: Uint8Array): void {
+  private sendBinaryFrame(frame: Uint8Array): Promise<void> {
     if (!this.transport || this.connectionState.status !== "connected") {
       if (this.config.suppressSendErrors) {
-        return;
+        return Promise.resolve();
       }
       throw new Error(`Transport not connected (status: ${this.connectionState.status})`);
     }
@@ -2152,10 +2160,10 @@ export class DaemonClient {
         envelopeType: "binary",
         messageType: "binary",
       });
-      this.sendTransportFrame(frame);
+      return this.sendTransportFrame(frame);
     } catch (error) {
       if (this.config.suppressSendErrors) {
-        return;
+        return Promise.resolve();
       }
       throw error instanceof Error ? error : new Error(String(error));
     }
@@ -2583,15 +2591,16 @@ export class DaemonClient {
     };
     this.pingProbe = probe;
 
-    try {
-      this.sendJsonMessage("ping", "ping", { type: "ping" });
-    } catch (error) {
-      this.clearPingProbe();
+    const rejectSend = (error: unknown) => {
+      if (this.pingProbe !== probe) return;
       const sendError = error instanceof Error ? error : new Error(String(error));
-      if (probe.drivesLivenessFailure) {
-        this.recordLivenessFailure(sendError);
-      }
-      return Promise.reject(sendError);
+      this.rejectPingProbe(sendError);
+      if (probe.drivesLivenessFailure) this.recordLivenessFailure(sendError);
+    };
+    try {
+      void this.sendJsonMessage("ping", "ping", { type: "ping" }).catch(rejectSend);
+    } catch (error) {
+      rejectSend(error);
     }
 
     return promise;
@@ -4246,7 +4255,7 @@ export class DaemonClient {
   }
 
   async sendVoiceAudioChunk(audio: string, format: string, isLast = false): Promise<void> {
-    this.sendSessionMessage({ type: "voice_audio_chunk", audio, format, isLast });
+    return this.sendSessionMessageStrict({ type: "voice_audio_chunk", audio, format, isLast });
   }
 
   async startDictationStream(dictationId: string, format: string): Promise<void> {
@@ -4287,8 +4296,13 @@ export class DaemonClient {
 
     const cleanupError = new Error("Cancelled dictation start waiter");
     try {
-      this.sendSessionMessageStrict({ type: "dictation_stream_start", dictationId, format });
-      await Promise.race([ackPromise, errorPromise]);
+      const sent = this.sendSessionMessageStrict({
+        type: "dictation_stream_start",
+        dictationId,
+        format,
+      });
+      const response = Promise.race([ackPromise, errorPromise]);
+      await Promise.race([response, sent.then(() => response)]);
     } finally {
       ack.cancel(cleanupError);
       streamError.cancel(cleanupError);
@@ -4423,12 +4437,17 @@ export class DaemonClient {
 
     const cleanupError = new Error("Cancelled dictation finish waiter");
     try {
-      this.sendSessionMessageStrict({ type: "dictation_stream_finish", dictationId, finalSeq });
-      const firstOutcome = await Promise.race([
+      const sent = this.sendSessionMessageStrict({
+        type: "dictation_stream_finish",
+        dictationId,
+        finalSeq,
+      });
+      const response = Promise.race([
         finalOutcomePromise,
         errorOutcomePromise,
         finishAcceptedOutcomePromise,
       ]);
+      const firstOutcome = await Promise.race([response, sent.then(() => response)]);
 
       if (firstOutcome.kind === "final") {
         return firstOutcome.payload;
@@ -5344,8 +5363,10 @@ export class DaemonClient {
         return undefined;
       },
     );
+    const sendFrame = (frame: Uint8Array) =>
+      Promise.race([this.sendBinaryFrame(frame), responsePromise]);
     try {
-      this.sendBinaryFrame(
+      await sendFrame(
         encodeFileTransferFrame({
           opcode: FileTransferOpcode.FileBegin,
           requestId: resolvedRequestId,
@@ -5359,6 +5380,7 @@ export class DaemonClient {
         }),
       );
 
+      if (settled) return await responsePromise;
       const chunkSize = input.chunkSize ?? 128 * 1024;
       for (let offset = 0; offset < bytes.byteLength; offset += chunkSize) {
         // Native WebSocket.send encodes binary synchronously. Let rendering and
@@ -5368,7 +5390,7 @@ export class DaemonClient {
         if (this.transport !== uploadTransport || this.connectionState.status !== "connected") {
           throw new DaemonConnectionError("Connection changed during file upload");
         }
-        this.sendBinaryFrame(
+        await sendFrame(
           encodeFileTransferFrame({
             opcode: FileTransferOpcode.FileChunk,
             requestId: resolvedRequestId,
@@ -5377,7 +5399,7 @@ export class DaemonClient {
         );
       }
 
-      this.sendBinaryFrame(
+      await sendFrame(
         encodeFileTransferFrame({
           opcode: FileTransferOpcode.FileEnd,
           requestId: resolvedRequestId,
@@ -7258,15 +7280,6 @@ export class DaemonClient {
     probe.resolve(perfNow() - probe.startedAt);
   }
 
-  private clearPingProbe(): void {
-    const probe = this.pingProbe;
-    if (!probe) {
-      return;
-    }
-    this.pingProbe = null;
-    clearTimeout(probe.timeoutHandle);
-  }
-
   private rejectPingProbe(error: Error): void {
     const probe = this.pingProbe;
     if (!probe) {

packages/client/src/hosts/index.ts

@@ -85,6 +85,9 @@ export function createHostChannelTransport(
       })
       .catch(close);
     return {
+      get supportsDrain() {
+        return origin.supportsDrain;
+      },
       maxFrameBytes: MAX_HOST_CHANNEL_FRAME_BYTES,
       send(data) {
         if (closed || !routeId)

packages/client/src/node/index.ts

@@ -172,6 +172,9 @@ export function createHostTransportFactory(
       close();
     });
     return {
+      get supportsDrain() {
+        return transport?.supportsDrain;
+      },
       get maxFrameBytes() {
         return transport?.maxFrameBytes;
       },

packages/protocol/src/chunked-transport.test.ts

@@ -368,3 +368,25 @@ it.each([false, true])("delivery errors leave framing usable (chunked=%s)", asyn
   expect(delivered).toEqual(["A"]);
   await transport.send("still connected");
 });
+
+it("unpaced transports retain ordinary outbound admission and accept inbound chunks", async () => {
+  const frames: unknown[] = [];
+  const delivered: unknown[] = [];
+  const transport = new ChunkedTransport({
+    frameBytes: 8,
+    supportsDrain: false,
+    send: (frame) => {
+      frames.push(frame);
+      if (typeof frame === "string" && frame.length > 8) throw new Error("single frame limit");
+    },
+    deliver: (frame) => {
+      delivered.push(frame);
+    },
+  });
+  transport.enabled = true;
+  expect(() => transport.send("too large for one frame")).toThrow("single frame limit");
+  expect(frames).toEqual(["too large for one frame"]);
+  transport.receive(new Uint8Array([0x31, 1, 0, 3, 65]));
+  expect(delivered).toEqual(["A"]);
+  await transport.send("small");
+});

packages/protocol/src/chunked-transport.ts

@@ -22,6 +22,7 @@ export class ChunkedTransport {
   constructor(
     private readonly io: {
       frameBytes: number;
+      supportsDrain?: boolean;
       send: (frame: Frame, waitForDrain: boolean) => void | Promise<void>;
       deliver: (frame: Frame) => void;
       onFailure?: (error: unknown) => void;
@@ -41,7 +42,8 @@ export class ChunkedTransport {
   send(frame: Frame, options: { waitForDrain?: boolean; byteLength?: number } = {}): Promise<void> {
     const byteLength = options.byteLength ?? frameByteLength(frame);
     if (byteLength > MAX_WEBSOCKET_MESSAGE_BYTES) throw new Error("Logical message too large");
-    const chunk = this.enabled && byteLength > this.io.frameBytes;
+    const chunk =
+      this.enabled && this.io.supportsDrain !== false && byteLength > this.io.frameBytes;
     if (this.closed) throw new Error("Transport closed");
     const write = () => {
       if (this.closed) throw new Error("Transport closed");

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

@@ -9,7 +9,7 @@ import {
 } from "@getpaseo/protocol/binary-frames/index";
 import {
   getDaemonAuthFailureReason,
-  type DaemonClient as HostClient,
+  DaemonClient as HostClient,
 } from "@getpaseo/client/internal/daemon-client";
 import type { SessionOutboundMessage } from "@getpaseo/protocol/messages";
 import { describe, expect, test } from "vitest";
@@ -593,41 +593,47 @@ async function waitForCapturedLog(
     }
   }, 90000);
 
-  test("large messages cross the relay in both directions and logical oversize rejects without disconnecting", async () => {
-    await startRelay({ useLocalRelay: true });
-    const daemon = await createTestPaseoDaemon({
-      relayEnabled: true,
-      relayEndpoint: `127.0.0.1:${relayPort}`,
-      relayUseTls: false,
-      relayPublicUseTls: false,
-    });
-    let client: DaemonClient | null = null;
-    try {
-      const offerUrl = await getPairingOfferUrl({
-        paseoHome: daemon.paseoHome,
+  test.each([true, false])(
+    "large messages cross the relay with unpaced upload admission preserved (supportsDrain=%s)",
+    async (supportsDrain) => {
+      await startRelay({ useLocalRelay: true });
+      const daemon = await createTestPaseoDaemon({
         relayEnabled: true,
-        relayEndpoint: daemon.config.relayEndpoint,
+        relayEndpoint: `127.0.0.1:${relayPort}`,
+        relayUseTls: false,
+        relayPublicUseTls: false,
       });
-      const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
-      client = new DaemonClient({
-        url: buildRelayWebSocketUrl({
-          endpoint: `127.0.0.1:${relayPort}`,
-          useTls: false,
-          serverId,
-          role: "client",
-        }),
-        e2ee: { enabled: true, daemonPublicKeyB64 },
-        reconnect: { baseDelayMs: 100, maxDelayMs: 100 },
-      });
-      await client.connect();
-      const statuses: string[] = [];
-      client.subscribeConnectionStatus((state) => statuses.push(state.status));
-      for (const imageMiB of [30, 61]) {
-        const images = [
-          { data: "A".repeat(4 * Math.ceil((imageMiB * 1024 * 1024) / 3)), mimeType: "image/png" },
-        ];
-        await withTimeout(
-          client.createWorkspace({
+      let client: HostClient | null = null;
+      try {
+        const offerUrl = await getPairingOfferUrl({
+          paseoHome: daemon.paseoHome,
+          relayEnabled: true,
+          relayEndpoint: daemon.config.relayEndpoint,
+        });
+        const { serverId, daemonPublicKeyB64 } = decodeOfferFromFragmentUrl(offerUrl);
+        client = new HostClient({
+          clientId: "paced-relay-regression",
+          webSocketFactory: Object.assign((url: string) => new WebSocket(url), { supportsDrain }),
+          url: buildRelayWebSocketUrl({
+            endpoint: `127.0.0.1:${relayPort}`,
+            useTls: false,
+            serverId,
+            role: "client",
+          }),
+          e2ee: { enabled: true, daemonPublicKeyB64 },
+          reconnect: { baseDelayMs: 100, maxDelayMs: 100 },
+        });
+        await client.connect();
+        const statuses: string[] = [];
+        client.subscribeConnectionStatus((state) => statuses.push(state.status));
+        for (const imageMiB of [30, 61]) {
+          const images = [
+            {
+              data: "A".repeat(4 * Math.ceil((imageMiB * 1024 * 1024) / 3)),
+              mimeType: "image/png",
+            },
+          ];
+          const creation = client.createWorkspace({
             source: { kind: "directory", path: daemon.staticDir },
             agent: {
               provider: "claude",
@@ -635,60 +641,64 @@ async function waitForCapturedLog(
               initialPrompt: "Describe these screens.",
               images,
             },
+          });
+          if (supportsDrain)
+            await withTimeout(creation, 30000, "Chunked workspace creation did not settle");
+          else await expect(creation).rejects.toThrow(/too large.*32\.0 MiB/);
+        }
+        expect((await client.fetchAgents()).entries).toHaveLength(supportsDrain ? 2 : 0);
+        await expect(
+          client.createWorkspace({
+            source: { kind: "directory", path: daemon.staticDir },
+            agent: {
+              provider: "claude",
+              cwd: daemon.staticDir,
+              initialPrompt: "Describe these screens.",
+              images: [{ data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }],
+            },
           }),
-          30000,
-          "Chunked workspace creation did not settle",
-        );
-      }
-      expect((await client.fetchAgents()).entries).toHaveLength(2);
-      await expect(
-        client.createWorkspace({
-          source: { kind: "directory", path: daemon.staticDir },
-          agent: {
-            provider: "claude",
-            cwd: daemon.staticDir,
-            initialPrompt: "Describe these screens.",
-            images: [{ data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }],
-          },
-        }),
-      ).rejects.toThrow(/too large.*100\.0 MiB/);
-      expect((await client.fetchAgents()).entries).toHaveLength(2);
-      const agent = await client.createAgent({ provider: "claude", cwd: daemon.staticDir });
-      await client.sendMessage(agent.id, "Describe these screens.", {
-        images: [{ data: "A".repeat(40 * 1024 * 1024), mimeType: "image/png" }],
-      });
-      await expect(
-        client.sendMessage(agent.id, "Describe these screens.", {
-          images: [{ data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }],
-        }),
-      ).rejects.toThrow(/too large.*100\.0 MiB/);
-      await client.sendMessage(agent.id, "A small follow-up still works.");
-      expect((await client.fetchAgents()).entries).toHaveLength(3);
-      // Respect the per-item content budget while exceeding the relay frame limit.
-      const text = "x".repeat(240 * 1024);
-      for (let row = 0; row < 220; row++)
-        await daemon.daemon.agentManager.appendTimelineItem(agent.id, {
-          type: "user_message",
-          text,
+        ).rejects.toThrow(/too large.*100\.0 MiB/);
+        expect((await client.fetchAgents()).entries).toHaveLength(supportsDrain ? 2 : 0);
+        const agent = await client.createAgent({ provider: "claude", cwd: daemon.staticDir });
+        const attachment = client.sendMessage(agent.id, "Describe these screens.", {
+          images: [{ data: "A".repeat(40 * 1024 * 1024), mimeType: "image/png" }],
         });
-      const timeline = await client.fetchAgentTimeline(agent.id, {
-        projection: "canonical",
-        limit: 0,
-        timeout: 30000,
-      });
-      const delivered = timeline.entries.filter(
-        (row) => row.item.type === "user_message" && row.item.text === text,
-      );
-      expect(delivered).toHaveLength(220);
-      expect(delivered.length * Buffer.byteLength(text)).toBeGreaterThan(50 * 1024 * 1024);
-      await client.ping();
-      expect(statuses).toEqual(["connected"]);
-    } finally {
-      await client?.close();
-      await daemon.close();
-      await stopRelay();
-    }
-  }, 120000);
+        if (supportsDrain) await attachment;
+        else await expect(attachment).rejects.toThrow(/too large.*32\.0 MiB/);
+        await expect(
+          client.sendMessage(agent.id, "Describe these screens.", {
+            images: [{ data: "A".repeat(101 * 1024 * 1024), mimeType: "image/png" }],
+          }),
+        ).rejects.toThrow(/too large.*100\.0 MiB/);
+        await client.sendMessage(agent.id, "A small follow-up still works.");
+        expect((await client.fetchAgents()).entries).toHaveLength(supportsDrain ? 3 : 1);
+        // Respect the per-item content budget while exceeding the relay frame limit.
+        const text = "x".repeat(240 * 1024);
+        for (let row = 0; row < 220; row++)
+          await daemon.daemon.agentManager.appendTimelineItem(agent.id, {
+            type: "user_message",
+            text,
+          });
+        const timeline = await client.fetchAgentTimeline(agent.id, {
+          projection: "canonical",
+          limit: 0,
+          timeout: 30000,
+        });
+        const delivered = timeline.entries.filter(
+          (row) => row.item.type === "user_message" && row.item.text === text,
+        );
+        expect(delivered).toHaveLength(220);
+        expect(delivered.length * Buffer.byteLength(text)).toBeGreaterThan(50 * 1024 * 1024);
+        await client.ping();
+        expect(statuses).toEqual(["connected"]);
+      } finally {
+        await client?.close();
+        await daemon.close();
+        await stopRelay();
+      }
+    },
+    120000,
+  );
 
   test("daemon connects to relay and client ping/pong works through relay", async () => {
     process.env.PASEO_PRIMARY_LAN_IP = "192.168.1.12";