feat: stream files as binary frames over WebSocket (#659)
Replace base64-in-JSON file transfers with binary WebSocket frames to
unblock the JS thread when assistant markdown messages contain images.
Receive cost (JSON.parse over multi-MB base64) and persist cost (base64
re-encode to disk) both removed.
- Split daemon-client `exploreFileSystem` into `listDirectory` + `readFile`;
`readFile` returns `{ bytes, mime, size, modifiedAt }`. Encoding stays
private to the seam.
- New `packages/server/src/shared/binary-frames/` module hosts terminal
frames (moved from `terminal-stream-protocol.ts`) and new file-transfer
opcodes (`FileBegin` / `FileChunk` / `FileEnd`) with request-id correlation.
- Add optional `acceptBinary` to `FileExplorerRequestSchema`. New daemons
emit binary frames when set; legacy JSON path remains for old clients.
Compat fallback lives only inside `readFile`.
- New `persistAttachmentFromBytes` writes bytes directly: native via
`expo-file-system` `File.write(Uint8Array)`, web via Blob -> IndexedDB,
desktop via new `write_attachment_bytes` IPC. `persistAttachmentFromBase64`
removed.
This commit is contained in:
@@ -1,5 +1,6 @@
|
||||
import { afterEach, expect, expectTypeOf, test, vi } from "vitest";
|
||||
import { DaemonClient, type DaemonTransport } from "./daemon-client";
|
||||
import { encodeFileTransferFrame, FileTransferOpcode } from "../shared/binary-frames/index.js";
|
||||
import {
|
||||
asUint8Array,
|
||||
decodeTerminalResizePayload,
|
||||
@@ -13,6 +14,9 @@ expectTypeOf<"getGitDiff" extends keyof DaemonClient ? true : false>().toEqualTy
|
||||
expectTypeOf<
|
||||
"getHighlightedDiff" extends keyof DaemonClient ? true : false
|
||||
>().toEqualTypeOf<false>();
|
||||
expectTypeOf<
|
||||
"exploreFileSystem" extends keyof DaemonClient ? true : false
|
||||
>().toEqualTypeOf<false>();
|
||||
|
||||
function createMockLogger() {
|
||||
return {
|
||||
@@ -235,6 +239,209 @@ test("does not reconnect after close when ensureConnected is called", async () =
|
||||
expect(client.getConnectionState().status).toBe("disposed");
|
||||
});
|
||||
|
||||
test("listDirectory sends a list file explorer request and returns directory entries", async () => {
|
||||
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 connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const responsePromise = client.listDirectory("/tmp/project", "src", "req-list");
|
||||
|
||||
expect(JSON.parse(String(mock.sent[0]))).toEqual({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "file_explorer_request",
|
||||
cwd: "/tmp/project",
|
||||
path: "src",
|
||||
mode: "list",
|
||||
requestId: "req-list",
|
||||
},
|
||||
});
|
||||
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "file_explorer_response",
|
||||
payload: {
|
||||
cwd: "/tmp/project",
|
||||
path: "src",
|
||||
mode: "list",
|
||||
directory: {
|
||||
path: "src",
|
||||
entries: [
|
||||
{
|
||||
name: "index.ts",
|
||||
path: "src/index.ts",
|
||||
kind: "file",
|
||||
size: 12,
|
||||
modifiedAt: "2026-05-02T00:00:00.000Z",
|
||||
},
|
||||
],
|
||||
},
|
||||
file: null,
|
||||
error: null,
|
||||
requestId: "req-list",
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
await expect(responsePromise).resolves.toEqual({
|
||||
path: "src",
|
||||
entries: [
|
||||
{
|
||||
name: "index.ts",
|
||||
path: "src/index.ts",
|
||||
kind: "file",
|
||||
size: 12,
|
||||
modifiedAt: "2026-05-02T00:00:00.000Z",
|
||||
},
|
||||
],
|
||||
});
|
||||
});
|
||||
|
||||
test("readFile hides legacy base64 behind bytes", async () => {
|
||||
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 connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const responsePromise = client.readFile("/tmp/project", "logo.png", "req-file");
|
||||
|
||||
expect(JSON.parse(String(mock.sent[0]))).toEqual({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "file_explorer_request",
|
||||
cwd: "/tmp/project",
|
||||
path: "logo.png",
|
||||
mode: "file",
|
||||
acceptBinary: true,
|
||||
requestId: "req-file",
|
||||
},
|
||||
});
|
||||
|
||||
mock.triggerMessage(
|
||||
wrapSessionMessage({
|
||||
type: "file_explorer_response",
|
||||
payload: {
|
||||
cwd: "/tmp/project",
|
||||
path: "logo.png",
|
||||
mode: "file",
|
||||
directory: null,
|
||||
file: {
|
||||
path: "logo.png",
|
||||
kind: "image",
|
||||
encoding: "base64",
|
||||
content: "aGVsbG8=",
|
||||
mimeType: "image/png",
|
||||
size: 5,
|
||||
modifiedAt: "2026-05-02T00:00:00.000Z",
|
||||
},
|
||||
error: null,
|
||||
requestId: "req-file",
|
||||
},
|
||||
}),
|
||||
);
|
||||
|
||||
const result = await responsePromise;
|
||||
expect(result).toMatchObject({
|
||||
mime: "image/png",
|
||||
size: 5,
|
||||
path: "logo.png",
|
||||
kind: "image",
|
||||
modifiedAt: "2026-05-02T00:00:00.000Z",
|
||||
});
|
||||
expect(new TextDecoder().decode(result.bytes)).toBe("hello");
|
||||
});
|
||||
|
||||
test("readFile resolves from binary file frames when the daemon supports them", async () => {
|
||||
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 connectPromise = client.connect();
|
||||
mock.triggerOpen();
|
||||
await connectPromise;
|
||||
|
||||
const responsePromise = client.readFile("/tmp/project", "logo.png", "req-binary");
|
||||
|
||||
expect(JSON.parse(String(mock.sent[0]))).toEqual({
|
||||
type: "session",
|
||||
message: {
|
||||
type: "file_explorer_request",
|
||||
cwd: "/tmp/project",
|
||||
path: "logo.png",
|
||||
mode: "file",
|
||||
acceptBinary: true,
|
||||
requestId: "req-binary",
|
||||
},
|
||||
});
|
||||
|
||||
mock.triggerMessage(
|
||||
encodeFileTransferFrame({
|
||||
opcode: FileTransferOpcode.FileBegin,
|
||||
requestId: "req-binary",
|
||||
metadata: {
|
||||
mime: "image/png",
|
||||
size: 5,
|
||||
encoding: "binary",
|
||||
modifiedAt: "2026-05-02T00:00:00.000Z",
|
||||
},
|
||||
}),
|
||||
);
|
||||
mock.triggerMessage(
|
||||
encodeFileTransferFrame({
|
||||
opcode: FileTransferOpcode.FileChunk,
|
||||
requestId: "req-binary",
|
||||
payload: new TextEncoder().encode("hello"),
|
||||
}),
|
||||
);
|
||||
mock.triggerMessage(
|
||||
encodeFileTransferFrame({
|
||||
opcode: FileTransferOpcode.FileEnd,
|
||||
requestId: "req-binary",
|
||||
}),
|
||||
);
|
||||
|
||||
const result = await responsePromise;
|
||||
expect(result).toMatchObject({
|
||||
mime: "image/png",
|
||||
size: 5,
|
||||
path: "logo.png",
|
||||
kind: "image",
|
||||
modifiedAt: "2026-05-02T00:00:00.000Z",
|
||||
});
|
||||
expect(new TextDecoder().decode(result.bytes)).toBe("hello");
|
||||
});
|
||||
|
||||
test("normalizes workspace_setup_progress into a workspace-scoped daemon event", async () => {
|
||||
const logger = createMockLogger();
|
||||
const mock = createMockTransport();
|
||||
|
||||
@@ -82,13 +82,16 @@ import type { MutableDaemonConfig, MutableDaemonConfigPatch } from "../shared/me
|
||||
import { isRelayClientWebSocketUrl } from "../shared/daemon-endpoints.js";
|
||||
import {
|
||||
asUint8Array,
|
||||
decodeFileTransferFrame,
|
||||
decodeTerminalSnapshotPayload,
|
||||
decodeTerminalStreamFrame,
|
||||
FileTransferOpcode,
|
||||
encodeTerminalResizePayload,
|
||||
encodeTerminalStreamFrame,
|
||||
TerminalStreamOpcode,
|
||||
type FileTransferFrame,
|
||||
type TerminalStreamFrame,
|
||||
} from "../shared/terminal-stream-protocol.js";
|
||||
} from "../shared/binary-frames/index.js";
|
||||
import {
|
||||
createRelayE2eeTransportFactory,
|
||||
createWebSocketTransportFactory,
|
||||
@@ -283,6 +286,16 @@ type CreatePaseoWorktreePayload = Extract<
|
||||
{ type: "create_paseo_worktree_response" }
|
||||
>["payload"];
|
||||
type FileExplorerPayload = FileExplorerResponse["payload"];
|
||||
export type FileExplorerDirectoryPayload = NonNullable<FileExplorerPayload["directory"]>;
|
||||
type LegacyFileExplorerFilePayload = NonNullable<FileExplorerPayload["file"]>;
|
||||
export interface FileReadResult {
|
||||
bytes: Uint8Array;
|
||||
mime: string;
|
||||
size: number;
|
||||
path: string;
|
||||
kind: LegacyFileExplorerFilePayload["kind"];
|
||||
modifiedAt: string;
|
||||
}
|
||||
type FileDownloadTokenPayload = FileDownloadTokenResponse["payload"];
|
||||
type ListProviderFeaturesPayload = ListProviderFeaturesResponseMessage["payload"];
|
||||
type ListProviderModelsPayload = ListProviderModelsResponseMessage["payload"];
|
||||
@@ -569,6 +582,22 @@ interface WaitHandle<T> {
|
||||
cancel: (error: Error) => void;
|
||||
}
|
||||
|
||||
interface PendingBinaryFileRead {
|
||||
cwd: string;
|
||||
path: string;
|
||||
}
|
||||
|
||||
interface BinaryFileTransferState extends PendingBinaryFileRead {
|
||||
mime: string;
|
||||
size: number;
|
||||
encoding: Extract<
|
||||
FileTransferFrame,
|
||||
{ opcode: typeof FileTransferOpcode.FileBegin }
|
||||
>["metadata"]["encoding"];
|
||||
modifiedAt: string;
|
||||
chunks: Uint8Array[];
|
||||
}
|
||||
|
||||
type RpcWaitResult<T> = { kind: "ok"; value: T } | { kind: "error"; error: DaemonRpcError };
|
||||
type GetDaemonConfigResponse = Extract<
|
||||
SessionOutboundMessage,
|
||||
@@ -627,6 +656,55 @@ function normalizeClientId(value: unknown): string | null {
|
||||
return trimmed.length > 0 ? trimmed : null;
|
||||
}
|
||||
|
||||
function decodeBase64ToBytes(base64: string): Uint8Array {
|
||||
const binary = globalThis.atob(base64);
|
||||
const bytes = new Uint8Array(binary.length);
|
||||
for (let index = 0; index < binary.length; index += 1) {
|
||||
bytes[index] = binary.charCodeAt(index);
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
function legacyExplorerFileToBytes(file: LegacyFileExplorerFilePayload): FileReadResult {
|
||||
let bytes: Uint8Array;
|
||||
if (file.encoding === "base64" && file.content) {
|
||||
bytes = decodeBase64ToBytes(file.content);
|
||||
} else if (file.encoding === "utf-8" && file.content) {
|
||||
bytes = new TextEncoder().encode(file.content);
|
||||
} else {
|
||||
bytes = new Uint8Array();
|
||||
}
|
||||
|
||||
return {
|
||||
bytes,
|
||||
mime: file.mimeType ?? "application/octet-stream",
|
||||
size: file.size,
|
||||
path: file.path,
|
||||
kind: file.kind,
|
||||
modifiedAt: file.modifiedAt,
|
||||
};
|
||||
}
|
||||
|
||||
function binaryFileKind(mime: string, encoding: string): FileReadResult["kind"] {
|
||||
if (mime.startsWith("image/")) {
|
||||
return "image";
|
||||
}
|
||||
if (encoding === "utf-8" || mime.startsWith("text/") || mime === "application/json") {
|
||||
return "text";
|
||||
}
|
||||
return "binary";
|
||||
}
|
||||
|
||||
function concatByteChunks(chunks: Uint8Array[], size: number): Uint8Array {
|
||||
const bytes = new Uint8Array(size);
|
||||
let offset = 0;
|
||||
for (const chunk of chunks) {
|
||||
bytes.set(chunk, offset);
|
||||
offset += chunk.byteLength;
|
||||
}
|
||||
return bytes;
|
||||
}
|
||||
|
||||
function hashForLog(value: string): string {
|
||||
let hash = 0;
|
||||
for (let index = 0; index < value.length; index += 1) {
|
||||
@@ -697,6 +775,9 @@ export class DaemonClient {
|
||||
private terminalDirectorySubscriptions = new Set<string>();
|
||||
private terminalSlots = new Map<string, number>();
|
||||
private slotTerminals = new Map<number, string>();
|
||||
private pendingBinaryFileReads = new Map<string, PendingBinaryFileRead>();
|
||||
private activeBinaryFileTransfers = new Map<string, BinaryFileTransferState>();
|
||||
private completedBinaryFileReads = new Map<string, FileReadResult>();
|
||||
private readonly terminalStreamListeners = new Set<(event: TerminalStreamEvent) => void>();
|
||||
private logger: Logger;
|
||||
private pendingSendQueue: PendingSend[] = [];
|
||||
@@ -2865,11 +2946,12 @@ export class DaemonClient {
|
||||
// File Explorer
|
||||
// ============================================================================
|
||||
|
||||
async exploreFileSystem(
|
||||
private async requestFileExplorer(
|
||||
cwd: string,
|
||||
path: string,
|
||||
mode: "list" | "file" = "list",
|
||||
mode: "list" | "file",
|
||||
requestId?: string,
|
||||
acceptBinary = false,
|
||||
): Promise<FileExplorerPayload> {
|
||||
return this.sendCorrelatedSessionRequest({
|
||||
requestId,
|
||||
@@ -2878,12 +2960,51 @@ export class DaemonClient {
|
||||
cwd,
|
||||
path,
|
||||
mode,
|
||||
...(acceptBinary ? { acceptBinary: true } : {}),
|
||||
},
|
||||
responseType: "file_explorer_response",
|
||||
timeout: 10000,
|
||||
});
|
||||
}
|
||||
|
||||
async listDirectory(
|
||||
cwd: string,
|
||||
path: string,
|
||||
requestId?: string,
|
||||
): Promise<FileExplorerDirectoryPayload> {
|
||||
const payload = await this.requestFileExplorer(cwd, path, "list", requestId);
|
||||
if (payload.error) {
|
||||
throw new Error(payload.error);
|
||||
}
|
||||
if (!payload.directory) {
|
||||
throw new Error("Directory listing unavailable.");
|
||||
}
|
||||
return payload.directory;
|
||||
}
|
||||
|
||||
async readFile(cwd: string, path: string, requestId?: string): Promise<FileReadResult> {
|
||||
const resolvedRequestId = this.createRequestId(requestId);
|
||||
this.pendingBinaryFileReads.set(resolvedRequestId, { cwd, path });
|
||||
try {
|
||||
const payload = await this.requestFileExplorer(cwd, path, "file", resolvedRequestId, true);
|
||||
if (payload.error) {
|
||||
throw new Error(payload.error);
|
||||
}
|
||||
const binaryResult = this.completedBinaryFileReads.get(resolvedRequestId);
|
||||
if (binaryResult) {
|
||||
this.completedBinaryFileReads.delete(resolvedRequestId);
|
||||
return binaryResult;
|
||||
}
|
||||
if (!payload.file) {
|
||||
throw new Error("File unavailable.");
|
||||
}
|
||||
return legacyExplorerFileToBytes(payload.file);
|
||||
} finally {
|
||||
this.pendingBinaryFileReads.delete(resolvedRequestId);
|
||||
this.activeBinaryFileTransfers.delete(resolvedRequestId);
|
||||
}
|
||||
}
|
||||
|
||||
async requestDownloadToken(
|
||||
cwd: string,
|
||||
path: string,
|
||||
@@ -3933,6 +4054,13 @@ export class DaemonClient {
|
||||
}
|
||||
|
||||
private tryHandleBinaryFrame(rawBytes: Uint8Array): boolean {
|
||||
const fileFrame = decodeFileTransferFrame(rawBytes);
|
||||
if (fileFrame) {
|
||||
this.handleFileTransferFrame(fileFrame);
|
||||
this.runtimeMetrics?.recordBinaryFrame("other", rawBytes.byteLength, 0);
|
||||
return true;
|
||||
}
|
||||
|
||||
const frame = decodeTerminalStreamFrame(rawBytes);
|
||||
if (!frame) {
|
||||
return false;
|
||||
@@ -3953,6 +4081,57 @@ export class DaemonClient {
|
||||
return true;
|
||||
}
|
||||
|
||||
private handleFileTransferFrame(frame: FileTransferFrame): void {
|
||||
if (frame.opcode === FileTransferOpcode.FileBegin) {
|
||||
const pending = this.pendingBinaryFileReads.get(frame.requestId);
|
||||
if (!pending) {
|
||||
return;
|
||||
}
|
||||
this.activeBinaryFileTransfers.set(frame.requestId, {
|
||||
...pending,
|
||||
mime: frame.metadata.mime,
|
||||
size: frame.metadata.size,
|
||||
encoding: frame.metadata.encoding,
|
||||
modifiedAt: frame.metadata.modifiedAt,
|
||||
chunks: [],
|
||||
});
|
||||
return;
|
||||
}
|
||||
|
||||
const transfer = this.activeBinaryFileTransfers.get(frame.requestId);
|
||||
if (!transfer) {
|
||||
return;
|
||||
}
|
||||
|
||||
if (frame.opcode === FileTransferOpcode.FileChunk) {
|
||||
transfer.chunks.push(frame.payload);
|
||||
return;
|
||||
}
|
||||
|
||||
const bytes = concatByteChunks(transfer.chunks, transfer.size);
|
||||
this.activeBinaryFileTransfers.delete(frame.requestId);
|
||||
this.completedBinaryFileReads.set(frame.requestId, {
|
||||
bytes,
|
||||
mime: transfer.mime,
|
||||
size: transfer.size,
|
||||
path: transfer.path,
|
||||
kind: binaryFileKind(transfer.mime, transfer.encoding),
|
||||
modifiedAt: transfer.modifiedAt,
|
||||
});
|
||||
this.handleSessionMessage({
|
||||
type: "file_explorer_response",
|
||||
payload: {
|
||||
cwd: transfer.cwd,
|
||||
path: transfer.path,
|
||||
mode: "file",
|
||||
directory: null,
|
||||
file: null,
|
||||
error: null,
|
||||
requestId: frame.requestId,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
private handleBinaryFrame(frame: TerminalStreamFrame): void {
|
||||
const terminalId = this.slotTerminals.get(frame.slot);
|
||||
if (!terminalId) {
|
||||
|
||||
Reference in New Issue
Block a user