Skip to content

Commit db346f5

Browse files
committed
feat(session): flag-gated append-only event log prototype (CL-9459)
Single append-only session event stream (id + parentId linkage, one JSON line per event) that coexists with the segmented turns.jsonl writer behind CORBITS_EVENT_LOG. No read-path change: the module is inert unless the flag is set, and nothing in the existing write/read path imports it. - createEventLogWriter: append-only writer, monotonic seq, reopen continues the on-disk line count - openEventLogWriter: flag-gated seam returning null unless CORBITS_EVENT_LOG is truthy - unit tests: append, order, linkage, reopen, flag gating Design pointer: https://linear.app/abklabs/issue/CL-9459
1 parent 40e19ce commit db346f5

2 files changed

Lines changed: 401 additions & 0 deletions

File tree

‎src/session/event-log.test.ts‎

Lines changed: 163 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,163 @@
1+
import { describe, test, expect } from "bun:test";
2+
import fs from "node:fs";
3+
import os from "node:os";
4+
import path from "node:path";
5+
import {
6+
EVENT_LOG_ENV,
7+
EVENT_LOG_FILE,
8+
createEventLogWriter,
9+
isEventLogEnabled,
10+
openEventLogWriter,
11+
} from "./event-log.js";
12+
13+
function tempDir(): string {
14+
return fs.mkdtempSync(path.join(os.tmpdir(), "event-log-"));
15+
}
16+
17+
function rawLines(dir: string, file: string = EVENT_LOG_FILE): string[] {
18+
const raw = fs.readFileSync(path.join(dir, file), "utf8");
19+
return raw.split("\n").filter((line) => line.length > 0);
20+
}
21+
22+
describe("createEventLogWriter", () => {
23+
test("appends one JSON line per event with id, type, seq, ts, and payload", async () => {
24+
const dir = tempDir();
25+
const writer = createEventLogWriter(dir, { now: () => 1_700_000_000_000 });
26+
const first = await writer.append({
27+
id: "evt-1",
28+
type: "session.started",
29+
payload: { task: "hello" },
30+
});
31+
const second = await writer.append({ id: "evt-2", type: "turn.done" });
32+
33+
expect(first).toEqual({
34+
id: "evt-1",
35+
type: "session.started",
36+
seq: 0,
37+
ts: 1_700_000_000_000,
38+
payload: { task: "hello" },
39+
});
40+
expect(second.seq).toBe(1);
41+
42+
const lines = rawLines(dir);
43+
expect(lines).toHaveLength(2);
44+
for (const line of lines) {
45+
expect(() => JSON.parse(line)).not.toThrow();
46+
expect(line.endsWith("\n")).toBe(false);
47+
}
48+
expect(lines[0]).toBe(JSON.stringify(first));
49+
});
50+
51+
test("preserves append order and assigns seq in order", async () => {
52+
const dir = tempDir();
53+
const writer = createEventLogWriter(dir);
54+
const ids = ["a", "b", "c", "d", "e"];
55+
for (const id of ids) {
56+
await writer.append({ id, type: "event" });
57+
}
58+
59+
const records = await writer.readAll();
60+
expect(records.map((r) => r.id)).toEqual(ids);
61+
expect(records.map((r) => r.seq)).toEqual([0, 1, 2, 3, 4]);
62+
63+
// File order matches append order.
64+
const lines = rawLines(dir).map(
65+
(line) => JSON.parse(line) as { id: string },
66+
);
67+
expect(lines.map((l) => l.id)).toEqual(ids);
68+
});
69+
70+
test("keeps parentId linkage resolvable across the stream", async () => {
71+
const dir = tempDir();
72+
const writer = createEventLogWriter(dir);
73+
await writer.append({ id: "root", type: "session.started" });
74+
await writer.append({ id: "turn-1", parentId: "root", type: "turn.done" });
75+
await writer.append({
76+
id: "turn-2",
77+
parentId: "turn-1",
78+
type: "turn.done",
79+
});
80+
81+
const records = await writer.readAll();
82+
expect(records.map((r) => r.parentId)).toEqual([
83+
undefined,
84+
"root",
85+
"turn-1",
86+
]);
87+
88+
// Walk the chain from the tail back to the root through parentId.
89+
const byId = new Map(records.map((r) => [r.id, r]));
90+
const chain: string[] = [];
91+
let cursor: string | undefined = "turn-2";
92+
while (cursor !== undefined) {
93+
const record = byId.get(cursor);
94+
if (record === undefined) throw new Error(`missing event ${cursor}`);
95+
chain.push(record.id);
96+
cursor = record.parentId;
97+
}
98+
expect(chain).toEqual(["turn-2", "turn-1", "root"]);
99+
});
100+
101+
test("a reopened writer continues seq and preserves prior events", async () => {
102+
const dir = tempDir();
103+
const first = createEventLogWriter(dir);
104+
await first.append({ id: "evt-1", type: "session.started" });
105+
await first.append({ id: "evt-2", type: "turn.done" });
106+
await first.close();
107+
108+
// New writer on the same file: seq continues from the on-disk line count.
109+
const reopened = createEventLogWriter(dir);
110+
const third = await reopened.append({ id: "evt-3", type: "turn.done" });
111+
expect(third.seq).toBe(2);
112+
113+
const records = await reopened.readAll();
114+
expect(records.map((r) => r.id)).toEqual(["evt-1", "evt-2", "evt-3"]);
115+
expect(records.map((r) => r.seq)).toEqual([0, 1, 2]);
116+
await reopened.close();
117+
});
118+
119+
test("appends reject after close", async () => {
120+
const dir = tempDir();
121+
const writer = createEventLogWriter(dir);
122+
await writer.close();
123+
await expect(writer.append({ id: "late", type: "event" })).rejects.toThrow(
124+
"closed",
125+
);
126+
});
127+
});
128+
129+
describe("event log flag gating", () => {
130+
test("isEventLogEnabled treats unset and falsy values as off", () => {
131+
expect(isEventLogEnabled({})).toBe(false);
132+
expect(isEventLogEnabled({ [EVENT_LOG_ENV]: "" })).toBe(false);
133+
expect(isEventLogEnabled({ [EVENT_LOG_ENV]: "0" })).toBe(false);
134+
expect(isEventLogEnabled({ [EVENT_LOG_ENV]: "false" })).toBe(false);
135+
expect(isEventLogEnabled({ [EVENT_LOG_ENV]: "no" })).toBe(false);
136+
expect(isEventLogEnabled({ [EVENT_LOG_ENV]: "1" })).toBe(true);
137+
expect(isEventLogEnabled({ [EVENT_LOG_ENV]: "true" })).toBe(true);
138+
});
139+
140+
test("openEventLogWriter returns null (inert) when the flag is off", () => {
141+
const dir = tempDir();
142+
expect(openEventLogWriter(dir, { flagEnv: {} })).toBeNull();
143+
expect(
144+
openEventLogWriter(dir, { flagEnv: { [EVENT_LOG_ENV]: "0" } }),
145+
).toBeNull();
146+
// Nothing is created on disk while the log is disabled.
147+
expect(fs.existsSync(path.join(dir, EVENT_LOG_FILE))).toBe(false);
148+
});
149+
150+
test("openEventLogWriter returns a live writer when the flag is on", async () => {
151+
const dir = tempDir();
152+
const writer = openEventLogWriter(dir, {
153+
flagEnv: { [EVENT_LOG_ENV]: "1" },
154+
});
155+
expect(writer).not.toBeNull();
156+
if (writer === null)
157+
throw new Error("flag-enabled writer should not be null");
158+
await writer.append({ id: "evt-1", type: "session.started" });
159+
expect(await writer.readAll()).toHaveLength(1);
160+
expect(fs.existsSync(path.join(dir, EVENT_LOG_FILE))).toBe(true);
161+
await writer.close();
162+
});
163+
});

0 commit comments

Comments
 (0)