refactor(coding-agent): extract the append-only event-log substrate from the RLM spawn ledger

packages/coding-agent/.changes/event-log-substrate.md

@@ -0,0 +1 @@
+- Extracted the RLM spawn ledger's crash-safety mechanics (single O_APPEND writes with optional fsync, bounded fail-closed replay, torn-final-line tolerance, repair-on-append) into a shared append-only event-log substrate; ledger behavior and public API are unchanged.

packages/coding-agent/src/core/event-log.ts

@@ -0,0 +1,184 @@
+import {
+	closeSync,
+	existsSync,
+	fstatSync,
+	fsyncSync,
+	ftruncateSync,
+	mkdirSync,
+	openSync,
+	readSync,
+	statSync,
+	writeSync,
+} from "node:fs";
+import { dirname } from "node:path";
+
+/**
+ * Append-only JSONL event log: the shared crash-safety substrate under the
+ * RLM spawn ledger and the ACP semantic-edge ledger.
+ *
+ * Appends are single O_APPEND writes (PIPE_BUF-scale sizes, whose atomicity
+ * multi-writer consumers rely on for interleaving), fsynced only when the
+ * caller needs durability. Replay tolerates exactly one torn FINAL line
+ * (rejected by the consumer's parser AND unterminated: a crashed writer's
+ * in-progress append) and fails closed on any malformed interior line.
+ * Repair happens only on append, never on read — a viewer may replay a live
+ * writer's log. EVERY unterminated tail is truncated at its byte offset,
+ * even one that parses as JSON: completing it with a newline would turn a
+ * line a strict consumer parser rejects into permanent fail-closed interior
+ * poison. Unifying consumers keeps the union of their safety behaviors.
+ */
+
+export interface EventLogOptions {
+	/** Fail closed beyond these bounds on every full read, including the repair path. */
+	maxBytes?: number;
+	maxRecords?: number;
+	log?: (message: string) => void;
+}
+
+/** Bounded read through the descriptor: the size check and the allocation see the same fd, so a concurrent grow cannot bypass the bound. */
+function readAllSync(fd: number, maxBytes: number | undefined, path: string): Buffer {
+	const size = fstatSync(fd).size;
+	if (maxBytes !== undefined && size > maxBytes) {
+		throw new Error(`event log ${path} exceeds ${maxBytes} bytes (${size}); refusing to read`);
+	}
+	const buffer = Buffer.alloc(size);
+	let offset = 0;
+	while (offset < size) {
+		const bytesRead = readSync(fd, buffer, offset, size - offset, offset);
+		if (bytesRead === 0) break;
+		offset += bytesRead;
+	}
+	return buffer.subarray(0, offset);
+}
+
+function serializeLine(event: unknown): string {
+	const serialized = JSON.stringify(event);
+	if (typeof serialized !== "string") {
+		throw new TypeError("event is not JSON-serializable");
+	}
+	return `${serialized}\n`;
+}
+
+export class EventLog {
+	constructor(
+		readonly path: string,
+		private readonly options: EventLogOptions = {},
+	) {}
+
+	/**
+	 * Replay every line through `parse`. `parse` throws for a line it rejects
+	 * (fail-closed for interior lines, tolerated for a torn final line) and
+	 * returns undefined for a line it deliberately skips.
+	 */
+	replaySync<T>(parse: (line: string, index: number) => T | undefined): T[] {
+		const { maxBytes, maxRecords } = this.options;
+		let fd: number;
+		try {
+			fd = openSync(this.path, "r");
+		} catch (error) {
+			if ((error as NodeJS.ErrnoException).code === "ENOENT") return [];
+			throw error;
+		}
+		let contents: string;
+		try {
+			contents = readAllSync(fd, maxBytes, this.path).toString("utf8");
+		} finally {
+			closeSync(fd);
+		}
+		const endsWithNewline = contents.endsWith("\n");
+		const rawLines = contents.split("\n");
+		const events: T[] = [];
+		let recordCount = 0;
+		for (let index = 0; index < rawLines.length; index++) {
+			const line = rawLines[index].trim();
+			if (!line) continue;
+			if (maxRecords !== undefined && ++recordCount > maxRecords) {
+				throw new Error(`event log ${this.path} exceeds ${maxRecords} records; refusing to read`);
+			}
+			let event: T | undefined;
+			try {
+				event = parse(line, index);
+			} catch (error) {
+				if (index === rawLines.length - 1 && !endsWithNewline) {
+					this.options.log?.(`ignored torn final line: ${error instanceof Error ? error.message : String(error)}`);
+					continue;
+				}
+				throw error;
+			}
+			if (event !== undefined) events.push(event);
+		}
+		return events;
+	}
+
+	/**
+	 * Append events as one write; `durable` fsyncs before returning. When the
+	 * file is created by this append, `onCreate`'s records lead the payload.
+	 * An unserializable event throws before any byte (including repair) is
+	 * written.
+	 */
+	appendSync(events: unknown[], options?: { durable?: boolean; onCreate?: () => unknown[] }): void {
+		const lines = events.map(serializeLine);
+		mkdirSync(dirname(this.path), { recursive: true, mode: 0o700 });
+		let leadLines: string[] = [];
+		if (existsSync(this.path)) {
+			this.repairTailSync();
+		} else {
+			leadLines = (options?.onCreate?.() ?? []).map(serializeLine);
+		}
+		const payload = [...leadLines, ...lines].join("");
+		const handle = openSync(this.path, "a", 0o600);
+		try {
+			writeSync(handle, payload);
+			if (options?.durable) fsyncSync(handle);
+		} finally {
+			closeSync(handle);
+		}
+	}
+
+	/**
+	 * Truncate a torn final line from a crashed writer before appending:
+	 * otherwise the append would turn a tolerable torn tail into a fail-closed
+	 * interior line. The torn bytes were never readable data.
+	 */
+	private repairTailSync(): void {
+		const { maxBytes } = this.options;
+		let size: number;
+		try {
+			size = statSync(this.path).size;
+		} catch {
+			return;
+		}
+		if (size === 0) return;
+		// Fail closed loudly at the read bound BEFORE the swallowing repair
+		// try-block: an oversized log must never trigger a file-sized
+		// allocation, and the error must not be silenced as a repair failure.
+		if (maxBytes !== undefined && size > maxBytes) {
+			throw new Error(`event log ${this.path} exceeds ${maxBytes} bytes (${size}); refusing to read`);
+		}
+		// All offsets are BYTE offsets on raw buffers: string indices diverge
+		// from byte offsets as soon as any record carries multi-byte UTF-8,
+		// and ftruncate takes bytes.
+		try {
+			const fd = openSync(this.path, "r+");
+			try {
+				const lastByte = Buffer.alloc(1);
+				if (readSync(fd, lastByte, 0, 1, size - 1) !== 1 || lastByte[0] === 0x0a) return;
+				// Truncate guarded by a double-read stability check (cheap
+				// cross-process hardening; a racing append between the check and
+				// the ftruncate stays in the same trust bucket as the documented
+				// O_APPEND small-write atomicity assumption).
+				const first = readAllSync(fd, maxBytes, this.path);
+				const second = readAllSync(fd, maxBytes, this.path);
+				if (second.length !== first.length || !second.equals(first)) return;
+				if (fstatSync(fd).size !== first.length) return;
+				const keep = first.lastIndexOf(0x0a) + 1;
+				ftruncateSync(fd, keep);
+				this.options.log?.(`truncated torn final line (${first.length - keep} bytes)`);
+			} finally {
+				closeSync(fd);
+			}
+		} catch {
+			// Leave the tail for the reader's torn-line tolerance.
+		}
+	}
+}

