diff --git a/packages/client/src/daemon-client.test.ts b/packages/client/src/daemon-client.test.ts index 9e23904c9f..dbe3ceef64 100644 --- a/packages/client/src/daemon-client.test.ts +++ b/packages/client/src/daemon-client.test.ts @@ -6906,6 +6906,70 @@ test("creation reconnect observation uses connection-owned subscriptions and rel expect(phases).toEqual(["accepted", "failed"]); }); +test("uploadFile started while connecting waits for the connection and uploads", async () => { + const mock = createMockTransport(); + const client = new DaemonClient({ + url: "ws://test", + clientId: "upload-while-connecting", + logger: createMockLogger(), + reconnect: { enabled: false }, + suppressSendErrors: true, + transportFactory: () => mock.transport, + }); + clients.push(client); + const connection = client.connect(); + expect(client.getConnectionState().status).toBe("connecting"); + + const upload = client.uploadFile({ + fileName: "notes.txt", + mimeType: "text/plain", + bytes: new TextEncoder().encode("hello world"), + modifiedAt: "2026-05-02T00:00:00.000Z", + requestId: "req-upload-connecting", + chunkSize: 5, + }); + void upload.catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 0)); + mock.triggerOpen(); + await connection; + + await vi.waitFor(() => { + const frames = mock.sent + .filter((frame) => typeof frame !== "string") + .map(assertUint8Array) + .map(decodeFileTransferFrame); + expect(frames.map((frame) => frame.opcode)).toEqual([ + FileTransferOpcode.FileBegin, + FileTransferOpcode.FileChunk, + FileTransferOpcode.FileChunk, + FileTransferOpcode.FileChunk, + FileTransferOpcode.FileEnd, + ]); + }); + const requestIndex = mock.sent.findIndex( + (frame) => + typeof frame === "string" && parseSentFrame(frame).requestId === "req-upload-connecting", + ); + expect(parseSentFrame(mock.sent[requestIndex])).toMatchObject({ type: "file.upload.request" }); + expect(requestIndex).toBeLessThan(mock.sent.findIndex((frame) => typeof frame !== "string")); + + const file = { + type: "uploaded_file" as const, + id: "upload_req-upload-connecting", + fileName: "notes.txt", + mimeType: "text/plain", + size: 11, + path: "/tmp/paseo-uploads/upload_req-upload-connecting/notes.txt", + }; + mock.triggerMessage( + wrapSessionMessage({ + type: "file.upload.response", + payload: { requestId: "req-upload-connecting", file, error: null }, + }), + ); + await expect(upload).resolves.toEqual({ requestId: "req-upload-connecting", file, error: null }); +}); + test("uploadFile stops sending chunks when the connection closes between sends", async () => { const mock = createMockTransport(); const client = new DaemonClient({ diff --git a/packages/client/src/daemon-client.ts b/packages/client/src/daemon-client.ts index c691898267..23664188b7 100644 --- a/packages/client/src/daemon-client.ts +++ b/packages/client/src/daemon-client.ts @@ -1169,7 +1169,7 @@ function toReasonCode(reason: string | null | undefined): string | null { } interface PendingSend { - message: SessionInboundMessage; + send: () => void; resolve: () => void; reject: (error: Error) => void; timeoutHandle: ReturnType; @@ -1797,12 +1797,23 @@ export class DaemonClient { * This prevents waiters from hanging forever when called during connection. */ private sendSessionMessageOrThrow(message: SessionInboundMessage): Promise { + return this.sendWhenConnected(() => { + const payload = SessionInboundMessageSchema.parse(message); + this.sendJsonMessage("session", payload.type, { type: "session", message: payload }); + }); + } + + /** Resolves once connected, waiting out a connection that is still being established. */ + private whenConnected(): Promise { + return this.sendWhenConnected(() => undefined); + } + + private sendWhenConnected(send: () => void): Promise { const status = this.connectionState.status; // If connected, send immediately if (this.transport && status === "connected") { - const payload = SessionInboundMessageSchema.parse(message); - this.sendJsonMessage("session", payload.type, { type: "session", message: payload }); + send(); return Promise.resolve(); } @@ -1823,7 +1834,7 @@ export class DaemonClient { ); }, DEFAULT_SEND_QUEUE_TIMEOUT_MS); - this.pendingSendQueue.push({ message, resolve, reject, timeoutHandle }); + this.pendingSendQueue.push({ send, resolve, reject, timeoutHandle }); }); } @@ -1842,8 +1853,7 @@ export class DaemonClient { clearTimeout(pending.timeoutHandle); try { if (this.transport && this.connectionState.status === "connected") { - const payload = SessionInboundMessageSchema.parse(pending.message); - this.sendJsonMessage("session", payload.type, { type: "session", message: payload }); + pending.send(); pending.resolve(); } else { pending.reject(new DaemonConnectionError("Connection lost before message could be sent")); @@ -4846,6 +4856,8 @@ export class DaemonClient { if (!bytes) { throw new Error("File bytes are required."); } + // The file frames bypass the send queue, so start only on an open connection. + await this.whenConnected(); const uploadTransport = this.transport; const resolvedRequestId = this.createRequestId(input.requestId); const modifiedAt = input.modifiedAt ?? new Date().toISOString();