@@ -4,69 +4,26 @@ import { join } from "node:path";
44import { generateSessionId } from "./index.js" ;
55import { withResolvedDirLock } from "./session-dir-lock.js" ;
66
7- /**
8- * Prototype for CL-9459: a single append-only session event stream (revives
9- * CL-4351 for the Hub). One JSON object per line in `event-log.jsonl` inside
10- * the session dir, coexisting with the segmented `turns.jsonl` writer without
11- * touching the existing read path.
12- *
13- * Contract:
14- * - Append-only: one event per line; the file is only grown by `appendFile`.
15- * A torn tail (an unterminated partial line from an interrupted append) is
16- * dropped on read and truncated before the next append so `seq` stays
17- * contiguous.
18- * - Ordering: `seq` is a monotonic per-file index assigned in append order;
19- * file order equals append order.
20- * - Serialization: every append runs inside `withResolvedDirLock`
21- * (`session-dir-lock.ts`), with `seq` derived from the on-disk complete
22- * line count, so concurrent writers on one dir can neither interleave lines
23- * nor duplicate seqs.
24- * - Linkage: events may carry `parentId` pointing at the `id` of the event
25- * they continue.
26- * - Ids: UUIDv7 via the existing `generateSessionId` pattern; a reopened
27- * writer continues an existing log from the on-disk line count.
28- *
29- * Flag-gated and inert by design: `settings.sessionEventLog` with env
30- * override `CORBITS_SESSION_EVENT_LOG`; an explicit "off" env value (`0`,
31- * `false`, `no`, `off`, empty) forces the feature off even when settings
32- * enable it. Nothing in the current write/read path imports this module;
33- * production would enter through `openEventStreamWriter`, which returns null
34- * when the flag is off.
35- */
36-
37- /** On-disk name of the append-only event stream within a session dir. */
387export const EVENT_LOG_FILE = "event-log.jsonl" ;
398
40- /** Env override for the flag (e.g. `CORBITS_SESSION_EVENT_LOG=1`). */
419export const SESSION_EVENT_LOG_ENV = "CORBITS_SESSION_EVENT_LOG" ;
4210
43- /** One event as persisted: a single JSON line in `EVENT_LOG_FILE`. */
4411export interface EventStreamRecord {
45- /** UUIDv7 event id; linkage target for child events (`parentId`). */
4612 id : string ;
47- /** id of the parent event this one continues, when any. */
4813 parentId ?: string ;
49- /** Event kind, e.g. `session.started`, `turn.done`, `turn.error`. */
5014 type : string ;
51- /** Monotonic per-file index assigned by the writer in append order. */
5215 seq : number ;
53- /** Unix ms timestamp assigned by the writer at append time. */
5416 ts : number ;
55- /** Free-form event payload. */
5617 payload ?: unknown ;
5718}
5819
59- /** Caller-supplied fields for one appended event; seq/ts are assigned. */
6020export interface AppendEventInput {
61- /** UUIDv7 id; generated (via `generateSessionId`) when omitted. */
6221 id ?: string ;
6322 parentId ?: string ;
6423 type : string ;
6524 payload ?: unknown ;
6625}
6726
68- /** The 7-method writer contract: append/flush/readFrom/latestCursor plus the
69- * Phase A projections and the legacy backfill. */
7027export interface EventStreamWriter {
7128 append ( event : AppendEventInput ) : Promise < EventStreamRecord > ;
7229 flush ( ) : Promise < void > ;
@@ -78,19 +35,14 @@ export interface EventStreamWriter {
7835}
7936
8037export interface EventStreamWriterOptions {
81- /** Clock override for `ts` (tests). */
8238 now ?: ( ) => number ;
8339}
8440
8541export interface OpenEventStreamWriterOptions extends EventStreamWriterOptions {
86- /** `settings.sessionEventLog` value; defaults to undefined (off). */
8742 settings ?: { sessionEventLog ?: boolean } ;
88- /** Env override for the flag check; defaults to `process.env`. */
8943 flagEnv ?: Record < string , string | undefined > ;
9044}
9145
92- /** Resolve the event-log flag from `settings.sessionEventLog` with an env
93- * override; an explicit "off" env value wins over settings. */
9446export function isSessionEventLogEnabled (
9547 sources : OpenEventStreamWriterOptions = { } ,
9648) : boolean {
@@ -117,7 +69,6 @@ function isENOENT(err: unknown): boolean {
11769 ) ;
11870}
11971
120- /** FNV-1a 32-bit hash of a legacy line; the idempotency key for backfill. */
12172function hashLine ( line : string ) : string {
12273 let hash = 0x811c9dc5 ;
12374 for ( let i = 0 ; i < line . length ; i ++ ) {
@@ -127,8 +78,7 @@ function hashLine(line: string): string {
12778 return ( hash >>> 0 ) . toString ( 16 ) . padStart ( 8 , "0" ) ;
12879}
12980
130- /** Complete, non-empty lines of `raw`; an unterminated final line (a torn
131- * tail) is dropped so reads never throw on it and `seq` stays contiguous. */
81+ // Drop an unterminated tail so a torn append does not shift seq.
13282function completeLines ( raw : string ) : string [ ] {
13383 if ( raw . length === 0 ) return [ ] ;
13484 const split = raw . split ( "\n" ) ;
@@ -149,20 +99,14 @@ async function readRawOrEmpty(filePath: string): Promise<string> {
14999 }
150100}
151101
152- /** Drop a torn tail from disk so the next append starts on a fresh line;
153- * no-op when the file ends with a newline (or is empty). */
102+ // Truncate the torn tail on disk so the next append cannot glue onto it.
154103async function dropTornTail ( filePath : string , raw : string ) : Promise < string > {
155104 if ( raw . length === 0 || raw . endsWith ( "\n" ) ) return raw ;
156105 const lastNl = raw . lastIndexOf ( "\n" ) ;
157106 await truncate ( filePath , lastNl + 1 ) ;
158107 return raw . slice ( 0 , lastNl + 1 ) ;
159108}
160109
161- /** Raw append-only event stream writer (not itself flag-gated; tests and
162- * `openEventStreamWriter` use this). Appends run inside
163- * `withResolvedDirLock` and derive `seq` from the on-disk complete line
164- * count, so every writer on a dir — including concurrent ones — serializes
165- * and assigns unique contiguous seqs. */
166110export function createEventStreamWriter (
167111 dir : string ,
168112 opts : EventStreamWriterOptions = { } ,
@@ -171,8 +115,7 @@ export function createEventStreamWriter(
171115 const filePath = join ( dir , EVENT_LOG_FILE ) ;
172116
173117 let dirReady : Promise < void > | null = null ;
174- // Appends issued by this writer; awaited by flush(). Cross-writer ordering
175- // is the dir lock's job; this only tracks this writer's own in-flight work.
118+ // flush() waits on this writer only; the dir lock orders concurrent writers.
176119 let pending : Promise < unknown > = Promise . resolve ( ) ;
177120
178121 const ensureDir = async ( ) : Promise < void > => {
@@ -182,7 +125,7 @@ export function createEventStreamWriter(
182125 await dirReady ;
183126 } ;
184127
185- // Caller must already hold the dir lock (append/backfillFromLegacy wrap it) .
128+ // Seq is the on-disk line count, so callers must already hold the dir lock.
186129 const appendLocked = async (
187130 event : AppendEventInput ,
188131 ) : Promise < EventStreamRecord > => {
@@ -259,9 +202,7 @@ export function createEventStreamWriter(
259202 if ( isENOENT ( err ) ) return 0 ;
260203 throw err ;
261204 }
262- // Idempotency: lines already backfilled (content hash present in the
263- // stream) are skipped, so re-running after a partial backfill cannot
264- // duplicate legacy events.
205+ // Content hash, not position, so a partial re-run cannot duplicate lines.
265206 const records = await readAll ( ) ;
266207 const existingKeys = new Set < string > ( ) ;
267208 for ( const record of records ) {
@@ -280,8 +221,6 @@ export function createEventStreamWriter(
280221 try {
281222 record = JSON . parse ( line ) ;
282223 } catch {
283- // A torn legacy line is not backfilled; the append-only log keeps
284- // whatever it has already persisted.
285224 continue ;
286225 }
287226 const event : AppendEventInput = {
@@ -308,11 +247,6 @@ export function createEventStreamWriter(
308247 } ;
309248}
310249
311- /** Flag-gated entry point: returns a live writer only when the
312- * `settings.sessionEventLog` / `CORBITS_SESSION_EVENT_LOG` flag is enabled,
313- * otherwise null (inert — nothing is written). This is the seam the
314- * segmented-writer path would call when the feature lands; today nothing
315- * calls it, so the prototype cannot affect the existing read path. */
316250export function openEventStreamWriter (
317251 dir : string ,
318252 opts : OpenEventStreamWriterOptions = { } ,
0 commit comments