From a24b1233a9f7e8f2fc3201b5a04a1a4d13700b50 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 11 Oct 2026 22:33:53 -0700 Subject: [PATCH 1/3] feat(session): flag-gated append-only session event log Single append-only session event stream (UUIDv7 id + parentId linkage, one JSON line per event) that coexists with the segmented turns.jsonl writer behind the settings.sessionEventLog / CORBITS_SESSION_EVENT_LOG flag. No read-path change: the module is inert unless the flag is on, and nothing in the existing write/read path imports it. - createEventStreamWriter: append-only writer with append, flush, readFrom, latestCursor, projectTurns, projectPrompt, backfillFromLegacy; a reopened writer continues the on-disk line count - appends serialized with withResolvedDirLock; seq derives from the on-disk complete line count inside the lock, so concurrent writers cannot interleave lines or duplicate seqs - torn tails (unterminated partial lines) are dropped on read and truncated before the next append, keeping seq contiguous - backfillFromLegacy is idempotent via a per-line content hash - openEventStreamWriter: flag-gated seam returning null unless the flag is enabled; explicit "off" env values win over settings - unit tests: append, order, linkage, reopen, cursor reads, concurrency, torn-tail recovery, projections, legacy backfill, flag gating --- src/session/event-log.test.ts | 366 ++++++++++++++++++++++++++++++++++ src/session/event-log.ts | 324 ++++++++++++++++++++++++++++++ 2 files changed, 690 insertions(+) create mode 100644 src/session/event-log.test.ts create mode 100644 src/session/event-log.ts diff --git a/src/session/event-log.test.ts b/src/session/event-log.test.ts new file mode 100644 index 000000000..baee86394 --- /dev/null +++ b/src/session/event-log.test.ts @@ -0,0 +1,366 @@ +import { describe, test, expect } from "bun:test"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { isSessionId } from "./index.js"; +import { + EVENT_LOG_FILE, + SESSION_EVENT_LOG_ENV, + createEventStreamWriter, + isSessionEventLogEnabled, + openEventStreamWriter, +} from "./event-log.js"; + +function tempDir(): string { + return fs.mkdtempSync(path.join(os.tmpdir(), "event-stream-")); +} + +function rawLines(dir: string, file: string = EVENT_LOG_FILE): string[] { + const raw = fs.readFileSync(path.join(dir, file), "utf8"); + return raw.split("\n").filter((line) => line.length > 0); +} + +describe("createEventStreamWriter", () => { + test("appends one JSON line per event with generated UUIDv7 id, seq, ts, payload", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir, { + now: () => 1_700_000_000_000, + }); + const first = await writer.append({ + type: "session.started", + payload: { task: "hello" }, + }); + const second = await writer.append({ id: "fixed-id", type: "turn.done" }); + + // Event ids are UUIDv7 when the caller does not supply one. + expect(isSessionId(first.id)).toBe(true); + expect(first).toEqual({ + id: first.id, + type: "session.started", + seq: 0, + ts: 1_700_000_000_000, + payload: { task: "hello" }, + }); + expect(second.id).toBe("fixed-id"); + expect(second.seq).toBe(1); + + const lines = rawLines(dir); + expect(lines).toHaveLength(2); + for (const line of lines) { + expect(() => JSON.parse(line)).not.toThrow(); + expect(line.endsWith("\n")).toBe(false); + } + expect(lines[0]).toBe(JSON.stringify(first)); + }); + + test("preserves append order and assigns seq in order", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + const ids = ["a", "b", "c", "d", "e"]; + for (const id of ids) { + await writer.append({ id, type: "event" }); + } + + const records = await writer.readFrom(-1); + expect(records.map((r) => r.id)).toEqual(ids); + expect(records.map((r) => r.seq)).toEqual([0, 1, 2, 3, 4]); + + // File order matches append order. + const lines = rawLines(dir).map( + (line) => JSON.parse(line) as { id: string }, + ); + expect(lines.map((l) => l.id)).toEqual(ids); + }); + + test("keeps parentId linkage resolvable across the stream", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + await writer.append({ id: "root", type: "session.started" }); + await writer.append({ id: "turn-1", parentId: "root", type: "turn.done" }); + await writer.append({ + id: "turn-2", + parentId: "turn-1", + type: "turn.done", + }); + + const records = await writer.readFrom(-1); + expect(records.map((r) => r.parentId)).toEqual([ + undefined, + "root", + "turn-1", + ]); + + // Walk the chain from the tail back to the root through parentId. + const byId = new Map(records.map((r) => [r.id, r])); + const chain: string[] = []; + let cursor: string | undefined = "turn-2"; + while (cursor !== undefined) { + const record = byId.get(cursor); + if (record === undefined) throw new Error(`missing event ${cursor}`); + chain.push(record.id); + cursor = record.parentId; + } + expect(chain).toEqual(["turn-2", "turn-1", "root"]); + }); + + test("a reopened writer continues seq and preserves prior events", async () => { + const dir = tempDir(); + const first = createEventStreamWriter(dir); + await first.append({ id: "evt-1", type: "session.started" }); + await first.append({ id: "evt-2", type: "turn.done" }); + await first.flush(); + + // New writer on the same file: seq continues from the on-disk line count. + const reopened = createEventStreamWriter(dir); + const third = await reopened.append({ id: "evt-3", type: "turn.done" }); + expect(third.seq).toBe(2); + + const records = await reopened.readFrom(-1); + expect(records.map((r) => r.id)).toEqual(["evt-1", "evt-2", "evt-3"]); + expect(records.map((r) => r.seq)).toEqual([0, 1, 2]); + expect(await reopened.latestCursor()).toBe(2); + }); + + test("flush awaits previously appended events", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + await writer.append({ id: "a", type: "event" }); + await writer.append({ id: "b", type: "event" }); + await writer.flush(); + expect(rawLines(dir)).toHaveLength(2); + }); + + test("concurrent writers on one dir cannot duplicate seqs or interleave", async () => { + const dir = tempDir(); + const writerA = createEventStreamWriter(dir); + const writerB = createEventStreamWriter(dir); + const events = await Promise.all( + Array.from({ length: 20 }, (_, i) => + (i % 2 === 0 ? writerA : writerB).append({ + id: `evt-${i}`, + type: "event", + }), + ), + ); + + const seqs = events.map((e) => e.seq); + expect(new Set(seqs).size).toBe(seqs.length); + expect(Math.max(...seqs)).toBe(seqs.length - 1); + + const records = await writerA.readFrom(-1); + expect(records).toHaveLength(20); + expect(new Set(records.map((r) => r.id)).size).toBe(20); + expect(records.map((r) => r.seq)).toEqual([...seqs].sort((a, b) => a - b)); + }); + + test("a torn tail on reopen is dropped: reads succeed and append gets a contiguous seq", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + await writer.append({ id: "evt-1", type: "session.started" }); + await writer.flush(); + + // Simulate a crash mid-append: a partial line with no trailing newline. + fs.appendFileSync( + path.join(dir, EVENT_LOG_FILE), + '{"id":"torn","type":"eve', + ); + + // Reads drop the torn tail instead of throwing on it. + const reader = createEventStreamWriter(dir); + expect((await reader.readFrom(-1)).map((r) => r.id)).toEqual(["evt-1"]); + expect(await reader.latestCursor()).toBe(0); + + // The next append truncates the torn bytes and continues the seq. + const reopened = createEventStreamWriter(dir); + const next = await reopened.append({ id: "evt-2", type: "turn.done" }); + expect(next.seq).toBe(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]); + }); + + test("readFrom returns only events past the cursor", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + for (const id of ["a", "b", "c", "d"]) { + await writer.append({ id, type: "event" }); + } + const after = await writer.readFrom(1); + expect(after.map((r) => r.id)).toEqual(["c", "d"]); + expect(await writer.readFrom(3)).toEqual([]); + }); + + test("latestCursor reports -1 on an empty log and the last seq after appends", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + expect(await writer.latestCursor()).toBe(-1); + await writer.append({ id: "a", type: "event" }); + await writer.append({ id: "b", type: "event" }); + expect(await writer.latestCursor()).toBe(1); + }); + + test("projectTurns returns turn events in order", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + await writer.append({ id: "s", type: "session.started" }); + await writer.append({ + id: "t1", + type: "turn.user", + payload: { text: "hi" }, + }); + await writer.append({ + id: "m", + type: "model.selected", + payload: { model: "x" }, + }); + await writer.append({ + id: "t2", + type: "turn.assistant", + payload: { text: "yo" }, + }); + + const turns = (await writer.projectTurns()) as { id: string }[]; + expect(turns.map((t) => t.id)).toEqual(["t1", "t2"]); + }); + + test("projectPrompt joins text payloads in order", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + await writer.append({ + id: "a", + type: "turn.user", + payload: { text: "first" }, + }); + await writer.append({ + id: "b", + type: "turn.user", + payload: { text: "second" }, + }); + await writer.append({ id: "c", type: "turn.user" }); + expect(await writer.projectPrompt()).toBe("first\nsecond"); + }); + + test("backfillFromLegacy appends turns.jsonl as chained turn.legacy events", async () => { + const dir = tempDir(); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + '{"role":"user","text":"one"}\n{"role":"assistant","text":"two"}\n', + ); + const writer = createEventStreamWriter(dir); + expect(await writer.backfillFromLegacy()).toBe(2); + + const records = await writer.readFrom(-1); + expect(records).toHaveLength(2); + expect(records.every((r) => r.type === "turn.legacy")).toBe(true); + const [first, second] = records; + if (first === undefined || second === undefined) { + throw new Error("expected two backfilled events"); + } + expect(first.parentId).toBeUndefined(); + expect(second.parentId).toBe(first.id); + expect((first.payload as { record: unknown }).record).toEqual({ + role: "user", + text: "one", + }); + }); + + test("backfillFromLegacy returns 0 when turns.jsonl is absent", async () => { + const dir = tempDir(); + const writer = createEventStreamWriter(dir); + expect(await writer.backfillFromLegacy()).toBe(0); + }); + + test("backfillFromLegacy is idempotent across re-runs", async () => { + const dir = tempDir(); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + '{"role":"user","text":"one"}\n{"role":"assistant","text":"two"}\n', + ); + const writer = createEventStreamWriter(dir); + expect(await writer.backfillFromLegacy()).toBe(2); + // A second run (e.g. after an interrupted first attempt) adds nothing. + expect(await writer.backfillFromLegacy()).toBe(0); + + const records = await writer.readFrom(-1); + expect(records).toHaveLength(2); + expect(records.map((r) => r.type)).toEqual(["turn.legacy", "turn.legacy"]); + }); + + test("backfillFromLegacy skips only already-backfilled lines", async () => { + const dir = tempDir(); + fs.writeFileSync( + path.join(dir, "turns.jsonl"), + '{"role":"user","text":"one"}\n{"role":"assistant","text":"two"}\n', + ); + const writer = createEventStreamWriter(dir); + expect(await writer.backfillFromLegacy()).toBe(2); + + // Grow the legacy file; only the new line is backfilled. + fs.appendFileSync( + path.join(dir, "turns.jsonl"), + '{"role":"user","text":"three"}\n', + ); + expect(await writer.backfillFromLegacy()).toBe(1); + + const records = await writer.readFrom(-1); + expect(records).toHaveLength(3); + }); +}); + +describe("event stream flag gating", () => { + test("isSessionEventLogEnabled honors settings and env, with explicit off", () => { + expect(isSessionEventLogEnabled({})).toBe(false); + expect( + isSessionEventLogEnabled({ settings: { sessionEventLog: true } }), + ).toBe(true); + expect( + isSessionEventLogEnabled({ settings: { sessionEventLog: false } }), + ).toBe(false); + expect( + isSessionEventLogEnabled({ flagEnv: { [SESSION_EVENT_LOG_ENV]: "1" } }), + ).toBe(true); + expect( + isSessionEventLogEnabled({ + flagEnv: { [SESSION_EVENT_LOG_ENV]: "true" }, + }), + ).toBe(true); + // An explicit "off" env value wins over settings. + for (const off of ["", "0", "false", "no", "off"]) { + expect( + isSessionEventLogEnabled({ + settings: { sessionEventLog: true }, + flagEnv: { [SESSION_EVENT_LOG_ENV]: off }, + }), + ).toBe(false); + } + }); + + test("openEventStreamWriter returns null (inert) when the flag is off", () => { + const dir = tempDir(); + expect( + openEventStreamWriter(dir, { settings: { sessionEventLog: false } }), + ).toBeNull(); + expect( + openEventStreamWriter(dir, { + flagEnv: { [SESSION_EVENT_LOG_ENV]: "off" }, + }), + ).toBeNull(); + // Nothing is created on disk while the flag is off. + expect(fs.existsSync(path.join(dir, EVENT_LOG_FILE))).toBe(false); + }); + + test("openEventStreamWriter returns a live writer when the flag is on", async () => { + const dir = tempDir(); + const writer = openEventStreamWriter(dir, { + settings: { sessionEventLog: true }, + }); + expect(writer).not.toBeNull(); + if (writer === null) + throw new Error("flag-enabled writer should not be null"); + await writer.append({ type: "session.started" }); + expect(await writer.readFrom(-1)).toHaveLength(1); + expect(fs.existsSync(path.join(dir, EVENT_LOG_FILE))).toBe(true); + }); +}); diff --git a/src/session/event-log.ts b/src/session/event-log.ts new file mode 100644 index 000000000..e7cf27f0a --- /dev/null +++ b/src/session/event-log.ts @@ -0,0 +1,324 @@ +import { appendFile, mkdir, readFile, truncate } from "node:fs/promises"; +import { join } from "node:path"; + +import { generateSessionId } from "./index.js"; +import { withResolvedDirLock } from "./session-dir-lock.js"; + +/** + * Prototype for CL-9459: a single append-only session event stream (revives + * CL-4351 for the Hub). One JSON object per line in `event-log.jsonl` inside + * the session dir, coexisting with the segmented `turns.jsonl` writer without + * touching the existing read path. + * + * Contract: + * - Append-only: one event per line; the file is only grown by `appendFile`. + * A torn tail (an unterminated partial line from an interrupted append) is + * dropped on read and truncated before the next append so `seq` stays + * contiguous. + * - Ordering: `seq` is a monotonic per-file index assigned in append order; + * file order equals append order. + * - Serialization: every append runs inside `withResolvedDirLock` + * (`session-dir-lock.ts`), with `seq` derived from the on-disk complete + * line count, so concurrent writers on one dir can neither interleave lines + * nor duplicate seqs. + * - Linkage: events may carry `parentId` pointing at the `id` of the event + * they continue. + * - Ids: UUIDv7 via the existing `generateSessionId` pattern; a reopened + * writer continues an existing log from the on-disk line count. + * + * Flag-gated and inert by design: `settings.sessionEventLog` with env + * override `CORBITS_SESSION_EVENT_LOG`; an explicit "off" env value (`0`, + * `false`, `no`, `off`, empty) forces the feature off even when settings + * enable it. Nothing in the current write/read path imports this module; + * production would enter through `openEventStreamWriter`, which returns null + * when the flag is off. + */ + +/** On-disk name of the append-only event stream within a session dir. */ +export const EVENT_LOG_FILE = "event-log.jsonl"; + +/** Env override for the flag (e.g. `CORBITS_SESSION_EVENT_LOG=1`). */ +export const SESSION_EVENT_LOG_ENV = "CORBITS_SESSION_EVENT_LOG"; + +/** One event as persisted: a single JSON line in `EVENT_LOG_FILE`. */ +export interface EventStreamRecord { + /** UUIDv7 event id; linkage target for child events (`parentId`). */ + id: string; + /** id of the parent event this one continues, when any. */ + parentId?: string; + /** Event kind, e.g. `session.started`, `turn.done`, `turn.error`. */ + type: string; + /** Monotonic per-file index assigned by the writer in append order. */ + seq: number; + /** Unix ms timestamp assigned by the writer at append time. */ + ts: number; + /** Free-form event payload. */ + payload?: unknown; +} + +/** Caller-supplied fields for one appended event; seq/ts are assigned. */ +export interface AppendEventInput { + /** UUIDv7 id; generated (via `generateSessionId`) when omitted. */ + id?: string; + parentId?: string; + type: string; + payload?: unknown; +} + +/** The 7-method writer contract: append/flush/readFrom/latestCursor plus the + * Phase A projections and the legacy backfill. */ +export interface EventStreamWriter { + append(event: AppendEventInput): Promise; + flush(): Promise; + readFrom(cursor: number): Promise; + latestCursor(): Promise; + projectTurns(): Promise; + projectPrompt(): Promise; + backfillFromLegacy(): Promise; +} + +export interface EventStreamWriterOptions { + /** Clock override for `ts` (tests). */ + now?: () => number; +} + +export interface OpenEventStreamWriterOptions extends EventStreamWriterOptions { + /** `settings.sessionEventLog` value; defaults to undefined (off). */ + settings?: { sessionEventLog?: boolean }; + /** Env override for the flag check; defaults to `process.env`. */ + flagEnv?: Record; +} + +/** Resolve the event-log flag from `settings.sessionEventLog` with an env + * override; an explicit "off" env value wins over settings. */ +export function isSessionEventLogEnabled( + sources: OpenEventStreamWriterOptions = {}, +): boolean { + const raw = (sources.flagEnv ?? process.env)[SESSION_EVENT_LOG_ENV]; + if (raw !== undefined) { + const normalized = raw.trim().toLowerCase(); + return ( + normalized !== "" && + normalized !== "0" && + normalized !== "false" && + normalized !== "no" && + normalized !== "off" + ); + } + return sources.settings?.sessionEventLog === true; +} + +function isENOENT(err: unknown): boolean { + return ( + typeof err === "object" && + err !== null && + "code" in err && + (err as { code?: unknown }).code === "ENOENT" + ); +} + +/** FNV-1a 32-bit hash of a legacy line; the idempotency key for backfill. */ +function hashLine(line: string): string { + let hash = 0x811c9dc5; + for (let i = 0; i < line.length; i++) { + hash ^= line.charCodeAt(i); + hash = Math.imul(hash, 0x01000193); + } + return (hash >>> 0).toString(16).padStart(8, "0"); +} + +/** Complete, non-empty lines of `raw`; an unterminated final line (a torn + * tail) is dropped so reads never throw on it and `seq` stays contiguous. */ +function completeLines(raw: string): string[] { + if (raw.length === 0) return []; + const split = raw.split("\n"); + if (!raw.endsWith("\n")) split.pop(); + return split.filter((line) => line.length > 0); +} + +function parseRecordLine(line: string): EventStreamRecord { + return JSON.parse(line) as EventStreamRecord; +} + +async function readRawOrEmpty(filePath: string): Promise { + try { + return await readFile(filePath, "utf8"); + } catch (err) { + if (isENOENT(err)) return ""; + throw err; + } +} + +/** Drop a torn tail from disk so the next append starts on a fresh line; + * no-op when the file ends with a newline (or is empty). */ +async function dropTornTail(filePath: string, raw: string): Promise { + 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); +} + +/** Raw append-only event stream writer (not itself flag-gated; tests and + * `openEventStreamWriter` use this). Appends run inside + * `withResolvedDirLock` and derive `seq` from the on-disk complete line + * count, so every writer on a dir — including concurrent ones — serializes + * and assigns unique contiguous seqs. */ +export function createEventStreamWriter( + dir: string, + opts: EventStreamWriterOptions = {}, +): EventStreamWriter { + const now = opts.now ?? Date.now; + const filePath = join(dir, EVENT_LOG_FILE); + + let dirReady: Promise | null = null; + // Appends issued by this writer; awaited by flush(). Cross-writer ordering + // is the dir lock's job; this only tracks this writer's own in-flight work. + let pending: Promise = Promise.resolve(); + + const ensureDir = async (): Promise => { + if (dirReady === null) { + dirReady = mkdir(dir, { recursive: true }).then(() => undefined); + } + await dirReady; + }; + + // Caller must already hold the dir lock (append/backfillFromLegacy wrap it). + const appendLocked = async ( + event: AppendEventInput, + ): Promise => { + await ensureDir(); + const raw = await readRawOrEmpty(filePath); + const cleaned = await dropTornTail(filePath, raw); + const seq = completeLines(cleaned).length; + const record: EventStreamRecord = { + id: event.id ?? generateSessionId(), + type: event.type, + seq, + ts: now(), + }; + if (event.parentId !== undefined) record.parentId = event.parentId; + if (event.payload !== undefined) record.payload = event.payload; + await appendFile(filePath, `${JSON.stringify(record)}\n`, "utf8"); + return record; + }; + + const append = (event: AppendEventInput): Promise => { + const run = withResolvedDirLock(dir, () => appendLocked(event)); + pending = run.then( + () => undefined, + () => undefined, + ); + return run; + }; + + const flush = async (): Promise => { + await pending; + }; + + const readAll = async (): Promise => { + const raw = await readRawOrEmpty(filePath); + return completeLines(raw).map((line) => parseRecordLine(line)); + }; + + const readFrom = async (cursor: number): Promise => { + const records = await readAll(); + return records.filter((record) => record.seq > cursor); + }; + + const latestCursor = async (): Promise => { + const records = await readAll(); + const last = records[records.length - 1]; + return last === undefined ? -1 : last.seq; + }; + + const projectTurns = async (): Promise => { + const records = await readAll(); + return records + .filter((record) => record.type.startsWith("turn.")) + .map((record) => ({ ...record, ...(record.payload as object) })); + }; + + const projectPrompt = async (): Promise => { + const records = await readAll(); + const texts: string[] = []; + for (const record of records) { + const payload = record.payload as { text?: unknown } | undefined; + if (typeof payload?.text === "string" && payload.text.length > 0) { + texts.push(payload.text); + } + } + return texts.join("\n"); + }; + + const backfillFromLegacy = (): Promise => + withResolvedDirLock(dir, async () => { + let raw: string; + try { + raw = await readFile(join(dir, "turns.jsonl"), "utf8"); + } catch (err) { + if (isENOENT(err)) return 0; + throw err; + } + // Idempotency: lines already backfilled (content hash present in the + // stream) are skipped, so re-running after a partial backfill cannot + // duplicate legacy events. + const records = await readAll(); + const existingKeys = new Set(); + for (const record of records) { + const payload = record.payload as { legacyKey?: unknown } | undefined; + if (typeof payload?.legacyKey === "string") { + existingKeys.add(payload.legacyKey); + } + } + 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 { + // A torn legacy line is not backfilled; the append-only log keeps + // whatever it has already persisted. + continue; + } + 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; + }); + + return { + append, + flush, + readFrom, + latestCursor, + projectTurns, + projectPrompt, + backfillFromLegacy, + }; +} + +/** Flag-gated entry point: returns a live writer only when the + * `settings.sessionEventLog` / `CORBITS_SESSION_EVENT_LOG` flag is enabled, + * otherwise null (inert — nothing is written). This is the seam the + * segmented-writer path would call when the feature lands; today nothing + * calls it, so the prototype cannot affect the existing read path. */ +export function openEventStreamWriter( + dir: string, + opts: OpenEventStreamWriterOptions = {}, +): EventStreamWriter | null { + if (!isSessionEventLogEnabled(opts)) return null; + const writerOpts: EventStreamWriterOptions = {}; + if (opts.now !== undefined) writerOpts.now = opts.now; + return createEventStreamWriter(dir, writerOpts); +} From ef26b9b0d98edfe2e308c20de4c886e25bbb6791 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 16:38:15 -0700 Subject: [PATCH 2/3] refactor(session): drop event-log narration --- src/session/event-log.test.ts | 11 ----- src/session/event-log.ts | 76 +++-------------------------------- 2 files changed, 5 insertions(+), 82 deletions(-) diff --git a/src/session/event-log.test.ts b/src/session/event-log.test.ts index baee86394..c62b848df 100644 --- a/src/session/event-log.test.ts +++ b/src/session/event-log.test.ts @@ -33,7 +33,6 @@ describe("createEventStreamWriter", () => { }); const second = await writer.append({ id: "fixed-id", type: "turn.done" }); - // Event ids are UUIDv7 when the caller does not supply one. expect(isSessionId(first.id)).toBe(true); expect(first).toEqual({ id: first.id, @@ -66,7 +65,6 @@ describe("createEventStreamWriter", () => { expect(records.map((r) => r.id)).toEqual(ids); expect(records.map((r) => r.seq)).toEqual([0, 1, 2, 3, 4]); - // File order matches append order. const lines = rawLines(dir).map( (line) => JSON.parse(line) as { id: string }, ); @@ -91,7 +89,6 @@ describe("createEventStreamWriter", () => { "turn-1", ]); - // Walk the chain from the tail back to the root through parentId. const byId = new Map(records.map((r) => [r.id, r])); const chain: string[] = []; let cursor: string | undefined = "turn-2"; @@ -111,7 +108,6 @@ describe("createEventStreamWriter", () => { await first.append({ id: "evt-2", type: "turn.done" }); await first.flush(); - // New writer on the same file: seq continues from the on-disk line count. const reopened = createEventStreamWriter(dir); const third = await reopened.append({ id: "evt-3", type: "turn.done" }); expect(third.seq).toBe(2); @@ -160,18 +156,15 @@ describe("createEventStreamWriter", () => { await writer.append({ id: "evt-1", type: "session.started" }); await writer.flush(); - // Simulate a crash mid-append: a partial line with no trailing newline. fs.appendFileSync( path.join(dir, EVENT_LOG_FILE), '{"id":"torn","type":"eve', ); - // Reads drop the torn tail instead of throwing on it. const reader = createEventStreamWriter(dir); expect((await reader.readFrom(-1)).map((r) => r.id)).toEqual(["evt-1"]); expect(await reader.latestCursor()).toBe(0); - // The next append truncates the torn bytes and continues the seq. const reopened = createEventStreamWriter(dir); const next = await reopened.append({ id: "evt-2", type: "turn.done" }); expect(next.seq).toBe(1); @@ -280,7 +273,6 @@ describe("createEventStreamWriter", () => { ); const writer = createEventStreamWriter(dir); expect(await writer.backfillFromLegacy()).toBe(2); - // A second run (e.g. after an interrupted first attempt) adds nothing. expect(await writer.backfillFromLegacy()).toBe(0); const records = await writer.readFrom(-1); @@ -297,7 +289,6 @@ describe("createEventStreamWriter", () => { const writer = createEventStreamWriter(dir); expect(await writer.backfillFromLegacy()).toBe(2); - // Grow the legacy file; only the new line is backfilled. fs.appendFileSync( path.join(dir, "turns.jsonl"), '{"role":"user","text":"three"}\n', @@ -326,7 +317,6 @@ describe("event stream flag gating", () => { flagEnv: { [SESSION_EVENT_LOG_ENV]: "true" }, }), ).toBe(true); - // An explicit "off" env value wins over settings. for (const off of ["", "0", "false", "no", "off"]) { expect( isSessionEventLogEnabled({ @@ -347,7 +337,6 @@ describe("event stream flag gating", () => { flagEnv: { [SESSION_EVENT_LOG_ENV]: "off" }, }), ).toBeNull(); - // Nothing is created on disk while the flag is off. expect(fs.existsSync(path.join(dir, EVENT_LOG_FILE))).toBe(false); }); diff --git a/src/session/event-log.ts b/src/session/event-log.ts index e7cf27f0a..d0b6bc295 100644 --- a/src/session/event-log.ts +++ b/src/session/event-log.ts @@ -4,69 +4,26 @@ import { join } from "node:path"; import { generateSessionId } from "./index.js"; import { withResolvedDirLock } from "./session-dir-lock.js"; -/** - * Prototype for CL-9459: a single append-only session event stream (revives - * CL-4351 for the Hub). One JSON object per line in `event-log.jsonl` inside - * the session dir, coexisting with the segmented `turns.jsonl` writer without - * touching the existing read path. - * - * Contract: - * - Append-only: one event per line; the file is only grown by `appendFile`. - * A torn tail (an unterminated partial line from an interrupted append) is - * dropped on read and truncated before the next append so `seq` stays - * contiguous. - * - Ordering: `seq` is a monotonic per-file index assigned in append order; - * file order equals append order. - * - Serialization: every append runs inside `withResolvedDirLock` - * (`session-dir-lock.ts`), with `seq` derived from the on-disk complete - * line count, so concurrent writers on one dir can neither interleave lines - * nor duplicate seqs. - * - Linkage: events may carry `parentId` pointing at the `id` of the event - * they continue. - * - Ids: UUIDv7 via the existing `generateSessionId` pattern; a reopened - * writer continues an existing log from the on-disk line count. - * - * Flag-gated and inert by design: `settings.sessionEventLog` with env - * override `CORBITS_SESSION_EVENT_LOG`; an explicit "off" env value (`0`, - * `false`, `no`, `off`, empty) forces the feature off even when settings - * enable it. Nothing in the current write/read path imports this module; - * production would enter through `openEventStreamWriter`, which returns null - * when the flag is off. - */ - -/** On-disk name of the append-only event stream within a session dir. */ export const EVENT_LOG_FILE = "event-log.jsonl"; -/** Env override for the flag (e.g. `CORBITS_SESSION_EVENT_LOG=1`). */ export const SESSION_EVENT_LOG_ENV = "CORBITS_SESSION_EVENT_LOG"; -/** One event as persisted: a single JSON line in `EVENT_LOG_FILE`. */ export interface EventStreamRecord { - /** UUIDv7 event id; linkage target for child events (`parentId`). */ id: string; - /** id of the parent event this one continues, when any. */ parentId?: string; - /** Event kind, e.g. `session.started`, `turn.done`, `turn.error`. */ type: string; - /** Monotonic per-file index assigned by the writer in append order. */ seq: number; - /** Unix ms timestamp assigned by the writer at append time. */ ts: number; - /** Free-form event payload. */ payload?: unknown; } -/** Caller-supplied fields for one appended event; seq/ts are assigned. */ export interface AppendEventInput { - /** UUIDv7 id; generated (via `generateSessionId`) when omitted. */ id?: string; parentId?: string; type: string; payload?: unknown; } -/** The 7-method writer contract: append/flush/readFrom/latestCursor plus the - * Phase A projections and the legacy backfill. */ export interface EventStreamWriter { append(event: AppendEventInput): Promise; flush(): Promise; @@ -78,19 +35,14 @@ export interface EventStreamWriter { } export interface EventStreamWriterOptions { - /** Clock override for `ts` (tests). */ now?: () => number; } export interface OpenEventStreamWriterOptions extends EventStreamWriterOptions { - /** `settings.sessionEventLog` value; defaults to undefined (off). */ settings?: { sessionEventLog?: boolean }; - /** Env override for the flag check; defaults to `process.env`. */ flagEnv?: Record; } -/** Resolve the event-log flag from `settings.sessionEventLog` with an env - * override; an explicit "off" env value wins over settings. */ export function isSessionEventLogEnabled( sources: OpenEventStreamWriterOptions = {}, ): boolean { @@ -117,7 +69,6 @@ function isENOENT(err: unknown): boolean { ); } -/** FNV-1a 32-bit hash of a legacy line; the idempotency key for backfill. */ function hashLine(line: string): string { let hash = 0x811c9dc5; for (let i = 0; i < line.length; i++) { @@ -127,8 +78,7 @@ function hashLine(line: string): string { return (hash >>> 0).toString(16).padStart(8, "0"); } -/** Complete, non-empty lines of `raw`; an unterminated final line (a torn - * tail) is dropped so reads never throw on it and `seq` stays contiguous. */ +// Drop an unterminated tail so a torn append does not shift seq. function completeLines(raw: string): string[] { if (raw.length === 0) return []; const split = raw.split("\n"); @@ -149,8 +99,7 @@ async function readRawOrEmpty(filePath: string): Promise { } } -/** Drop a torn tail from disk so the next append starts on a fresh line; - * no-op when the file ends with a newline (or is empty). */ +// Truncate the torn tail on disk so the next append cannot glue onto it. async function dropTornTail(filePath: string, raw: string): Promise { if (raw.length === 0 || raw.endsWith("\n")) return raw; const lastNl = raw.lastIndexOf("\n"); @@ -158,11 +107,6 @@ async function dropTornTail(filePath: string, raw: string): Promise { return raw.slice(0, lastNl + 1); } -/** Raw append-only event stream writer (not itself flag-gated; tests and - * `openEventStreamWriter` use this). Appends run inside - * `withResolvedDirLock` and derive `seq` from the on-disk complete line - * count, so every writer on a dir — including concurrent ones — serializes - * and assigns unique contiguous seqs. */ export function createEventStreamWriter( dir: string, opts: EventStreamWriterOptions = {}, @@ -171,8 +115,7 @@ export function createEventStreamWriter( const filePath = join(dir, EVENT_LOG_FILE); let dirReady: Promise | null = null; - // Appends issued by this writer; awaited by flush(). Cross-writer ordering - // is the dir lock's job; this only tracks this writer's own in-flight work. + // flush() waits on this writer only; the dir lock orders concurrent writers. let pending: Promise = Promise.resolve(); const ensureDir = async (): Promise => { @@ -182,7 +125,7 @@ export function createEventStreamWriter( await dirReady; }; - // Caller must already hold the dir lock (append/backfillFromLegacy wrap it). + // Seq is the on-disk line count, so callers must already hold the dir lock. const appendLocked = async ( event: AppendEventInput, ): Promise => { @@ -259,9 +202,7 @@ export function createEventStreamWriter( if (isENOENT(err)) return 0; throw err; } - // Idempotency: lines already backfilled (content hash present in the - // stream) are skipped, so re-running after a partial backfill cannot - // duplicate legacy events. + // Content hash, not position, so a partial re-run cannot duplicate lines. const records = await readAll(); const existingKeys = new Set(); for (const record of records) { @@ -280,8 +221,6 @@ export function createEventStreamWriter( try { record = JSON.parse(line); } catch { - // A torn legacy line is not backfilled; the append-only log keeps - // whatever it has already persisted. continue; } const event: AppendEventInput = { @@ -308,11 +247,6 @@ export function createEventStreamWriter( }; } -/** Flag-gated entry point: returns a live writer only when the - * `settings.sessionEventLog` / `CORBITS_SESSION_EVENT_LOG` flag is enabled, - * otherwise null (inert — nothing is written). This is the seam the - * segmented-writer path would call when the feature lands; today nothing - * calls it, so the prototype cannot affect the existing read path. */ export function openEventStreamWriter( dir: string, opts: OpenEventStreamWriterOptions = {}, From 904438bf067374f99718d1e1771b51e6e3474ef8 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 16:56:44 -0700 Subject: [PATCH 3/3] fix(session): repair event-log tail, backfill, and flush --- src/session/event-log.test.ts | 92 +++++++++++++++++++++++++++++++++-- src/session/event-log.ts | 83 ++++++++++++++++++------------- 2 files changed, 138 insertions(+), 37 deletions(-) diff --git a/src/session/event-log.test.ts b/src/session/event-log.test.ts index c62b848df..d6ef8a33b 100644 --- a/src/session/event-log.test.ts +++ b/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", () => { diff --git a/src/session/event-log.ts b/src/session/event-log.ts index d0b6bc295..ea5c2b56d 100644 --- a/src/session/event-log.ts +++ b/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 { + if (parsed !== null && typeof parsed === "object" && !Array.isArray(parsed)) { + return { ...(parsed as Record), 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 { async function dropTornTail(filePath: string, raw: string): Promise { 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 => { 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 => { - await pending; + const current = pending; + try { + await current; + } catch (err) { + if (pending === current) pending = Promise.resolve(); + throw err; + } }; const readAll = async (): Promise => { @@ -195,14 +211,8 @@ export function createEventStreamWriter( const backfillFromLegacy = (): Promise => 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(); 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(); + 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; });