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;
@@ -135,11 +131,7 @@ export class EventLog {
 		}
 	}
 
-	/**
-	 * 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). */
 	private repairTailSync(): void {
 		const { maxBytes } = this.options;
 		let size: number;

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.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", () => {