diff --git a/.changeset/computer-bounded-filesystem.md b/.changeset/computer-bounded-filesystem.md new file mode 100644 index 00000000..0b5f7249 --- /dev/null +++ b/.changeset/computer-bounded-filesystem.md @@ -0,0 +1,5 @@ +--- +"@cloudflare/computer": minor +--- + +Expose bounded workspace byte reads through RPC and return paginated directory listings with file metadata. diff --git a/packages/computer/src/stub.test.ts b/packages/computer/src/stub.test.ts index 8d7dad13..84f42362 100644 --- a/packages/computer/src/stub.test.ts +++ b/packages/computer/src/stub.test.ts @@ -205,6 +205,24 @@ describe("WorkspaceStub", () => { }); }); + it("fs.readFile forwards ranged stream options", async () => { + await withStub(async (ws) => { + const stub = ws.stub(); + await stub.fs.writeFile("/bin", new Uint8Array([1, 2, 3, 4, 5])); + const stream = await stub.fs.readFile("/bin", { byteOffset: 1, byteLength: 3 }); + const bytes = new Uint8Array(await new Response(stream).arrayBuffer()); + expect(Array.from(bytes)).toEqual([2, 3, 4]); + }); + }); + + it("fs.readRange forwards bounded byte reads", async () => { + await withStub(async (ws) => { + const stub = ws.stub(); + await stub.fs.writeFile("/bin", new Uint8Array([1, 2, 3, 4, 5])); + expect(Array.from(await stub.fs.readRange("/bin", 1, 3))).toEqual([2, 3, 4]); + }); + }); + it("fs.readdir forwards bounded-read options", async () => { await withStub(async (ws) => { const stub = ws.stub(); diff --git a/packages/computer/src/stub.ts b/packages/computer/src/stub.ts index 844d8452..90eda3b1 100644 --- a/packages/computer/src/stub.ts +++ b/packages/computer/src/stub.ts @@ -104,6 +104,11 @@ export class WorkspaceFilesystemStub extends RpcTarget { readFile(path: string): Promise>; readFile(path: string, encoding: "utf8"): Promise; + readFile( + path: string, + options: ReadFileOptions & { encoding?: undefined }, + ): Promise>; + readFile(path: string, options: ReadFileOptions & { encoding: "utf8" }): Promise; readFile(path: string, options: ReadFileOptions): Promise>; readFile( path: string, @@ -114,6 +119,15 @@ export class WorkspaceFilesystemStub extends RpcTarget { ); } + readRange(path: string, offset: number, length: number): Promise { + return withSpan( + this.#ws.observer, + "workspace.fs.readRange", + { "workspace.fs.path": path, "workspace.fs.offset": offset, "workspace.fs.length": length }, + () => this.#ws.fs.readRange(path, offset, length), + ); + } + exists(path: string): Promise { return withSpan( this.#ws.observer, diff --git a/packages/computer/src/tools/ai.test.ts b/packages/computer/src/tools/ai.test.ts index b2636f94..e58c1054 100644 --- a/packages/computer/src/tools/ai.test.ts +++ b/packages/computer/src/tools/ai.test.ts @@ -179,28 +179,109 @@ function memoryStore(options: { } describe("WorkspaceFileStore", () => { - it("slices byte ranges while reading chunks from Workspace.fs", async () => { - const workspace = makeWorkspace(); - await workspace.fs.mkdir("/workspace", { recursive: true }); - await workspace.fs.writeFile("/workspace/range.txt", bytes("abcdefghij")); + it("opens one ranged stream instead of issuing repeated range calls", async () => { + const calls: Array<{ byteOffset?: number; byteLength?: number }> = []; + const content = bytes("abcdefghij"); + const workspace = { + fs: { + async stat() { + throw new Error("stat must not be called by readChunks"); + }, + async readRange() { + throw new Error("readRange must not be called by readChunks"); + }, + async readFile( + _path: string, + options: { byteOffset?: number; byteLength?: number } = {}, + ): Promise> { + calls.push(options); + const start = options.byteOffset ?? 0; + const end = options.byteLength === undefined ? undefined : start + options.byteLength; + return new ReadableStream({ + start(controller) { + controller.enqueue(content.slice(start, end)); + controller.close(); + }, + }); + }, + async writeFile() {}, + async mkdir() {}, + async rm() {}, + }, + }; const store = new WorkspaceFileStore(workspace); await expect( drainChunks(store.readChunks("/workspace/range.txt", 2, 5)).then(decode), ).resolves.toBe("cdefg"); + expect(calls).toEqual([{ byteOffset: 2, byteLength: 5 }]); + }); + + it("still validates the path for a zero-length read", async () => { + const store = new WorkspaceFileStore(makeWorkspace()); + + await expect(drainChunks(store.readChunks("/missing", 0, 0))).rejects.toMatchObject({ + code: "ENOENT", + }); + }); + + it("rejects directories instead of treating them as empty files", async () => { + const workspace = makeWorkspace(); + await workspace.fs.mkdir("/directory"); + const store = new WorkspaceFileStore(workspace); + + await expect(drainChunks(store.readChunks("/directory"))).rejects.toMatchObject({ + code: "EISDIR", + }); + }); + + it("keeps a real multi-chunk workspace read on one snapshot", async () => { + const workspace = makeWorkspace(); + await workspace.fs.mkdir("/workspace", { recursive: true }); + const original = new Uint8Array(600_000); + original.fill(0x41, 0, 500_000); + original.fill(0x42, 500_000); + await workspace.fs.writeFile("/workspace/large.bin", original); + const store = new WorkspaceFileStore(workspace); + + const chunks = store.readChunks("/workspace/large.bin")[Symbol.asyncIterator](); + const first = await chunks.next(); + expect(first.done).toBe(false); + await workspace.fs.writeFile( + "/workspace/large.bin", + new Uint8Array(original.length).fill(0x43), + ); + + const parts = [first.value]; + while (true) { + const next = await chunks.next(); + if (next.done) break; + parts.push(next.value); + } + const result = await drainChunks( + (async function* () { + yield* parts; + })(), + ); + expect(result.byteLength).toBe(original.byteLength); + expect(result.every((value, index) => value === original[index])).toBe(true); }); - it("cancels read streams when a byte range stops before EOF", async () => { + it("cancels a ranged stream when its consumer stops early", async () => { let cancelled = false; const workspace = { fs: { async stat() { - return { size: 10, mtime: 1, mode: 0o100644, isFile: true, isDirectory: false }; + throw new Error("stat must not be called by readChunks"); + }, + async readRange() { + throw new Error("readRange must not be called by readChunks"); }, - async readFile() { - return new ReadableStream({ + async readFile(): Promise> { + return new ReadableStream({ start(controller) { - controller.enqueue(bytes("abcdefghij")); + controller.enqueue(bytes("first")); + controller.enqueue(bytes("second")); }, cancel() { cancelled = true; @@ -210,16 +291,11 @@ describe("WorkspaceFileStore", () => { async writeFile() {}, async mkdir() {}, async rm() {}, - async readdir() { - return []; - }, }, }; const store = new WorkspaceFileStore(workspace); - await expect( - drainChunks(store.readChunks("/workspace/range.txt", 2, 5)).then(decode), - ).resolves.toBe("cdefg"); + for await (const _chunk of store.readChunks("/workspace/range.txt")) break; expect(cancelled).toBe(true); }); }); @@ -255,7 +331,17 @@ describe("createAITools filesystem tools", () => { ); await expect(executeTool(tools.ls, { path: "/workspace/notes" })).resolves.toEqual({ path: "/workspace/notes", - entries: [{ name: "todo.txt", isFile: true, isDirectory: false }], + count: 1, + entries: [ + { + name: "todo.txt", + size: 8, + mtime: 1_700_000_000_000, + isFile: true, + isDirectory: false, + isSymbolicLink: false, + }, + ], }); await expect( executeTool(tools.read, { path: "/workspace/notes/todo.txt", limit: 1 }), @@ -279,6 +365,32 @@ describe("createAITools filesystem tools", () => { ); }); + it("paginates ls results and reports a continuation offset", async () => { + const workspace = makeWorkspace(); + await workspace.fs.mkdir("/workspace", { recursive: true }); + for (const name of ["a", "b", "c"]) { + await workspace.fs.writeFile(`/workspace/${name}`, name); + } + const tools = createAITools({ workspace }); + + await expect( + executeTool(tools.ls, { path: "/workspace", limit: 2, offset: 0 }), + ).resolves.toMatchObject({ + count: 2, + entries: [ + { name: "a", size: 1 }, + { name: "b", size: 1 }, + ], + nextOffset: 2, + }); + await expect( + executeTool(tools.ls, { path: "/workspace", limit: 2, offset: 2 }), + ).resolves.toMatchObject({ + count: 1, + entries: [{ name: "c", size: 1 }], + }); + }); + it("preserves file mode when write overwrites an existing file", async () => { const writes: Array<{ path: string; content: string; mode?: number }> = []; const tool = createWriteTool({ diff --git a/packages/computer/src/tools/fs/list.ts b/packages/computer/src/tools/fs/list.ts index d5e91490..eafe7097 100644 --- a/packages/computer/src/tools/fs/list.ts +++ b/packages/computer/src/tools/fs/list.ts @@ -3,7 +3,19 @@ import { z } from "zod"; export interface ListWorkspaceLike { fs: { - readdir(path: string): Promise>; + readdir( + path: string, + options?: { limit?: number; offset?: number }, + ): Promise< + Array<{ + name: string; + size: number; + mtime: number; + isFile: boolean; + isDirectory: boolean; + isSymbolicLink: boolean; + }> + >; }; } @@ -11,26 +23,55 @@ export interface ListToolOptions { workspace: ListWorkspaceLike; } +const DEFAULT_LIMIT = 200; +const MAX_LIMIT = 1000; + const inputSchema = z.object({ path: z.string().describe("Absolute directory path to list, e.g. /workspace/src."), + limit: z + .number() + .int() + .min(1) + .max(MAX_LIMIT) + .optional() + .describe(`Maximum entries to return. Defaults to ${DEFAULT_LIMIT}.`), + offset: z.number().int().min(0).optional().describe("Number of entries to skip in name order."), }); export function createListTool(options: ListToolOptions): Tool> { return tool({ description: - "List entries in a workspace directory. Returns each entry name and whether it is a file or directory.", + "List entries in a workspace directory with file sizes and modification times. Use limit and offset to page through large directories.", inputSchema, - execute: async ({ path }) => { + execute: async ({ path, limit, offset }) => { try { - const entries = await options.workspace.fs.readdir(path); - return { + const pageSize = limit ?? DEFAULT_LIMIT; + const pageOffset = offset ?? 0; + const entries = await options.workspace.fs.readdir(path, { + limit: pageSize + 1, + offset: pageOffset, + }); + const truncated = entries.length > pageSize; + const page = (truncated ? entries.slice(0, pageSize) : entries).map((entry) => ({ + name: entry.name, + size: entry.size, + mtime: entry.mtime, + isFile: entry.isFile, + isDirectory: entry.isDirectory, + isSymbolicLink: entry.isSymbolicLink, + })); + const result: { + path: string; + count: number; + entries: typeof page; + nextOffset?: number; + } = { path, - entries: entries.map((entry) => ({ - name: entry.name, - isFile: entry.isFile, - isDirectory: entry.isDirectory, - })), + count: page.length, + entries: page, }; + if (truncated) result.nextOffset = pageOffset + pageSize; + return result; } catch (err) { return { error: err instanceof Error ? err.message : String(err) }; } diff --git a/packages/computer/src/tools/fs/store.ts b/packages/computer/src/tools/fs/store.ts index 739d97cd..442de1c4 100644 --- a/packages/computer/src/tools/fs/store.ts +++ b/packages/computer/src/tools/fs/store.ts @@ -6,9 +6,9 @@ * class. This adapter is the bridge from that contract to the public * `workspace.fs` surface. * - * Reads go through `fs.readFile(path)` as a `ReadableStream` - * and are stitched together either chunk-by-chunk (`readChunks`) or all - * at once (`readAll`). + * Chunked and ranged reads use one `fs.readFile` stream so remote workspaces + * keep one snapshot and one RPC invocation. Whole-file reads used by edit and + * multimodal output drain the same stream interface. */ import type { FileStat, FileStore } from "./types.js"; @@ -26,11 +26,26 @@ export interface WorkspaceLike { isFile: boolean; isDirectory: boolean; }>; - readFile(path: string): Promise>; + readFile( + path: string, + options?: { byteOffset?: number; byteLength?: number }, + ): Promise>; writeFile(path: string, content: Uint8Array, options?: { mode?: number }): Promise; mkdir(path: string, options?: { recursive?: boolean }): Promise; rm(path: string, options?: { recursive?: boolean; force?: boolean }): Promise; - readdir(path: string): Promise>; + readdir( + path: string, + options?: { limit?: number; offset?: number }, + ): Promise< + Array<{ + name: string; + size: number; + mtime: number; + isFile: boolean; + isDirectory: boolean; + isSymbolicLink: boolean; + }> + >; }; } @@ -64,55 +79,30 @@ export class WorkspaceFileStore implements FileStore { } async *readChunks(path: string, byteOffset = 0, byteLength?: number): AsyncIterable { - if (byteOffset < 0) throw new Error("readChunks: byteOffset must be non-negative"); - if (byteLength !== undefined && byteLength < 0) { - throw new Error("readChunks: byteLength must be non-negative"); + if (!Number.isSafeInteger(byteOffset) || byteOffset < 0) { + throw new Error("readChunks: byteOffset must be a non-negative safe integer"); } - if (byteLength === 0) return; - - const stream = await this.ws.fs.readFile(path); + if (byteLength !== undefined && (!Number.isSafeInteger(byteLength) || byteLength < 0)) { + throw new Error("readChunks: byteLength must be a non-negative safe integer"); + } + const stream = await this.ws.fs.readFile(path, { byteOffset, byteLength }); const reader = stream.getReader(); - let skipped = 0; - let yielded = 0; let completed = false; try { while (true) { const { value, done } = await reader.read(); if (done) { completed = true; - break; - } - if (!value || value.byteLength === 0) continue; - - let start = 0; - if (skipped < byteOffset) { - const needed = byteOffset - skipped; - if (value.byteLength <= needed) { - skipped += value.byteLength; - continue; - } - start = needed; - skipped = byteOffset; - } - - let end = value.byteLength; - if (byteLength !== undefined) { - const remaining = byteLength - yielded; - if (remaining <= 0) break; - end = Math.min(end, start + remaining); - } - - if (end > start) { - const chunk = value.slice(start, end); - yielded += chunk.byteLength; - yield chunk; + return; } - - if (byteLength !== undefined && yielded >= byteLength) break; + if (value !== undefined && value.byteLength > 0) yield value; } } finally { - if (!completed) await reader.cancel(); - reader.releaseLock(); + try { + if (!completed) await reader.cancel(); + } finally { + reader.releaseLock(); + } } } }