Merge pull request #563 from corbitsdev/cl-5935-display-hydrate-skips-non-null-malformed-turns-lines

Skip malformed turns lines in display-only resume hydrate (CL-5935)

Author: TheGreatAxios

src/session/optimized-context-store.test.ts

@@ -344,6 +344,60 @@ describe("loadRecentTurns", () => {
     const loaded = await loadRecentTurns(dir, 5);
     expect(loaded.map((t) => (t.content[0] as { text: string }).text)).toEqual(["a", "b"]);
   });
+
+  test("skips a non-tail malformed line in the newest segment", async () => {
+    const dir = tempDir();
+    fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl([turn("a")]));
+    fs.writeFileSync(
+      path.join(dir, segmentFileName(TURNS_FILE, 1)),
+      jsonl([turn("b")]) + '{"role":"user","content":[{"type":"te\n' + jsonl([turn("c")]),
+    );
+
+    const loaded = await loadRecentTurns(dir, 5);
+    expect(loaded.map((t) => (t.content[0] as { text: string }).text)).toEqual(["a", "b", "c"]);
+  });
+
+  test("skips a malformed line in an older sealed segment", async () => {
+    const dir = tempDir();
+    fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl([turn("a")]));
+    fs.writeFileSync(
+      path.join(dir, segmentFileName(TURNS_FILE, 1)),
+      jsonl([turn("b")]) + '{"role":"user","content":[{"type":"te\n' + jsonl([turn("c")]),
+    );
+    fs.writeFileSync(path.join(dir, segmentFileName(TURNS_FILE, 2)), jsonl([turn("d")]));
+
+    const loaded = await loadRecentTurns(dir, 5);
+    expect(loaded.map((t) => (t.content[0] as { text: string }).text)).toEqual([
+      "a",
+      "b",
+      "c",
+      "d",
+    ]);
+  });
+
+  test("skips a line that parses as JSON but fails the turn schema", async () => {
+    const dir = tempDir();
+    fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl([turn("a")]));
+    const badTurn = JSON.stringify({ role: "user", content: "not-an-array", timestamp: 1 });
+    fs.writeFileSync(
+      path.join(dir, segmentFileName(TURNS_FILE, 1)),
+      jsonl([turn("b")]) + badTurn + "\n" + jsonl([turn("c")]),
+    );
+
+    const loaded = await loadRecentTurns(dir, 5);
+    expect(loaded.map((t) => (t.content[0] as { text: string }).text)).toEqual(["a", "b", "c"]);
+  });
+
+  test("the reactor's load() stays strict on the same corrupt fixture and names the segment", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    fs.writeFileSync(
+      path.join(dir, TURNS_FILE),
+      jsonl([turn("a")]) + '{"role":"user","content":[{"type":"te\n' + jsonl([turn("b")]),
+    );
+
+    await expect(store.load()).rejects.toThrow(TURNS_FILE);
+  });
 });
 
 describe("createOptimizedContextStore checkpoint", () => {

src/session/optimized-context-store.ts

@@ -79,11 +79,18 @@ function sanitizeCallId(callId: string): string {
  * so a poisoned segment can still yield its usable turns on resume. Errors name
  * `fileName` when provided so diagnostics point at the on-disk file, not a bare
  * Bun JSON token.
+ *
+ * `skipMalformed` is for display-only reads (see loadRecentTurns): a bad line
+ * anywhere in any segment drops that line and keeps the surrounding history,
+ * because a blank transcript is a worse answer than a transcript with a hole in
+ * it. The reactor's own load() must never use it — there, history *is* the live
+ * conversation state and silently dropping a turn would corrupt it (CL-5935).
  */
 function parseSegmentTurns(
   text: string,
   tolerateTornTail: boolean,
   fileName = "turns segment",
+  skipMalformed = false,
 ): ConversationTurn[] {
   if (text.length === 0) return [];
   // POSIX truncate past EOF pads with `\0`. Strip them so the rest of the JSONL
@@ -103,10 +110,18 @@ function parseSegmentTurns(
       raw = JSON.parse(line);
     } catch (cause) {
       if (tolerateTornTail && isLast) break;
+      if (skipMalformed) {
+        log.warn?.(`skipping malformed JSON at ${fileName} line ${i + 1}`);
+        continue;
+      }
       throw new Error(`${fileName} has malformed JSON at line ${i + 1}`, { cause });
     }
     const result = ConversationTurnSchema(raw);
     if (result instanceof type.errors) {
+      if (skipMalformed) {
+        log.warn?.(`skipping unexpected structure at ${fileName} line ${i + 1}`);
+        continue;
+      }
       throw new Error(`${fileName} has unexpected structure at line ${i + 1}: ${result.summary}`);
     }
     turns.push(result);
@@ -233,9 +248,13 @@ export async function loadRecentTurns(dir: string, minTurns: number): Promise<Co
   const collectedNewestFirst: ConversationTurn[][] = [];
   let total = 0;
   for (let i = segments.length - 1; i >= 0; i--) {
-    const text = await fs.promises.readFile(path.join(dir, segments[i]!), "utf-8");
+    const name = segments[i]!;
+    const text = await fs.promises.readFile(path.join(dir, name), "utf-8");
     // Only the active (last) segment can be mid-write; sealed ones are complete.
-    const turns = parseSegmentTurns(text, i === segments.length - 1);
+    // Display-only: skip lines that will not parse rather than losing the whole
+    // transcript to one bad line, and name the segment in any error that does
+    // escape (CL-5935).
+    const turns = parseSegmentTurns(text, i === segments.length - 1, name, true);
     collectedNewestFirst.push(turns);
     total += turns.length;
     if (total >= minTurns) break;