From c82700ab21fe2efb3cdc96ba5c908fffc1ac907a Mon Sep 17 00:00:00 2001 From: 7Sageer <7sageer@djwcb.cn> Date: Mon, 22 Jun 2026 15:43:12 +0800 Subject: [PATCH] perf(kaos): stream UTF-8 readLines --- .changeset/stream-readlines-utf8.md | 5 ++ packages/kaos/src/local.ts | 56 ++++++++++--- packages/kaos/test/local.test.ts | 121 ++++++++++++++++++++++++++++ 3 files changed, 172 insertions(+), 10 deletions(-) create mode 100644 .changeset/stream-readlines-utf8.md diff --git a/.changeset/stream-readlines-utf8.md b/.changeset/stream-readlines-utf8.md new file mode 100644 index 0000000000..771e0113cf --- /dev/null +++ b/.changeset/stream-readlines-utf8.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +Reduce memory use when reading large UTF-8 text files while preserving non-UTF-8 line reading behavior. diff --git a/packages/kaos/src/local.ts b/packages/kaos/src/local.ts index 3a4dbf7340..03e0001bb4 100644 --- a/packages/kaos/src/local.ts +++ b/packages/kaos/src/local.ts @@ -447,17 +447,39 @@ export class LocalKaos implements Kaos { const resolved = this._resolvePath(path); const encoding = options?.encoding ?? 'utf-8'; const errors = options?.errors ?? 'strict'; - const buf = await readFile(resolved); - const content = decodeTextWithErrors(buf, encoding, errors); - const lines = content.split('\n'); - for (let i = 0; i < lines.length; i++) { - const line = lines[i]; - if (line === undefined) continue; - if (i < lines.length - 1) { - yield line + '\n'; - } else if (line !== '') { - yield line; + + if (encoding !== 'utf-8' && encoding !== 'utf8') { + const content = decodeTextWithErrors(await readFile(resolved), encoding, errors); + yield* splitLinesKeepingTerminator(content); + return; + } + + const fh = await open(resolved, 'r'); + try { + const chunkSize = 64 * 1024; + const buf = Buffer.alloc(chunkSize); + let pending = Buffer.alloc(0); + while (true) { + const { bytesRead } = await fh.read(buf, 0, buf.length, null); + if (bytesRead === 0) break; + const data = + pending.length > 0 + ? Buffer.concat([pending, buf.subarray(0, bytesRead)]) + : buf.subarray(0, bytesRead); + let lineStart = 0; + for (let i = 0; i < data.length; i += 1) { + if (data[i] === 0x0a) { + yield decodeTextWithErrors(data.subarray(lineStart, i + 1), encoding, errors); + lineStart = i + 1; + } + } + pending = lineStart < data.length ? Buffer.from(data.subarray(lineStart)) : Buffer.alloc(0); } + if (pending.length > 0) { + yield decodeTextWithErrors(pending, encoding, errors); + } + } finally { + await fh.close(); } } @@ -589,6 +611,20 @@ export class LocalKaos implements Kaos { } } +function* splitLinesKeepingTerminator(text: string): Generator { + if (text.length === 0) return; + let start = 0; + for (let i = 0; i < text.length; i += 1) { + if (text.codePointAt(i) === 0x0a) { + yield text.slice(start, i + 1); + start = i + 1; + } + } + if (start < text.length) { + yield text.slice(start); + } +} + // Wait for a freshly spawned ChildProcess to either emit 'spawn' (success) or // 'error' (ENOENT / EACCES / etc.). Until this resolves, callers should not // assume the child is running — they may otherwise write to the stdin of a diff --git a/packages/kaos/test/local.test.ts b/packages/kaos/test/local.test.ts index 01f192f88c..a3de2fcb98 100644 --- a/packages/kaos/test/local.test.ts +++ b/packages/kaos/test/local.test.ts @@ -176,6 +176,127 @@ describe('LocalKaos', () => { }); }); + describe('readLines streaming', () => { + it('preserves content exactly across representative line endings', async () => { + const fixtures: Array<[string, string]> = [ + ['multiline', 'line1\nline2\nline3\n'], + ['no trailing newline', 'line1\nline2'], + ['single line', 'only'], + ['single newline', '\n'], + ['empty', ''], + ['crlf', 'a\r\nb\r\n'], + ['lone cr', 'a\rB\n'], + ]; + for (const [name, content] of fixtures) { + const filePath = join(tempDir, `${name}.txt`); + await kaos.writeText(filePath, content); + const lines: string[] = []; + for await (const line of kaos.readLines(filePath)) { + lines.push(line); + } + expect(lines.join('')).toBe(content); + } + }); + + it('reads a large multi-chunk file line by line', async () => { + const filePath = join(tempDir, 'large.txt'); + const lineCount = 5000; + const content = + Array.from({ length: lineCount }, (_, i) => `line ${String(i)} ${'x'.repeat(40)}`).join( + '\n', + ) + '\n'; + await kaos.writeText(filePath, content); + + const lines: string[] = []; + for await (const line of kaos.readLines(filePath)) { + lines.push(line); + } + expect(lines.length).toBe(lineCount); + expect(lines.join('')).toBe(content); + }); + + it('preserves a multibyte character straddling the 64KiB chunk boundary', async () => { + const filePath = join(tempDir, 'boundary.txt'); + const emoji = '😀'; + const content = `${'a'.repeat(65535)}${emoji}\n`; + await kaos.writeText(filePath, content); + + const lines: string[] = []; + for await (const line of kaos.readLines(filePath)) { + lines.push(line); + } + expect(lines).toEqual([content]); + }); + + it('keeps utf16le line splitting on the decoded text path', async () => { + const filePath = join(tempDir, 'utf16le.txt'); + await kaos.writeBytes(filePath, Buffer.from('a\n\u0A41\n', 'utf16le')); + + const lines: string[] = []; + for await (const line of kaos.readLines(filePath, { encoding: 'utf16le' })) { + lines.push(line); + } + expect(lines).toEqual(['a\n', 'āŠ\n']); + }); + + it('keeps lossless encodings on the decoded text path', async () => { + const filePath = join(tempDir, 'hex.txt'); + await kaos.writeBytes(filePath, Buffer.from('a\nb')); + + const lines: string[] = []; + for await (const line of kaos.readLines(filePath, { encoding: 'hex' })) { + lines.push(line); + } + expect(lines).toEqual(['610a62']); + }); + + describe('errors parameter', () => { + const invalidBytes = Buffer.concat([ + Buffer.from([0xe4, 0xb8, 0xad]), + Buffer.from([0xff]), + Buffer.from([0xe6, 0x96, 0x87]), + Buffer.from([0x0a]), + ]); + + it('throws on invalid utf-8 with errors="strict"', async () => { + const filePath = join(tempDir, 'invalid-strict.txt'); + await kaos.writeBytes(filePath, invalidBytes); + await expect( + (async () => { + const lines: string[] = []; + for await (const line of kaos.readLines(filePath)) { + lines.push(line); + } + return lines; + })(), + ).rejects.toThrow(); + }); + + it('replaces invalid bytes with errors="replace"', async () => { + const filePath = join(tempDir, 'invalid-replace.txt'); + await kaos.writeBytes(filePath, invalidBytes); + const lines: string[] = []; + for await (const line of kaos.readLines(filePath, { errors: 'replace' })) { + lines.push(line); + } + const output = lines.join(''); + expect(output).toContain('\uFFFD'); + expect(output).toContain('中'); + expect(output).toContain('文'); + }); + + it('drops invalid bytes with errors="ignore"', async () => { + const filePath = join(tempDir, 'invalid-ignore.txt'); + await kaos.writeBytes(filePath, invalidBytes); + const lines: string[] = []; + for await (const line of kaos.readLines(filePath, { errors: 'ignore' })) { + lines.push(line); + } + expect(lines.join('')).toBe('中文\n'); + }); + }); + }); + describe('readText errors parameter (Python compat)', () => { // A file with a valid UTF-8 prefix "中", an invalid standalone byte 0xff, // and a valid UTF-8 suffix "文". Under strict decoding this throws.