packages/coding-agent/src/modes/daemon/rlm-ledger.ts

@@ -2,21 +2,17 @@ import { createHash } from "node:crypto";
 import {
 	closeSync,
 	existsSync,
-	fstatSync,
 	fsyncSync,
-	ftruncateSync,
 	linkSync,
 	mkdirSync,
 	openSync,
-	readFileSync,
-	readSync,
 	realpathSync,
 	rmSync,
-	statSync,
 	writeSync,
 } from "node:fs";
 import { readdir, readFile, stat } from "node:fs/promises";
 import { basename, dirname, join, resolve } from "node:path";
+import { EventLog } from "../../core/event-log.js";
 import { canonicalSessionPath } from "../../core/session-lease.js";
 import { getSessionArtifactPathForFile, readSessionInfo, type SessionInfo } from "../../core/session-manager.js";
 import { readFirstLineSync } from "../../utils/file-lines.js";
@@ -288,23 +284,6 @@ function parseLedgerLine(line: string, index: number): RlmLedgerRecord | RlmLedg
 	}
 }
 
-function readAllSync(fd: number): Buffer {
-	const size = fstatSync(fd).size;
-	// Never allocate beyond the read bound: every full read of the ledger,
-	// including the repair path, is bounded the same way replaySync is.
-	if (size > RLM_LEDGER_MAX_BYTES) {
-		throw new Error(`RLM ledger exceeds ${RLM_LEDGER_MAX_BYTES} bytes (${size}); refusing to read`);
-	}
-	const buffer = Buffer.alloc(size);
-	let offset = 0;
-	while (offset < size) {
-		const bytesRead = readSync(fd, buffer, offset, size - offset, offset);
-		if (bytesRead === 0) break;
-		offset += bytesRead;
-	}
-	return buffer.subarray(0, offset);
-}
-
 function edgeKey(childId: string, child: string): string {
 	return `${childId}\u0000${canonicalSessionPath(child)}`;
 }
