From d661a308b6a5812dcb0ee5b41d4bcc07d1769fa5 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 00:17:30 +0200 Subject: [PATCH 1/5] refactor(coding-agent): move the semantic-edge ledger onto the event-log substrate MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The recorder's private append/replay/repair IO is deleted; EventLog owns it, the same move #1987 made for the RLM spawn ledger. One durability rule is unified in the substrate rather than dropped: an unterminated final line is an uncommitted append, skipped on read and truncated before the next append — never newline-completed and never surfaced to a consumer whose next append destroys it. --- .../res-1260-semantic-edges-on-event-log.md | 1 + packages/coding-agent/src/core/event-log.ts | 38 ++++----- .../coding-agent/src/core/semantic-edges.ts | 77 ++++--------------- packages/coding-agent/test/event-log.test.ts | 6 +- .../coding-agent/test/semantic-edges.test.ts | 9 ++- 5 files changed, 45 insertions(+), 86 deletions(-) create mode 100644 packages/coding-agent/.changes/res-1260-semantic-edges-on-event-log.md 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..83d175d0af 100644 --- a/packages/coding-agent/src/core/event-log.ts +++ b/packages/coding-agent/src/core/event-log.ts @@ -18,14 +18,15 @@ import { dirname } from "node:path"; * * 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. + * caller needs durability. An unterminated final line is a crashed writer's + * uncommitted append: replay never surfaces it (even when it parses — data + * the next append truncates must never be acted on) 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 { @@ -66,9 +67,9 @@ 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`. `parse` throws for a line + * it rejects (fail-closed) and returns undefined for a line it deliberately + * skips; an unterminated final line never reaches it. */ replaySync(parse: (line: string, index: number) => T | undefined): T[] { const { maxBytes, maxRecords } = this.options; @@ -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; diff --git a/packages/coding-agent/src/core/semantic-edges.ts b/packages/coding-agent/src/core/semantic-edges.ts index be2dc567d5..2bd66926a2 100644 --- a/packages/coding-agent/src/core/semantic-edges.ts +++ b/packages/coding-agent/src/core/semantic-edges.ts @@ -1,7 +1,8 @@ import { createHash, randomUUID } from "node:crypto"; -import { appendFileSync, existsSync, mkdirSync, readFileSync, truncateSync } from "node:fs"; -import { dirname, join } from "node:path"; +import { statSync } from "node:fs"; +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 +159,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 +175,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 +332,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 +351,18 @@ 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. + statSync(path); + return new EventLog(path).replaySync(parseSemanticEdgeLine); } interface FoldSession { diff --git a/packages/coding-agent/test/event-log.test.ts b/packages/coding-agent/test/event-log.test.ts index 1a37d1ebf7..09726ddd7c 100644 --- a/packages/coding-agent/test/event-log.test.ts +++ b/packages/coding-agent/test/event-log.test.ts @@ -28,8 +28,12 @@ describe("event log substrate", () => { 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. + // permanent fail-closed interior poison; truncation must win, and replay + // must never surface bytes the next append destroys. 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..33c3153b34 100644 --- a/packages/coding-agent/test/semantic-edges.test.ts +++ b/packages/coding-agent/test/semantic-edges.test.ts @@ -347,20 +347,23 @@ 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)); + // Never surfaced even though it parses: the next append destroys these bytes, + // so acting on them would derive edges from a request the ledger disowns. + 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", () => { From 02f1cdd1dfffb4cf7ab2bfdec8d717b11b2bf320 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 11:43:37 +0200 Subject: [PATCH 2/5] fix(coding-agent): make the explicit ledger reader's ENOENT contract atomic readSemanticEdgeLedger probed with statSync before reading through EventLog, which swallows ENOENT; a ledger deleted between the two returned [] instead of throwing. The missing-file decision now lives at the single open (replaySync missingFileThrows), so no check-then-read window exists. --- packages/coding-agent/src/core/event-log.ts | 11 ++++++++--- packages/coding-agent/src/core/semantic-edges.ts | 4 +--- 2 files changed, 9 insertions(+), 6 deletions(-) diff --git a/packages/coding-agent/src/core/event-log.ts b/packages/coding-agent/src/core/event-log.ts index 83d175d0af..a4f5e38089 100644 --- a/packages/coding-agent/src/core/event-log.ts +++ b/packages/coding-agent/src/core/event-log.ts @@ -69,15 +69,20 @@ export class EventLog { /** * Replay every terminated line through `parse`. `parse` throws for a line * it rejects (fail-closed) and returns undefined for a line it deliberately - * skips; an unterminated final line never reaches it. + * skips; an unterminated final line never reaches it. A missing log is an + * empty history for owners and an error for explicit readers; the choice 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; diff --git a/packages/coding-agent/src/core/semantic-edges.ts b/packages/coding-agent/src/core/semantic-edges.ts index 2bd66926a2..198ea40721 100644 --- a/packages/coding-agent/src/core/semantic-edges.ts +++ b/packages/coding-agent/src/core/semantic-edges.ts @@ -1,5 +1,4 @@ import { createHash, randomUUID } from "node:crypto"; -import { statSync } from "node:fs"; import { join } from "node:path"; import type { StreamFn } from "@earendil-works/pi-agent-core"; import { EventLog } from "./event-log.js"; @@ -361,8 +360,7 @@ function parseSemanticEdgeLine(line: string, index: number): SemanticEdgeLedgerE export function readSemanticEdgeLedger(path: string): SemanticEdgeLedgerEvent[] { // A missing ledger stays loud for explicit readers; the recorder treats absence as empty. - statSync(path); - return new EventLog(path).replaySync(parseSemanticEdgeLine); + return new EventLog(path).replaySync(parseSemanticEdgeLine, { missingFileThrows: true }); } interface FoldSession { From 9e6f95989d07dd3634217ed427fac50488d7bc75 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 12:15:09 +0200 Subject: [PATCH 3/5] docs(coding-agent): state the event-log tail rule once The unterminated-tail contract was restated four times (module doc, replaySync doc, two test comments). It now lives once in the module doc; the method doc keeps only its own parse/missing-file semantics and the test comments reference the contract. --- packages/coding-agent/src/core/event-log.ts | 33 +++++++------------ packages/coding-agent/test/event-log.test.ts | 4 +-- .../coding-agent/test/semantic-edges.test.ts | 3 +- 3 files changed, 14 insertions(+), 26 deletions(-) diff --git a/packages/coding-agent/src/core/event-log.ts b/packages/coding-agent/src/core/event-log.ts index a4f5e38089..ab00647017 100644 --- a/packages/coding-agent/src/core/event-log.ts +++ b/packages/coding-agent/src/core/event-log.ts @@ -16,17 +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. An unterminated final line is a crashed writer's - * uncommitted append: replay never surfaces it (even when it parses — data - * the next append truncates must never be acted on) 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 { @@ -67,11 +64,9 @@ export class EventLog { ) {} /** - * Replay every terminated line through `parse`. `parse` throws for a line - * it rejects (fail-closed) and returns undefined for a line it deliberately - * skips; an unterminated final line never reaches it. A missing log is an - * empty history for owners and an error for explicit readers; the choice is - * made at the open so no check-then-read window exists. + * 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, @@ -136,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/test/event-log.test.ts b/packages/coding-agent/test/event-log.test.ts index 09726ddd7c..bdde28c4ae 100644 --- a/packages/coding-agent/test/event-log.test.ts +++ b/packages/coding-agent/test/event-log.test.ts @@ -27,9 +27,7 @@ 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, and replay - // must never surface bytes the next append destroys. + // 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 }, diff --git a/packages/coding-agent/test/semantic-edges.test.ts b/packages/coding-agent/test/semantic-edges.test.ts index 33c3153b34..181d77eee7 100644 --- a/packages/coding-agent/test/semantic-edges.test.ts +++ b/packages/coding-agent/test/semantic-edges.test.ts @@ -355,8 +355,7 @@ describe("SemanticEdgeRecorder", () => { rmSync(path); appendFileSync(path, raw.slice(0, -1)); - // Never surfaced even though it parses: the next append destroys these bytes, - // so acting on them would derive edges from a request the ledger disowns. + // 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(); From ebc1a94a3804b6fb93e92598b5138908899c4b2f Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 16:08:58 +0200 Subject: [PATCH 4/5] fix(coding-agent): write event-log appends fully and gate appends on tail repair writeSync may write short (ENOSPC after a prefix); appendSync now loops until the payload is fully on disk so write-before-action callers never act on a torn record reported as success. A tail-repair failure (e.g. append-only ACL permitting O_APPEND but not r+) now propagates instead of being swallowed: writing through an unrepaired torn tail would weld it to the new record as permanent interior corruption. ENOENT and the concurrent-writer instability path keep their existing semantics. --- packages/coding-agent/src/core/event-log.ts | 58 +++++++++-------- .../test/event-log-faults.test.ts | 64 +++++++++++++++++++ 2 files changed, 95 insertions(+), 27 deletions(-) create mode 100644 packages/coding-agent/test/event-log-faults.test.ts diff --git a/packages/coding-agent/src/core/event-log.ts b/packages/coding-agent/src/core/event-log.ts index ab00647017..5ba797a1fd 100644 --- a/packages/coding-agent/src/core/event-log.ts +++ b/packages/coding-agent/src/core/event-log.ts @@ -124,53 +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 an unterminated tail before appending (the module-doc tail rule). */ + /** + * 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); } } } diff --git a/packages/coding-agent/test/event-log-faults.test.ts b/packages/coding-agent/test/event-log-faults.test.ts new file mode 100644 index 0000000000..57b5f31531 --- /dev/null +++ b/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(); + 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); + }); +}); From e88fc9886fdef22f316dc7d2328bc31cf8268013 Mon Sep 17 00:00:00 2001 From: Sebastian Date: Fri, 4 Sep 2026 16:18:03 +0200 Subject: [PATCH 5/5] fix(coding-agent): reclaim short event-log writes instead of completing them The rlm spawn ledger is multi-writer by documented design (supervisor plus each worker over one file), so completing a short O_APPEND write with a second write could interleave with a rival append and weld two records. A short write now truncates its own torn prefix back off (only while this writer still owns the tail) and fails the append; a torn tail is read-tolerated, a weld is permanent corruption. The append fd opens a+ so the ownership check can read the tail. --- packages/coding-agent/src/core/event-log.ts | 25 +++++++++++++------ .../test/event-log-faults.test.ts | 11 +++++--- 2 files changed, 26 insertions(+), 10 deletions(-) diff --git a/packages/coding-agent/src/core/event-log.ts b/packages/coding-agent/src/core/event-log.ts index 5ba797a1fd..47b65c75d9 100644 --- a/packages/coding-agent/src/core/event-log.ts +++ b/packages/coding-agent/src/core/event-log.ts @@ -122,14 +122,25 @@ export class EventLog { leadLines = (options?.onCreate?.() ?? []).map(serializeLine); } const payload = [...leadLines, ...lines].join(""); - const handle = openSync(this.path, "a", 0o600); + const handle = openSync(this.path, "a+", 0o600); try { - // 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)); + const buffer = Buffer.from(payload, "utf8"); + const written = writeSync(handle, buffer); + if (written < buffer.length) { + // A short O_APPEND write (ENOSPC-class) reported as success would + // break write-before-action callers, and completing it with a second + // write could interleave with a rival process's append (rlm-ledger is + // multi-writer), welding two records. Reclaim the torn prefix while + // we still own the tail, then fail: a torn tail is read-tolerated. + const size = fstatSync(handle).size; + const tail = Buffer.alloc(written); + if ( + readSync(handle, tail, 0, written, size - written) === written && + tail.equals(buffer.subarray(0, written)) + ) { + ftruncateSync(handle, size - written); + } + throw new Error(`event log ${this.path}: short write (${written} of ${buffer.length} bytes)`); } if (options?.durable) fsyncSync(handle); } finally { diff --git a/packages/coding-agent/test/event-log-faults.test.ts b/packages/coding-agent/test/event-log-faults.test.ts index 57b5f31531..6f088a8efa 100644 --- a/packages/coding-agent/test/event-log-faults.test.ts +++ b/packages/coding-agent/test/event-log-faults.test.ts @@ -38,14 +38,19 @@ describe("event log fault injection", () => { rmSync(dir, { recursive: true, force: true }); }); - it("persists the full payload even when the kernel writes short", () => { + it("reclaims its torn prefix and fails the append on a short write", () => { const path = join(dir, "log.jsonl"); const log = new EventLog(path); + log.appendSync([{ v: 1, id: "committed" }]); + const before = readFileSync(path, "utf8"); faults.shortWriteOnce = true; - log.appendSync([{ v: 1, id: "short-write-survivor" }]); + // Completing the write could interleave with a rival process's append; + // failing with a clean file is the only safe terminal state. + expect(() => log.appendSync([{ v: 1, id: "short-write" }])).toThrow(/short write/); + expect(readFileSync(path, "utf8")).toBe(before); expect(new EventLog(path).replaySync((line) => JSON.parse(line) as { id?: string })).toEqual([ - { v: 1, id: "short-write-survivor" }, + { v: 1, id: "committed" }, ]); });