Merge remote-tracking branch 'origin/farm/61175132/fix-windows-session-tail-loss'

2026-07-02 23:33:20 +02:00
parent 2c8daf0578 01ab7e26d7
commit 5de45fa975
9 changed files with 1111 additions and 67 deletions
@@ -14,6 +14,9 @@
 - Improved reliability of edits when file snapshots share identical 16-bit hash tags
 - Fixed ACP `terminal/create` sending the bash tool's full shell line in `command` with no `args`, which broke spec-conformant clients that spawn `command`+`args` directly (no implicit shell) — any command containing a space, pipe, `&&`, redirect, or `$(...)` failed with `ENOENT` and the agent silently degraded to read-only tools. The bash tool now wraps the shell line before calling `clientBridge.createTerminal`, reusing the same shell binary + args the local `bash-executor` resolves via `settings.getShellConfig()` (Git Bash / `bash.exe` on Windows, `$SHELL` with `sh` fallback on POSIX) so bash semantics — `$VAR`, `$(...)`, `source`, POSIX quoting, `-l` — are preserved on both platforms. ([#4333](https://github.com/can1357/oh-my-pi/issues/4333))
 - Fixed inference worker subprocesses (TTS, STT, tiny-model, mnemopi embeddings) discarding stderr, which left every unexpected exit — most visibly the local Kokoro TTS worker's recurring `exit code 7` crash loop — undiagnosable from the parent's logs. `createWorkerSubprocess` now pipes stderr without starting a live read while the worker is idle, then drains the stream after `onExit`, emits captured lines to `logger.debug` under an `<exitLabel> stderr` message, and keeps the last 16 KiB in a bounded ring that gets appended to the `Error` surfaced through `onError`. The exit surface is synchronized with the post-exit drain via `SpawnedSubprocess.stderrDrained`, so the full native trace shows up on the `tts: worker error` line without reintroducing event-loop liveness from unref'd workers. ([#4324](https://github.com/can1357/oh-my-pi/issues/4324))