@@ -317,6 +296,7 @@ function edgeKey(childId: string, child: string): string {
  */
 export class RlmSpawnLedger {
 	private readonly path: string;
+	private readonly eventLog: EventLog;
 	private readonly canonicalSessionsDir: string;
 	private queue: Promise<unknown> = Promise.resolve();
 	private seedAttempted = false;
@@ -329,6 +309,11 @@ export class RlmSpawnLedger {
 	) {
 		this.canonicalSessionsDir = canonicalizeDirPath(sessionsDir);
 		this.path = rlmLedgerPath(agentDir, sessionsDir);
+		this.eventLog = new EventLog(this.path, {
+			maxBytes: RLM_LEDGER_MAX_BYTES,
+			maxRecords: RLM_LEDGER_MAX_RECORDS,
+			log: (message) => this.log(`RLM ledger: ${message}`),
+		});
 	}
 
 	get ledgerPath(): string {
@@ -726,109 +711,24 @@ export class RlmSpawnLedger {
 	}
 
 	private appendRecord(record: RlmLedgerRecord): void {
-		const dir = dirname(this.path);
-		mkdirSync(dir, { recursive: true, mode: 0o700 });
-		const isNew = !existsSync(this.path);
-		if (!isNew) {
-			// Repair a torn final line from a crashed writer before appending:
-			// otherwise the next append would turn a tolerable torn tail into a
-			// fail-closed interior line. The torn bytes were never readable data.
-			this.truncateTornTailSync();
-		}
-		const handle = openSync(this.path, "a", 0o600);
-		try {
-			if (isNew) {
-				const meta: RlmLedgerMetaRecord = {
-					v: 1,
-					op: "meta",
-					at: nowIso(),
-					sessionsDir: this.canonicalSessionsDir,
-				};
-				writeSync(handle, `${JSON.stringify(meta)}\n`);
-			}
-			writeSync(handle, `${JSON.stringify(record)}\n`);
-			fsyncSync(handle);
-		} finally {
-			closeSync(handle);
-		}
-	}
-
-	private truncateTornTailSync(): void {
-		// Fail closed loudly at the read bound BEFORE the swallowing repair
-		// try-block: an oversized ledger must never trigger a file-sized
-		// allocation, and the error must not be silenced as a repair failure.
-		let size: number;
-		try {
-			size = statSync(this.path).size;
-		} catch {
-			return;
-		}
-		if (size > RLM_LEDGER_MAX_BYTES) {
-			throw new Error(`RLM ledger ${this.path} exceeds ${RLM_LEDGER_MAX_BYTES} bytes (${size}); refusing to read`);
-		}
-		// All offsets are BYTE offsets on raw buffers: string indices diverge
-		// from byte offsets as soon as any record carries multi-byte UTF-8
-		// (session names do, in real data), and ftruncate takes bytes.
-		try {
-			const fd = openSync(this.path, "r+");
-			try {
-				const first = readAllSync(fd);
-				if (first.length === 0 || first[first.length - 1] === 0x0a) return;
-				const lastNewline = first.lastIndexOf(0x0a);
-				// Cheap cross-process hardening: only truncate when the bytes are
-				// stable across two reads and the size has not moved under us
-				// (same fd for stat and truncate). A racing append between this
-				// check and the ftruncate remains possible — same trust bucket as
-				// the documented O_APPEND small-write atomicity assumption.
-				const second = readAllSync(fd);
-				if (second.length !== first.length || !second.equals(first)) return;
-				if (fstatSync(fd).size !== first.length) return;
-				ftruncateSync(fd, lastNewline + 1);
-				this.log(`RLM ledger: truncated torn final line (${first.length - lastNewline - 1} bytes)`);
-			} finally {
-				closeSync(fd);
-			}
-		} catch {
-			// Leave the tail for the reader's torn-line tolerance.
-		}
+		this.eventLog.appendSync([record], {
+			durable: true,
+			onCreate: () => [
+				{ v: 1, op: "meta", at: nowIso(), sessionsDir: this.canonicalSessionsDir } satisfies RlmLedgerMetaRecord,
+			],
+		});
 	}
 
 	private replaySync(): Map<string, RlmLedgerEdge> {
 		const edges = new Map<string, RlmLedgerEdge>();
-		if (!existsSync(this.path)) return edges;
-		const size = statSync(this.path).size;
-		if (size > RLM_LEDGER_MAX_BYTES) {
-			throw new Error(`RLM ledger ${this.path} exceeds ${RLM_LEDGER_MAX_BYTES} bytes (${size}); refusing to read`);
-		}
-		const contents = readFileSync(this.path, "utf8");
-		const endsWithNewline = contents.endsWith("\n");
-		const rawLines = contents.split("\n");
-		let recordCount = 0;
-		for (let index = 0; index < rawLines.length; index++) {
-			const line = rawLines[index].trim();
-			if (!line) continue;
-			if (++recordCount > RLM_LEDGER_MAX_RECORDS) {
-				throw new Error(`RLM ledger ${this.path} exceeds ${RLM_LEDGER_MAX_RECORDS} records; refusing to read`);
-			}
-			let record: RlmLedgerRecord | RlmLedgerMetaRecord | undefined;
-			try {
-				record = parseLedgerLine(line, index);
-			} catch (error) {
-				// Exactly one unparseable FINAL line without a trailing newline is
-				// an in-progress or crashed append: log and ignore it. Interior
-				// malformed lines stay fail-closed.
-				if (index === rawLines.length - 1 && !endsWithNewline) {
-					this.log(
-						`RLM ledger: ignored torn final line: ${error instanceof Error ? error.message : String(error)}`,
-					);
-					continue;
-				}
-				throw error;
-			}
+		const records = this.eventLog.replaySync((line, index) => {
+			const record = parseLedgerLine(line, index);
 			if (record === undefined) {
 				this.log(`RLM ledger: skipped record with unknown op on line ${index + 1}`);
-				continue;
 			}
+			return record;
+		});
+		for (const record of records) {
 			if (record.op === "meta") continue;
 			const key = edgeKey(record.childId, record.child);
 			switch (record.op) {