refactor(coding-agent): move the semantic-edge ledger onto the event-log substrate

packages/coding-agent/.changes/res-1260-semantic-edges-on-event-log.md

@@ -0,0 +1 @@
+- Moved the semantic-edge ledger's append and replay IO onto the shared event-log substrate. One behavior unified across both ledgers: an unterminated final line is an uncommitted append — skipped on read and truncated before the next append, never newline-completed.

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

@@ -16,16 +16,14 @@ 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.
+ * Appends are single O_APPEND writes (PIPE_BUF-scale atomicity), fsynced only
+ * when the caller needs durability. Tail rule (union of every consumer's
+ * safety): an unterminated final line is an uncommitted append — skipped on
+ * read even when it parses, truncated at its byte offset on the next append,
+ * never newline-completed (completion turns a line a strict parser rejects
+ * into permanent fail-closed interior poison). Interior malformed lines fail
+ * closed. Repair runs only on append, never on read: a viewer may replay a
+ * live writer's log.
  */
 
 export interface EventLogOptions {
@@ -66,17 +64,20 @@ export class EventLog {
 	) {}
 
 	/**
-	 * 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.
+	 * Replay every terminated line through `parse`: throw to reject a line,
+	 * return undefined to skip one. The missing-file decision is made at the
+	 * open, so no check-then-read window exists.
 	 */
-	replaySync<T>(parse: (line: string, index: number) => T | undefined): T[] {
+	replaySync<T>(
+		parse: (line: string, index: number) => T | undefined,
+		options?: { missingFileThrows?: boolean },
+	): 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 [];
+			if (!options?.missingFileThrows && (error as NodeJS.ErrnoException).code === "ENOENT") return [];
 			throw error;
 		}
 		let contents: string;
@@ -92,19 +93,14 @@ export class EventLog {
 		for (let index = 0; index < rawLines.length; index++) {
 			const line = rawLines[index].trim();
 			if (!line) continue;
+			if (index === rawLines.length - 1 && !endsWithNewline) {
+				this.options.log?.("ignored torn final 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;
-			}
+			const event = parse(line, index);
 			if (event !== undefined) events.push(event);
 		}
 		return events;
@@ -128,57 +124,57 @@ export class EventLog {
 		const payload = [...leadLines, ...lines].join("");
 		const handle = openSync(this.path, "a", 0o600);
 		try {
-			writeSync(handle, payload);
+			// writeSync may write short (e.g. ENOSPC after a prefix); a partial
+			// append reported as success would break write-before-action callers.
+			// TODO(unify): lift to utils/atomic-file writeFullySync when #2035 lands.
+			let buffer = Buffer.from(payload, "utf8");
+			while (buffer.length > 0) {
+				buffer = buffer.subarray(writeSync(handle, buffer));
+			}
 			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.
+	 * Truncate an unterminated tail before appending (the module-doc tail
+	 * rule). A repair failure propagates and gates the append: writing through
+	 * an unrepaired tail would weld it to the new record as permanent
+	 * fail-closed interior corruption.
 	 */
 	private repairTailSync(): void {
 		const { maxBytes } = this.options;
 		let size: number;
 		try {
 			size = statSync(this.path).size;
-		} catch {
-			return;
+		} catch (error) {
+			if ((error as NodeJS.ErrnoException).code === "ENOENT") return;
+			throw error;
 		}
 		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.
+		const fd = openSync(this.path, "r+");
 		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.
+			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: unstable bytes
+			// mean a live concurrent writer whose own append terminates the tail
+			// (the documented O_APPEND small-write atomicity trust bucket).
+			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);
 		}
 	}
 }

packages/coding-agent/src/core/semantic-edges.ts

@@ -1,7 +1,7 @@
 import { createHash, randomUUID } from "node:crypto";
-import { appendFileSync, existsSync, mkdirSync, readFileSync, truncateSync } from "node:fs";
-import { dirname, join } from "node:path";
+import { join } from "node:path";
 import type { StreamFn } from "@earendil-works/pi-agent-core";
+import { EventLog } from "./event-log.js";
 
 /**
  * ACP semantic-edges-v1 producer: a durable per-agent ledger of model-request
@@ -158,7 +158,7 @@ export function hashTurnBody(
 export class SemanticEdgeRecorder {
 	readonly sessionId: string;
 	private readonly _ledgerPath?: string;
-	private _pendingRepair?: { truncateToBytes: number } | { terminateLine: true };
+	private readonly _eventLog?: EventLog;
 	private _disabled = false;
 	private _epoch = 0;
 	private _lastTurn?: { requestId: string; epoch: number; bodyHash?: string };
@@ -174,10 +174,11 @@ export class SemanticEdgeRecorder {
 	}) {
 		this.sessionId = options.sessionId;
 		this._ledgerPath = options.ledgerPath;
+		this._eventLog = options.ledgerPath ? new EventLog(options.ledgerPath) : undefined;
 
 		let existing: SemanticEdgeLedgerEvent[] = [];
 		try {
-			existing = this._loadExisting();
+			existing = this._eventLog?.replaySync(parseSemanticEdgeLine) ?? [];
 		} catch (error) {
 			this._disable(error);
 			return;
@@ -330,41 +331,15 @@ export class SemanticEdgeRecorder {
 		}
 	}
 
-	// Construction never mutates the file: a viewer may be reading a live
-	// writer's ledger. Torn-tail repair is deferred to this recorder's first append.
-	private _loadExisting(): SemanticEdgeLedgerEvent[] {
-		if (!this._ledgerPath || !existsSync(this._ledgerPath)) {
-			return [];
-		}
-		const raw = readFileSync(this._ledgerPath, "utf8");
-		const parsed = parseLedgerContent(raw);
-		if (parsed.validLength < raw.length) {
-			this._pendingRepair = { truncateToBytes: Buffer.byteLength(raw.slice(0, parsed.validLength)) };
-		} else if (raw.length > 0 && !raw.endsWith("\n")) {
-			this._pendingRepair = { terminateLine: true };
-		}
-		return parsed.events;
-	}
-
 	// Durable append first, in-memory state second: a failed write must not leave
 	// commit state pointing at events that never reached the ledger.
 	private _append(event: SemanticEdgeLedgerEvent): boolean {
 		if (this._disabled) {
 			return false;
 		}
-		if (this._ledgerPath) {
+		if (this._eventLog) {
 			try {
-				mkdirSync(dirname(this._ledgerPath), { recursive: true });
-				if (this._pendingRepair) {
-					if ("truncateToBytes" in this._pendingRepair) {
-						// Discard the torn tail line so it never becomes mid-file corruption.
-						truncateSync(this._ledgerPath, this._pendingRepair.truncateToBytes);
-					} else {
-						appendFileSync(this._ledgerPath, "\n");
-					}
-					this._pendingRepair = undefined;
-				}
-				appendFileSync(this._ledgerPath, `${JSON.stringify(event)}\n`);
+				this._eventLog.appendSync([event]);
 			} catch (error) {
 				this._disable(error);
 				return false;
@@ -375,39 +350,17 @@ export class SemanticEdgeRecorder {
 	}
 }
 
-/**
- * Parse a ledger, tolerating only a torn final line: malformed AND
- * unterminated (a killed mid-append). A newline-terminated malformed line is
- * real corruption anywhere in the file and throws.
- */
-function parseLedgerContent(raw: string): { events: SemanticEdgeLedgerEvent[]; validLength: number } {
-	const events: SemanticEdgeLedgerEvent[] = [];
-	let offset = 0;
-	let validLength = 0;
-	let lineNumber = 0;
-	while (offset < raw.length) {
-		const newlineIndex = raw.indexOf("\n", offset);
-		const end = newlineIndex === -1 ? raw.length : newlineIndex + 1;
-		const line = raw.slice(offset, end);
-		lineNumber += 1;
-		if (line.trim().length > 0) {
-			try {
-				events.push(JSON.parse(line) as SemanticEdgeLedgerEvent);
-			} catch (error) {
-				if (newlineIndex === -1) {
-					return { events, validLength };
-				}
-				throw new Error(`corrupt semantic-edge ledger line ${lineNumber}: ${String(error)}`);
-			}
-		}
-		offset = end;
-		validLength = end;
+function parseSemanticEdgeLine(line: string, index: number): SemanticEdgeLedgerEvent {
+	try {
+		return JSON.parse(line) as SemanticEdgeLedgerEvent;
+	} catch (error) {
+		throw new Error(`corrupt semantic-edge ledger line ${index + 1}: ${String(error)}`);
 	}
-	return { events, validLength };
 }
 
 export function readSemanticEdgeLedger(path: string): SemanticEdgeLedgerEvent[] {
-	return parseLedgerContent(readFileSync(path, "utf8")).events;
+	// A missing ledger stays loud for explicit readers; the recorder treats absence as empty.
+	return new EventLog(path).replaySync(parseSemanticEdgeLine, { missingFileThrows: true });
 }
 
 interface FoldSession {

packages/coding-agent/test/event-log-faults.test.ts

@@ -0,0 +1,64 @@
+import { appendFileSync, mkdtempSync, readFileSync, rmSync } from "node:fs";
+import { tmpdir } from "node:os";
+import { join } from "node:path";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
+import { EventLog } from "../src/core/event-log.js";
+
+/** Armable fs faults; everything passes through to the real fs by default. */
+const faults: { shortWriteOnce?: boolean; truncateError?: Error } = {};
+
+vi.mock("node:fs", async (importOriginal) => {
+	const actual = await importOriginal<typeof import("node:fs")>();
+	return {
+		...actual,
+		writeSync: ((fd: number, data: Uint8Array) => {
+			if (faults.shortWriteOnce && data.length > 1) {
+				faults.shortWriteOnce = false;
+				return actual.writeSync(fd, data.subarray(0, Math.floor(data.length / 2)));
+			}
+			return actual.writeSync(fd, data);
+		}) as typeof actual.writeSync,
+		ftruncateSync: ((fd: number, len?: number) => {
+			if (faults.truncateError) throw faults.truncateError;
+			return actual.ftruncateSync(fd, len);
+		}) as typeof actual.ftruncateSync,
+	};
+});
+
+describe("event log fault injection", () => {
+	let dir: string;
+
+	beforeEach(() => {
+		dir = mkdtempSync(join(tmpdir(), "prime-event-log-faults-"));
+	});
+
+	afterEach(() => {
+		faults.shortWriteOnce = undefined;
+		faults.truncateError = undefined;
+		rmSync(dir, { recursive: true, force: true });
+	});
+
+	it("persists the full payload even when the kernel writes short", () => {
+		const path = join(dir, "log.jsonl");
+		const log = new EventLog(path);
+		faults.shortWriteOnce = true;
+		log.appendSync([{ v: 1, id: "short-write-survivor" }]);
+
+		expect(new EventLog(path).replaySync((line) => JSON.parse(line) as { id?: string })).toEqual([
+			{ v: 1, id: "short-write-survivor" },
+		]);
+	});
+
+	it("refuses to append through a tail it could not repair", () => {
+		const path = join(dir, "log.jsonl");
+		const log = new EventLog(path);
+		log.appendSync([{ v: 1, id: "committed" }]);
+		appendFileSync(path, '{"torn');
+		const before = readFileSync(path, "utf8");
+
+		faults.truncateError = new Error("EPERM: append-only file");
+		// Writing through would weld the torn tail to the new record forever.
+		expect(() => log.appendSync([{ v: 1, id: "next" }])).toThrow(/EPERM/);
+		expect(readFileSync(path, "utf8")).toBe(before);
+	});
+});

packages/coding-agent/test/event-log.test.ts

@@ -27,9 +27,11 @@ describe("event log substrate", () => {
 		const path = join(dir, "log.jsonl");
 		const log = new EventLog(path);
 		log.appendSync([{ v: 1, keep: true }]);
-		// A newline-completion here would hand this line to strict parsers as
-		// permanent fail-closed interior poison; truncation must win.
+		// Tail rule: uncommitted append — see the EventLog module doc.
 		writeFileSync(path, `${readFileSync(path, "utf8")}{"not":"a valid record"}`);
+		expect(new EventLog(path).replaySync((line) => JSON.parse(line) as { v?: number })).toEqual([
+			{ v: 1, keep: true },
+		]);
 		log.appendSync([{ v: 1, second: true }]);
 		const strict = new EventLog(path).replaySync((line, index) => {
 			const value = JSON.parse(line) as { v?: number };

packages/coding-agent/test/semantic-edges.test.ts

@@ -347,20 +347,22 @@ describe("SemanticEdgeRecorder", () => {
 		expect(requestIds).toEqual([originalId, firstId, secondId]);
 	});
 
-	it("newline-terminates a valid unterminated final line before appending", () => {
+	it("treats a valid unterminated final line as uncommitted: skipped on read, truncated on append", () => {
 		const recorder = createRecorder();
-		const firstId = recorder.startTurnRequest();
+		recorder.startTurnRequest();
 		const path = join(tempDir, "semantic-edges.jsonl");
 		const raw = readFileSync(path, "utf8");
 		rmSync(path);
 		appendFileSync(path, raw.slice(0, -1));
 
+		// Tail rule: uncommitted append — see the EventLog module doc.
+		expect(readSemanticEdgeLedger(path).filter((event) => event.type === "request_started")).toEqual([]);
 		const resumed = createRecorder();
 		const secondId = resumed.startTurnRequest();
 		const requestIds = readSemanticEdgeLedger(path)
 			.filter((event) => event.type === "request_started")
 			.map((event) => (event.type === "request_started" ? event.request_id : ""));
-		expect(requestIds).toEqual([firstId, secondId]);
+		expect(requestIds).toEqual([secondId]);
 	});
 
 	it("treats a newline-terminated malformed final line as corruption, not a torn append", () => {