diff --git a/scripts/proof/transcripts-store-stream-errors.mts b/scripts/proof/transcripts-store-stream-errors.mts new file mode 100644 index 000000000000..a16b9b6a2050 --- /dev/null +++ b/scripts/proof/transcripts-store-stream-errors.mts @@ -0,0 +1,54 @@ +// Real behavior proof: TranscriptsStore rejects non-ENOENT stream errors +// gracefully instead of leaking an unhandled rejection or returning partial data. +// +// The proof creates a real transcript session directory where `transcript.jsonl` +// is a directory instead of a file. `fs.createReadStream` on a directory emits +// an EISDIR error on the stream. With the fix, `readUtterancesFromSessionDir` +// rejects with that error after the stream closes. Missing files (ENOENT) still +// resolve to an empty array. Before the fix the unhandled stream error would +// reject the promise before listeners were attached. + +import fs from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { fileURLToPath } from "node:url"; + +const repoRoot = path.dirname(path.dirname(path.dirname(fileURLToPath(import.meta.url)))); +const { TranscriptsStore } = await import(path.join(repoRoot, "src/transcripts/store.js")); + +const tmpDir = await fs.mkdtemp(path.join(os.tmpdir(), "openclaw-proof-transcripts-")); + +const session = { + sessionId: "proof-session", + startedAt: "2026-07-01T00:00:00Z", +}; + +const store = new TranscriptsStore(tmpDir); +const sessionDir = store.sessionDir(session); +await fs.mkdir(sessionDir, { recursive: true }); + +// Make transcript.jsonl a directory. createReadStream on a directory emits +// EISDIR, which exercises the stream error handler in readUtterancesFromDir. +const transcriptPath = path.join(sessionDir, "transcript.jsonl"); +await fs.mkdir(transcriptPath); + +console.log("=== Proof: transcripts store stream error catch ===\n"); +console.log(`Created directory-as-file at: ${transcriptPath}`); +console.log("Calling readUtterancesFromSessionDir with maxUtterances...\n"); + +try { + await store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 10 }); + console.log("\nFAIL: readUtterancesFromSessionDir should have rejected for EISDIR."); + process.exitCode = 1; +} catch (err) { + const message = err instanceof Error ? err.message : String(err); + console.log(`Rejected with: ${message}`); + if (message.includes("EISDIR") || message.includes("is a directory")) { + console.log("\nPASS: EISDIR stream error was caught and rejected after stream close."); + } else { + console.log("\nFAIL: unexpected rejection reason."); + process.exitCode = 1; + } +} finally { + await fs.rm(tmpDir, { recursive: true, force: true }); +} diff --git a/src/transcripts/store.test.ts b/src/transcripts/store.test.ts index d1492198517d..146a88e28fa8 100644 --- a/src/transcripts/store.test.ts +++ b/src/transcripts/store.test.ts @@ -1,7 +1,18 @@ -// Tests TranscriptsStore stream cleanup and transcript reading behavior. import fs from "node:fs"; import path from "node:path"; -import { afterEach, describe, expect, it } from "vitest"; +// Tests TranscriptsStore stream cleanup and transcript reading behavior. +import { PassThrough } from "node:stream"; +import { afterEach, describe, expect, it, vi } from "vitest"; + +const createReadStreamMock = vi.hoisted(() => vi.fn()); +vi.mock("node:fs", async (importOriginal) => { + const actual = await importOriginal(); + return { + ...actual, + createReadStream: (...args: Parameters) => + createReadStreamMock(...args) ?? actual.createReadStream(...args), + }; +}); import { listOpenFileDescriptorsForPath } from "../../src/infra/open-file-descriptors.test-support.js"; import { cleanupTempDirs, makeTempDir } from "../../test/helpers/temp-dir.js"; import { TranscriptsStore } from "./store.js"; @@ -105,4 +116,25 @@ describe("TranscriptsStore.readUtterancesFromSessionDir", () => { expect(leaked).toHaveLength(0); }, ); + + it("rejects non-ENOENT read stream errors", async () => { + const tmpDir = makeTempDir(tempRoots, "openclaw-transcript-test-"); + const store = new TranscriptsStore(tmpDir); + const sessionDir = path.join(tmpDir, "2026-07-01", "session-1"); + fs.mkdirSync(sessionDir, { recursive: true }); + fs.writeFileSync(path.join(sessionDir, "transcript.jsonl"), ""); + + createReadStreamMock.mockImplementation(() => { + const stream = new PassThrough(); + setTimeout(() => { + stream.write(JSON.stringify({ text: "hello", sessionId: "session-1" }) + "\n"); + stream.destroy(new Error("read failed")); + }, 10); + return stream; + }); + + await expect( + store.readUtterancesFromSessionDir(sessionDir, { maxUtterances: 10 }), + ).rejects.toThrow("read failed"); + }); }); diff --git a/src/transcripts/store.ts b/src/transcripts/store.ts index b5fc58815e12..3e5bf389837f 100644 --- a/src/transcripts/store.ts +++ b/src/transcripts/store.ts @@ -188,40 +188,72 @@ export class TranscriptsStore { const transcriptPath = path.join(dir, "transcript.jsonl"); const maxUtterances = resolveOptionalIntegerOption(options.maxUtterances, { min: 1 }); if (maxUtterances !== undefined) { - const utterances: TranscriptUtterance[] = []; - try { + return await new Promise((resolve, reject) => { + const utterances: TranscriptUtterance[] = []; const stream = createReadStream(transcriptPath, { encoding: "utf8" }); const lines = createInterface({ input: stream, crlfDelay: Infinity, }); - try { - for await (const line of lines) { - if (!line) { - continue; - } - utterances.push(JSON.parse(line) as TranscriptUtterance); - if (utterances.length > maxUtterances) { - // Stream and keep only the tail so large transcripts do not require full-file memory. - utterances.shift(); - } + let settled = false; + let emptyForENOENT = false; + let pendingError: Error | undefined; + + const settle = () => { + if (settled) { + return; } - } finally { + settled = true; lines.close(); stream.destroy(); - if (!stream.closed) { - await new Promise((resolve) => { - stream.once("close", () => resolve()); - }); + if (pendingError) { + reject(pendingError); + } else if (emptyForENOENT) { + resolve([]); + } else { + resolve(utterances); } - } - } catch (err) { - if (err && typeof err === "object" && "code" in err && err.code === "ENOENT") { - return []; - } - throw err; - } - return utterances; + }; + const setError = (err: unknown) => { + if (!pendingError) { + pendingError = err instanceof Error ? err : new Error(String(err)); + } + }; + + stream.on("close", settle); + stream.on("error", (err) => { + if (err && typeof err === "object" && "code" in err && err.code === "ENOENT") { + emptyForENOENT = true; + return; + } + setError(err); + stream.destroy(); + }); + lines.on("error", (err) => { + if (err && typeof err === "object" && "code" in err && err.code === "ENOENT") { + emptyForENOENT = true; + return; + } + setError(err); + stream.destroy(); + }); + lines.on("line", (line) => { + if (!line) { + return; + } + try { + utterances.push(JSON.parse(line) as TranscriptUtterance); + } catch (err) { + setError(err); + stream.destroy(); + return; + } + if (utterances.length > maxUtterances) { + // Stream and keep only the tail so large transcripts do not require full-file memory. + utterances.shift(); + } + }); + }); } let raw: string; try {