fix(memory): yield while parsing session transcripts

steipete · steipete · commit 58d60ba3db16 · 2026-05-10T07:21:57.000+01:00
diff --git a/packages/memory-host-sdk/src/host/session-files-yield.test.ts b/packages/memory-host-sdk/src/host/session-files-yield.test.ts
@@ -0,0 +1,53 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+
+const { fileState } = vi.hoisted(() => ({
+  fileState: { raw: "" },
+}));
+
+vi.mock("./fs-utils.js", () => ({
+  readRegularFile: vi.fn(async () => ({
+    buffer: Buffer.from(fileState.raw, "utf-8"),
+  })),
+  statRegularFile: vi.fn(async () => ({
+    missing: false,
+    stat: {
+      mtimeMs: 1,
+      size: Buffer.byteLength(fileState.raw, "utf-8"),
+    },
+  })),
+}));
+
+import { buildSessionEntry } from "./session-files.js";
+
+describe("buildSessionEntry responsiveness", () => {
+  afterEach(() => {
+    fileState.raw = "";
+    vi.clearAllMocks();
+  });
+
+  it("yields while parsing a single large transcript", async () => {
+    fileState.raw = Array.from({ length: 25 }, (_value, index) =>
+      JSON.stringify({
+        type: "message",
+        message: { role: "user", content: `message ${index}` },
+      }),
+    ).join("\n");
+    let immediateRan = false;
+    const immediate = new Promise<void>((resolve) => {
+      setImmediate(() => {
+        immediateRan = true;
+        resolve();
+      });
+    });
+
+    const entry = await buildSessionEntry("/tmp/session.jsonl", {
+      generatedByCronRun: false,
+      generatedByDreamingNarrative: false,
+      parseYieldEveryLines: 10,
+    });
+
+    expect(entry?.lineMap).toHaveLength(25);
+    expect(immediateRan).toBe(true);
+    await immediate;
+  });
+});
diff --git a/packages/memory-host-sdk/src/host/session-files.ts b/packages/memory-host-sdk/src/host/session-files.ts
@@ -27,6 +27,7 @@ const DREAMING_NARRATIVE_RUN_PREFIX = "dreaming-narrative-";
 // toxic line. Wrapped continuation lines still map back to the same JSONL line.
 // This limit applies to content only; the role label adds up to 11 chars.
 const SESSION_EXPORT_CONTENT_WRAP_CHARS = 800;
+const SESSION_ENTRY_PARSE_YIELD_LINES = 250;
 const DIRECT_CRON_PROMPT_RE = /^\[cron:[^\]]+\]\s*/;
 
 export type SessionFileEntry = {
@@ -51,6 +52,8 @@ export type BuildSessionEntryOptions = {
   generatedByDreamingNarrative?: boolean;
   /** Optional preclassification from a caller-managed cron transcript lookup. */
   generatedByCronRun?: boolean;
+  /** Override for tests or specialized callers that need a tighter parse yield cadence. */
+  parseYieldEveryLines?: number;
 };
 
 export type SessionTranscriptClassification = {
@@ -520,6 +523,25 @@ function parseSessionTimestampMs(
   return 0;
 }
 
+function resolveSessionEntryParseYieldLines(opts: BuildSessionEntryOptions): number {
+  const configured = opts.parseYieldEveryLines;
+  if (typeof configured === "number" && Number.isFinite(configured)) {
+    return Math.max(1, Math.floor(configured));
+  }
+  return SESSION_ENTRY_PARSE_YIELD_LINES;
+}
+
+async function yieldSessionEntryParseIfNeeded(
+  lineIndex: number,
+  everyLines: number,
+): Promise<void> {
+  if (lineIndex > 0 && lineIndex % everyLines === 0) {
+    await new Promise<void>((resolve) => {
+      setImmediate(resolve);
+    });
+  }
+}
+
 export async function buildSessionEntry(
   absPath: string,
   opts: BuildSessionEntryOptions = {},
@@ -543,10 +565,10 @@ export async function buildSessionEntry(
       };
     }
     const raw = (await readRegularFile({ filePath: absPath })).buffer.toString("utf-8");
-    const lines = raw.split("\n");
     const collected: string[] = [];
     const lineMap: number[] = [];
     const messageTimestampsMs: number[] = [];
+    const parseYieldEveryLines = resolveSessionEntryParseYieldLines(opts);
     const sessionStoreClassification =
       opts.generatedByDreamingNarrative === undefined || opts.generatedByCronRun === undefined
         ? classifySessionTranscriptFromSessionStore(absPath)
@@ -559,8 +581,12 @@ export async function buildSessionEntry(
       opts.generatedByCronRun ?? sessionStoreClassification?.generatedByCronRun ?? false;
     const allowArchiveContentCronClassification =
       isUsageCountedSessionArchiveTranscriptPath(absPath);
-    for (let jsonlIdx = 0; jsonlIdx < lines.length; jsonlIdx++) {
-      const line = lines[jsonlIdx];
+    for (let jsonlIdx = 0, lineStart = 0; lineStart <= raw.length; jsonlIdx++) {
+      await yieldSessionEntryParseIfNeeded(jsonlIdx, parseYieldEveryLines);
+      const newlineIndex = raw.indexOf("\n", lineStart);
+      const lineEnd = newlineIndex === -1 ? raw.length : newlineIndex;
+      const line = raw.slice(lineStart, lineEnd);
+      lineStart = newlineIndex === -1 ? raw.length + 1 : newlineIndex + 1;
       if (!line.trim()) {
         continue;
       }
