From b044201f6294bff78a81870bb6c3c7304ef34605 Mon Sep 17 00:00:00 2001 From: Mohamed Boudra Date: Fri, 30 Oct 2026 17:41:57 +0100 Subject: [PATCH] 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 | 5 +- .../src/daemon-client-relay-e2ee-transport.ts | 1 + .../src/daemon-client-transport-types.ts | 13 +- .../src/daemon-client-transport.test.ts | 15 ++ .../src/daemon-client-websocket-transport.ts | 4 + packages/client/src/daemon-client.test.ts | 118 ++++++++++++ packages/client/src/daemon-client.ts | 71 ++++--- packages/client/src/hosts/index.ts | 3 + packages/client/src/node/index.ts | 3 + .../protocol/src/chunked-transport.test.ts | 22 +++ packages/protocol/src/chunked-transport.ts | 4 +- .../daemon-e2e/relay-transport.e2e.test.ts | 182 +++++++++--------- 12 files changed, 319 insertions(+), 122 deletions(-) diff --git a/docs/protocol-compatibility.md b/docs/protocol-compatibility.md index 4d719bc15f..74650255a5 100644 --- a/docs/protocol-compatibility.md +++ b/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 diff --git a/packages/client/src/daemon-client-relay-e2ee-transport.ts b/packages/client/src/daemon-client-relay-e2ee-transport.ts index 27081046d0..0faecdcb5e 100644 --- a/packages/client/src/daemon-client-relay-e2ee-transport.ts +++ b/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) { diff --git a/packages/client/src/daemon-client-transport-types.ts b/packages/client/src/daemon-client-transport-types.ts index 76afb52094..64f84cf8c1 100644 --- a/packages/client/src/daemon-client-transport-types.ts +++ b/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; 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; protocols?: string[] }, -) => WebSocketLike; +export interface WebSocketFactory { + ( + url: string, + options?: { headers?: Record; protocols?: string[] }, + ): WebSocketLike; + supportsDrain?: boolean; +} export interface WebSocketLike { bufferedAmount?: number; diff --git a/packages/client/src/daemon-client-transport.test.ts b/packages/client/src/daemon-client-transport.test.ts index 963a8b541c..c95c534129 100644 --- a/packages/client/src/daemon-client-transport.test.ts +++ b/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); +}); diff --git a/packages/client/src/daemon-client-websocket-transport.ts b/packages/client/src/daemon-client-websocket-transport.ts index f0667720f0..c7ec87555d 100644 --- a/packages/client/src/daemon-client-websocket-transport.ts +++ b/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})`); diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 1d6da7fb45..6ef72a0e1a 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/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((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((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((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; + 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(); +}); diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index a83b3b89b8..9152159f1c 100644 --- a/packages/client/src/daemon-client.ts +++ b/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 { 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 { - 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 { @@ -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) { diff --git a/packages/client/src/hosts/index.ts b/packages/client/src/hosts/index.ts index b1c6451365..aa5c910158 100644 --- a/packages/client/src/hosts/index.ts +++ b/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) diff --git a/packages/client/src/node/index.ts b/packages/client/src/node/index.ts index 2516946a07..30d2b8744f 100644 --- a/packages/client/src/node/index.ts +++ b/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; }, diff --git a/packages/protocol/src/chunked-transport.test.ts b/packages/protocol/src/chunked-transport.test.ts index 696c0507a0..3979b48885 100644 --- a/packages/protocol/src/chunked-transport.test.ts +++ b/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"); +}); diff --git a/packages/protocol/src/chunked-transport.ts b/packages/protocol/src/chunked-transport.ts index 6bfc8aabfc..a32d68424f 100644 --- a/packages/protocol/src/chunked-transport.ts +++ b/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; deliver: (frame: Frame) => void; onFailure?: (error: unknown) => void; @@ -41,7 +42,8 @@ export class ChunkedTransport { send(frame: Frame, options: { waitForDrain?: boolean; byteLength?: number } = {}): Promise { 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"); diff --git a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts index 0670bd2898..f47accaa55 100644 --- a/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts +++ b/packages/server/src/server/daemon-e2e/relay-transport.e2e.test.ts @@ -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";