📖 feat: Stream Bounded Workspace File Reads

packages/code/README.md

@@ -729,9 +729,10 @@ native SRT command backend. This operator switch controls availability;
 LibreChat tool approval hooks remain the user-facing allow/deny boundary for
 each invocation.
 
-Reads reject absolute paths, traversal, escaping symlinks, non-regular files,
-and files larger than 1 MiB. The opened file is checked against its canonical
-in-workspace inode before it is read. Text search uses `rg` only to enumerate a
+Reads reject absolute paths, traversal, escaping symlinks, and non-regular files.
+The opened file is checked against its canonical in-workspace inode before it is
+read. Ordinary reads stream bounded line windows even from files larger than 1 MiB;
+see the [read contract](WORKSPACE-READS.md). Text search uses `rg` only to enumerate a
 bounded set of ignored-aware candidates with configuration and symlink following
 disabled. It then opens and verifies each candidate through the same confined
 1 MiB read boundary before matching locally. File listing invokes `rg` without

packages/code/WORKSPACE-READS.md

@@ -0,0 +1,32 @@
+# Workspace file read contract
+
+Ordinary `read_file` streams complete LF-delimited line windows through the verified
+file descriptor. It defaults to 200 lines, accepts at most 500, and returns at most
+1 MiB of UTF-8 content, including inter-line separators, regardless of file size.
+CR in CRLF is preserved; the final LF does not add a phantom line. An empty file
+has one empty line. A start beyond EOF returns an empty, non-truncated window.
+
+Byte-limited windows stop before the first line that cannot fit and return
+`nextStartLine = endLine + 1`. If the first requested line exceeds 1 MiB, the read
+fails with `READ_LIMIT_EXCEEDED`, without a repeating continuation. Scanning and
+collection check cancellation and share a 10-second deadline. Far-away starts
+can fail with `READ_LIMIT_EXCEEDED`; use an earlier start or `search_text`.
+Cancellation and timeout settle the request without waiting for queued I/O.
+Node retains an interrupted read's descriptor and chunk until that I/O drains;
+file and held-root cleanup are initiated immediately and finish asynchronously.
+This releases the request lane, but does not cancel kernel I/O.
+
+Memory is bounded by 64 KiB chunks and the returned window. Reads stop at the
+opened file's initial size, so concurrent growth cannot extend the scan. Truncation
+ends the scan at observed EOF; pathname replacement never switches the open
+handle. In-place writes can change observed content. Pagination is not a snapshot
+across requests and assumes the file stays unchanged.
+
+UTF-8 and BOM-marked UTF-16LE/BE are decoded incrementally. One leading BOM is
+removed; malformed sequences use replacement characters, as before. Binary bytes
+are decoded as text, including NUL, not classified or rejected. Neither a successful
+window nor continuation validates unread content.
+
+Request/result shapes, protocol version, and capabilities are unchanged. Search,
+preview, edit, and write limits remain unchanged. Digest-checked repository
+instruction snapshots retain their separate byte-bounded, newline-preserving path.

packages/code/src/root-access.ts

@@ -378,6 +378,12 @@ export class WorkspaceRootAccess {
 }
 
 const context = new AsyncLocalStorage<WorkspaceRootAccess>();
