diff --git a/packages/coding-agent/.changes/res-1260-semantic-edges-on-event-log.md b/packages/coding-agent/.changes/res-1260-semantic-edges-on-event-log.md new file mode 100644 index 0000000000..81ba62115e --- /dev/null +++ b/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. diff --git a/packages/coding-agent/src/core/event-log.ts b/packages/coding-agent/src/core/event-log.ts index 1c856c402c..ab00647017 100644 --- a/packages/coding-agent/src/core/event-log.ts +++ b/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(parse: (line: string, index: number) => T | undefined): T[] { + replaySync( + 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; diff --git a/packages/coding-agent/src/core/semantic-edges.ts b/packages/coding-agent/src/core/semantic-edges.ts index be2dc567d5..198ea40721 100644 --- a/packages/coding-agent/src/core/semantic-edges.ts +++ b/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 { diff --git a/packages/coding-agent/test/event-log.test.ts b/packages/coding-agent/test/event-log.test.ts index 1a37d1ebf7..bdde28c4ae 100644 --- a/packages/coding-agent/test/event-log.test.ts +++ b/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 }; diff --git a/packages/coding-agent/test/semantic-edges.test.ts b/packages/coding-agent/test/semantic-edges.test.ts index c2938d94f5..181d77eee7 100644 --- a/packages/coding-agent/test/semantic-edges.test.ts +++ b/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", () => {