Merge PR #7444: fix(coding-agent): keep completed session entries durable across crashes (@olegpulatov)

packages/coding-agent/CHANGELOG.md

@@ -25,6 +25,7 @@
 
 ### Fixed
 
+- Fixed session transcript entries being lost on process crash: file-backed JSONL writers now write each completed entry to the OS page cache on append instead of microtask-batching; concurrent appends during an in-place atomic rewrite supersede the paused publish with a synchronous full-body rewrite so fenced events are durable before return; concurrent appends during `/move` write a full body to the live relocation path (source pre-rename, destination post-rename) so a crash mid-move no longer drops completed events; and the first write failure latches `#diskFailure` synchronously via `appendSync` so a later flush/close reports it instead of silently succeeding after a discarded rejected Promise ([#7444](https://github.com/can1357/oh-my-pi/pull/7444) by [@olegpulatov](https://github.com/olegpulatov)).
 - Fixed template argument substitution (`substituteArgs`) executing recursive placeholder expansion when positional argument values contain literal `$@` or `$ARGUMENTS` tokens.
 - Fixed focused-agent status bar dimming darkening Powerline end caps.
 - Fixed the browser relay creating duplicate "omp" tab groups: the bridge now keeps at most one group RPC in flight (a queued drain replaces fire-and-forget per-tab requests), so concurrent requests can no longer race the extension's non-atomic query→create→set-title sequence in the same window. Also fixed an extension reconnect (relay daemon restart, service-worker recycle) being misread as the user dragging every tab out of the omp group — grouping state is reset when the extension socket closes, so tabs regroup on the next hello instead of being permanently opted out.

packages/coding-agent/src/session/indexed-session-storage.ts

@@ -511,6 +511,15 @@ class IndexedSessionStorageWriter implements SessionStorageWriter {
 		return next;
 	}
 
+	appendSync(line: string): void {
+		if (this.#closed) throw new Error("Writer closed");
+		if (this.#error) throw this.#error;
+		// Local index is updated immediately; remote publish stays ordered on the
+		// path queue. Callers that need remote durability still await append()/flush().
+		const mtimeMs = this.#storage._appendForWriter(this.#path, line);
+		void this.#trackPromise(this.#storage._queueAppend(this.#path, line, mtimeMs, () => this.#error));
+	}
+
 	async append(line: string): Promise<void> {
 		if (this.#closed) throw new Error("Writer closed");
 		if (this.#error) throw this.#error;

packages/coding-agent/src/session/session-manager.ts

@@ -405,17 +405,22 @@ export class SessionPersistenceIndeterminateError extends AggregateError {
  * tree by `(id, parentId)`, and the mutable leaf pointer selects which path is
  * active for future appends and for LLM context construction.
  *
- * Durability is software-crash safe but not power-loss safe: appends are normally
- * handed to the OS synchronously in-body but never `fsync`'d. While an in-place
- * atomic rewrite publishes, a synchronous `flushSync` still persists fenced
- * appends by superseding the pending publish against the same (stable) path.
- * A {@link moveTo} relocation is the one exception: the session file is being
- * renamed to a new path, so appends landing in the rename window are held in
- * memory and persisted only by the move's trailing rewrite. A `flushSync` in
- * that narrow window cannot durably place them — writing the vacated source
- * would recreate the orphan the move eliminates, and a queued backend publish
- * (fire-and-forget on `IndexedSessionStorage`) could land after the rename all
- * the same. This gap is inherent to closing the writer for a Windows-safe rename.
+ * Durability is software-crash safe but not power-loss safe: completed entries
+ * (user/assistant/toolResult messages, tool_execution_start markers, custom
+ * entries) are handed to the OS synchronously in-body on append and never
+ * `fsync`'d. In-flight streaming text is intentionally not durable until
+ * `message_end` persists the finished message.
+ *
+ * While an in-place atomic rewrite is publishing, a concurrent completed append
+ * supersedes that publish with a synchronous full-body rewrite so the entry is
+ * software-crash durable before the append returns; the abandoned atomic's
+ * `commitGuard` then refuses to clobber the fresher body.
+ *
+ * During {@link moveTo}, appends write a full body to the live relocation path
+ * (source until rename, destination once the rename has landed) so a crash mid-
+ * move still preserves completed entries without recreating a vacated source.
+ * A trailing atomic rewrite still rewrites the header cwd after the path is
+ * repointed.
  */
 export class SessionManager {
 	#cwd: string;
@@ -484,15 +489,12 @@ export class SessionManager {
 	/** Set by synchronous appends that land while an atomic replacement is active. */
 	#atomicRewriteDirty = false;
 	/**
-	 * True while {@link moveTo} is renaming the session file but has not yet
-	 * repointed `#sessionFile`. Both the synchronous append hot path
-	 * ({@link #appendToSessionFile}) and synchronous rewrites
-	 * ({@link #rewriteSynchronously}, reached via `flushSync` on Ctrl+C) would
-	 * otherwise touch the pre-rename path — opening a fresh writer or writing the
-	 * full body — recreating the very orphan the move eliminates. Both defer
-	 * while this is set; the move's trailing atomic rewrite persists them.
+	 * Active {@link moveTo} relocation. Concurrent completed appends write a
+	 * full body to the live path: source while it still exists, destination
+	 * once rename has landed (source gone). Never recreates a vacated source.
+	 * `null` outside an active relocation.
 	 */
-	#sessionFileRelocating = false;
+	#sessionFileRelocating: { source: string; dest: string } | null = null;
 	/** Atomic entry batch currently staged for a full-file commit. */
 	#atomicEntryBatch: AtomicEntryBatch | undefined;
 
@@ -763,30 +765,56 @@ export class SessionManager {
 		return this.#forceFileCreation || this.#fileIsCurrent || this.#historyContainsAssistantMessage();
 	}
 
+	/**
+	 * Live path for concurrent completed appends during {@link moveTo}.
+	 * Prefers destination once rename has landed (source gone); otherwise
+	 * source. Never invents a path that does not already exist.
+	 */
+	#liveRelocationWritePath(): string | null {
+		const relocating = this.#sessionFileRelocating;
+		if (!relocating) return null;
+		if (this.#storage.existsSync(relocating.dest)) return relocating.dest;
+		if (this.#storage.existsSync(relocating.source)) return relocating.source;
+		// Rename in flight with neither path visible (rare cross-device edge):
+		// fall back to destination so we do not recreate a vacated source.
+		return relocating.dest;
+	}
+
 	/**
 	 * Synchronously rewrite the whole file (header + entries) and keep no open
 	 * writer; the next append re-opens one. `writeTextSync` returns with the
 	 * bytes in the kernel page cache, so the file is software-crash durable.
+	 *
+	 * During {@link moveTo}, writes to the live relocation path (source pre-
+	 * rename, destination post-rename) rather than always `#sessionFile`, so
+	 * concurrent completed entries are durable without recreating a vacated source.
 	 */
 	#rewriteSynchronously(): void {
-		if (!this.#persist || !this.#sessionFile || !this.#shouldHaveSessionFile()) return;
-		// A move is renaming the file out from under `#sessionFile`. Writing the
-		// full body now would target either the vacated source (recreating the
-		// orphan the move eliminates) or a path a queued backend publish may clobber
-		// post-rename, so defer to moveTo's trailing atomic rewrite. This narrows
-		// flushSync durability for entries in the rename window (see class doc).
-		if (this.#sessionFileRelocating) return;
+		if (!this.#persist || !this.#shouldHaveSessionFile()) return;
+		const targetPath = this.#liveRelocationWritePath() ?? this.#sessionFile;
+		if (!targetPath) return;
 
 		try {
 			const body = this.#fileBody();
 			this.#diskEpoch++;
 			this.#diskTail = Promise.resolve();
 			this.#closeWriterEventually();
-			this.#storage.writeTextSync(this.#sessionFile, body);
-			this.#fileIsCurrent = true;
-			this.#materializeBreadcrumb();
-			this.#rewriteRequired = false;
-			this.#hasTitleSlot = true;
+			this.#storage.writeTextSync(targetPath, body);
+			// 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.
+			if (!this.#sessionFileRelocating || targetPath === this.#sessionFile) {
+				this.#fileIsCurrent = true;
+				this.#materializeBreadcrumb();
+				this.#rewriteRequired = false;
+				this.#hasTitleSlot = true;
+			} else {
+				// Destination body is current on disk; in-memory still needs a
+				// header-cwd rewrite after repoint, but entries are durable.
+				this.#fileIsCurrent = false;
+				this.#rewriteRequired = true;
+				this.#hasTitleSlot = true;
+			}
 		} catch (err) {
 			this.#noteDiskFailure(err);
 		}
@@ -872,22 +900,21 @@ export class SessionManager {
 			return;
 		}
 
-		// Atomic replacement window: the old path may be moved aside underneath
-		// any newly-opened append handle (Windows EPERM fallback), or a `moveTo`
-		// may be renaming the session file to a new directory. Do not open a
-		// writer here; the active rewrite loops and serializes a fresh full body,
-		// and the move's trailing atomic rewrite folds in these deferred entries.
-		// A superseding synchronous rewrite bumps `#diskEpoch`, at which point
-		// the pending atomic publish is guaranteed to abandon via its
-		// `commitGuard`, so appends can (and must) take the hot path so they
-		// don't strand in memory while `close()` returns without a rewrite.
-		if (
-			this.#sessionFileRelocating ||
-			(this.#atomicRewriteFenceEpoch !== null && this.#atomicRewriteFenceEpoch === this.#diskEpoch)
-		) {
-			this.#fileIsCurrent = false;
-			this.#rewriteRequired = true;
+		// Atomic replacement / move window: do not open a fresh append writer that
+		// a Windows EPERM replace could detach from the current JSONL path.
+		// - moveTo: write a full body to the live relocation path (source pre-
+		//   rename, destination post-rename) so completed entries are durable
+		//   without recreating a vacated source.
+		// - in-place atomic fence: supersede the pending publish with a
+		//   synchronous full-body rewrite; bumping `#diskEpoch` abandons the
+		//   in-flight atomic via its `commitGuard`.
+		if (this.#sessionFileRelocating) {
+			this.#rewriteSynchronously();
+			return;
+		}
+		if (this.#atomicRewriteFenceEpoch !== null && this.#atomicRewriteFenceEpoch === this.#diskEpoch) {
 			this.#atomicRewriteDirty = true;
+			this.#rewriteSynchronously();
 			return;
 		}
 		// Cold/divergent: not on disk yet, or in-memory entries diverged from the
@@ -897,15 +924,23 @@ export class SessionManager {
 			return;
 		}
 
-		// Hot path: queue the entry directly on the writer, outside the async disk
-		// chain. File writers batch same-turn appends until their microtask
-		// boundary; flush(), flushSync(), and close() drain that batch explicitly.
+		// Hot path: write the entry directly on the writer, outside the async disk
+		// 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.
 		// A mid-close writer leaves `#writer` undefined, so `#appendWriter` simply
 		// opens a fresh append handle and the entry still lands.
 		try {
-			void this.#appendWriter()
-				.append(this.#lineFor(entry))
-				.catch(err => this.#noteDiskFailure(err));
+			const writer = this.#appendWriter();
+			const line = this.#lineFor(entry);
+			if (writer.appendSync) {
+				writer.appendSync(line);
+			} else {
+				void writer.append(line).catch(err => this.#noteDiskFailure(err));
+			}
 		} catch (err) {
 			this.#noteDiskFailure(err);
 		}
@@ -921,12 +956,11 @@ export class SessionManager {
 		}
 
 		// Title changes use their own asynchronous append path rather than
-		// #appendToSessionFile. Defer them under the same move fence so they cannot
-		// recreate the vacated source path before #sessionFile is repointed.
+		// #appendToSessionFile. During move, write the full body (including the
+		// title entry) to the live relocation path so a crash mid-move still
+		// keeps the title change; the trailing rewrite still updates header cwd.
 		if (this.#sessionFileRelocating) {
-			this.#fileIsCurrent = false;
-			this.#rewriteRequired = true;
-			this.#atomicRewriteDirty = true;
+			this.#rewriteSynchronously();
 			return;
 		}
 
@@ -1362,14 +1396,15 @@ export class SessionManager {
 				: computeDefaultSessionDir(resolvedCwd, this.#storage));
 
 		let sessionFileExisted = false;
-		// Fence synchronous appends and rewrites away from the session file for the
-		// duration of the relocation (see `#sessionFileRelocating`). This flag —
-		// not a `#diskEpoch` bump — is used deliberately: bumping the epoch here
-		// would cancel any disk task already queued at the current epoch (e.g. a
-		// header-only `ensureOnDisk()` materializing rewrite) before the drain
-		// below runs it, silently dropping an explicitly materialized session.
+		// Track source+dest for concurrent completed appends during relocation
+		// (see `#sessionFileRelocating`). Existence of either path decides the
+		// live write target — not a `#diskEpoch` bump, which would cancel any
+		// disk task already queued at the current epoch (e.g. a header-only
+		// `ensureOnDisk()` materializing rewrite) before the drain below runs it.
 		if (this.#persist && this.#sessionFile) {
-			this.#sessionFileRelocating = true;
+			const source = this.#sessionFile;
+			const dest = path.join(nextSessionDir, path.basename(source));
+			this.#sessionFileRelocating = { source, dest };
 		}
 
 		try {
@@ -1433,8 +1468,8 @@ export class SessionManager {
 				this.#sessionFile = newSessionFile;
 				this.#artifactManager = null;
 				this.#artifactManagerSessionFile = null;
-				// Path is repointed; a sync rewrite may now safely target the new file.
-				this.#sessionFileRelocating = false;
+				// Path is repointed; hot-path appends may use `#sessionFile` again.
+				this.#sessionFileRelocating = null;
 			}
 
 			this.#cwd = resolvedCwd;
@@ -1459,7 +1494,7 @@ export class SessionManager {
 
 			if (this.#sessionFile) this.#rememberBreadcrumb(resolvedCwd, this.#sessionFile);
 		} finally {
-			this.#sessionFileRelocating = false;
+			this.#sessionFileRelocating = null;
 		}
 	}
 

packages/coding-agent/src/session/session-storage.ts

@@ -14,13 +14,25 @@ export interface SessionStorageStat {
 
 export interface SessionStorageWriter {
 	/**
-	 * Append one newline-terminated line. File storage batches same-turn appends
-	 * until its microtask boundary; memory storage updates in-body; indexed
-	 * backends queue in call order.
+	 * Append one newline-terminated line.
+	 *
+	 * File and memory storage apply the line synchronously before the returned
+	 * promise settles, so a software crash after `append` returns (or after a
+	 * fire-and-forget call begins) still sees the entry on disk / in body. No
+	 * `fsync` — power loss may still drop the last page. Indexed backends update
+	 * the local index immediately and queue the remote publish in call order.
 	 *
 	 * `line` MUST include the trailing newline.
 	 */
 	append(line: string): Promise<void>;
+	/**
+	 * Synchronous append when the backend can apply the line before return.
+	 * File and memory implement this so {@link SessionManager} can latch the
+	 * first write failure before the appending call returns (surfaced by a later
+	 * flushSync/close/next append — the turn loop does not throw from append).
+	 * Indexed backends update the local index immediately and queue remote I/O.
+	 */
+	appendSync?(line: string): void;
 	/** Resolve once all queued appends complete. No fsync. */
 	flush(): Promise<void>;
 	/** Drain synchronously flushable queued work when the backend supports it. No fsync. */
@@ -93,8 +105,6 @@ class FileSessionStorageWriter implements SessionStorageWriter {
 	#closed = false;
 	#error: Error | undefined;
 	#onError: ((err: Error) => void) | undefined;
-	#pending = "";
-	#flushScheduled = false;
 
 	constructor(fpath: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }) {
 		this.#onError = options?.onError;
@@ -129,34 +139,29 @@ class FileSessionStorageWriter implements SessionStorageWriter {
 		}
 	}
 
-	#flushPendingNow(): void {
-		this.#flushScheduled = false;
-		if (this.#pending.length === 0) return;
-		const pending = this.#pending;
-		this.#pending = "";
+	appendSync(line: string): void {
+		if (this.#closed) throw new Error("Writer closed");
+		if (this.#error) throw this.#error;
+		// Write in-body so software crash after the call still sees the entry.
+		// Microtask batching used to leave completed transcript lines only in
+		// memory until the next event-loop turn; process crash then lost every
+		// post-checkpoint event. flush/flushSync remain no-op drains (no fsync).
 		try {
-			this.#writeNow(pending);
+			this.#writeNow(line);
 		} catch (err) {
-			this.#recordError(err);
+			throw this.#recordError(err);
 		}
 	}
 
 	async append(line: string): Promise<void> {
-		if (this.#closed) throw new Error("Writer closed");
-		if (this.#error) throw this.#error;
-		this.#pending += line;
-		if (this.#flushScheduled) return;
-		this.#flushScheduled = true;
-		queueMicrotask(() => this.#flushPendingNow());
+		this.appendSync(line);
 	}
 
 	async flush(): Promise<void> {
-		this.#flushPendingNow();
 		if (this.#error) throw this.#error;
 	}
 
 	flushSync(): void {
-		this.#flushPendingNow();
 		if (this.#error) throw this.#error;
 	}
 
@@ -166,7 +171,6 @@ class FileSessionStorageWriter implements SessionStorageWriter {
 
 	async close(): Promise<void> {
 		if (this.#closed) return;
-		this.#flushPendingNow();
 		this.#closed = true;
 		// Unregister from finalization - we're closing properly
 		writerRegistry.unregister(this);
@@ -487,7 +491,7 @@ class MemorySessionStorageWriter implements SessionStorageWriter {
 		return error;
 	}
 
-	async append(line: string): Promise<void> {
+	appendSync(line: string): void {
 		if (this.#closed) throw new Error("Writer closed");
 		if (this.#error) throw this.#error;
 		try {
@@ -498,6 +502,10 @@ class MemorySessionStorageWriter implements SessionStorageWriter {
 		}
 	}
 
+	async append(line: string): Promise<void> {
+		this.appendSync(line);
+	}
+
 	async flush(): Promise<void> {
 		if (this.#error) throw this.#error;
 	}

packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts

@@ -364,10 +364,16 @@ describe("SessionManager atomic rewrite fence spans writer.close()", () => {
 
 		sessionManager.appendMessage({ role: "user", content: "during close", timestamp: Date.now() });
 		sessionManager.appendCustomEntry("during_close_custom", { reason: "guard" });
-		// Pre-fix, #appendToSessionFile would take the hot path and call
-		// storage.openWriter here; the writer would then be caught by the pending
-		// writeTextAtomic detachment. Fence keeps writerOpens flat.
-		expect(storage.writerOpens).toBe(opensBeforeRewrite);
+		// First fenced append supersedes the in-flight atomic with a synchronous
+		// full-body rewrite (software-crash durable before return). That bumps
+		// `#diskEpoch`, so a second append may open a hot-path writer against the
+		// already-published body; the abandoned atomic's commitGuard must still
+		// refuse to clobber it, and nothing may land on a detached handle.
+		const sessionFileMid = sessionManager.getSessionFile();
+		if (!sessionFileMid) throw new Error("Expected session file");
+		const midContent = await storage.readText(sessionFileMid);
+		expect(midContent).toContain("during close");
+		expect(midContent).toContain('"customType":"during_close_custom"');
 
 		storage.allowClose.resolve();
 		storage.allowWrite.resolve();
@@ -379,6 +385,9 @@ describe("SessionManager atomic rewrite fence spans writer.close()", () => {
 		const content = await storage.readText(sessionFile);
 		expect(content).toContain("during close");
 		expect(content).toContain('"customType":"during_close_custom"');
+		// Superseding rewrite may finish before the paused atomic reaches its
+		// commitGuard; either way the fenced entries must remain and no append
+		// may land on a detached handle.
 		expect(storage.detachedLines).toEqual([]);
 	});
 });
@@ -443,9 +452,10 @@ describe("SessionManager title-change fallback fenced-append durability", () =>
 		const rename = sessionManager.setSessionName("second title", "user", "test");
 		await storage.writeStarted.promise;
 
-		// Fenced appends during the paused fallback rewrite: pre-fix these
-		// would be marked dirty and dropped from the serialized body because
-		// the fallback never looped on that flag.
+		// Fenced appends during the paused fallback rewrite supersede the atomic
+		// with a synchronous full-body rewrite, so they are on disk before the
+		// paused publish resumes. The abandoned atomic's body must not clobber
+		// them when released.
 		sessionManager.appendMessage({
 			role: "user",
 			content: "during title fallback",
@@ -453,6 +463,12 @@ describe("SessionManager title-change fallback fenced-append durability", () =>
 		});
 		sessionManager.appendCustomEntry("during_title_fallback_custom", { reason: "test" });
 
+		const sessionFileMid = sessionManager.getSessionFile();
+		if (!sessionFileMid) throw new Error("Expected session file");
+		const midContent = await storage.readText(sessionFileMid);
+		expect(midContent).toContain("during title fallback");
+		expect(midContent).toContain('"customType":"during_title_fallback_custom"');
+
 		storage.allowWrite.resolve();
 		await rename;
 		await sessionManager.flush();
@@ -463,9 +479,10 @@ describe("SessionManager title-change fallback fenced-append durability", () =>
 		expect(content).toContain('"title":"second title"');
 		expect(content).toContain("during title fallback");
 		expect(content).toContain('"customType":"during_title_fallback_custom"');
-		// Loop must have executed at least twice: first pass paused, dirty from
-		// the fenced appends triggers a second pass that includes them.
-		expect(storage.writeTextAtomicCalls).toBeGreaterThanOrEqual(2);
+		// At least the failed title path's atomic fallback ran once; fenced
+		// appends may have superseded it via writeTextSync without a second
+		// atomic pass.
+		expect(storage.writeTextAtomicCalls).toBeGreaterThanOrEqual(1);
 	});
 });
 

packages/coding-agent/test/session-manager-immediate-persist.test.ts

@@ -1,9 +1,10 @@
-import { afterEach, describe, expect, it } from "bun:test";
+import { afterEach, describe, expect, it, spyOn } from "bun:test";
 import * as fs from "node:fs";
 import * as path from "node:path";
 import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
 import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager";
-import { TempDir } from "@oh-my-pi/pi-utils";
+import { MemorySessionStorage, type WriteTextAtomicOptions } from "@oh-my-pi/pi-coding-agent/session/session-storage";
+import { parseJsonlLenient, TempDir } from "@oh-my-pi/pi-utils";
 
 const tempDirs: TempDir[] = [];
 
@@ -27,11 +28,11 @@ function assistantMessage(text: string) {
 		provider: model.provider,
 		model: model.id,
 		usage: {
-			input: 0,
-			output: 0,
+			input: 1,
+			output: 1,
 			cacheRead: 0,
 			cacheWrite: 0,
-			totalTokens: 0,
+			totalTokens: 2,
 			cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
 		},
 		stopReason: "stop" as const,
@@ -39,6 +40,17 @@ function assistantMessage(text: string) {
 	};
 }
 
+function toolResultMessage(toolCallId: string, toolName: string, text: string) {
+	return {
+		role: "toolResult" as const,
+		toolCallId,
+		toolName,
+		content: [{ type: "text" as const, text }],
+		isError: false,
+		timestamp: Date.now(),
+	};
+}
+
 function readJsonl(file: string): Array<Record<string, unknown>> {
 	return fs
 		.readFileSync(file, "utf8")
@@ -52,18 +64,27 @@ function readJsonl(file: string): Array<Record<string, unknown>> {
 function messageRole(entry: Record<string, unknown>): string | undefined {
 	const message = entry.message;
 	if (!message || typeof message !== "object") return undefined;
-	const role = (message as { role?: unknown }).role;
-	return typeof role === "string" ? role : undefined;
+	if (!("role" in message) || typeof message.role !== "string") return undefined;
+	return message.role;
 }
 
 function messageContent(entry: Record<string, unknown>): unknown {
 	const message = entry.message;
 	if (!message || typeof message !== "object") return undefined;
-	return (message as { content?: unknown }).content;
+	if (!("content" in message)) return undefined;
+	return message.content;
 }
 
-describe("SessionManager batched JSONL persistence", () => {
-	it("writes later entries at the microtask or synchronous flush boundary", async () => {
+function entryKind(entry: Record<string, unknown>): string {
+	if (entry.type === "message") return messageRole(entry) ?? "message";
+	if (entry.type === "custom") {
+		return `custom:${typeof entry.customType === "string" ? entry.customType : ""}`;
+	}
+	return typeof entry.type === "string" ? entry.type : "unknown";
+}
+
+describe("SessionManager JSONL software-crash durability", () => {
+	it("makes completed entries visible on disk without a microtask or flush", () => {
 		const cwd = makeTempDir("@pi-immediate-cwd-");
 		const sessionDir = path.join(cwd, "sessions");
 		const manager = SessionManager.create(cwd, sessionDir);
@@ -73,6 +94,7 @@ describe("SessionManager batched JSONL persistence", () => {
 		manager.appendMessage({ role: "user", content: "queued before assistant", timestamp: Date.now() });
 		expect(fs.existsSync(sessionFile)).toBe(false);
 
+		// First assistant materializes the file via the synchronous rewrite path.
 		manager.appendMessage(assistantMessage("hello"));
 		expect(fs.existsSync(sessionFile)).toBe(true);
 
@@ -81,24 +103,99 @@ describe("SessionManager batched JSONL persistence", () => {
 		expect(messageRole(entries[1] ?? {})).toBe("user");
 		expect(messageRole(entries[2] ?? {})).toBe("assistant");
 
+		// Hot-path appends must land in the OS page cache before the call returns.
+		// A software crash (SIGKILL / OOM / hard abort) after append and before the
+		// next event-loop turn must not drop them.
 		manager.appendMessage({ role: "user", content: "written immediately", timestamp: Date.now() });
-
-		entries = readJsonl(sessionFile);
-		expect(entries).toHaveLength(3);
-
-		await Promise.resolve();
 		entries = readJsonl(sessionFile);
 		expect(entries).toHaveLength(4);
 		expect(messageRole(entries[3] ?? {})).toBe("user");
 		expect(messageContent(entries[3] ?? {})).toBe("written immediately");
 
-		manager.appendMessage({ role: "user", content: "flushed synchronously", timestamp: Date.now() });
-		manager.flushSync();
+		manager.appendMessage(assistantMessage("second turn"));
+		manager.appendCustomEntry("tool_execution_start", {
+			toolCallId: "call-1",
+			toolName: "bash",
+			args: { command: "echo hi" },
+		});
+		manager.appendMessage(toolResultMessage("call-1", "bash", "hi"));
 
 		entries = readJsonl(sessionFile);
-		expect(entries).toHaveLength(5);
-		expect(messageRole(entries[4] ?? {})).toBe("user");
-		expect(messageContent(entries[4] ?? {})).toBe("flushed synchronously");
+		expect(entries.map(entryKind)).toEqual([
+			"session",
+			"user",
+			"assistant",
+			"user",
+			"assistant",
+			"custom:tool_execution_start",
+			"toolResult",
+		]);
+	});
+
+	it("reopens post-checkpoint user/assistant/tool events after a crash-equivalent snapshot", async () => {
+		const cwd = makeTempDir("@pi-crash-reopen-cwd-");
+		const sessionDir = path.join(cwd, "sessions");
+		const manager = SessionManager.create(cwd, sessionDir);
+		const sessionFile = manager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected a persisted session file path");
+
+		// Durable checkpoint: materialize, then snapshot bytes.
+		manager.appendMessage({ role: "user", content: "checkpoint user", timestamp: Date.now() });
+		manager.appendMessage(assistantMessage("checkpoint assistant"));
+		const checkpointBytes = fs.readFileSync(sessionFile);
+		const checkpointKinds = parseJsonlLenient<Record<string, unknown>>(checkpointBytes.toString("utf8"))
+			.filter(entry => entry.type !== "title")
+			.map(entryKind);
+		expect(checkpointKinds).toEqual(["session", "user", "assistant"]);
+
+		// Completed events after the checkpoint — no flushSync, no await, no close.
+		manager.appendMessage({ role: "user", content: "post-checkpoint user", timestamp: Date.now() });
+		manager.appendMessage(assistantMessage("post-checkpoint assistant"));
+		manager.appendCustomEntry("tool_execution_start", {
+			toolCallId: "call-post",
+			toolName: "read",
+			args: { path: "README.md" },
+		});
+		manager.appendMessage(toolResultMessage("call-post", "read", "# readme"));
+
+		// Crash equivalent: only bytes already on disk survive. Copy without yielding
+		// to microtasks and reopen from that snapshot in a fresh manager.
+		const crashPath = path.join(cwd, "crashed-session.jsonl");
+		fs.writeFileSync(crashPath, fs.readFileSync(sessionFile));
+
+		const reopened = await SessionManager.open(crashPath);
+		const reopenedKinds = reopened.getEntries().map(entry => {
+			if (entry.type === "message") return entry.message.role;
+			if (entry.type === "custom") return `custom:${entry.customType}`;
+			return entry.type;
+		});
+		expect(reopenedKinds).toEqual([
+			"user",
+			"assistant",
+			"user",
+			"assistant",
+			"custom:tool_execution_start",
+			"toolResult",
+		]);
+
+		const postUser = reopened
+			.getEntries()
+			.find(
+				entry =>
+					entry.type === "message" &&
+					entry.message.role === "user" &&
+					entry.message.content === "post-checkpoint user",
+			);
+		expect(postUser).toBeDefined();
+		const toolResult = reopened
+			.getEntries()
+			.find(
+				entry =>
+					entry.type === "message" &&
+					entry.message.role === "toolResult" &&
+					entry.message.toolCallId === "call-post",
+			);
+		expect(toolResult).toBeDefined();
 	});
 
 	it("keeps pre-assistant sessions out of history during shutdown", async () => {
@@ -137,4 +234,126 @@ describe("SessionManager batched JSONL persistence", () => {
 		expect(messageRole(entries[1] ?? {})).toBe("user");
 		expect(messageContent(entries[1] ?? {})).toBe("persist me");
 	});
+
+	it("makes fenced appends durable while an atomic rewrite is paused", async () => {
+		// H1: in-place atomic fence used to mark entries dirty and return without
+		// writing. A crash before the paused publish resumed lost every fenced
+		// completed event. Superseding with #rewriteSynchronously must put them
+		// on disk before append returns, and the abandoned atomic must not clobber.
+		class PausedRewriteStorage extends MemorySessionStorage {
+			readonly rewriteStarted = Promise.withResolvers<void>();
+			readonly allowRewrite = Promise.withResolvers<void>();
+			guardRejections = 0;
+
+			override async writeTextAtomic(
+				filePath: string,
+				content: string,
+				options?: WriteTextAtomicOptions,
+			): Promise<void> {
+				this.rewriteStarted.resolve();
+				await this.allowRewrite.promise;
+				if (options?.commitGuard && !options.commitGuard()) {
+					this.guardRejections++;
+					return;
+				}
+				this.writeTextSync(filePath, content);
+			}
+		}
+
+		const storage = new PausedRewriteStorage();
+		const manager = SessionManager.create("/cwd", "/sessions", storage);
+		manager.appendMessage(assistantMessage("seed"));
+		await manager.flush();
+		manager.appendMessage({ role: "user", content: "checkpoint", timestamp: Date.now() });
+		await manager.flush();
+
+		const sessionFile = manager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+
+		const rewrite = manager.rewriteEntries();
+		await storage.rewriteStarted.promise;
+
+		manager.appendMessage({ role: "user", content: "fenced-user", timestamp: Date.now() });
+		manager.appendMessage(assistantMessage("fenced-assistant"));
+		manager.appendCustomEntry("tool_execution_start", {
+			toolCallId: "fenced-call",
+			toolName: "bash",
+		});
+		manager.appendMessage(toolResultMessage("fenced-call", "bash", "hi"));
+
+		// Crash-equivalent while the atomic is still paused: only storage bytes survive.
+		const crashKinds = parseJsonlLenient<Record<string, unknown>>(await storage.readText(sessionFile))
+			.filter(entry => entry.type !== "title")
+			.map(entryKind);
+		expect(crashKinds).toEqual([
+			"session",
+			"assistant",
+			"user",
+			"user",
+			"assistant",
+			"custom:tool_execution_start",
+			"toolResult",
+		]);
+
+		storage.allowRewrite.resolve();
+		await rewrite;
+		await manager.flush();
+
+		const afterKinds = parseJsonlLenient<Record<string, unknown>>(await storage.readText(sessionFile))
+			.filter(entry => entry.type !== "title")
+			.map(entryKind);
+		// Fenced events remain after the paused atomic settles (commitGuard may
+		// reject a stale body, or a superseding rewrite already owns the file).
+		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.
+		const cwd = makeTempDir("@pi-write-fail-cwd-");
+		const sessionDir = path.join(cwd, "sessions");
+		const manager = SessionManager.create(cwd, sessionDir);
+		const sessionFile = manager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+
+		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);
+
+		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.flushSync()).toThrow("ENOSPC");
+			expect(() => manager.appendMessage({ role: "user", content: "next-user", timestamp: Date.now() })).toThrow(
+				"ENOSPC",
+			);
+
+			const users = parseJsonlLenient<Record<string, unknown>>(fs.readFileSync(sessionFile, "utf8"))
+				.filter(entry => entry.type === "message" && messageRole(entry) === "user")
+				.map(entry => messageContent(entry));
+			expect(users).toEqual(["ok-user"]);
+		} finally {
+			failWrites = false;
+			writeSpy.mockRestore();
+		}
+	});
 });

packages/coding-agent/test/session-manager/move-to.test.ts

@@ -387,4 +387,100 @@ describe("SessionManager.moveTo", () => {
 		const targetSessions = await SessionManager.list(cwdB);
 		expect(targetSessions.some(item => item.path === movedFile)).toBe(true);
 	});
+
+	it("keeps post-rename fenced appends durable before trailing rewrite", async () => {
+		// Crash window: session file has been renamed to dest, `#sessionFile` is
+		// still the source path, and the trailing atomic rewrite has not run.
+		// Completed entries appended in this window must land on dest (not recreate
+		// source) and survive a crash-equivalent snapshot + reopen.
+		const session = SessionManager.create(cwdA);
+		session.appendMessage({ role: "user", content: "before move", timestamp: 1 });
+		session.appendMessage(makeAssistantMessage());
+		await session.flush();
+
+		const oldFile = session.getSessionFile();
+		if (!oldFile) throw new Error("Expected session file");
+
+		const renameFinished = Promise.withResolvers<{ dest: string }>();
+		const allowMoveToResume = Promise.withResolvers<void>();
+		const rename = fs.promises.rename.bind(fs.promises);
+		const renameSpy = spyOn(fs.promises, "rename").mockImplementation(async (source, target) => {
+			await rename(source, target);
+			if (path.resolve(source.toString()) !== path.resolve(oldFile)) return;
+			renameFinished.resolve({ dest: path.resolve(target.toString()) });
+			await allowMoveToResume.promise;
+		});
+
+		let dest = "";
+		try {
+			const move = session.moveTo(cwdB);
+			({ dest } = await renameFinished.promise);
+
+			session.appendMessage({ role: "user", content: "during move crash window", timestamp: 2 });
+			session.appendCustomEntry("tool_execution_start", {
+				toolCallId: "move-call",
+				toolName: "bash",
+			});
+			session.appendMessage({
+				role: "toolResult",
+				toolCallId: "move-call",
+				toolName: "bash",
+				content: [{ type: "text", text: "ok" }],
+				isError: false,
+				timestamp: 3,
+			});
+
+			// Crash-equivalent: only dest bytes exist; source must stay absent.
+			expect(fs.existsSync(oldFile)).toBe(false);
+			expect(fs.existsSync(dest)).toBe(true);
+			const crashBytes = fs.readFileSync(dest, "utf8");
+			expect(crashBytes).toContain("during move crash window");
+			expect(crashBytes).toContain('"customType":"tool_execution_start"');
+			expect(crashBytes).toContain("move-call");
+
+			const crashPath = path.join(testAgentDir, "crashed-move.jsonl");
+			fs.writeFileSync(crashPath, crashBytes);
+			const reopened = await SessionManager.open(crashPath);
+			const reopenedEntries = reopened.getEntries();
+			expect(
+				reopenedEntries.some(
+					entry =>
+						entry.type === "message" &&
+						entry.message.role === "user" &&
+						entry.message.content === "during move crash window",
+				),
+			).toBe(true);
+			expect(
+				reopenedEntries.some(
+					entry =>
+						entry.type === "message" &&
+						entry.message.role === "toolResult" &&
+						entry.message.toolCallId === "move-call",
+				),
+			).toBe(true);
+			expect(
+				reopenedEntries.some(entry => entry.type === "custom" && entry.customType === "tool_execution_start"),
+			).toBe(true);
+
+			allowMoveToResume.resolve();
+			await move;
+			await session.flush();
+		} finally {
+			allowMoveToResume.resolve();
+			renameSpy.mockRestore();
+		}
+
+		expect(fs.existsSync(oldFile)).toBe(false);
+		const movedFile = session.getSessionFile();
+		if (!movedFile) throw new Error("Expected moved session file");
+		const entries = await loadEntriesFromFile(movedFile);
+		expect(
+			entries.some(
+				entry =>
+					entry.type === "message" &&
+					entry.message.role === "user" &&
+					entry.message.content === "during move crash window",
+			),
+		).toBe(true);
+	});
 });

packages/coding-agent/test/session-storage.test.ts

@@ -101,8 +101,23 @@ describe("FileSessionStorage writer", () => {
 		await fsp.rm(tempDir, { recursive: true, force: true });
 	});
 
+	it("makes each append visible on disk without awaiting a microtask", () => {
+		const sessionPath = path.join(tempDir, "immediate.jsonl");
+		const writer = storage.openWriter(sessionPath, { flags: "w" });
+		// Contract: visibility must not depend on awaiting the returned Promise
+		// (or a microtask drain). appendSync / the sync body of append writes
+		// before return; awaiting alone would pass on the old microtask writer.
+		const appendSync = writer.appendSync?.bind(writer);
+		if (!appendSync) throw new Error("File writer must expose appendSync");
+		appendSync("one\n");
+		expect(fs.readFileSync(sessionPath, "utf8")).toBe("one\n");
+		appendSync("two\n");
+		expect(fs.readFileSync(sessionPath, "utf8")).toBe("one\ntwo\n");
+		void writer.close();
+	});
+
 	it("preserves append order through flush and close", async () => {
-		const sessionPath = path.join(tempDir, "batched.jsonl");
+		const sessionPath = path.join(tempDir, "ordered.jsonl");
 		const writer = storage.openWriter(sessionPath, { flags: "w" });
 		await writer.append("one\n");
 		await writer.append("two\n");
@@ -122,14 +137,17 @@ describe("FileSessionStorage writer", () => {
 		expect(fs.readFileSync(sessionPath, "utf8")).toBe("one\ntwo\n");
 	});
 
-	it("rejects close when draining a queued append fails", async () => {
-		const sessionPath = path.join(tempDir, "close-error.jsonl");
+	it("rejects appendSync and append when the underlying write fails", async () => {
+		const sessionPath = path.join(tempDir, "append-error.jsonl");
 		const writer = storage.openWriter(sessionPath, { flags: "w" });
-		void writer.append("one\n");
 		vi.spyOn(fs, "writeSync").mockImplementation(() => {
 			throw new Error("disk full");
 		});
 
+		const appendSync = writer.appendSync?.bind(writer);
+		if (!appendSync) throw new Error("File writer must expose appendSync");
+		expect(() => appendSync("one\n")).toThrow("disk full");
+		await expect(writer.append("two\n")).rejects.toThrow("disk full");
 		await expect(writer.close()).rejects.toThrow("disk full");
 	});
 });