From d1733b9697ed267156929e0e49efea2a385b0551 Mon Sep 17 00:00:00 2001 From: Thomas Kosiewski Date: Sat, 12 Sep 2026 11:35:59 +0200 Subject: [PATCH] =?UTF-8?q?=F0=9F=A4=96=20fix:=20isolate=20scanner=20callb?= =?UTF-8?q?acks=20and=20preserve=20late=20cancellation?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Thomas Kosiewski --- _Generated with `xum` • Model: `unavailable` • Thinking: `unavailable` • Cost: `$unavailable`_ Change-Id: I8e8213dd9ce3c49524397757d99ebad2af2d6164 --- bun.lock | 5 + flake.nix | 2 +- package.json | 1 + src/node/services/historyRowScanner.test.ts | 739 ++++++++++++++++++++ src/node/services/historyRowScanner.ts | 238 +++++++ 5 files changed, 984 insertions(+), 1 deletion(-) create mode 100644 src/node/services/historyRowScanner.test.ts create mode 100644 src/node/services/historyRowScanner.ts diff --git a/bun.lock b/bun.lock index fb9961f1486..8e62b04e748 100644 --- a/bun.lock +++ b/bun.lock @@ -96,6 +96,7 @@ "ssh-config": "^5.0.4", "ssh2": "^1.17.0", "sshpk": "^1.18.0", + "stream-json": "3.6.0", "streamdown": "2.0.0-canary.2", "trpc-cli": "^0.12.1", "turndown": "^7.2.2", @@ -3441,6 +3442,10 @@ "storybook": ["storybook@10.3.3", "", { "dependencies": { "@storybook/global": "^5.0.0", "@storybook/icons": "^2.0.1", "@testing-library/jest-dom": "^6.9.1", "@testing-library/user-event": "^14.6.1", "@vitest/expect": "3.2.4", "@vitest/spy": "3.2.4", "esbuild": "^0.18.0 || ^0.19.0 || ^0.20.0 || ^0.21.0 || ^0.22.0 || ^0.23.0 || ^0.24.0 || ^0.25.0 || ^0.26.0 || ^0.27.0", "open": "^10.2.0", "recast": "^0.23.5", "semver": "^7.7.3", "use-sync-external-store": "^1.5.0", "ws": "^8.18.0" }, "peerDependencies": { "prettier": "^2 || ^3" }, "optionalPeers": ["prettier"], "bin": "./dist/bin/dispatcher.js" }, "sha512-tMoRAts9EVqf+mEMPLC6z1DPyHbcPe+CV1MhLN55IKsl0HxNjvVGK44rVPSePbltPE6vIsn4bdRj6CCUt8SJwQ=="], + "stream-chain": ["stream-chain@4.2.5", "", {}, "sha512-Wtyq3bNE3ggLR0v2vftqvuhltym3WbZAkZpfIrkr5F/6vpeUmWmwTgXa16zD87gpahwJ/Qulq3zVfUlgIc0J2A=="], + + "stream-json": ["stream-json@3.6.0", "", { "dependencies": { "stream-chain": "^4.2.5" } }, "sha512-NiJdqxKyau579z/E8vfqcjWfSDWxW/AT99javFXdPXF147Z5za85LRXSHEmSX9TKOakB7gaIccfD0fOIctb7KQ=="], + "streamdown": ["streamdown@2.0.0-canary.2", "", { "dependencies": { "clsx": "^2.1.1", "esbuild": "^0.27.2", "hast-util-to-jsx-runtime": "^2.3.6", "html-url-attributes": "^3.0.1", "katex": "^0.16.27", "marked": "^17.0.1", "rehype-harden": "^1.1.7", "rehype-katex": "^7.0.1", "rehype-raw": "^7.0.0", "rehype-sanitize": "^6.0.0", "remark-cjk-friendly": "^1.2.3", "remark-cjk-friendly-gfm-strikethrough": "^1.2.3", "remark-gfm": "^4.0.1", "remark-math": "^6.0.0", "remark-parse": "^11.0.0", "remark-rehype": "^11.1.2", "remend": "1.0.2-canary.0", "shiki": "^3.19.0", "tailwind-merge": "^3.4.0", "unified": "^11.0.5", "unist-util-visit": "^5.0.0" }, "peerDependencies": { "react": "^18.0.0 || ^19.0.0" } }, "sha512-0gKCgPBsMUiB1FimimzZrgRP9vwMVtt7l9NTuptau3T6767cEU0DGLCICLdxJukebzNsH00ccQvrVs1o3bkPEQ=="], "string-length": ["string-length@6.0.0", "", { "dependencies": { "strip-ansi": "^7.1.0" } }, "sha512-1U361pxZHEQ+FeSjzqRpV+cu2vTzYeWeafXFLykiFlv4Vc0n3njgU8HrMbyik5uwm77naWMuVG8fhEF+Ovb1Kg=="], diff --git a/flake.nix b/flake.nix index 2950f39164a..43e67cf33e4 100644 --- a/flake.nix +++ b/flake.nix @@ -84,7 +84,7 @@ outputHashMode = "recursive"; # Marker used by scripts/update_flake_hash.sh to update this hash in place. - outputHash = "sha256-sO6bGZQoHXmiS/lMgKy/p0/nTa8WByUFIUAz8V6WiBU="; # xum-offline-cache-hash + outputHash = "sha256-5V2PO5L0AlCoSLgxjXYDRq/aCzmXYnCRiPss/IDOtYM="; # xum-offline-cache-hash }; configurePhase = '' diff --git a/package.json b/package.json index e2f5ddf51c7..5e32bd8c2da 100644 --- a/package.json +++ b/package.json @@ -140,6 +140,7 @@ "ssh-config": "^5.0.4", "ssh2": "^1.17.0", "sshpk": "^1.18.0", + "stream-json": "3.6.0", "streamdown": "2.0.0-canary.2", "trpc-cli": "^0.12.1", "turndown": "^7.2.2", diff --git a/src/node/services/historyRowScanner.test.ts b/src/node/services/historyRowScanner.test.ts new file mode 100644 index 00000000000..1b8b1fca9cc --- /dev/null +++ b/src/node/services/historyRowScanner.test.ts @@ -0,0 +1,739 @@ +import { afterEach, beforeEach, describe, expect, it, mock, spyOn } from "bun:test"; +import { createHash } from "node:crypto"; +import * as fs from "node:fs/promises"; +import * as os from "node:os"; +import * as path from "node:path"; +import { SESSION_HISTORY_SCAN_CHUNK_BYTES } from "@/common/constants/contextBudget"; +import { + scanHistoryRows, + type HistoryRowDescriptor, + type HistoryRowToken, +} from "./historyRowScanner"; + +// Small-fixture oracle only; production visitors must not assemble unbounded values. +function assemble(tokens: HistoryRowToken[]): unknown { + const stack: Array<{ value: Record | unknown[]; key?: string }> = []; + let result: unknown; + let scalar = ""; + const put = (value: unknown) => { + const parent = stack.at(-1); + if (!parent) result = value; + else if (Array.isArray(parent.value)) parent.value.push(value); + else parent.value[parent.key!] = value; + }; + for (const token of tokens) { + switch (token.name) { + case "startObject": + case "startArray": { + const value = token.name === "startObject" ? {} : []; + put(value); + stack.push({ value }); + break; + } + case "endObject": + case "endArray": + stack.pop(); + break; + case "startKey": + case "startString": + case "startNumber": + scalar = ""; + break; + case "stringChunk": + case "numberChunk": + scalar += token.value; + break; + case "endKey": + stack.at(-1)!.key = scalar; + break; + case "endString": + put(scalar); + break; + case "endNumber": + put(Number(scalar)); + break; + case "trueValue": + case "falseValue": + case "nullValue": + put(token.value); + break; + default: + throw new Error(`Unexpected packed token: ${token.name}`); + } + } + return result; +} + +describe("raw history row scanner", () => { + let directory: string; + let filePath: string; + beforeEach(async () => { + directory = await fs.mkdtemp(path.join(os.tmpdir(), "history-row-scanner-")); + filePath = path.join(directory, "chat.jsonl"); + }); + afterEach(async () => { + mock.restore(); + await fs.rm(directory, { recursive: true, force: true }); + }); + const digest = (value: string | Buffer) => createHash("sha256").update(value).digest("hex"); + async function collect(options?: Parameters[2]) { + const rows: Array<{ descriptor: HistoryRowDescriptor; tokens: HistoryRowToken[] }> = []; + await scanHistoryRows( + filePath, + () => { + const tokens: HistoryRowToken[] = []; + return { + token: (token) => { + tokens.push(token); + }, + finish: (descriptor) => { + rows.push({ descriptor, tokens }); + }, + }; + }, + options + ); + return rows; + } + function shortReads(limit: number) { + const open = fs.open; + spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + const handle = await open(...args); + if (args[0] !== filePath) return handle; + const read = handle.read.bind(handle); + spyOn(handle, "read").mockImplementation( + new Proxy(read, { + apply(target, receiver, values: unknown[]) { + if (typeof values[2] === "number") values[2] = Math.min(values[2], limit); + return Reflect.apply(target, receiver, values) as ReturnType; + }, + }) + ); + return handle; + }); + } + + it.each([1, 2, 3, 7, 16, 64, SESSION_HISTORY_SCAN_CHUNK_BYTES])( + "preserves nested values, escapes and UTF-8 with %d-byte reads", + async (limit) => { + shortReads(limit); + const texts = [ + JSON.stringify({ + escapedkey: '€😀\ud800\n\\"', + nested: [null, true, false, -12.5e2, { n: 0 }], + }) + .replace("escapedkey", "escaped\\u006bey") + .replace("😀", "😀\\ud83d\\ude00"), + '"root string"', + "true", + "false", + "null", + "-0", + "1.125e-15", + "[]", + "{}", + ]; + await fs.writeFile(filePath, texts.join("\n")); + const rows = await collect(); + expect(rows.map(({ tokens }) => assemble(tokens))).toEqual( + texts.map((text) => JSON.parse(text) as unknown) + ); + expect( + rows.every( + ({ descriptor }) => + descriptor.validJson && descriptor.validUtf8 && !descriptor.hasDuplicateKeys + ) + ).toBe(true); + let start = 0; + rows.forEach(({ descriptor }, i) => { + const byteLength = Buffer.byteLength(texts[i]); + const terminatedByLf = i < texts.length - 1; + expect(descriptor).toMatchObject({ + start, + end: start + byteLength + Number(terminatedByLf), + byteLength, + sha256: digest(texts[i]), + terminatedByLf, + }); + start = descriptor.end; + }); + } + ); + + it("uses the same content digest for LF and complete EOF rows without trimming whitespace", async () => { + const text = ' {"a":1}\r'; + await fs.writeFile(filePath, text + "\n" + text); + const rows = await collect(); + expect(rows.map(({ descriptor }) => descriptor.sha256)).toEqual([digest(text), digest(text)]); + expect(rows.map(({ descriptor }) => descriptor.terminatedByLf)).toEqual([true, false]); + expect(rows.every(({ descriptor }) => descriptor.validJson)).toBe(true); + expect(rows[0].descriptor.sha256).not.toBe(digest(text.trim())); + }); + + it.each([ + '{"a":', + "[1,]", + '"unterminated', + '"\\u12"', + "01", + "truex", + "", + " ", + "\uFEFF{}", + "[\u00a01]", + "[\v1]", + ])("reports malformed row %j and recovers at the next LF", async (text) => { + shortReads(1); + await fs.writeFile(filePath, text + '\n{"after":true}\n'); + const rows = await collect(); + expect(rows).toHaveLength(2); + expect(rows[0].descriptor).toMatchObject({ + validJson: false, + validUtf8: true, + sha256: digest(text), + }); + expect(rows[1].descriptor.validJson).toBe(true); + expect(assemble(rows[1].tokens)).toEqual({ after: true }); + }); + + it.each([[0xc3, 0x28], [0xed, 0xa0, 0x80], [0xf0, 0x9f], [0xff]].map((bytes) => ({ bytes })))( + "rejects invalid UTF-8 bytes %j without hiding the next row", + async ({ bytes }) => { + shortReads(1); + const invalid = Buffer.concat([Buffer.from('"'), Buffer.from(bytes), Buffer.from('"')]); + await fs.writeFile(filePath, Buffer.concat([invalid, Buffer.from("\n{}")])); + const rows = await collect(); + expect(rows[0].descriptor).toMatchObject({ + validUtf8: false, + validJson: false, + sha256: digest(invalid), + }); + expect(rows[1].descriptor.validJson).toBe(true); + } + ); + + it("rejects a partial UTF-8 code point at EOF", async () => { + await fs.writeFile(filePath, Buffer.from([34, 0xf0, 0x9f])); + const rows = await collect(); + expect(rows).toHaveLength(1); + expect(rows[0].descriptor).toMatchObject({ + validUtf8: false, + validJson: false, + terminatedByLf: false, + }); + }); + + it("detects duplicate decoded keys per object across split surrogate chunks", async () => { + shortReads(1); + const texts = [ + '{"a":1,"\\u0061":2}', + '{"😀":1,"\\ud83d\\ude00":2}', + '{"\\ud800":1,"\\ud801":2}', + '{"x":{"a":1},"y":{"a":2}}', + ]; + await fs.writeFile(filePath, texts.join("\n")); + const rows = await collect(); + expect(rows.map(({ descriptor }) => descriptor.hasDuplicateKeys)).toEqual([ + true, + true, + false, + false, + ]); + expect(rows.every(({ descriptor }) => descriptor.validJson)).toBe(true); + }); + + it("streams oversized keys, tool-like strings and numbers without full-row allocation", async () => { + const chunk = Buffer.alloc(SESSION_HISTORY_SCAN_CHUNK_BYTES, 120); + const repeats = 64; + const hash = createHash("sha256"); + await using writer = await fs.open(filePath, "w"); + const write = async (bytes: Buffer | string) => { + hash.update(bytes); + await writer.writeFile(bytes); + }; + await write('{"'); + for (let i = 0; i < repeats; i++) await write(chunk); + await write('":{"type":"dynamic-tool","output":"'); + for (let i = 0; i < repeats; i++) await write(chunk); + await write('","number":'); + chunk.fill(49); + for (let i = 0; i < repeats; i++) await write(chunk); + await write("}}"); // A complete crash tail without LF is valid too. + let maxAllocation = 0; + let maxRawCopy = 0; + let maxToken = 0; + let scalarUnits = 0; + let packedScalar = false; + let descriptor: HistoryRowDescriptor | undefined; + for (const allocator of ["alloc", "allocUnsafe", "allocUnsafeSlow"] as const) { + const allocate = Buffer[allocator]; + spyOn(Buffer, allocator).mockImplementation( + new Proxy(allocate, { + apply(target, receiver, args: unknown[]) { + const result = Reflect.apply(target, receiver, args) as Buffer; + maxAllocation = Math.max(maxAllocation, result.length); + return result; + }, + }) + ); + } + const from = Buffer.from.bind(Buffer); + spyOn(Buffer, "from").mockImplementation( + new Proxy(from, { + apply(target, receiver, args: unknown[]) { + const result = Reflect.apply(target, receiver, args) as Buffer; + maxAllocation = Math.max(maxAllocation, result.length); + return result; + }, + }) + ); + const concat = Buffer.concat.bind(Buffer); + spyOn(Buffer, "concat").mockImplementation((...args: Parameters) => { + const result = concat(...args); + maxAllocation = Math.max(maxAllocation, result.length); + return result; + }); + expect( + await scanHistoryRows(filePath, () => ({ + raw(bytes) { + maxRawCopy = Math.max(maxRawCopy, bytes.buffer.byteLength); + new Uint8Array(bytes.buffer).fill(255); + }, + token(token) { + if (["stringValue", "keyValue", "numberValue"].includes(token.name)) packedScalar = true; + if (token.name === "stringChunk" || token.name === "numberChunk") { + maxToken = Math.max(maxToken, token.value.length); + scalarUnits += token.value.length; + } + }, + finish(row) { + descriptor = row; + }, + })) + ).toBe(true); + expect(descriptor).toMatchObject({ + validJson: true, + validUtf8: true, + terminatedByLf: false, + sha256: hash.digest("hex"), + }); + expect(scalarUnits).toBeGreaterThanOrEqual(3 * repeats * chunk.length); + expect(packedScalar).toBe(false); + expect(maxToken).toBeLessThanOrEqual(SESSION_HISTORY_SCAN_CHUNK_BYTES); + expect(maxAllocation).toBeLessThanOrEqual(2 * SESSION_HISTORY_SCAN_CHUNK_BYTES); + expect(maxRawCopy).toBeGreaterThan(0); + expect(maxRawCopy).toBeLessThanOrEqual(SESSION_HISTORY_SCAN_CHUNK_BYTES); + }); + + it.each(["token", "finish", "abort", "throw", "raw-stop", "raw-throw"] as const)( + "closes its file on %s early termination", + async (how) => { + await fs.writeFile(filePath, '{"text":"value"}\n{}'); + const controller = new AbortController(); + const open = fs.open; + let opened: fs.FileHandle | undefined; + spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + const handle = await open(...args); + if (args[0] === filePath) opened = handle; + return handle; + }); + const scanning = scanHistoryRows( + filePath, + () => ({ + raw() { + if (how === "raw-stop") return false; + if (how === "raw-throw") throw new Error("visitor failed"); + }, + token() { + if (how === "token") return false; + if (how === "abort") controller.abort(new Error("stop scan")); + if (how === "throw") throw new Error("visitor failed"); + }, + finish() { + return false; + }, + }), + { signal: controller.signal } + ); + if (how === "abort" || how === "throw" || how === "raw-throw") { + const failure: unknown = await scanning.catch((error: unknown) => error); + expect(failure).toMatchObject({ + message: how === "abort" ? "stop scan" : "visitor failed", + }); + } else expect(await scanning).toBe(false); + const scannerHandle = opened; + expect(scannerHandle?.fd).toBe(-1); + // Other services can open files before this assertion; they do not own the scanner handle. + await using unrelated = await fs.open(`${filePath}.unrelated`, "w+"); + expect(unrelated.fd).toBeGreaterThanOrEqual(0); + expect(opened?.fd).toBe(-1); + } + ); + + it.each(["strict", "replacement"] as const)( + "%s raw taps preserve every content byte and offset after parse/decode errors", + async (decoding) => { + shortReads(3); + const contents = [ + Buffer.from('{"broken": bad, "later":true}'), + Buffer.concat([ + Buffer.from('{"text":"'), + Buffer.from([255]), + Buffer.from('","id":"after"}'), + ]), + Buffer.from("{}"), + ]; + await fs.writeFile( + filePath, + Buffer.concat( + contents.flatMap((row, i) => (i < contents.length - 1 ? [row, Buffer.from("\n")] : [row])) + ) + ); + const observed: Buffer[] = []; + const descriptors: HistoryRowDescriptor[] = []; + await scanHistoryRows( + filePath, + (start) => { + const chunks: Buffer[] = []; + let next = start; + return { + raw(bytes, offset) { + expect(offset).toBe(next); + next += bytes.length; + chunks.push(Buffer.from(bytes)); // This small fixture deliberately collects bounded raw copies. + }, + token() { + // This test inspects framing and completion rather than tokens. + }, + finish(row) { + observed.push(Buffer.concat(chunks)); + descriptors.push(row); + }, + }; + }, + { decoding } + ); + expect(observed).toEqual(contents); + expect(descriptors.map((row) => row.sha256)).toEqual(contents.map(digest)); + expect(descriptors.map((row) => row.validJson)).toEqual([false, false, true]); + } + ); + + it.each([ + { name: "before identity", prefix: '{"text":"', suffix: '","id":"after"}' }, + { name: "inside identity", prefix: '{"id":"a', suffix: 'b"}' }, + { name: "after identity", prefix: '{"id":"before","text":"', suffix: '"}' }, + ])( + "replacement decoding retains values with invalid bytes $name without declaring valid JSON", + async ({ prefix, suffix }) => { + shortReads(1); + const raw = Buffer.concat([ + Buffer.from(prefix), + Buffer.from([237, 160, 128]), + Buffer.from(suffix), + ]); + await fs.writeFile(filePath, raw); + const [strict] = await collect(); + expect(strict.descriptor).toMatchObject({ + validUtf8: false, + validJson: false, + decodedJsonComplete: false, + }); + const [replacement] = await collect({ decoding: "replacement" }); + expect(replacement.descriptor).toMatchObject({ + validUtf8: false, + validJson: false, + decodedJsonComplete: true, + }); + expect(assemble(replacement.tokens)).toEqual(JSON.parse(raw.toString("utf8")) as unknown); + } + ); + + it("replacement decoding still rejects malformed JSON and detects replacement-equivalent keys", async () => { + const duplicate = Buffer.concat([ + Buffer.from('{"'), + Buffer.from([255]), + Buffer.from('":1,"'), + Buffer.from([254]), + Buffer.from('":2}'), + ]); + await fs.writeFile( + filePath, + Buffer.concat([duplicate, Buffer.from('\n{"id":"provisional",oops}')]) + ); + const rows = await collect({ decoding: "replacement" }); + expect(rows[0].descriptor).toMatchObject({ + validUtf8: false, + validJson: false, + decodedJsonComplete: true, + hasDuplicateKeys: true, + }); + expect(assemble(rows[0].tokens)).toEqual(JSON.parse(duplicate.toString("utf8")) as unknown); + expect(rows[1].descriptor).toMatchObject({ + validUtf8: true, + validJson: false, + decodedJsonComplete: false, + }); + }); + + it.each(["strict", "replacement"] as const)( + "snapshots %s decoding before opening the file", + async (decoding) => { + const raw = Buffer.from([34, 255, 34]); + await fs.writeFile(filePath, Buffer.concat([raw, Buffer.from("\n"), raw])); + const options: { decoding: "strict" | "replacement" } = { decoding }; + const rows: HistoryRowDescriptor[] = []; + await scanHistoryRows( + filePath, + () => { + options.decoding = decoding === "strict" ? "replacement" : "strict"; + return { + token() { + // This test inspects framing and completion rather than tokens. + }, + finish(row) { + rows.push(row); + }, + }; + }, + options + ); + expect(rows.map((row) => row.decodedJsonComplete)).toEqual([ + decoding === "replacement", + decoding === "replacement", + ]); + expect(rows.every((row) => !row.validUtf8 && !row.validJson)).toBe(true); + } + ); + + it("isolates raw visitor mutations from tokens, hashes, and buffered later rows", async () => { + const contents = ['{"id":"original"}', '{"id":"next"}']; + await fs.writeFile(filePath, contents.join("\n")); + const values: unknown[] = []; + const descriptors: HistoryRowDescriptor[] = []; + await scanHistoryRows(filePath, () => { + const tokens: HistoryRowToken[] = []; + return { + raw(bytes) { + expect(bytes.byteLength).toBeLessThanOrEqual(SESSION_HISTORY_SCAN_CHUNK_BYTES); + // Even access to the backing buffer must not expose the scanner's unread bytes. + new Uint8Array(bytes.buffer).fill(255); + }, + token(token) { + tokens.push(token); + }, + finish(row) { + values.push(assemble(tokens)); + descriptors.push(row); + }, + }; + }); + expect(values).toEqual(contents.map((row) => JSON.parse(row) as unknown)); + expect(descriptors.map((row) => row.sha256)).toEqual(contents.map(digest)); + expect(descriptors.every((row) => row.validJson && row.validUtf8)).toBe(true); + }); + + it.each([ + { content: "true\nfalse", keepGoing: true }, + { content: "true\n", keepGoing: true }, + { content: "true", keepGoing: true }, + { content: "true\nfalse", keepGoing: false }, + { content: "true\n", keepGoing: false }, + { content: "true", keepGoing: false }, + ])( + "rejects late completion cancellation ($content, continue=$keepGoing)", + async ({ content, keepGoing }) => { + await fs.writeFile(filePath, content); + const controller = new AbortController(); + const reason = { canceled: "after inner completion check" }; + const starts: number[] = []; + const result = await scanHistoryRows( + filePath, + (start) => { + starts.push(start); + return { + token() { + /* Completion ordering is under test. */ + }, + finish() { + const completion = Promise.resolve(keepGoing); + // Attach after the scanner's await reaction: its inner check already ran when abort fires. + queueMicrotask(() => { + completion.then( + () => controller.abort(reason), + (error: unknown) => controller.abort(error) + ); + }); + return completion; + }, + }; + }, + { signal: controller.signal } + ).catch((error: unknown) => error); + expect(result).toBe(reason); + expect(starts).toEqual([0]); + } + ); + + it.each([true, false])( + "observes cancellation during file disposal (continue=%s)", + async (keepGoing) => { + await fs.writeFile(filePath, "true\n"); + const controller = new AbortController(); + const reason = { canceled: "while disposing file" }; + const open = fs.open; + let opened: fs.FileHandle | undefined; + spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + const handle = await open(...args); + if (args[0] !== filePath) return handle; + opened = handle; + const dispose = handle[Symbol.asyncDispose].bind(handle); + spyOn(handle, Symbol.asyncDispose).mockImplementation(async () => { + await dispose(); + controller.abort(reason); + }); + return handle; + }); + const result = await scanHistoryRows( + filePath, + () => ({ + token() { + /* Resource cleanup is under test. */ + }, + finish: () => keepGoing, + }), + { signal: controller.signal } + ).catch((error: unknown) => error); + expect(result).toBe(reason); + expect(opened?.fd).toBe(-1); + } + ); + + it.each(["continue", "stop", "abort", "reject"] as const)( + "awaits row completion before proceeding (%s)", + async (outcome) => { + await fs.writeFile(filePath, "true\nfalse"); + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const controller = new AbortController(); + const starts: number[] = []; + const scanning = scanHistoryRows( + filePath, + (start) => { + starts.push(start); + return { + token() { + // This test inspects framing and completion rather than tokens. + }, + async finish() { + if (start !== 0) return true; + entered.resolve(); + return await release.promise; + }, + }; + }, + { signal: controller.signal } + ).catch((error: unknown) => error); + await entered.promise; + try { + await new Promise((resolve) => setImmediate(resolve)); + expect(starts).toEqual([0]); + } finally { + if (outcome === "abort") controller.abort(new Error("abort during completion")); + if (outcome === "reject") release.reject(new Error("completion failed")); + else release.resolve(outcome !== "stop"); + } + const result = await scanning; + if (outcome === "abort" || outcome === "reject") + expect(result).toMatchObject({ + message: outcome === "abort" ? "abort during completion" : "completion failed", + }); + else expect(result).toBe(outcome === "continue"); + expect(starts).toEqual(outcome === "continue" ? [0, 5] : [0]); + } + ); + + it.each([false, undefined])( + "rejects beginRow cancellation before callbacks returning %s", + async (rawResult) => { + await fs.writeFile(filePath, "true"); + const controller = new AbortController(); + const reason = new Error("canceled while opening row"); + const callbacks: string[] = []; + const result = await scanHistoryRows( + filePath, + () => { + controller.abort(reason); + return { + raw() { + callbacks.push("raw"); + return rawResult; + }, + token() { + callbacks.push("token"); + }, + finish() { + callbacks.push("finish"); + }, + }; + }, + { signal: controller.signal } + ).catch((error: unknown) => error); + expect(result).toBe(reason); + expect(callbacks).toEqual([]); + } + ); + + it.each(["open", "stat"] as const)( + "rejects cancellation during empty-file %s and closes the handle", + async (stage) => { + await fs.writeFile(filePath, ""); + const controller = new AbortController(); + const reason = { canceledDuring: stage }; + const entered = Promise.withResolvers(); + const release = Promise.withResolvers(); + const open = fs.open; + let opened: fs.FileHandle | undefined; + spyOn(fs, "open").mockImplementation(async (...args: Parameters) => { + const handle = await open(...args); + if (args[0] !== filePath) return handle; + opened = handle; + if (stage === "open") { + entered.resolve(); + await release.promise; + } else { + const stat = handle.stat.bind(handle); + spyOn(handle, "stat").mockImplementation( + new Proxy(stat, { + async apply(target, _receiver, args: Parameters) { + entered.resolve(); + await release.promise; + return target(...args); + }, + }) + ); + } + return handle; + }); + const begin = mock(() => { + throw new Error("empty file must not begin a row"); + }); + const scanning = scanHistoryRows(filePath, begin, { signal: controller.signal }).catch( + (error: unknown) => error + ); + await entered.promise; + controller.abort(reason); + release.resolve(); + expect(await scanning).toBe(reason); + expect(begin).not.toHaveBeenCalled(); + expect(opened?.fd).toBe(-1); + } + ); + + it("emits no phantom row for an empty file or a trailing LF", async () => { + await fs.writeFile(filePath, ""); + expect(await collect()).toEqual([]); + await fs.writeFile(filePath, "{}\n"); + expect(await collect()).toHaveLength(1); + }); +}); diff --git a/src/node/services/historyRowScanner.ts b/src/node/services/historyRowScanner.ts new file mode 100644 index 00000000000..80a18acec2b --- /dev/null +++ b/src/node/services/historyRowScanner.ts @@ -0,0 +1,238 @@ +import { createHash, type Hash } from "node:crypto"; +import * as fs from "node:fs/promises"; +import { finished } from "node:stream/promises"; +import { parser, type Token } from "stream-json/parser.js"; +import { SESSION_HISTORY_SCAN_CHUNK_BYTES } from "@/common/constants/contextBudget"; + +export type HistoryRowToken = Readonly; + +export interface HistoryRowDescriptor { + /** Half-open raw byte range, including the delimiter when present. */ + start: number; + end: number; + /** Raw bytes and their SHA-256 exclude only the optional final LF. */ + byteLength: number; + sha256: string; + terminatedByLf: boolean; + validUtf8: boolean; + /** Completion under the selected decoder; replacement decoding does not validate raw JSON. */ + decodedJsonComplete: boolean; + validJson: boolean; + hasDuplicateKeys: boolean; +} + +export interface HistoryRowVisitor { + /** Tokens are provisional until finish confirms a complete, valid row. False stops the scan. */ + token(token: HistoryRowToken): boolean | void; + /** Isolated bounded bytes excluding LF; visitors must bound any retained copies. */ + raw?(bytes: Uint8Array, absoluteOffset: number): boolean | void; + /** Completion is awaited before another row or file read begins. */ + finish(row: HistoryRowDescriptor): boolean | void | Promise; +} + +/** + * Raw JSONL framing only: no message/schema projection or authority is inferred here. + * Scalar strings, keys and numbers are never assembled. Memory is bounded by the read + * chunk plus parser depth and per-object key fingerprints, not strictly constant for + * arbitrarily deep/wide JSON. Visitors must keep their own accumulation bounded too. + * A caller using descriptors as evidence must revalidate file stamps against concurrent writes. + */ +export async function scanHistoryRows( + filePath: string, + beginRow: (start: number) => HistoryRowVisitor, + options: { signal?: AbortSignal; decoding?: "strict" | "replacement" } = {} +): Promise { + const { signal, decoding = "strict" } = options; + signal?.throwIfAborted(); + try { + await using handle = await fs.open(filePath, "r"); + signal?.throwIfAborted(); + const { size } = await handle.stat(); + signal?.throwIfAborted(); + const buffer = Buffer.alloc(SESSION_HISTORY_SCAN_CHUNK_BYTES); + let position = 0; + let row: ReturnType | undefined; + try { + while (position < size) { + signal?.throwIfAborted(); + const { bytesRead } = await handle.read( + buffer, + 0, + Math.min(buffer.length, size - position), + position + ); + if (bytesRead === 0) throw new Error("History ended before its captured size"); + signal?.throwIfAborted(); + let offset = 0; + while (offset < bytesRead) { + signal?.throwIfAborted(); + row ??= createRow(position + offset, beginRow(position + offset), decoding, signal); + const newline = buffer.subarray(0, bytesRead).indexOf(10, offset); + const end = newline === -1 ? bytesRead : newline; + const keepGoing = await row.push(buffer.subarray(offset, end)); + signal?.throwIfAborted(); + if (!keepGoing) return false; + if (newline !== -1) { + if (!(await row.finish(position + newline + 1, true))) return false; + await row.close(); + row = undefined; + } + offset = end + (newline === -1 ? 0 : 1); + } + position += bytesRead; + } + if (row && !(await row.finish(position, false))) return false; + return true; + } finally { + await row?.close(); + } + } finally { + // Resource disposal is asynchronous too; cancellation during cleanup must remain observable. + signal?.throwIfAborted(); + } +} + +function createRow( + start: number, + visitor: HistoryRowVisitor, + decoding: "strict" | "replacement", + signal?: AbortSignal +) { + // Ignore BOM handling so a BOM is passed to the strict JSON parser, never silently erased. + const decoder = new TextDecoder("utf-8", { fatal: decoding === "strict", ignoreBOM: true }); + const utf8Probe = + decoding === "replacement" + ? new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }) + : undefined; + const rawHash = createHash("sha256"); + let byteLength = 0; + let validUtf8 = true; + let decodedJsonComplete = true; + let hasDuplicateKeys = false; + let stopped = false; + let visitorFailed = false; + let visitorError: unknown; + const objects: Array | null> = []; + let keyHash: Hash | undefined; + const stream = parser.asStream({ packValues: false, streamValues: true }); + // Observe errors immediately, including failures before the next awaited write/end. + const completed = finished(stream, { cleanup: true }).catch(() => { + decodedJsonComplete = false; + }); + stream.on("data", (token: Token) => { + if (stopped || visitorFailed || signal?.aborted) return; + switch (token.name) { + case "startObject": + objects.push(new Set()); + break; + case "startArray": + objects.push(null); + break; + case "endObject": + case "endArray": + objects.pop(); + break; + case "startKey": + keyHash = createHash("sha256"); + break; + case "stringChunk": + // UTF-16 code units make escaped/raw keys and split surrogate pairs equivalent, + // while keeping distinct lone surrogates distinct (UTF-8 replacement would not). + keyHash?.update(Buffer.from(token.value, "utf16le")); + break; + case "endKey": { + const key = keyHash!.digest("hex"); + keyHash = undefined; + const keys = objects.at(-1)!; + if (keys.has(key)) hasDuplicateKeys = true; + keys.add(key); + break; + } + default: + break; + } + try { + if (visitor.token(token) === false) { + stopped = true; + stream.destroy(); + } + } catch (error) { + visitorFailed = true; + visitorError = error; + stream.destroy(); + } + }); + const abort = () => stream.destroy(); + signal?.addEventListener("abort", abort, { once: true }); + const check = () => { + signal?.throwIfAborted(); + if (visitorFailed) throw visitorError; + return !stopped; + }; + const write = async (text: string) => { + if (!decodedJsonComplete || !text) return; + await new Promise((resolve) => { + stream.write(text, (error) => { + if (error) decodedJsonComplete = false; + resolve(); + }); + }); + }; + const decode = (bytes?: Uint8Array, streaming = false): string | undefined => { + if (utf8Probe && validUtf8) { + try { + utf8Probe.decode(bytes, { stream: streaming }); + } catch { + validUtf8 = false; + } + } + if (!validUtf8 && decoding === "strict") return; + try { + return decoder.decode(bytes, { stream: streaming }); + } catch { + validUtf8 = false; + decodedJsonComplete = false; + stream.destroy(); + return; + } + }; + return { + async push(bytes: Buffer) { + if (!check()) return false; + const absoluteOffset = start + byteLength; + rawHash.update(bytes); + byteLength += bytes.length; + if (bytes.length && visitor.raw?.(Uint8Array.from(bytes), absoluteOffset) === false) + return false; + if (!check()) return false; + const text = decode(bytes, true); + if (text !== undefined) await write(text); + return check(); + }, + async finish(end: number, terminatedByLf: boolean) { + if (!check()) return false; + const text = decode(); + if (text !== undefined) await write(text); + if (!stream.destroyed) stream.end(); + await completed; + if (!check()) return false; + const result = await visitor.finish({ + start, + end, + byteLength, + sha256: rawHash.digest("hex"), + terminatedByLf, + validUtf8, + decodedJsonComplete, + validJson: validUtf8 && decodedJsonComplete, + hasDuplicateKeys, + }); + return check() && result !== false; + }, + async close() { + signal?.removeEventListener("abort", abort); + stream.destroy(); + await completed; + }, + }; +}