+const deferredCleanup = new WeakSet<WorkspaceRootAccess>();
+/** Interrupted I/O must not hold a request open while Node drains its handles. */
+export function deferWorkspaceRootCleanup(): void {
+    const access = context.getStore();
+    if (access) deferredCleanup.add(access);
+}
 /** Whether filesystem adapters in this call are anchored to a held root descriptor. */
 export const holdsWorkspaceRoot = (): boolean => context.getStore() != null;
 export async function withWorkspaceRoot<T>(
@@ -390,7 +396,9 @@ export async function withWorkspaceRoot<T>(
     try {
         return await context.run(access, action);
     } finally {
-        await access.close();
+        const closing = access.close();
+        if (deferredCleanup.delete(access)) void closing.catch(() => {});
+        else await closing;
     }
 }
 

packages/code/src/workspace-read-io.test.ts

@@ -0,0 +1,147 @@
+import assert from 'node:assert/strict';
+import { execFile } from 'node:child_process';
+import * as fs from 'node:fs/promises';
+import { tmpdir } from 'node:os';
+import { join } from 'node:path';
+import test from 'node:test';
+import { promisify } from 'node:util';
+
+const execFileAsync = promisify(execFile);
+
+// Isolate libuv saturation from the test runner and sibling tests.
+const queuedRead = `
+import assert from 'node:assert/strict';
+import * as fs from 'node:fs';
+import * as fsp from 'node:fs/promises';
+import { execFileSync } from 'node:child_process';
+import { getEventListeners } from 'node:events';
+import { join } from 'node:path';
+import { mock } from 'node:test';
+import { LocalWorkspaceTools } from ${JSON.stringify(
+    new URL('./workspace.js', import.meta.url).href
+)};
+import { captureWorkspaceRootIdentity } from ${JSON.stringify(
+    new URL('./root-identity.js', import.meta.url).href
+)};
+import { WorkspaceRootAccess } from ${JSON.stringify(
+    new URL('./root-access.js', import.meta.url).href
+)};
+const [root, cause, held] = process.argv.slice(1);
+const tools = await LocalWorkspaceTools.create({ workspaces: [{
+    id: 'primary', root,
+    ...(held === 'true' ? { identity: await captureWorkspaceRootIdentity(root) } : {}),
+}] });
+const fifo = join(root, 'pool');
+execFileSync('mkfifo', [fifo]);
+const pipe = fs.openSync(fifo, fs.constants.O_RDWR);
+const probe = await fsp.open(join(root, 'file'), 'r');
+const prototype = Object.getPrototypeOf(probe);
+await probe.close();
+const originalRead = prototype.read;
+const controller = new AbortController();
+let entered, blocker, pendingIo, physicalFd, readSettled = false;
+const started = new Promise(resolve => { entered = resolve; });
+const closing = [];
+const descriptors = [];
+let elapsed = 0;
+const now = performance.now();
+mock.method(performance, 'now', () => now + elapsed);
+function trackClose(handle) {
+    const originalClose = handle.close;
+    mock.method(handle, 'close', function (...args) {
+        descriptors.push(this.fd);
+        const promise = originalClose.apply(this, args);
+        closing.push(promise);
+        return promise;
+    });
+}
+const originalOpen = WorkspaceRootAccess.open;
+mock.method(WorkspaceRootAccess, 'open', async (...args) => {
+    const access = await originalOpen(...args);
+    trackClose(access.handle);
+    return access;
+});
+mock.method(prototype, 'read', function (...args) {
+    physicalFd = this.fd;
+    trackClose(this);
+    // This actual pipe read occupies the sole libuv worker until explicitly drained.
+    blocker = new Promise((resolve, reject) => fs.read(pipe, Buffer.alloc(1), 0, 1, null,
+        error => error ? reject(error) : resolve()));
+    // Invoke Node's original FileHandle.read, including its active-I/O references.
+    pendingIo = originalRead.apply(this, args).finally(() => { readSettled = true; });
+    if (cause === 'deadline') elapsed = 10_000;
+    if (cause === 'early-deadline') elapsed = 9_999.75;
+    entered();
+    return pendingIo;
+});
+let released = false;
+function drain() {
+    if (!released) { released = true; fs.writeSync(pipe, Buffer.from('x')); }
+}
+const wallStart = Date.now();
+const pending = tools.execute({ protocolVersion: 1, operation: 'read_file', workspaceId: 'primary', path: 'file' }, controller.signal);
+const rejected = assert.rejects(pending, error => error.code === (cause === 'abort' ? 'EXECUTION_ABORTED' : 'READ_LIMIT_EXCEEDED'));
+await started;
+if (cause === 'abort') controller.abort();
+// Always drain, even if a regression prevents bounded settlement.
+const fallback = setTimeout(drain, 2_000);
+try {
+    await rejected;
+    assert.equal(released, false, 'request settled before pool was drained');
+    if (cause === 'early-deadline') assert.ok(performance.now() < now + 10_000, 'timer settled before the monotonic deadline');
+    assert.equal(readSettled, false, 'real descriptor read remained pending');
+    assert.ok(fs.fstatSync(physicalFd).isFile(), 'Node still owns the physical descriptor');
+    assert.equal(closing.length, held === 'true' ? 2 : 1, 'file and held-root closes initiated');
+    assert.deepEqual(getEventListeners(controller.signal, 'abort'), []);
+    console.log(JSON.stringify({ cause, held: held === 'true', settledBeforeDrain: true, readSettled, elapsedMs: Date.now() - wallStart }));
+} finally {
+    clearTimeout(fallback);
+    drain();
+    await blocker;
+    await pendingIo;
+    await Promise.all(closing);
+    for (const fd of descriptors) assert.throws(() => fs.fstatSync(fd), { code: 'EBADF' });
+    mock.restoreAll();
+    fs.closeSync(pipe);
+}
+`;
+
+for (const cause of ['abort', 'deadline', 'early-deadline'] as const) {
+    for (const held of [false, true]) {
+        test(`${cause} settles before real queued I/O drains (${
+            held ? 'held' : 'legacy'
+        } root)`, async t => {
+            if (!['linux', 'darwin'].includes(process.platform))
+                return t.skip('requires POSIX FIFO and descriptor access');
+            const root = await fs.realpath(
+                await fs.mkdtemp(join(tmpdir(), 'workspace-read-io-'))
+            );
+            t.after(() => fs.rm(root, { recursive: true, force: true }));
+            await fs.writeFile(join(root, 'file'), 'line\n'.repeat(300_000));
+            const { stdout } = await execFileAsync(
+                process.execPath,
+                [
+                    '--input-type=module',
+                    '--eval',
+                    queuedRead,
+                    root,
+                    cause,
+                    String(held),
+                ],
+                {
+                    // Node 20 can bypass the pool for regular files via io_uring.
+                    env: {
+                        ...process.env,
+                        UV_THREADPOOL_SIZE: '1',
+                        UV_USE_IO_URING: '0',
+                    },
+                    timeout: 10_000,
+                }
+            );
+            const observed = JSON.parse(stdout);
+            assert.equal(observed.settledBeforeDrain, true);
+            assert.equal(observed.readSettled, false);
+            t.diagnostic(stdout.trim());
+        });
+    }
+}