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

@@ -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

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();

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<RawFileEntry>(body) as FileEntry[];
-	return { entries: foldTitleSlot(entries, slot), titleSlot: slot };
+	let malformedRecords = 0;
+	const entries = parseJsonlLenient<RawFileEntry>(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<SessionLoadResult> {
 	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<FileEntry[]> {
-	let loaded: { entries: FileEntry[]; titleSlot: SessionTitleUpdate | undefined };
+): Promise<SessionLoadResult> {
+	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<FileEntry[]> {
+	const loaded = await loadSessionFile(filePath, storage);
+	return loaded.entries;
 }
 
 /**

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<void> {
 		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<void> {
+	async #setSessionFile(sessionFile: string, loadedSession?: SessionLoadResult): Promise<void> {
 		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<SessionManager> {
-		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)

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;
 		}
 	}
 

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", {});
+	});
 });

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);
 	});
 });

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<typeof fs.writeSync>) => {
-			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<Record<string, unknown>>(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();
 		}
 	});

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

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<MyType>(fileContents);
  * ```
  */
-export function parseJsonlLenient<T>(buffer: string): T[] {
+export function parseJsonlLenient<T>(buffer: string, options: { onMalformedRecord?: () => void } = {}): T[] {
 	let entries: T[] | undefined;
 
 	while (buffer.length > 0) {
@@ -466,11 +467,16 @@ export function parseJsonlLenient<T>(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;
 	}