fix(transcripts): handle read stream errors gracefully in TranscriptsStore (#100524)

* fix(transcripts): handle read stream errors gracefully in TranscriptsStore

* proof(transcripts): add real behavior proof script for store stream error catch

* fix(transcripts): reject with Error in stream error handler

* proof(transcripts): replace wrapper with real EISDIR stream error proof

* fix(transcripts): reject non-ENOENT stream errors after stream close

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 });
+}

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<typeof import("node:fs")>();
+  return {
+    ...actual,
+    createReadStream: (...args: Parameters<typeof actual.createReadStream>) =>
+      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");
+  });
 });

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<TranscriptUtterance[]>((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<void>((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 {