From 9ae089d71bca00786a24eb2ffc5aadb3ebfefb12 Mon Sep 17 00:00:00 2001 From: roboomp Date: Sat, 15 Aug 2026 00:33:48 +0000 Subject: [PATCH] fix(session): repaired torn JSONL appends Rolled partial file appends back to their pre-write size and marked malformed resumed sessions for an atomic rewrite. Retried transient persistence failures from in-memory state and surfaced the first failure in the interactive TUI. Fixes #8596 --- packages/coding-agent/CHANGELOG.md | 4 + .../src/modes/interactive-mode.ts | 10 +++ .../src/session/session-loader.ts | 82 +++++++++++++------ .../src/session/session-manager.ts | 68 ++++++++++++--- .../src/session/session-storage.ts | 23 ++++-- .../test/interactive-mode-lsp-startup.test.ts | 21 +++++ .../test/session-loader-stream.test.ts | 3 + .../session-manager-immediate-persist.test.ts | 69 +++++++++------- .../coding-agent/test/session-storage.test.ts | 18 ++++ packages/utils/src/stream.ts | 10 ++- 10 files changed, 233 insertions(+), 75 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 5597d222b..86482d09f 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Repaired torn session JSONL appends after disk-write failures, rewrote malformed resumed files before their next append, retried transient persistence failures, and surfaced failures in the TUI ([#8596](https://github.com/can1357/oh-my-pi/issues/8596)). + ## [17.3.4] - 2026-08-14 ### Changed diff --git a/packages/coding-agent/src/modes/interactive-mode.ts b/packages/coding-agent/src/modes/interactive-mode.ts index 88c0d14dc..07b02f1ad 100644 --- a/packages/coding-agent/src/modes/interactive-mode.ts +++ b/packages/coding-agent/src/modes/interactive-mode.ts @@ -50,6 +50,7 @@ import { logger, postmortem, prompt, + sanitizeText, setProjectDir, } from "@oh-my-pi/pi-utils"; import chalk from "@oh-my-pi/pi-utils/chalk"; @@ -1108,6 +1109,15 @@ export class InteractiveMode implements InteractiveModeContext { // before initHooksAndCustomTools/#reconcileModeFromSession/#enterPlanMode — // all of which can reach setSessionName during init. this.#eventBusUnsubscribers.push( + this.sessionManager.onPersistenceError(error => { + const detail = truncateToWidth( + replaceTabs(sanitizeText(error.message)).replace(/[\r\n]+/g, " "), + TRUNCATE_LENGTHS.LINE, + ); + this.showWarning( + `Session persistence failed: ${detail}. Unsaved entries remain in memory; persistence will retry on the next entry.`, + ); + }), this.sessionManager.onSessionNameChanged(() => { setSessionTerminalTitle(this.sessionManager.getSessionName(), this.sessionManager.getCwd()); this.#handleSessionAccentInputsChanged(); diff --git a/packages/coding-agent/src/session/session-loader.ts b/packages/coding-agent/src/session/session-loader.ts index 2c2f0fab2..899bb6ae0 100644 --- a/packages/coding-agent/src/session/session-loader.ts +++ b/packages/coding-agent/src/session/session-loader.ts @@ -33,6 +33,15 @@ export interface VisitEntriesFromFileStreamOptions { yieldEveryBytes?: number; /** Yield to the macrotask queue after this many entries have been visited. */ yieldEveryEntries?: number; + /** Called once for every malformed JSONL record skipped by the stream. */ + onMalformedRecord?: () => void; +} + +/** Parsed session entries plus corruption metadata needed by writable loaders. */ +export interface SessionLoadResult { + entries: FileEntry[]; + titleSlot: SessionTitleUpdate | undefined; + malformedRecords: number; } function splitTitleSlot(content: string): { body: string; slot: SessionTitleUpdate | undefined } { @@ -60,13 +69,15 @@ function foldTitleSlot(entries: FileEntry[], slot: SessionTitleUpdate | undefine } /** Parse session JSONL while stripping and folding the optional fixed title slot. */ -export function parseSessionContent(content: string): { - entries: FileEntry[]; - titleSlot: SessionTitleUpdate | undefined; -} { +export function parseSessionContent(content: string): SessionLoadResult { const { body, slot } = splitTitleSlot(content); - const entries = parseJsonlLenient(body) as FileEntry[]; - return { entries: foldTitleSlot(entries, slot), titleSlot: slot }; + let malformedRecords = 0; + const entries = parseJsonlLenient(body, { + onMalformedRecord: () => { + malformedRecords++; + }, + }) as FileEntry[]; + return { entries: foldTitleSlot(entries, slot), titleSlot: slot, malformedRecords }; } /** Parse session JSONL and visit each entry without retaining prior entries. */ @@ -143,6 +154,15 @@ export async function visitEntriesFromFileStream( // Malformed record: skip past the next newline and continue. const nextNewline = buffer.indexOf(0x0a, read); if (nextNewline === -1) break; // rest of the bad line not yet received + let nonWhitespace = false; + for (let index = read; index < nextNewline; index++) { + const byte = buffer[index]; + if (byte !== 0x09 && byte !== 0x0d && byte !== 0x20) { + nonWhitespace = true; + break; + } + } + if (nonWhitespace) options.onMalformedRecord?.(); recordsSeen++; buffer = buffer.subarray(nextNewline + 1); if (recordsSeen >= maxRecords) { @@ -203,15 +223,21 @@ export async function visitEntriesFromFileStream( } /** Exported for testing — the ≥8MiB streaming path (works on any file size). */ -export async function loadEntriesFromFileStream(filePath: string): Promise<{ - entries: FileEntry[]; - titleSlot: SessionTitleUpdate | undefined; -}> { +export async function loadEntriesFromFileStream(filePath: string): Promise { const entries: FileEntry[] = []; - const titleSlot = await visitEntriesFromFileStream(filePath, entry => { - entries.push(entry); - }); - return { entries: foldTitleSlot(entries, titleSlot), titleSlot }; + let malformedRecords = 0; + const titleSlot = await visitEntriesFromFileStream( + filePath, + entry => { + entries.push(entry); + }, + { + onMalformedRecord: () => { + malformedRecords++; + }, + }, + ); + return { entries: foldTitleSlot(entries, titleSlot), titleSlot, malformedRecords }; } /** Read only the fixed-size head window to detect a physical title slot. */ @@ -235,12 +261,12 @@ export function parseSessionEntries(content: string): FileEntry[] { return parseSessionContent(content).entries; } -/** Exported for testing */ -export async function loadEntriesFromFile( +/** Load and validate a session while retaining malformed-record diagnostics. */ +export async function loadSessionFile( filePath: string, storage: SessionStorage = new FileSessionStorage(), -): Promise { - let loaded: { entries: FileEntry[]; titleSlot: SessionTitleUpdate | undefined }; +): Promise { + let loaded: SessionLoadResult; try { const stat = storage.statSync(filePath); loaded = @@ -248,19 +274,25 @@ export async function loadEntriesFromFile( ? await loadEntriesFromFileStream(filePath) : parseSessionContent(await storage.readText(filePath)); } catch (err) { - if (isEnoent(err)) return []; + if (isEnoent(err)) return { entries: [], titleSlot: undefined, malformedRecords: 0 }; throw err; } - const { entries } = loaded; // Validate session header - if (entries.length === 0) return entries; - const header = entries[0] as SessionHeader; - if (header.type !== "session" || typeof header.id !== "string") { - return []; + const header = loaded.entries[0] as SessionHeader | undefined; + if (header?.type !== "session" || typeof header.id !== "string") { + return { ...loaded, entries: [] }; } + return loaded; +} - return entries; +/** Load the valid entries from a session file, skipping malformed records. */ +export async function loadEntriesFromFile( + filePath: string, + storage: SessionStorage = new FileSessionStorage(), +): Promise { + const loaded = await loadSessionFile(filePath, storage); + return loaded.entries; } /** diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index 37a50703d..dcb5ac417 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -59,7 +59,12 @@ import { type UsageStatistics, } from "./session-entries"; import { findMostRecentSession, listAllSessions, listSessions, type SessionInfo } from "./session-listing"; -import { loadEntriesFromFile, readTitleSlotFromFile, resolveBlobRefsInEntries } from "./session-loader"; +import { + loadEntriesFromFile, + loadSessionFile, + resolveBlobRefsInEntries, + type SessionLoadResult, +} from "./session-loader"; import { generateId, migrateToCurrentVersion } from "./session-migrations"; import { computeDefaultSessionDir, @@ -516,6 +521,7 @@ export class SessionManager { */ #breadcrumbFresh = false; #sessionNameChangedCallbacks = new Set<() => void>(); + #persistenceErrorCallbacks = new Set<(error: Error) => void>(); private constructor(cwd: string, sessionDir: string, persist: boolean, storage: SessionStorage) { this.#cwd = cwd; @@ -557,6 +563,15 @@ export class SessionManager { error: error.message, stack: error.stack, }); + for (const callback of this.#persistenceErrorCallbacks) { + try { + callback(error); + } catch (callbackError) { + logger.warn("Session persistence error observer failed", { + error: toError(callbackError).message, + }); + } + } } return this.#diskFailure; @@ -814,6 +829,7 @@ export class SessionManager { this.#diskTail = Promise.resolve(); this.#closeWriterEventually(); this.#storage.writeTextSync(targetPath, body); + this.#clearDiskError(); // Only mark the manager current when writing the active session path. // Mid-move writes update the live relocation path; `#sessionFile` is // still the pre-repoint source until moveTo repoints it. @@ -906,7 +922,13 @@ export class SessionManager { this.#atomicRewriteDirty = true; return; } - if (this.#diskFailure) throw this.#diskFailure; + if (this.#diskFailure) { + // The failed entry and any later entries remain in memory. A full + // replacement is the writability probe and restores all of them once + // transient storage pressure clears. + this.#fileIsCurrent = false; + this.#rewriteRequired = true; + } // Lazy gate: a brand-new session is not written until it has an assistant // message (or someone forced creation), so sessions that never produce @@ -944,9 +966,9 @@ export class SessionManager { // chain. Prefer appendSync so write failures latch `#diskFailure` before // this call returns (not via a discarded rejected Promise after a later // microtask). Callers stay non-throwing here — the core turn loop invokes - // appendMessage/appendCustomEntry without try/catch; flushSync/close and - // subsequent appends still throw the latched error. File writers apply - // each line to the OS page cache before return. + // appendMessage/appendCustomEntry without try/catch. A later entry retries + // all in-memory state through a full rewrite. File writers apply each line + // to the OS page cache before return. // A mid-close writer leaves `#writer` undefined, so `#appendWriter` simply // opens a fresh append handle and the entry still lands. try { @@ -955,16 +977,28 @@ export class SessionManager { if (writer.appendSync) { writer.appendSync(line); } else { - void writer.append(line).catch(err => this.#noteDiskFailure(err)); + void writer.append(line).catch(err => { + this.#fileIsCurrent = false; + this.#rewriteRequired = true; + this.#noteDiskFailure(err); + }); } } catch (err) { + this.#fileIsCurrent = false; + this.#rewriteRequired = true; this.#noteDiskFailure(err); } } async #persistTitleChangeEntry(entry: TitleChangeEntry, update: SessionTitleUpdate): Promise { if (!this.#persist || !this.#sessionFile) return; - if (this.#diskFailure) throw this.#diskFailure; + if (this.#diskFailure) { + this.#fileIsCurrent = false; + this.#rewriteRequired = true; + this.#rewriteSynchronously(); + if (this.#diskFailure) throw this.#diskFailure; + return; + } if (!this.#shouldHaveSessionFile()) { this.#fileIsCurrent = false; @@ -1288,7 +1322,7 @@ export class SessionManager { await this.#setSessionFile(sessionFile); } - async #setSessionFile(sessionFile: string, loadedEntries?: FileEntry[]): Promise { + async #setSessionFile(sessionFile: string, loadedSession?: SessionLoadResult): Promise { await this.#drainAndCloseWriter(); this.#clearDiskError(); this.#draftOnlySessionCleanupArmed = false; @@ -1297,8 +1331,8 @@ export class SessionManager { this.#sessionFile = resolvedSessionFile; this.#rememberBreadcrumb(this.#cwd, resolvedSessionFile); - const titleSlot = await readTitleSlotFromFile(resolvedSessionFile, this.#storage); - const fileEntries = loadedEntries ?? (await loadEntriesFromFile(resolvedSessionFile, this.#storage)); + const loaded = loadedSession ?? (await loadSessionFile(resolvedSessionFile, this.#storage)); + const { entries: fileEntries, titleSlot } = loaded; if (fileEntries.length === 0) { // Explicit but empty/missing path (e.g. --session flag): start fresh but // keep the requested path and materialize the header immediately. @@ -1332,7 +1366,7 @@ export class SessionManager { this.#titleUpdatedAt = titleSlot?.updatedAt ?? header.timestamp; this.#hasTitleSlot = titleSlot !== undefined; this.#fileIsCurrent = true; - this.#rewriteRequired = migrated; + this.#rewriteRequired = migrated || loaded.malformedRecords > 0; this.#forceFileCreation = true; this.#artifactManager = null; this.#artifactManagerSessionFile = null; @@ -1985,6 +2019,14 @@ export class SessionManager { }; } + /** Subscribe to persistence failures so hosts can surface lost-durability state. */ + onPersistenceError(cb: (error: Error) => void): () => void { + this.#persistenceErrorCallbacks.add(cb); + return () => { + this.#persistenceErrorCallbacks.delete(cb); + }; + } + /** * Set the session display name. * @param source "user" for explicit renames; "auto" for generated titles. @@ -2589,8 +2631,8 @@ export class SessionManager { storage: SessionStorage = new FileSessionStorage(), options?: { initialCwd?: string; suppressBreadcrumb?: boolean }, ): Promise { - const loaded = await loadEntriesFromFile(filePath, storage); - const header = loaded.find(entry => entry.type === "session") as SessionHeader | undefined; + const loaded = await loadSessionFile(filePath, storage); + const header = loaded.entries.find(entry => entry.type === "session") as SessionHeader | undefined; // Resume into the session's recorded cwd only when that directory still // exists. A deleted project dir would make the constructor's #cwd — and the // `setProjectDir` chdir interactive mode runs next — point at (and fail on) diff --git a/packages/coding-agent/src/session/session-storage.ts b/packages/coding-agent/src/session/session-storage.ts index a10e8dc7c..e3b913f92 100644 --- a/packages/coding-agent/src/session/session-storage.ts +++ b/packages/coding-agent/src/session/session-storage.ts @@ -128,14 +128,27 @@ class FileSessionStorageWriter implements SessionStorageWriter { } #writeNow(line: string): void { + const originalSize = fs.fstatSync(this.#fd).size; const buf = Buffer.from(line, "utf-8"); let offset = 0; - while (offset < buf.length) { - const written = fs.writeSync(this.#fd, buf, offset, buf.length - offset); - if (written === 0) { - throw new Error("Short write"); + try { + while (offset < buf.length) { + const written = fs.writeSync(this.#fd, buf, offset, buf.length - offset); + if (written === 0) { + throw new Error("Short write"); + } + offset += written; } - offset += written; + } catch (writeError) { + try { + fs.ftruncateSync(this.#fd, originalSize); + } catch (rollbackError) { + throw new AggregateError( + [toError(writeError), toError(rollbackError)], + "Session append failed and its partial bytes could not be rolled back", + ); + } + throw writeError; } } diff --git a/packages/coding-agent/test/interactive-mode-lsp-startup.test.ts b/packages/coding-agent/test/interactive-mode-lsp-startup.test.ts index de6d15781..ab166c6e7 100644 --- a/packages/coding-agent/test/interactive-mode-lsp-startup.test.ts +++ b/packages/coding-agent/test/interactive-mode-lsp-startup.test.ts @@ -1,4 +1,5 @@ import { afterEach, beforeAll, beforeEach, describe, expect, it, vi } from "bun:test"; +import * as fs from "node:fs"; import * as path from "node:path"; import { Agent } from "@oh-my-pi/pi-agent-core"; import { ModelRegistry } from "@oh-my-pi/pi-coding-agent/config/model-registry"; @@ -127,4 +128,24 @@ describe("InteractiveMode LSP startup welcome banner", () => { } satisfies LspStartupEvent); expect(showWarningSpy).not.toHaveBeenCalled(); }); + + it("surfaces a sanitized warning when session persistence fails", async () => { + await mode.init(); + await session.sessionManager.ensureOnDisk(); + const showWarning = vi.spyOn(mode, "showWarning").mockImplementation(() => {}); + const writeFailure = vi.spyOn(fs, "writeSync").mockImplementation(() => { + throw Object.assign(new Error("ENOSPC:\tdisk full\n\u001b[31mretry later\u001b[0m"), { code: "ENOSPC" }); + }); + session.sessionManager.appendCustomEntry("persistence-failure-probe", {}); + + expect(showWarning).toHaveBeenCalledTimes(1); + const warning = showWarning.mock.calls[0]?.[0] ?? ""; + expect(warning).toContain("Session persistence failed: ENOSPC:"); + expect(warning).toContain("Unsaved entries remain in memory"); + expect(warning).not.toContain("\t"); + expect(warning).not.toContain("\n"); + expect(warning).not.toContain("\u001b"); + writeFailure.mockRestore(); + session.sessionManager.appendCustomEntry("persistence-recovery-probe", {}); + }); }); diff --git a/packages/coding-agent/test/session-loader-stream.test.ts b/packages/coding-agent/test/session-loader-stream.test.ts index a0900f904..2790bbf51 100644 --- a/packages/coding-agent/test/session-loader-stream.test.ts +++ b/packages/coding-agent/test/session-loader-stream.test.ts @@ -92,6 +92,7 @@ describe("loadEntriesFromFileStream (Bun.JSONL parity)", () => { expect(titleSlot?.title).toBe("Visitor"); expect(entryIds(visited)).toEqual(["s1", "m1", "m2"]); + expect((await sessionLoader.loadEntriesFromFileStream(file)).malformedRecords).toBe(1); }); it("does not revisit entries before a malformed line spanning stream chunks", async () => { const content = [ @@ -167,6 +168,7 @@ describe("loadEntriesFromFileStream (Bun.JSONL parity)", () => { expect(entryTypes(stream.entries)).toEqual(["session", "message", "message"]); const ids = messageIds(stream.entries); expect(ids).toEqual(["m1", "m2"]); // valid entries kept in order, malformed skipped + expect(stream.malformedRecords).toBe(1); }); it("matches parseSessionContent when there is no title slot (header is the first line)", async () => { @@ -217,5 +219,6 @@ describe("loadEntriesFromFileStream (Bun.JSONL parity)", () => { const stream = await sessionLoader.loadEntriesFromFileStream(missing); expect(stream.entries).toEqual([]); expect(stream.titleSlot).toBeUndefined(); + expect(stream.malformedRecords).toBe(0); }); }); diff --git a/packages/coding-agent/test/session-manager-immediate-persist.test.ts b/packages/coding-agent/test/session-manager-immediate-persist.test.ts index 045ac116c..d9c15d8b3 100644 --- a/packages/coding-agent/test/session-manager-immediate-persist.test.ts +++ b/packages/coding-agent/test/session-manager-immediate-persist.test.ts @@ -198,6 +198,24 @@ describe("SessionManager JSONL software-crash durability", () => { expect(toolResult).toBeDefined(); }); + it("rewrites a malformed resumed tail before appending another entry", async () => { + const cwd = makeTempDir("@pi-malformed-tail-cwd-"); + const manager = SessionManager.create(cwd, path.join(cwd, "sessions")); + manager.appendMessage(assistantMessage("seed")); + const sessionFile = manager.getSessionFile(); + if (!sessionFile) throw new Error("Expected session file"); + await manager.close(); + + fs.appendFileSync(sessionFile, '{"type":"message","id":"torn","message":{"role":"user","content":"lost'); + const resumed = await SessionManager.open(sessionFile); + resumed.appendMessage({ role: "user", content: "after resume", timestamp: Date.now() }); + + const entries = readJsonl(sessionFile); + expect(entries.map(entryKind)).toEqual(["session", "assistant", "user"]); + expect(messageContent(entries[2] ?? {})).toBe("after resume"); + await resumed.close(); + }); + it("keeps pre-assistant sessions out of history during shutdown", async () => { const cwd = makeTempDir("@pi-empty-session-cwd-"); const sessionDir = path.join(cwd, "sessions"); @@ -307,10 +325,7 @@ describe("SessionManager JSONL software-crash durability", () => { expect(afterKinds).toEqual(crashKinds); }); - it("latches the first hot-path write failure before append returns", () => { - // H2: async append + discarded Promise meant ENOSPC/EIO only latched after a - // later flush/append. appendSync must latch `#diskFailure` on the first call - // without throwing through unguarded turn-loop callers. + it("alerts once and retries all in-memory entries after a transient write failure", () => { const cwd = makeTempDir("@pi-write-fail-cwd-"); const sessionDir = path.join(cwd, "sessions"); const manager = SessionManager.create(cwd, sessionDir); @@ -320,39 +335,33 @@ describe("SessionManager JSONL software-crash durability", () => { manager.appendMessage(assistantMessage("seed")); manager.appendMessage({ role: "user", content: "ok-user", timestamp: Date.now() }); - let failWrites = true; - const origWrite = fs.writeSync.bind(fs) as typeof fs.writeSync; - const writeSpy = spyOn(fs, "writeSync").mockImplementation(((...args: Parameters) => { - if (failWrites) { - const err = new Error("ENOSPC: no space left on device") as NodeJS.ErrnoException; - err.code = "ENOSPC"; - throw err; - } - return origWrite(...args); - }) as typeof fs.writeSync); + const writeSpy = spyOn(fs, "writeSync").mockImplementation(() => { + throw Object.assign(new Error("ENOSPC: no space left on device"), { code: "ENOSPC" }); + }); + const failures: Error[] = []; + manager.onPersistenceError(error => { + failures.push(error); + }); try { - let threw = false; - try { - manager.appendMessage({ role: "user", content: "should-fail-user", timestamp: Date.now() }); - } catch { - threw = true; - } - // Unguarded turn-loop contract: appendMessage itself must not throw. - expect(threw).toBe(false); - - // Failure is latched before return — flushSync / next append surface it. + expect(() => + manager.appendMessage({ role: "user", content: "failed-user", timestamp: Date.now() }), + ).not.toThrow(); expect(() => manager.flushSync()).toThrow("ENOSPC"); - expect(() => manager.appendMessage({ role: "user", content: "next-user", timestamp: Date.now() })).toThrow( - "ENOSPC", - ); + expect(failures).toHaveLength(1); - const users = parseJsonlLenient>(fs.readFileSync(sessionFile, "utf8")) + writeSpy.mockRestore(); + expect(() => + manager.appendMessage({ role: "user", content: "recovered-user", timestamp: Date.now() }), + ).not.toThrow(); + expect(() => manager.flushSync()).not.toThrow(); + + const users = readJsonl(sessionFile) .filter(entry => entry.type === "message" && messageRole(entry) === "user") .map(entry => messageContent(entry)); - expect(users).toEqual(["ok-user"]); + expect(users).toEqual(["ok-user", "failed-user", "recovered-user"]); + expect(failures).toHaveLength(1); } finally { - failWrites = false; writeSpy.mockRestore(); } }); diff --git a/packages/coding-agent/test/session-storage.test.ts b/packages/coding-agent/test/session-storage.test.ts index 9c4d0f131..59b92dad6 100644 --- a/packages/coding-agent/test/session-storage.test.ts +++ b/packages/coding-agent/test/session-storage.test.ts @@ -150,6 +150,24 @@ describe("FileSessionStorage writer", () => { await expect(writer.append("two\n")).rejects.toThrow("disk full"); await expect(writer.close()).rejects.toThrow("disk full"); }); + + it("rolls back bytes from a partial append before surfacing the error", () => { + const sessionPath = path.join(tempDir, "partial-append.jsonl"); + fs.writeFileSync(sessionPath, "complete\n"); + const writer = storage.openWriter(sessionPath); + vi.spyOn(fs, "writeSync") + .mockImplementationOnce(() => { + fs.appendFileSync(sessionPath, "par"); + return 3; + }) + .mockImplementation(() => { + throw Object.assign(new Error("ENOSPC: no space left on device"), { code: "ENOSPC" }); + }); + const appendSync = writer.appendSync?.bind(writer); + if (!appendSync) throw new Error("File writer must expose appendSync"); + expect(() => appendSync("partial entry\n")).toThrow("ENOSPC"); + expect(fs.readFileSync(sessionPath, "utf8")).toBe("complete\n"); + }); }); describe("FileSessionStorage.deleteSessionWithArtifacts", () => { diff --git a/packages/utils/src/stream.ts b/packages/utils/src/stream.ts index 36ff5380e..02dd5433b 100644 --- a/packages/utils/src/stream.ts +++ b/packages/utils/src/stream.ts @@ -446,12 +446,13 @@ export async function* readSseEvents( * Uses `Bun.JSONL.parseChunk` internally. On parse errors, the malformed * region is skipped up to the next newline and parsing continues. * + * @param options.onMalformedRecord Called once for every skipped JSONL record. * @example * ```ts * const entries = parseJsonlLenient(fileContents); * ``` */ -export function parseJsonlLenient(buffer: string): T[] { +export function parseJsonlLenient(buffer: string, options: { onMalformedRecord?: () => void } = {}): T[] { let entries: T[] | undefined; while (buffer.length > 0) { @@ -466,11 +467,16 @@ export function parseJsonlLenient(buffer: string): T[] { } if (error) { const nextNewline = buffer.indexOf("\n", read); + const malformedEnd = nextNewline === -1 ? buffer.length : nextNewline; + if (buffer.substring(read, malformedEnd).trim().length > 0) options.onMalformedRecord?.(); if (nextNewline === -1) break; buffer = buffer.substring(nextNewline + 1); continue; } - if (read === 0) break; + if (read === 0) { + if (buffer.trim().length > 0) options.onMalformedRecord?.(); + break; + } buffer = buffer.substring(read); if (done) break; }