fix(session): repair event-log tail, backfill, and flush

Author: TheGreatAxios

src/session/event-log.test.ts

@@ -174,6 +174,40 @@ describe("createEventStreamWriter", () => {
     expect(records.map((r) => r.seq)).toEqual([0, 1]);
   });
 
+  test("a non-ASCII line plus a torn tail stays two parseable records with distinct seqs", async () => {
+    const dir = tempDir();
+    const writer = createEventStreamWriter(dir);
+    await writer.append({
+      id: "evt-1",
+      type: "session.started",
+      payload: { text: "é" },
+    });
+    await writer.flush();
+
+    fs.appendFileSync(
+      path.join(dir, EVENT_LOG_FILE),
+      '{"id":"torn","type":"eve',
+    );
+
+    const reopened = createEventStreamWriter(dir);
+    const next = await reopened.append({ id: "evt-2", type: "turn.done" });
+    expect(next.seq).toBe(1);
+
+    const lines = rawLines(dir);
+    expect(lines).toHaveLength(2);
+    const parsed = lines.map(
+      (line) => JSON.parse(line) as { id: string; seq: number },
+    );
+    expect(parsed.map((record) => record.id)).toEqual(["evt-1", "evt-2"]);
+    expect(parsed.map((record) => record.seq)).toEqual([0, 1]);
+
+    const records = await reopened.readFrom(-1);
+    expect(records.map((r) => r.id)).toEqual(["evt-1", "evt-2"]);
+    expect(records.map((r) => r.seq)).toEqual([0, 1]);
+    const again = await reopened.append({ id: "evt-3", type: "turn.done" });
+    expect(again.seq).toBe(2);
+  });
+
   test("readFrom returns only events past the cursor", async () => {
     const dir = tempDir();
     const writer = createEventStreamWriter(dir);
@@ -253,10 +287,7 @@ describe("createEventStreamWriter", () => {
     }
     expect(first.parentId).toBeUndefined();
     expect(second.parentId).toBe(first.id);
-    expect((first.payload as { record: unknown }).record).toEqual({
-      role: "user",
-      text: "one",
-    });
+    expect(first.payload).toMatchObject({ role: "user", text: "one" });
   });
 
   test("backfillFromLegacy returns 0 when turns.jsonl is absent", async () => {
@@ -298,6 +329,59 @@ describe("createEventStreamWriter", () => {
     const records = await writer.readFrom(-1);
     expect(records).toHaveLength(3);
   });
+
+  test("backfillFromLegacy keeps identical lines and does not duplicate them on re-run", async () => {
+    const dir = tempDir();
+    const line = '{"role":"user","text":"same"}\n';
+    fs.writeFileSync(path.join(dir, "turns.jsonl"), line + line);
+    const writer = createEventStreamWriter(dir);
+    expect(await writer.backfillFromLegacy()).toBe(2);
+    expect(await writer.backfillFromLegacy()).toBe(0);
+    expect(await writer.readFrom(-1)).toHaveLength(2);
+  });
+
+  test("backfillFromLegacy includes rolled turn segments in order", async () => {
+    const dir = tempDir();
+    fs.writeFileSync(
+      path.join(dir, "turns.jsonl"),
+      '{"role":"user","text":"one"}\n',
+    );
+    fs.writeFileSync(
+      path.join(dir, "turns-0001.jsonl"),
+      '{"role":"user","text":"two"}\n',
+    );
+    const writer = createEventStreamWriter(dir);
+    expect(await writer.backfillFromLegacy()).toBe(2);
+    const records = await writer.readFrom(-1);
+    expect(
+      records.map((record) => (record.payload as { text?: string }).text),
+    ).toEqual(["one", "two"]);
+  });
+
+  test("backfillFromLegacy stores user text where projectPrompt reads it", async () => {
+    const dir = tempDir();
+    fs.writeFileSync(
+      path.join(dir, "turns.jsonl"),
+      '{"role":"user","text":"one"}\n',
+    );
+    const writer = createEventStreamWriter(dir);
+    expect(await writer.backfillFromLegacy()).toBe(1);
+    expect(await writer.projectPrompt()).toBe("one");
+  });
+
+  test("flush rejects when the append it covers rejects", async () => {
+    const dir = tempDir();
+    const writer = createEventStreamWriter(dir);
+    const payload: { self?: unknown } = {};
+    payload.self = payload;
+    const appending = writer.append({ type: "event", payload });
+    const flushing = writer.flush();
+    const settled = await Promise.allSettled([appending, flushing]);
+    expect(settled[0]?.status).toBe("rejected");
+    expect(settled[1]?.status).toBe("rejected");
+    await expect(appending).rejects.toThrow(TypeError);
+    await expect(flushing).rejects.toThrow(TypeError);
+  });
 });
 
 describe("event stream flag gating", () => {

src/session/event-log.ts

@@ -2,6 +2,7 @@ import { appendFile, mkdir, readFile, truncate } from "node:fs/promises";
 import { join } from "node:path";
 
 import { generateSessionId } from "./index.js";
+import { listSegmentFiles } from "./incremental-jsonl.js";
 import { withResolvedDirLock } from "./session-dir-lock.js";
 
 export const EVENT_LOG_FILE = "event-log.jsonl";
@@ -86,6 +87,16 @@ function completeLines(raw: string): string[] {
   return split.filter((line) => line.length > 0);
 }
 
+function legacyPayload(
+  parsed: unknown,
+  legacyKey: string,
+): Record<string, unknown> {
+  if (parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)) {
+    return { ...(parsed as Record<string, unknown>), legacyKey };
+  }
+  return { record: parsed, legacyKey };
+}
+
 function parseRecordLine(line: string): EventStreamRecord {
   return JSON.parse(line) as EventStreamRecord;
 }
@@ -103,8 +114,9 @@ async function readRawOrEmpty(filePath: string): Promise<string> {
 async function dropTornTail(filePath: string, raw: string): Promise<string> {
   if (raw.length === 0 || raw.endsWith("\n")) return raw;
   const lastNl = raw.lastIndexOf("\n");
-  await truncate(filePath, lastNl + 1);
-  return raw.slice(0, lastNl + 1);
+  const kept = raw.slice(0, lastNl + 1);
+  await truncate(filePath, Buffer.byteLength(kept));
+  return kept;
 }
 
 export function createEventStreamWriter(
@@ -147,15 +159,19 @@ export function createEventStreamWriter(
 
   const append = (event: AppendEventInput): Promise<EventStreamRecord> => {
     const run = withResolvedDirLock(dir, () => appendLocked(event));
-    pending = run.then(
-      () => undefined,
-      () => undefined,
-    );
+    pending = run.then(() => undefined);
+    void pending.catch(() => undefined);
     return run;
   };
 
   const flush = async (): Promise<void> => {
-    await pending;
+    const current = pending;
+    try {
+      await current;
+    } catch (err) {
+      if (pending === current) pending = Promise.resolve();
+      throw err;
+    }
   };
 
   const readAll = async (): Promise<EventStreamRecord[]> => {
@@ -195,14 +211,8 @@ export function createEventStreamWriter(
 
   const backfillFromLegacy = (): Promise<number> =>
     withResolvedDirLock(dir, async () => {
-      let raw: string;
-      try {
-        raw = await readFile(join(dir, "turns.jsonl"), "utf8");
-      } catch (err) {
-        if (isENOENT(err)) return 0;
-        throw err;
-      }
-      // Content hash, not position, so a partial re-run cannot duplicate lines.
+      const names = await listSegmentFiles(dir, "turns.jsonl");
+      if (names.length === 0) return 0;
       const records = await readAll();
       const existingKeys = new Set<string>();
       for (const record of records) {
@@ -213,25 +223,32 @@ export function createEventStreamWriter(
       }
       let lastId = records[records.length - 1]?.id;
       let count = 0;
-      for (const line of raw.split("\n")) {
-        if (line.length === 0) continue;
-        const key = hashLine(line);
-        if (existingKeys.has(key)) continue;
-        let record: unknown;
-        try {
-          record = JSON.parse(line);
-        } catch {
-          continue;
+      const occurrences = new Map<string, number>();
+      for (const name of names) {
+        const raw = await readFile(join(dir, name), "utf8");
+        for (const line of raw.split("\n")) {
+          if (line.length === 0) continue;
+          const hash = hashLine(line);
+          const index = occurrences.get(hash) ?? 0;
+          occurrences.set(hash, index + 1);
+          const key = `${hash}:${index}`;
+          if (existingKeys.has(key)) continue;
+          let parsed: unknown;
+          try {
+            parsed = JSON.parse(line);
+          } catch {
+            continue;
+          }
+          const event: AppendEventInput = {
+            type: "turn.legacy",
+            payload: legacyPayload(parsed, key),
+          };
+          if (lastId !== undefined) event.parentId = lastId;
+          const appended = await appendLocked(event);
+          existingKeys.add(key);
+          lastId = appended.id;
+          count += 1;
         }
-        const event: AppendEventInput = {
-          type: "turn.legacy",
-          payload: { record, legacyKey: key },
-        };
-        if (lastId !== undefined) event.parentId = lastId;
-        const appended = await appendLocked(event);
-        existingKeys.add(key);
-        lastId = appended.id;
-        count += 1;
       }
       return count;
     });