+### Fixed
+
+- Fixed Windows session tail loss after atomic compaction rewrites by fencing append writers during full-file replacement and gating the atomic publish on a `commitGuard` that the storage backend checks synchronously before rename, so a concurrent `flushSync` (Ctrl+C / session-exit) is not overwritten by the stale body serialized before it ran. Covers post-compaction prompts, tool results, title changes, and exit diagnostics on the current JSONL path ([#4338](https://github.com/can1357/oh-my-pi/issues/4338)).

 ## [16.3.2] - 2026-07-02

@@ -1,5 +1,10 @@
 import { toError } from "@oh-my-pi/pi-utils";
-import type { SessionStorage, SessionStorageStat, SessionStorageWriter } from "./session-storage";
+import type {
+	SessionStorage,
+	SessionStorageStat,
+	SessionStorageWriter,
+	WriteTextAtomicOptions,
+} from "./session-storage";
 import {
 	overlayTitleSlotContent,
 	overlayTitleSlotPrefix,
@@ -234,8 +239,40 @@ export class IndexedSessionStorage implements SessionStorage {
 		}
 	}

-	writeTextAtomic(path: string, content: string): Promise<void> {
-		return this.writeText(path, content);
+	async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		const commitGuard = options?.commitGuard;
+		if (commitGuard && !commitGuard()) return;
+		await this.#awaitPath(path);
+		// A concurrent flushSync (writeTextSync) may have taken over during the
+		// awaitPath yield and bumped the epoch. Re-check before touching the
+		// index or enqueueing the backend publish.
+		if (commitGuard && !commitGuard()) return;
+		const previous = this.#index.get(path);
+		const mtimeMs = this.#allocMtimeMs();
+		const title = titleUpdateFromSlot(parseTitleSlotFromContent(content));
+		this.#setIndex(path, byteLength(content), mtimeMs, title ?? null);
+		try {
+			await this.#enqueuePath(
+				path,
+				async () => {
+					// Final guard immediately before the backend actually publishes.
+					// If a concurrent writer has advanced the index past our
+					// optimistic entry, leave that newer state alone; otherwise
+					// restore the pre-write snapshot so readers do not observe a
+					// body we never wrote.
+					if (commitGuard && !commitGuard()) {
+						const current = this.#index.get(path);
+						if (current?.mtimeMs === mtimeMs) this.#restoreIndex(path, previous);
+						return;
+					}
+					await this.#backend.writeFull(path, content, mtimeMs, title);
+				},
+				{ trackDrain: false },
+			);
+		} catch (err) {
+			this.#restoreIndex(path, previous);
+			throw toError(err);
+		}
 	}

 	async rename(src: string, dst: string): Promise<void> {
@@ -383,6 +383,17 @@ export class SessionManager {
 	#diskFailureLogged = false;
 	/** Bumped on every sync rewrite / chain reset so stale queued tasks become no-ops. */
 	#diskEpoch = 0;
+	/**
+	 * Epoch of the in-flight atomic rewrite, or `null` when no rewrite is running.
+	 * The fence in {@link #appendToSessionFile} only applies while this matches
+	 * `#diskEpoch`: once a synchronous rewrite (`flushSync` → `#rewriteSynchronously`)
+	 * bumps the epoch, the pending atomic publish is guaranteed to abandon via
+	 * its `commitGuard`, and appends can safely take the hot path against the
+	 * freshly-published file.
+	 */
+	#atomicRewriteFenceEpoch: number | null = null;
+	/** Set by synchronous appends that land while an atomic replacement is active. */
+	#atomicRewriteDirty = false;

 	#artifactManager: ArtifactManager | null = null;
 	#artifactManagerSessionFile: string | null = null;
@@ -555,25 +566,63 @@ export class SessionManager {

 	/**
 	 * Rewrite the whole file atomically (temp-write + rename, EPERM-safe) on the
-	 * disk chain. The body is serialized inside the task — after the writer is
-	 * closed — so entries appended before the task runs are included.
+	 * disk chain. The body is serialized after the writer is closed. The fence
+	 * is enabled BEFORE `#closeWriterHandle()` and stays active until the last
+	 * atomic publish returns, so a sync append landing in the close-yield window
+	 * cannot open a fresh writer that the pending replacement would then detach
+	 * from the current JSONL path. A `commitGuard` also prevents a superseding
+	 * synchronous rewrite from being overwritten by the stale body serialized
+	 * before it ran.
 	 */
 	async #rewriteAtomically(): Promise<void> {
 		if (!this.#persist || !this.#sessionFile) return;

-		const epoch = this.#diskEpoch;
+		const startEpoch = this.#diskEpoch;
 		await this.#scheduleDiskWork(
 			async () => {
+				if (await this.#runFencedAtomicRewrite(startEpoch)) {
+					this.#fileIsCurrent = true;
+					this.#rewriteRequired = false;
+					this.#hasTitleSlot = true;
+				}
+			},
+			{ epoch: startEpoch },
+		);
+	}
+
+	/**
+	 * Shared fenced atomic-rewrite loop used by `#rewriteAtomically` and the
+	 * `#persistTitleChangeEntry` fallback. Holds `#atomicRewriteActive` across
+	 * the writer close and the full-file replace, and loops on
+	 * `#atomicRewriteDirty` so any fenced append that lands during the rewrite
+	 * is captured before the task resolves. Returns `false` when the disk epoch
+	 * moved (a superseding synchronous rewrite has taken over) so callers skip
+	 * their post-publish state updates.
+	 */
+	async #runFencedAtomicRewrite(epoch: number): Promise<boolean> {
+		this.#atomicRewriteFenceEpoch = epoch;
+		try {
+			do {
+				this.#atomicRewriteDirty = false;
 				await this.#closeWriterHandle();
 				const sessionFile = this.#sessionFile;
-				if (!sessionFile) return;
-				await this.#storage.writeTextAtomic(sessionFile, this.#fileBody());
-				this.#fileIsCurrent = true;
-				this.#rewriteRequired = false;
-				this.#hasTitleSlot = true;
-			},
-			{ epoch },
-		);
+				if (!sessionFile) return false;
+				if (this.#diskEpoch !== epoch) return false;
+				await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), {
+					commitGuard: () => this.#diskEpoch === epoch,
+				});
+				if (this.#diskEpoch !== epoch) return false;
+			} while (this.#atomicRewriteDirty);
+			return true;
+		} finally {
+			// Only relinquish the fence if we still own it. A superseding
+			// synchronous rewrite (`flushSync` → `#rewriteSynchronously`) may
+			// have reset `#diskTail`, scheduled a fresh atomic task at the new
+			// epoch, and that task may have taken ownership of the fence while
+			// this stale rewrite was still awaiting storage. Clearing it here
+			// unconditionally would strand appends during the newer publish.
+			if (this.#atomicRewriteFenceEpoch === epoch) this.#atomicRewriteFenceEpoch = null;
+		}
 	}

 	#appendToSessionFile(entry: SessionEntry): void {
@@ -588,6 +637,19 @@ export class SessionManager {
 			return;
 		}

+		// Atomic replacement window: the old path may be moved aside underneath
+		// any newly-opened append handle (Windows EPERM fallback). Do not open a
+		// writer here; the active rewrite loops and serializes a fresh full body.
+		// 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.#atomicRewriteFenceEpoch !== null && this.#atomicRewriteFenceEpoch === this.#diskEpoch) {
+			this.#fileIsCurrent = false;
+			this.#rewriteRequired = true;
+			this.#atomicRewriteDirty = true;
+			return;
+		}
 		// Cold/divergent: not on disk yet, or in-memory entries diverged from the
 		// file → rewrite the whole file synchronously and keep going.
 		if (!this.#fileIsCurrent || this.#rewriteRequired) {
@@ -637,10 +699,9 @@ export class SessionManager {
 				try {
 					await this.#appendWriter().append(line);
 					await this.#storage.updateSessionTitle(sessionFile, update);
-					this.#fileIsCurrent = true;
+					if (this.#diskEpoch === epoch) this.#fileIsCurrent = true;
 				} catch {
-					await this.#closeWriterHandle();
-					await this.#storage.writeTextAtomic(sessionFile, this.#fileBody());
+					if (!(await this.#runFencedAtomicRewrite(epoch))) return;
 					this.#clearDiskError();
 					this.#fileIsCurrent = true;
 					this.#rewriteRequired = false;
@@ -1049,6 +1110,10 @@ export class SessionManager {
 		await this.#scheduleDiskWork(async () => {
 			if (this.#writer?.isOpen()) await this.#writer.flush();
 		});
+		// Drain any fire-and-forget backing writes (e.g. `writeTextSync` queued
+		// on IndexedSessionStorage during `flushSync`) so callers relying on
+		// flush() see the write durably visible to readers.
+		await this.#storage.drain();
 		if (this.#diskFailure) throw this.#diskFailure;
 	}

@@ -1078,6 +1143,10 @@ export class SessionManager {
 			if (hadWriter || (this.#sessionFile && this.#storage.existsSync(this.#sessionFile)))
 				this.#fileIsCurrent = true;
 		});
+		// Wait for any queued backing writes (IndexedSessionStorage per-path
+		// tail) to become durable so a graceful shutdown does not exit while
+		// a fire-and-forget publish is still on the wire.
+		await this.#storage.drain();
 		if (this.#diskFailure) throw this.#diskFailure;
 	}

@@ -28,6 +28,18 @@ export interface SessionStorageWriter {
 	getError(): Error | undefined;
 }

+/**
+ * Optional guard applied by {@link SessionStorage.writeTextAtomic}. The
+ * backend MUST call `commitGuard()` synchronously immediately before it makes
+ * the staged content visible at `path`. If it returns `false`, the staged
+ * write is discarded and the target is left untouched. Backends MUST NOT
+ * yield between calling the guard and publishing the write, so a concurrent
+ * synchronous rewrite that took over cannot be overwritten by a stale body.
+ */
+export interface WriteTextAtomicOptions {
+	commitGuard?: () => boolean;
+}
+
 export interface SessionStorage {
 	ensureDirSync(dir: string): void;
 	existsSync(path: string): boolean;
@@ -48,11 +60,20 @@ export interface SessionStorage {
 	/** Read the requested UTF-8 byte windows from the head and tail of the file. */
 	readTextSlices(path: string, prefixBytes: number, suffixBytes: number): Promise<[string, string]>;
 	writeText(path: string, content: string): Promise<void>;
-	writeTextAtomic(path: string, content: string): Promise<void>;
+	writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void>;
 	rename(path: string, nextPath: string): Promise<void>;
 	unlink(path: string): Promise<void>;
 	deleteSessionWithArtifacts(sessionPath: string): Promise<void>;
 	openWriter(path: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }): SessionStorageWriter;
+	/**
+	 * Wait for every backing write scheduled by this storage to become durably
+	 * visible. Sync backends (file, memory) return immediately because their
+	 * writes complete in-body; async backends (Redis/SQL via
+	 * {@link IndexedSessionStorage}) await their per-path queues so a caller
+	 * driving a graceful shutdown does not exit while a fire-and-forget
+	 * `writeTextSync` publish is still on the wire.
+	 */
+	drain(): Promise<void>;
 }

 // FinalizationRegistry to clean up leaked file descriptors
@@ -229,53 +250,107 @@ export class FileSessionStorage implements SessionStorage {
 		await Bun.write(path, content, { createPath: true });
 	}

-	async writeTextAtomic(fpath: string, content: string): Promise<void> {
+	async writeTextAtomic(fpath: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
 		const dir = path.resolve(fpath, "..");
 		const tempPath = path.join(dir, `.${path.basename(fpath)}.${Snowflake.next()}.tmp`);
 		await fs.promises.mkdir(dir, { recursive: true });
 		try {
 			await fs.promises.writeFile(tempPath, content);
-			try {
-				await this.rename(tempPath, fpath);
-				return;
-			} catch (err) {
-				if (!hasFsCode(err, "EPERM")) throw toError(err);
-				await this.#replaceSessionFileAfterEperm(tempPath, fpath, err);
-				return;
-			}
 		} catch (err) {
-			try {
-				await this.unlink(tempPath);
-			} catch (cleanupErr) {
-				if (!isEnoent(cleanupErr)) {
-					logger.warn("Failed to remove session rewrite temp file", {
-						sessionFile: fpath,
-						tempPath,
-						error: toError(cleanupErr).message,
-					});
-				}
-			}
+			this.#discardTemp(tempPath, fpath);
 			throw toError(err);
 		}
+		// Guard-check + rename MUST NOT be separated by an await. A concurrent
+		// synchronous rewrite (flushSync -> #rewriteSynchronously) can otherwise
+		// publish a fresh body between the check and the rename, and this stale
+		// staged body would overwrite it. Sync rename closes that window.
+		if (options?.commitGuard && !options.commitGuard()) {
+			this.#discardTemp(tempPath, fpath);
+			return;
+		}
+		try {
+			this.renameSync(tempPath, fpath);
+			return;
+		} catch (err) {
+			if (!hasFsCode(err, "EPERM")) {
+				this.#discardTemp(tempPath, fpath);
+				throw toError(err);
+			}
+			try {
+				this.#replaceSessionFileAfterEpermSync(tempPath, fpath, err, options?.commitGuard);
+			} catch (fallbackErr) {
+				this.#discardTemp(tempPath, fpath);
+				throw fallbackErr;
+			}
+		}
 	}

-	async #replaceSessionFileAfterEperm(tempPath: string, targetPath: string, renameError: unknown): Promise<void> {
+	/**
+	 * Sync rename hook. Split from `rename` so `writeTextAtomic` can perform its
+	 * guard-then-publish step without a yield, and so tests can inject
+	 * Windows-style EPERM at the sync layer used by the atomic path.
+	 */
+	renameSync(source: string, target: string): void {
+		fs.renameSync(source, target);
+	}
+
+	#discardTemp(tempPath: string, targetPath: string): void {
+		try {
+			fs.unlinkSync(tempPath);
+		} catch (err) {
+			if (!isEnoent(err)) {
+				logger.warn("Failed to remove session rewrite temp file", {
+					sessionFile: targetPath,
+					tempPath,
+					error: toError(err).message,
+				});
+			}
+		}
+	}
+
+	#replaceSessionFileAfterEpermSync(
+		tempPath: string,
+		targetPath: string,
+		renameError: unknown,
+		commitGuard?: () => boolean,
+	): void {
 		const dir = path.resolve(targetPath, "..");
 		const backupPath = path.join(dir, `${path.basename(targetPath)}.${Snowflake.next()}.bak`);
 		try {
-			await this.rename(targetPath, backupPath);
+			this.renameSync(targetPath, backupPath);
 		} catch (moveAsideError) {
 			if (isEnoent(moveAsideError)) {
-				await this.rename(tempPath, targetPath);
+				if (commitGuard && !commitGuard()) {
+					this.#discardTemp(tempPath, targetPath);
+					return;
+				}
+				this.renameSync(tempPath, targetPath);
 				return;
 			}
 			throw toError(renameError);
 		}
+		if (commitGuard && !commitGuard()) {
+			// A concurrent synchronous rewrite published a fresh body between the
+			// move-aside and this point. Restore the moved-aside file so we do
+			// not overwrite it with our staged (stale) body, and drop the temp
+			// so `writeTextAtomic`'s "discard on abandon" contract holds.
+			try {
+				this.renameSync(backupPath, targetPath);
+			} catch (restoreErr) {
+				logger.warn("Failed to restore backup after commitGuard rejection", {
+					sessionFile: targetPath,
+					backupPath,
+					error: toError(restoreErr).message,
+				});
+			}
+			this.#discardTemp(tempPath, targetPath);
+			return;
+		}
 		try {
-			await this.rename(tempPath, targetPath);
+			this.renameSync(tempPath, targetPath);
 		} catch (replaceError) {
 			try {
-				await this.rename(backupPath, targetPath);
+				this.renameSync(backupPath, targetPath);
 			} catch (rollbackErr) {
 				const rollbackError = toError(rollbackErr);
 				throw new Error(
@@ -288,7 +363,7 @@ export class FileSessionStorage implements SessionStorage {
 			throw toError(replaceError);
 		}
 		try {
-			await this.unlink(backupPath);
+			fs.unlinkSync(backupPath);
 		} catch (err) {
 			if (!isEnoent(err)) {
 				logger.warn("Failed to remove session rewrite backup", {
@@ -312,6 +387,12 @@ export class FileSessionStorage implements SessionStorage {
 		return fs.promises.unlink(path);
 	}

+	drain(): Promise<void> {
+		// File writes complete synchronously in-body via fs.writeFileSync /
+		// fs.renameSync, so there is no queued work to await.
+		return Promise.resolve();
+	}
+
 	openWriter(path: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }): SessionStorageWriter {
 		return new FileSessionStorageWriter(path, options);
 	}
@@ -621,7 +702,8 @@ export class MemorySessionStorage implements SessionStorage {
 		return Promise.resolve();
 	}

-	writeTextAtomic(path: string, content: string): Promise<void> {
+	writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		if (options?.commitGuard && !options.commitGuard()) return Promise.resolve();
 		this.writeTextSync(path, content);
 		return Promise.resolve();
 	}
@@ -642,6 +724,10 @@ export class MemorySessionStorage implements SessionStorage {
 		return Promise.resolve();
 	}

+	drain(): Promise<void> {
+		return Promise.resolve();
+	}
+
 	openWriter(path: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }): SessionStorageWriter {
 		return new MemorySessionStorageWriter(this, path, options);
 	}
@@ -0,0 +1,681 @@
+import { describe, expect, it } from "bun:test";
+import { getBundledModel } from "@oh-my-pi/pi-catalog/models";
+import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager";
+import {
+	MemorySessionStorage,
+	type SessionStorageWriter,
+	type WriteTextAtomicOptions,
+} from "@oh-my-pi/pi-coding-agent/session/session-storage";
+import type { SessionTitleUpdate } from "@oh-my-pi/pi-coding-agent/session/session-title-slot";
+
+interface DetachableWriter extends SessionStorageWriter {
+	detach(): void;
+}
+
+class DetachingRewriteStorage extends MemorySessionStorage {
+	readonly detachedLines: string[] = [];
+	readonly rewriteStarted = Promise.withResolvers<void>();
+	readonly allowRewrite = Promise.withResolvers<void>();
+	pausedRewrites = 0;
+	guardRejections = 0;
+	readonly #writers = new Set<DetachableWriter>();
+
+	openWriter(path: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }): SessionStorageWriter {
+		const inner = super.openWriter(path, options);
+		const writers = this.#writers;
+		const detachedLines = this.detachedLines;
+		let detached = false;
+		const writer: DetachableWriter = {
+			async append(line: string): Promise<void> {
+				if (detached) {
+					detachedLines.push(line);
+					return;
+				}
+				await inner.append(line);
+			},
+			async flush(): Promise<void> {
+				await inner.flush();
+			},
+			isOpen(): boolean {
+				const open = inner.isOpen();
+				return open;
+			},
+			async close(): Promise<void> {
+				writers.delete(writer);
+				await inner.close();
+			},
+			getError(): Error | undefined {
+				const error = inner.getError();
+				return error;
+			},
+			detach(): void {
+				if (detached) return;
+				detached = true;
+			},
+		};
+		writers.add(writer);
+		return writer;
+	}
+
+	override async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		this.pausedRewrites++;
+		this.rewriteStarted.resolve();
+		await this.allowRewrite.promise;
+		if (options?.commitGuard && !options.commitGuard()) {
+			this.guardRejections++;
+			return;
+		}
+		for (const writer of this.#writers) writer.detach();
+		this.writeTextSync(path, content);
+	}
+}
+
+class CloseGatedRewriteStorage extends MemorySessionStorage {
+	readonly closeStarted = Promise.withResolvers<void>();
+	readonly allowClose = Promise.withResolvers<void>();
+	readonly writeStarted = Promise.withResolvers<void>();
+	readonly allowWrite = Promise.withResolvers<void>();
+	readonly detachedLines: string[] = [];
+	writerOpens = 0;
+	guardRejections = 0;
+	readonly #detachables = new Set<DetachableWriter>();
+
+	override openWriter(
+		path: string,
+		options?: { flags?: "a" | "w"; onError?: (err: Error) => void },
+	): SessionStorageWriter {
+		this.writerOpens++;
+		const inner = super.openWriter(path, options);
+		const closeStarted = this.closeStarted;
+		const allowClose = this.allowClose;
+		const detachedLines = this.detachedLines;
+		const detachables = this.#detachables;
+		let detached = false;
+		const writer: DetachableWriter = {
+			async append(line: string): Promise<void> {
+				if (detached) {
+					detachedLines.push(line);
+					return;
+				}
+				await inner.append(line);
+			},
+			async flush(): Promise<void> {
+				await inner.flush();
+			},
+			isOpen(): boolean {
+				return inner.isOpen();
+			},
+			async close(): Promise<void> {
+				closeStarted.resolve();
+				await allowClose.promise;
+				detachables.delete(writer);
+				await inner.close();
+			},
+			getError(): Error | undefined {
+				return inner.getError();
+			},
+			detach(): void {
+				detached = true;
+			},
+		};
+		detachables.add(writer);
+		return writer;
+	}
+
+	override async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		this.writeStarted.resolve();
+		await this.allowWrite.promise;
+		if (options?.commitGuard && !options.commitGuard()) {
+			this.guardRejections++;
+			return;
+		}
+		// Emulate the Windows post-EPERM fallback: writers opened against the
+		// pre-replacement target end up attached to the moved-aside file after
+		// this call returns, so their future appends are detached from `path`.
+		for (const w of this.#detachables) w.detach();
+		this.writeTextSync(path, content);
+	}
+}
+
+describe("SessionManager atomic rewrite race", () => {
+	it("keeps post-compaction appends on the current JSONL path", async () => {
+		const storage = new DetachingRewriteStorage();
+		const sessionManager = SessionManager.create("/cwd", "/sessions", storage);
+		const model = getBundledModel("anthropic", "claude-sonnet-4-5");
+		if (!model) throw new Error("Expected built-in anthropic model");
+
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "seed response" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.flush();
+		sessionManager.appendMessage({ role: "user", content: "before compaction", timestamp: Date.now() });
+		await sessionManager.flush();
+
+		const firstKeptEntryId = sessionManager.getBranch()[0]?.id;
+		if (!firstKeptEntryId) throw new Error("Expected seeded branch entry");
+		sessionManager.appendCompaction("older summary", "older", firstKeptEntryId, 100);
+		await sessionManager.flush();
+		sessionManager.appendCompaction("newer summary", "newer", firstKeptEntryId, 80);
+		await storage.rewriteStarted.promise;
+
+		sessionManager.appendMessage({ role: "user", content: "during rewrite prompt", timestamp: Date.now() });
+		sessionManager.appendCustomMessageEntry("during_rewrite_custom", "during rewrite custom", false);
+		sessionManager.appendCustomEntry("session_exit", { reason: "dispose", kind: "normal" });
+		const titlePersisted = sessionManager.setSessionName("Post rewrite title", "user", "test");
+
+		storage.allowRewrite.resolve();
+		await titlePersisted;
+		await sessionManager.flush();
+		sessionManager.appendMessage({
+			role: "toolResult",
+			toolCallId: "toolu_after_rewrite",
+			toolName: "bash",
+			content: [{ type: "text", text: "after rewrite tool" }],
+			isError: false,
+			timestamp: Date.now(),
+		});
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "after rewrite assistant" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.close();
+
+		const sessionFile = sessionManager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+		const content = await storage.readText(sessionFile);
+		const [titleSlot] = content.split("\n");
+		expect(JSON.parse(titleSlot ?? "{}")).toMatchObject({
+			type: "title",
+			title: "Post rewrite title",
+			source: "user",
+		});
+		expect(content).toContain("newer summary");
+		expect(content).toContain("during rewrite prompt");
+		expect(content).toContain("during rewrite custom");
+		expect(content).toContain('"customType":"session_exit"');
+		expect(content).toContain('"type":"title_change"');
+		expect(content).toContain("after rewrite tool");
+		expect(content).toContain("after rewrite assistant");
+		expect(storage.detachedLines).toEqual([]);
+
+		const reloaded = await SessionManager.open(sessionFile, "/sessions", storage, {
+			initialCwd: "/cwd",
+			suppressBreadcrumb: true,
+		});
+		const branch = reloaded.getBranch();
+		expect(branch.some(entry => entry.type === "compaction" && entry.summary === "newer summary")).toBe(true);
+		expect(
+			branch.some(
+				entry =>
+					entry.type === "message" &&
+					entry.message.role === "user" &&
+					entry.message.content === "during rewrite prompt",
+			),
+		).toBe(true);
+		expect(
+			branch.some(
+				entry =>
+					entry.type === "message" &&
+					entry.message.role === "assistant" &&
+					entry.message.content.some(part => part.type === "text" && part.text === "after rewrite assistant"),
+			),
+		).toBe(true);
+		expect(reloaded.getSessionName()).toBe("Post rewrite title");
+	});
+
+	it("flushSync during an in-flight atomic rewrite durably publishes the exit record", async () => {
+		const storage = new DetachingRewriteStorage();
+		const sessionManager = SessionManager.create("/cwd", "/sessions", storage);
+		const model = getBundledModel("anthropic", "claude-sonnet-4-5");
+		if (!model) throw new Error("Expected built-in anthropic model");
+
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "seed response" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.flush();
+		sessionManager.appendMessage({ role: "user", content: "before compaction", timestamp: Date.now() });
+		await sessionManager.flush();
+
+		const firstKeptEntryId = sessionManager.getBranch()[0]?.id;
+		if (!firstKeptEntryId) throw new Error("Expected seeded branch entry");
+		sessionManager.appendCompaction("older summary", "older", firstKeptEntryId, 100);
+		await sessionManager.flush();
+		// Second compaction elides the first, scheduling a full-file rewrite that
+		// parks inside the fake storage until we release it.
+		sessionManager.appendCompaction("newer summary", "newer", firstKeptEntryId, 80);
+		await storage.rewriteStarted.promise;
+
+		// Simulate a Ctrl+C teardown: append a session_exit custom entry (fenced
+		// because the atomic rewrite is active) and flushSync it.
+		sessionManager.appendCustomEntry("session_exit", { reason: "sigterm", kind: "signal" });
+		expect(() => sessionManager.flushSync()).not.toThrow();
+
+		const sessionFile = sessionManager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+		const afterFlush = await storage.readText(sessionFile);
+		expect(afterFlush).toContain('"customType":"session_exit"');
+		expect(afterFlush).toContain("newer summary");
+
+		// Release the in-flight atomic rewrite. Its commitGuard MUST reject the
+		// stale body serialized before flushSync bumped the disk epoch; otherwise
+		// the async publish would overwrite the durable exit record.
+		storage.allowRewrite.resolve();
+		await Promise.resolve();
+		await Promise.resolve();
+
+		const afterRelease = await storage.readText(sessionFile);
+		expect(afterRelease).toContain('"customType":"session_exit"');
+		expect(afterRelease).toContain("newer summary");
+		expect(storage.guardRejections).toBeGreaterThanOrEqual(1);
+		expect(storage.detachedLines).toEqual([]);
+	});
+});
+
+describe("SessionManager atomic rewrite fence spans writer.close()", () => {
+	it("blocks a fresh writer from opening while an in-flight rewrite awaits writer.close()", async () => {
+		const storage = new CloseGatedRewriteStorage();
+		const sessionManager = SessionManager.create("/cwd", "/sessions", storage);
+		const model = getBundledModel("anthropic", "claude-sonnet-4-5");
+		if (!model) throw new Error("Expected built-in anthropic model");
+
+		// Seed an assistant message so the session materializes on disk without
+		// opening a persistent writer (cold-path #rewriteSynchronously).
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "seed response" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.flush();
+		// Second append takes the hot path and opens a persistent writer that
+		// the atomic rewrite task must close before publishing the replacement.
+		sessionManager.appendMessage({ role: "user", content: "before rewrite", timestamp: Date.now() });
+		await sessionManager.flush();
+		const opensBeforeRewrite = storage.writerOpens;
+		expect(opensBeforeRewrite).toBeGreaterThan(0);
+
+		// Schedule an atomic rewrite; the task opens by closing the current
+		// writer, which parks on the fake's close gate. The fence must be active
+		// throughout the entire close-yield window so no fresh writer opens.
+		const rewrite = sessionManager.rewriteEntries();
+		await storage.closeStarted.promise;
+
+		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);
+
+		storage.allowClose.resolve();
+		storage.allowWrite.resolve();
+		await rewrite;
+		await sessionManager.flush();
+
+		const sessionFile = sessionManager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+		const content = await storage.readText(sessionFile);
+		expect(content).toContain("during close");
+		expect(content).toContain('"customType":"during_close_custom"');
+		expect(storage.detachedLines).toEqual([]);
+	});
+});
+
+class TitleFallbackPausingStorage extends MemorySessionStorage {
+	readonly writeStarted = Promise.withResolvers<void>();
+	readonly allowWrite = Promise.withResolvers<void>();
+	writeTextAtomicCalls = 0;
+	failNextUpdateTitle = false;
+
+	override async updateSessionTitle(path: string, update: SessionTitleUpdate): Promise<void> {
+		if (this.failNextUpdateTitle) {
+			this.failNextUpdateTitle = false;
+			throw new Error("updateSessionTitle forced failure");
+		}
+		return super.updateSessionTitle(path, update);
+	}
+
+	override async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		this.writeTextAtomicCalls += 1;
+		this.writeStarted.resolve();
+		await this.allowWrite.promise;
+		if (options?.commitGuard && !options.commitGuard()) return;
+		this.writeTextSync(path, content);
+	}
+}
+
+describe("SessionManager title-change fallback fenced-append durability", () => {
+	it("loops on the dirty flag so fenced appends during the fallback rewrite persist", async () => {
+		const storage = new TitleFallbackPausingStorage();
+		const sessionManager = SessionManager.create("/cwd", "/sessions", storage);
+		const model = getBundledModel("anthropic", "claude-sonnet-4-5");
+		if (!model) throw new Error("Expected built-in anthropic model");
+
+		// Materialize the session on disk with a title slot present so a later
+		// setSessionName takes the append-then-updateSessionTitle try branch
+		// instead of the up-front #rewriteAtomically fallback.
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "seed response" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.flush();
+		await sessionManager.setSessionName("initial title", "user", "seed");
+		await sessionManager.flush();
+		expect(storage.writeTextAtomicCalls).toBe(0);
+
+		// Force the try branch to fail so the catch runs the atomic-rewrite loop.
+		storage.failNextUpdateTitle = true;
+		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.
+		sessionManager.appendMessage({
+			role: "user",
+			content: "during title fallback",
+			timestamp: Date.now(),
+		});
+		sessionManager.appendCustomEntry("during_title_fallback_custom", { reason: "test" });
+
+		storage.allowWrite.resolve();
+		await rename;
+		await sessionManager.flush();
+
+		const sessionFile = sessionManager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+		const content = await storage.readText(sessionFile);
+		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);
+	});
+});
+
+describe("SessionManager fence relaxes when flushSync supersedes the atomic rewrite", () => {
+	it("routes post-flushSync appends onto the hot path so they land on disk before close()", async () => {
+		const storage = new DetachingRewriteStorage();
+		const sessionManager = SessionManager.create("/cwd", "/sessions", storage);
+		const model = getBundledModel("anthropic", "claude-sonnet-4-5");
+		if (!model) throw new Error("Expected built-in anthropic model");
+
+		// Materialize a session on disk so subsequent rewrites are meaningful.
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "seed response" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.flush();
+		sessionManager.appendMessage({ role: "user", content: "before rewrite", timestamp: Date.now() });
+		await sessionManager.flush();
+
+		// Schedule an atomic rewrite that parks inside writeTextAtomic.
+		const rewrite = sessionManager.rewriteEntries();
+		await storage.rewriteStarted.promise;
+
+		// (1) Append X1 while the fence epoch is still current: fenced into memory
+		// and captured by flushSync's #fileBody() below.
+		sessionManager.appendCustomEntry("during_active_atomic", { data: "X1" });
+
+		// (2) flushSync supersedes the pending atomic (bumps #diskEpoch) and
+		// publishes a synchronous body containing X1.
+		expect(() => sessionManager.flushSync()).not.toThrow();
+
+		// (3) Post-flushSync append MUST take the hot path: pre-fix, the fence
+		// stayed active and this entry was only marked dirty, then dropped when
+		// the pending atomic returned false and close() published nothing.
+		sessionManager.appendMessage({
+			role: "user",
+			content: "post_flush_sync_prompt",
+			timestamp: Date.now(),
+		});
+		sessionManager.appendCustomEntry("post_flush_sync_custom", { data: "X2" });
+
+		const sessionFile = sessionManager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+		const midFlight = await storage.readText(sessionFile);
+		expect(midFlight).toContain('"customType":"during_active_atomic"');
+		expect(midFlight).toContain("post_flush_sync_prompt");
+		expect(midFlight).toContain('"customType":"post_flush_sync_custom"');
+
+		// Release the paused atomic rewrite. Its commitGuard MUST reject — a
+		// stale publish now would clobber the hot-path appends written above.
+		storage.allowRewrite.resolve();
+		await rewrite;
+		await sessionManager.close();
+
+		const afterClose = await storage.readText(sessionFile);
+		expect(afterClose).toContain('"customType":"during_active_atomic"');
+		expect(afterClose).toContain("post_flush_sync_prompt");
+		expect(afterClose).toContain('"customType":"post_flush_sync_custom"');
+		expect(storage.guardRejections).toBeGreaterThanOrEqual(1);
+		expect(storage.detachedLines).toEqual([]);
+	});
+});
+
+interface PauseHandle {
+	started: PromiseWithResolvers<void>;
+	allow: PromiseWithResolvers<void>;
+}
+
+class SequencedRewriteStorage extends MemorySessionStorage {
+	readonly detachedLines: string[] = [];
+	readonly pauses: PauseHandle[] = [];
+	guardRejections = 0;
+	writerOpens = 0;
+	pauseCount = 0;
+	#calls = 0;
+	readonly #writers = new Set<DetachableWriter>();
+
+	override openWriter(
+		path: string,
+		options?: { flags?: "a" | "w"; onError?: (err: Error) => void },
+	): SessionStorageWriter {
+		this.writerOpens++;
+		const inner = super.openWriter(path, options);
+		const writers = this.#writers;
+		const detachedLines = this.detachedLines;
+		let detached = false;
+		const writer: DetachableWriter = {
+			async append(line: string): Promise<void> {
+				if (detached) {
+					detachedLines.push(line);
+					return;
+				}
+				await inner.append(line);
+			},
+			async flush(): Promise<void> {
+				await inner.flush();
+			},
+			isOpen(): boolean {
+				return inner.isOpen();
+			},
+			async close(): Promise<void> {
+				writers.delete(writer);
+				await inner.close();
+			},
+			getError(): Error | undefined {
+				return inner.getError();
+			},
+			detach(): void {
+				detached = true;
+			},
+		};
+		writers.add(writer);
+		return writer;
+	}
+
+	override async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		const index = this.#calls++;
+		if (index < this.pauseCount) {
+			const pause: PauseHandle = {
+				started: Promise.withResolvers<void>(),
+				allow: Promise.withResolvers<void>(),
+			};
+			this.pauses.push(pause);
+			pause.started.resolve();
+			await pause.allow.promise;
+		}
+		if (options?.commitGuard && !options.commitGuard()) {
+			this.guardRejections++;
+			return;
+		}
+		for (const w of this.#writers) w.detach();
+		this.writeTextSync(path, content);
+	}
+}
+
+describe("SessionManager fence handoff across superseded rewrites", () => {
+	it("preserves the newer fence when a stale rewrite unwinds after flushSync", async () => {
+		const storage = new SequencedRewriteStorage();
+		storage.pauseCount = 2;
+		const sessionManager = SessionManager.create("/cwd", "/sessions", storage);
+		const model = getBundledModel("anthropic", "claude-sonnet-4-5");
+		if (!model) throw new Error("Expected built-in anthropic model");
+
+		sessionManager.appendMessage({
+			role: "assistant",
+			content: [{ type: "text", text: "seed response" }],
+			api: model.api,
+			provider: model.provider,
+			model: model.id,
+			usage: {
+				input: 0,
+				output: 0,
+				cacheRead: 0,
+				cacheWrite: 0,
+				totalTokens: 0,
+				cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+			},
+			stopReason: "stop",
+			timestamp: Date.now(),
+		});
+		await sessionManager.flush();
+		sessionManager.appendMessage({ role: "user", content: "before rewrite", timestamp: Date.now() });
+		await sessionManager.flush();
+		expect(storage.writerOpens).toBeGreaterThan(0);
+
+		// Stale rewrite parks at pauses[0]. Fence epoch = 0.
+		const stale = sessionManager.rewriteEntries();
+		while (storage.pauses.length < 1) await Promise.resolve();
+		await storage.pauses[0].started.promise;
+
+		// A fenced append flips fileIsCurrent so flushSync actually publishes,
+		// bumping the epoch to 1 with the fenced entry captured in the body.
+		sessionManager.appendCustomEntry("during_stale", { data: "X1" });
+		expect(() => sessionManager.flushSync()).not.toThrow();
+
+		// Newer rewrite scheduled at epoch=1. Parks at pauses[1]. Fence epoch = 1.
+		const newer = sessionManager.rewriteEntries();
+		while (storage.pauses.length < 2) await Promise.resolve();
+		await storage.pauses[1].started.promise;
+
+		const opensBeforeUnwind = storage.writerOpens;
+
+		// Release the stale rewrite. Its `finally` MUST NOT clear the newer
+		// fence — pre-fix an unconditional reset stranded the newer rewrite's
+		// epoch bookkeeping so subsequent appends took the hot path and were
+		// then detached by the newer publish.
+		storage.pauses[0].allow.resolve();
+		for (let i = 0; i < 20; i++) await Promise.resolve();
+
+		// Sync append during the newer rewrite: MUST still be fenced.
+		sessionManager.appendCustomEntry("during_newer", { data: "X2" });
+		expect(storage.writerOpens).toBe(opensBeforeUnwind);
+
+		// Release the newer rewrite. Its dirty-loop second iteration is not
+		// paused (pauseCount=2) and captures X2 into the published body.
+		storage.pauses[1].allow.resolve();
+
+		await stale;
+		await newer;
+		await sessionManager.close();
+
+		const sessionFile = sessionManager.getSessionFile();
+		if (!sessionFile) throw new Error("Expected session file");
+		const content = await storage.readText(sessionFile);
+		expect(content).toContain('"customType":"during_stale"');
+		expect(content).toContain('"customType":"during_newer"');
+		expect(storage.guardRejections).toBeGreaterThanOrEqual(1);
+		expect(storage.detachedLines).toEqual([]);
+	});
+});
@@ -30,6 +30,7 @@ import {
 	MemorySessionStorage,
 	type SessionStorage,
 	type SessionStorageWriter,
+	type WriteTextAtomicOptions,
 } from "@oh-my-pi/pi-coding-agent/session/session-storage";
 import type { SessionTitleUpdate } from "@oh-my-pi/pi-coding-agent/session/session-title-slot";

@@ -104,8 +105,8 @@ class CloseHoldingStorage implements SessionStorage {
 	writeText(p: string, content: string): Promise<void> {
 		return this.#inner.writeText(p, content);
 	}
-	writeTextAtomic(p: string, content: string): Promise<void> {
-		return this.#inner.writeTextAtomic(p, content);
+	writeTextAtomic(p: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
+		return this.#inner.writeTextAtomic(p, content, options);
 	}
 	rename(p: string, nextPath: string): Promise<void> {
 		return this.#inner.rename(p, nextPath);
@@ -116,6 +117,9 @@ class CloseHoldingStorage implements SessionStorage {
 	deleteSessionWithArtifacts(sessionPath: string): Promise<void> {
 		return this.#inner.deleteSessionWithArtifacts(sessionPath);
 	}
+	drain(): Promise<void> {
+		return this.#inner.drain();
+	}
 }

 /** Drive microtasks while releasing every parked close until `promise` settles. */
@@ -16,14 +16,15 @@ class FsCodeError extends Error {
 }

 // The atomic-write + EPERM `.bak` move-aside/rollback dance lives in
-// FileSessionStorage.writeTextAtomic, so these tests must drive a real
-// file-backed storage (with a temp dir) and override `rename` to simulate the
-// Windows EPERM-on-replace failure.
+// FileSessionStorage.writeTextAtomic, which calls `renameSync` for the
+// guard-then-publish step so a concurrent synchronous rewrite cannot be
+// overwritten between the guard and the rename. These tests inject the
+// Windows-style EPERM at the sync layer used by the atomic path.
 class RenameEpermOnceStorage extends FileSessionStorage {
 	failNextSessionReplace = false;
-	backupCleanupPath: string | undefined;
+	backupPath: string | undefined;

-	async rename(source: string, target: string): Promise<void> {
+	override renameSync(source: string, target: string): void {
 		if (
 			this.failNextSessionReplace &&
 			source.includes(".tmp") &&
@@ -33,14 +34,10 @@ class RenameEpermOnceStorage extends FileSessionStorage {
 			this.failNextSessionReplace = false;
 			throw new FsCodeError("EPERM", `EPERM: operation not permitted, rename '${source}' -> '${target}'`);
 		}
-		return super.rename(source, target);
-	}
-
-	async unlink(target: string): Promise<void> {
-		if (target.endsWith(".bak")) {
-			this.backupCleanupPath = target;
+		if (source.endsWith(".jsonl") && target.endsWith(".bak")) {
+			this.backupPath = target;
 		}
-		return super.unlink(target);
+		super.renameSync(source, target);
 	}
 }

@@ -71,7 +68,7 @@ describe("SessionManager rewrite EPERM replacement fallback", () => {

 		const rewritten = await storage.readText(sessionFile);
 		expect(rewritten).toContain('"title":"renamed session"');
-		const backupPath = storage.backupCleanupPath;
+		const backupPath = storage.backupPath;
 		if (!backupPath) throw new Error("Expected EPERM fallback to create a rollback backup");
 		expect(storage.existsSync(backupPath)).toBe(false);

@@ -96,10 +93,10 @@ describe("SessionManager rewrite EPERM rollback failure", () => {
 			failureMode = false;
 			tempRenameAttempts = 0;

-			async rename(source: string, target: string): Promise<void> {
-				if (!this.failureMode) return super.rename(source, target);
+			override renameSync(source: string, target: string): void {
+				if (!this.failureMode) return super.renameSync(source, target);
 				// Every temp -> target rename fails with EPERM (both the upstream attempt in
-				// writeTextAtomic and the retry inside #replaceSessionFileAfterEperm).
+				// writeTextAtomic and the retry inside #replaceSessionFileAfterEpermSync).
 				if (source.includes(".tmp") && target.endsWith(".jsonl")) {
 					this.tempRenameAttempts++;
 					const tag = this.tempRenameAttempts === 1 ? "original" : "retry";
@@ -109,7 +106,7 @@ describe("SessionManager rewrite EPERM rollback failure", () => {
 				if (source.endsWith(".bak") && target.endsWith(".jsonl")) {
 					throw new FsCodeError("EIO", `EIO rollback: rename '${source}' -> '${target}'`);
 				}
-				return super.rename(source, target);
+				super.renameSync(source, target);
 			}
 		}

@@ -139,6 +136,93 @@ describe("SessionManager rewrite EPERM rollback failure", () => {
 	});
 });

+describe("FileSessionStorage.writeTextAtomic commitGuard cleanup", () => {
+	let sessionDir: string;
+
+	beforeEach(async () => {
+		sessionDir = await fsp.mkdtemp(path.join(os.tmpdir(), "omp-guard-cleanup-"));
+	});
+
+	afterEach(async () => {
+		await fsp.rm(sessionDir, { recursive: true, force: true });
+	});
+
+	async function listTempFiles(): Promise<string[]> {
+		const names = await fsp.readdir(sessionDir);
+		return names.filter(name => name.endsWith(".tmp"));
+	}
+
+	it("discards the staged temp when commitGuard rejects on the direct rename path", async () => {
+		const storage = new FileSessionStorage();
+		const target = path.join(sessionDir, "session.jsonl");
+		await storage.writeTextAtomic(target, "existing\n", { commitGuard: () => false });
+		expect(await listTempFiles()).toEqual([]);
+		expect(await Bun.file(target).exists()).toBe(false);
+	});
+
+	it("discards the staged temp when the EPERM move-aside fallback's commitGuard rejects", async () => {
+		let epermAttempted = false;
+		let guardCalls = 0;
+		class EpermThenGuardStorage extends FileSessionStorage {
+			override renameSync(source: string, targetPath: string): void {
+				if (source.includes(".tmp") && targetPath.endsWith(".jsonl") && !epermAttempted) {
+					epermAttempted = true;
+					throw new FsCodeError("EPERM", `EPERM: operation not permitted, rename '${source}' -> '${targetPath}'`);
+				}
+				super.renameSync(source, targetPath);
+			}
+		}
+		const storage = new EpermThenGuardStorage();
+		const target = path.join(sessionDir, "session.jsonl");
+		await fsp.writeFile(target, "seed\n");
+
+		await storage.writeTextAtomic(target, "next\n", {
+			commitGuard: () => {
+				guardCalls += 1;
+				// First call (before primary rename): pass so we hit EPERM.
+				// Second call (inside EPERM fallback, after move-aside): reject.
+				return guardCalls === 1;
+			},
+		});
+
+		expect(epermAttempted).toBe(true);
+		expect(guardCalls).toBe(2);
+		expect(await listTempFiles()).toEqual([]);
+		// Backup was restored, so target still holds the seed content.
+		expect(await Bun.file(target).text()).toBe("seed\n");
+		const backups = (await fsp.readdir(sessionDir)).filter(name => name.endsWith(".bak"));
+		expect(backups).toEqual([]);
+	});
+
+	it("discards the staged temp when the ENOENT move-aside branch's commitGuard rejects", async () => {
+		let epermAttempted = false;
+		let guardCalls = 0;
+		class EpermMissingTargetStorage extends FileSessionStorage {
+			override renameSync(source: string, targetPath: string): void {
+				if (source.includes(".tmp") && targetPath.endsWith(".jsonl") && !epermAttempted) {
+					epermAttempted = true;
+					throw new FsCodeError("EPERM", `EPERM: operation not permitted, rename '${source}' -> '${targetPath}'`);
+				}
+				super.renameSync(source, targetPath);
+			}
+		}
+		const storage = new EpermMissingTargetStorage();
+		const target = path.join(sessionDir, "session.jsonl");
+		// Target does not exist, so the move-aside step raises ENOENT.
+		await storage.writeTextAtomic(target, "next\n", {
+			commitGuard: () => {
+				guardCalls += 1;
+				return guardCalls === 1;
+			},
+		});
+
+		expect(epermAttempted).toBe(true);
+		expect(guardCalls).toBe(2);
+		expect(await listTempFiles()).toEqual([]);
+		expect(await Bun.file(target).exists()).toBe(false);
+	});
+});
+
 describe("recoverOrphanedBackups", () => {
 	it("promotes an orphaned <basename>.jsonl.<snowflake>.bak back to the primary path when the primary is missing", async () => {
 		const storage = new MemorySessionStorage();
@@ -10,7 +10,7 @@ import {
 } from "@oh-my-pi/pi-coding-agent/session/session-entries";
 import { loadEntriesFromFile } from "@oh-my-pi/pi-coding-agent/session/session-loader";
 import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager";
-import { FileSessionStorage } from "@oh-my-pi/pi-coding-agent/session/session-storage";
+import { FileSessionStorage, type WriteTextAtomicOptions } from "@oh-my-pi/pi-coding-agent/session/session-storage";
 import type { SessionTitleUpdate } from "@oh-my-pi/pi-coding-agent/session/session-title-slot";
 import { getConfigRootDir, removeSyncWithRetries, setAgentDir } from "@oh-my-pi/pi-utils";

@@ -31,9 +31,9 @@ class CountingTitleSlotStorage extends FileSessionStorage {
 		super.writeTextSync(filePath, content);
 	}

-	override async writeTextAtomic(filePath: string, content: string): Promise<void> {
+	override async writeTextAtomic(filePath: string, content: string, options?: WriteTextAtomicOptions): Promise<void> {
 		this.atomicWrites++;
-		await super.writeTextAtomic(filePath, content);
+		await super.writeTextAtomic(filePath, content, options);
 	}

 	resetCounts(): void {
@@ -231,3 +231,83 @@ describe("IndexedSessionStorage.updateSessionTitle", () => {
 		expect(JSON.parse(slotLine)).toMatchObject({ type: "title", title: "Second", source: "user", updatedAt: "t2" });
 	});
 });
+
+class PausableWriteFullBackend implements SessionStorageBackend {
+	readonly writeFullCalls: Array<{ content: string; mtimeMs: number }> = [];
+	readonly firstWriteStarted = Promise.withResolvers<void>();
+	readonly firstWriteRelease = Promise.withResolvers<void>();
+	#firstReleased = false;
+
+	init(): Promise<void> {
+		return Promise.resolve();
+	}
+	loadIndex(): Promise<Iterable<SessionStorageIndexEntry>> {
+		return Promise.resolve([]);
+	}
+	readFull(): Promise<string | null> {
+		return Promise.resolve(null);
+	}
+	readSlices(): Promise<[string, string]> {
+		return Promise.resolve(["", ""]);
+	}
+	async writeFull(_path: string, content: string, mtimeMs: number): Promise<void> {
+		if (!this.#firstReleased) {
+			this.#firstReleased = true;
+			this.firstWriteStarted.resolve();
+			await this.firstWriteRelease.promise;
+		}
+		this.writeFullCalls.push({ content, mtimeMs });
+	}
+	append(): Promise<void> {
+		return Promise.resolve();
+	}
+	updateSessionTitle(): Promise<void> {
+		return Promise.resolve();
+	}
+	truncate(): Promise<void> {
+		return Promise.resolve();
+	}
+	remove(): Promise<void> {
+		return Promise.resolve();
+	}
+	move(): Promise<void> {
+		return Promise.resolve();
+	}
+}
+
+describe("IndexedSessionStorage.writeTextAtomic commitGuard", () => {
+	it("aborts before touching the backend when the guard rejects up front", async () => {
+		const backend = new PausableWriteFullBackend();
+		const storage = new IndexedSessionStorage(backend);
+		await storage.initialize();
+
+		await storage.writeTextAtomic("/sessions/s.jsonl", "stale", { commitGuard: () => false });
+		expect(backend.writeFullCalls).toEqual([]);
+		expect(storage.existsSync("/sessions/s.jsonl")).toBe(false);
+	});
+
+	it("re-checks the guard inside the enqueued task so a concurrent write cannot be overwritten", async () => {
+		const backend = new PausableWriteFullBackend();
+		const storage = new IndexedSessionStorage(backend);
+		await storage.initialize();
+
+		// First write parks the backend inside writeFull, holding the per-path
+		// tail. The second write awaits behind it. When the first releases,
+		// the second's awaitPath resumes — but by then the guard has flipped
+		// (simulated flushSync epoch bump), and the backend MUST NOT see the
+		// stale second body.
+		const first = storage.writeTextAtomic("/sessions/s.jsonl", "seed", {});
+		let epochBumped = false;
+		const second = storage.writeTextAtomic("/sessions/s.jsonl", "stale", {
+			commitGuard: () => !epochBumped,
+		});
+
+		await backend.firstWriteStarted.promise;
+		epochBumped = true;
+		backend.firstWriteRelease.resolve();
+		await first;
+		await second;
+
+		expect(backend.writeFullCalls.map(call => call.content)).toEqual(["seed"]);
+	});
+});