From 69db80fbe78168aa95a14a726ad2b3cb57080931 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 18:36:56 +0000 Subject: [PATCH 1/8] fix(session): preserved tail after atomic compaction rewrites Fenced synchronous session appends while an atomic full-file replacement is active so Windows EPERM fallback cannot detach the append writer from the current JSONL path. Added a deterministic storage fake regression covering superseded compaction rewrites, title changes, session-exit diagnostics, resume, and post-rewrite tool/assistant tail persistence. Fixes #4338 --- packages/coding-agent/CHANGELOG.md | 4 + .../src/session/session-manager.ts | 35 +++- ...ession-manager-atomic-rewrite-race.test.ts | 173 ++++++++++++++++++ 3 files changed, 206 insertions(+), 6 deletions(-) create mode 100644 packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index bb87143c3..36ffec862 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -2,6 +2,10 @@ ## [Unreleased] +### Fixed + +- Fixed Windows session tail loss after atomic compaction rewrites by fencing append writers during full-file replacement so post-compaction prompts, tool results, title changes, and exit diagnostics persist to the current JSONL path ([#4338](https://github.com/can1357/oh-my-pi/issues/4338)). + ## [16.3.2] - 2026-07-02 ### Breaking Changes diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index 7637e0448..dca050eb7 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -383,6 +383,10 @@ export class SessionManager { #diskFailureLogged = false; /** Bumped on every sync rewrite / chain reset so stale queued tasks become no-ops. */ #diskEpoch = 0; + /** True while an atomic full-file replacement can detach append handles opened to the old path. */ + #atomicRewriteActive = false; + /** Set by synchronous appends that land while an atomic replacement is active. */ + #atomicRewriteDirty = false; #artifactManager: ArtifactManager | null = null; #artifactManagerSessionFile: string | null = null; @@ -555,8 +559,10 @@ 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. Appends that + * arrive while the storage backend is replacing the path are fenced out of the + * append hot path and force the rewrite task to loop with a fresh body before + * it resolves. */ async #rewriteAtomically(): Promise { if (!this.#persist || !this.#sessionFile) return; @@ -564,10 +570,18 @@ export class SessionManager { const epoch = this.#diskEpoch; await this.#scheduleDiskWork( async () => { - await this.#closeWriterHandle(); - const sessionFile = this.#sessionFile; - if (!sessionFile) return; - await this.#storage.writeTextAtomic(sessionFile, this.#fileBody()); + do { + this.#atomicRewriteDirty = false; + await this.#closeWriterHandle(); + const sessionFile = this.#sessionFile; + if (!sessionFile) return; + this.#atomicRewriteActive = true; + try { + await this.#storage.writeTextAtomic(sessionFile, this.#fileBody()); + } finally { + this.#atomicRewriteActive = false; + } + } while (this.#atomicRewriteDirty); this.#fileIsCurrent = true; this.#rewriteRequired = false; this.#hasTitleSlot = true; @@ -588,6 +602,15 @@ 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. + if (this.#atomicRewriteActive) { + 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) { diff --git a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts new file mode 100644 index 000000000..367503732 --- /dev/null +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -0,0 +1,173 @@ +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 } from "@oh-my-pi/pi-coding-agent/session/session-storage"; + +interface DetachableWriter extends SessionStorageWriter { + detach(): void; +} + +class DetachingRewriteStorage extends MemorySessionStorage { + readonly detachedLines: string[] = []; + readonly rewriteStarted = Promise.withResolvers(); + readonly allowRewrite = Promise.withResolvers(); + readonly #writers = new Set(); + + 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 { + if (detached) { + detachedLines.push(line); + return; + } + await inner.append(line); + }, + async flush(): Promise { + await inner.flush(); + }, + isOpen(): boolean { + const open = inner.isOpen(); + return open; + }, + async close(): Promise { + 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; + } + + async writeTextAtomic(path: string, content: string): Promise { + this.rewriteStarted.resolve(); + await this.allowRewrite.promise; + for (const writer of this.#writers) writer.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"); + }); +}); From 326a3406a8fdfc1e9665944d6a6121ad5f5049d4 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 19:00:27 +0000 Subject: [PATCH 2/8] fix(session): guarded atomic rewrite against flushSync overwrites SessionStorage.writeTextAtomic now accepts a commitGuard the backend calls synchronously immediately before publishing the staged body. FileSessionStorage performs the guard check and rename in the same tick via fs.renameSync (both on the direct path and the EPERM move-aside fallback), so a concurrent #rewriteSynchronously (flushSync -> Ctrl+C / session exit) that bumps the disk epoch cannot be overwritten by the stale body serialized before it ran. MemorySessionStorage and IndexedSessionStorage honor the same guard. SessionManager.#rewriteAtomically threads a guard that returns false when the disk epoch changes, and re-checks the epoch after every writeTextAtomic before touching #fileIsCurrent / #rewriteRequired. #persistTitleChangeEntry's atomic fallback wires the same guard. Added a regression that pauses the fake storage's writeTextAtomic mid-flight, appends a session_exit custom entry (which the fence records in memory), calls flushSync, releases the paused rewrite, and asserts the exit record is still on the JSONL path and the atomic publish was rejected by the guard. Fixes #4338 --- packages/coding-agent/CHANGELOG.md | 2 +- .../src/session/indexed-session-storage.ts | 7 +- .../src/session/session-manager.ts | 21 +++- .../src/session/session-storage.ts | 118 +++++++++++++----- ...ession-manager-atomic-rewrite-race.test.ts | 76 ++++++++++- .../test/session-manager-close-race.test.ts | 5 +- .../rewrite-rename-eperm.test.ts | 31 +++-- .../title-source-persistence.test.ts | 6 +- 8 files changed, 206 insertions(+), 60 deletions(-) diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 36ffec862..4c4fbae67 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -4,7 +4,7 @@ ### Fixed -- Fixed Windows session tail loss after atomic compaction rewrites by fencing append writers during full-file replacement so post-compaction prompts, tool results, title changes, and exit diagnostics persist to the current JSONL path ([#4338](https://github.com/can1357/oh-my-pi/issues/4338)). +- 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 diff --git a/packages/coding-agent/src/session/indexed-session-storage.ts b/packages/coding-agent/src/session/indexed-session-storage.ts index 4ae9fb9df..cab099d87 100644 --- a/packages/coding-agent/src/session/indexed-session-storage.ts +++ b/packages/coding-agent/src/session/indexed-session-storage.ts @@ -234,7 +234,12 @@ export class IndexedSessionStorage implements SessionStorage { } } - writeTextAtomic(path: string, content: string): Promise { + writeTextAtomic( + path: string, + content: string, + options?: import("./session-storage").WriteTextAtomicOptions, + ): Promise { + if (options?.commitGuard && !options.commitGuard()) return Promise.resolve(); return this.writeText(path, content); } diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index dca050eb7..02fa84ec9 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -562,12 +562,14 @@ export class SessionManager { * disk chain. The body is serialized after the writer is closed. Appends that * arrive while the storage backend is replacing the path are fenced out of the * append hot path and force the rewrite task to loop with a fresh body before - * it resolves. + * it resolves. Passes a `commitGuard` to the storage layer so a synchronous + * rewrite (`flushSync` → `#rewriteSynchronously`) that supersedes this task + * cannot be overwritten by the stale body serialized before it ran. */ async #rewriteAtomically(): Promise { if (!this.#persist || !this.#sessionFile) return; - const epoch = this.#diskEpoch; + const startEpoch = this.#diskEpoch; await this.#scheduleDiskWork( async () => { do { @@ -575,18 +577,22 @@ export class SessionManager { await this.#closeWriterHandle(); const sessionFile = this.#sessionFile; if (!sessionFile) return; + if (this.#diskEpoch !== startEpoch) return; this.#atomicRewriteActive = true; try { - await this.#storage.writeTextAtomic(sessionFile, this.#fileBody()); + await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { + commitGuard: () => this.#diskEpoch === startEpoch, + }); } finally { this.#atomicRewriteActive = false; } + if (this.#diskEpoch !== startEpoch) return; } while (this.#atomicRewriteDirty); this.#fileIsCurrent = true; this.#rewriteRequired = false; this.#hasTitleSlot = true; }, - { epoch }, + { epoch: startEpoch }, ); } @@ -660,10 +666,13 @@ 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()); + await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { + commitGuard: () => this.#diskEpoch === epoch, + }); + if (this.#diskEpoch !== epoch) return; this.#clearDiskError(); this.#fileIsCurrent = true; this.#rewriteRequired = false; diff --git a/packages/coding-agent/src/session/session-storage.ts b/packages/coding-agent/src/session/session-storage.ts index 5afef3ea4..f7e2bcc60 100644 --- a/packages/coding-agent/src/session/session-storage.ts +++ b/packages/coding-agent/src/session/session-storage.ts @@ -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,7 +60,7 @@ 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; - writeTextAtomic(path: string, content: string): Promise; + writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise; rename(path: string, nextPath: string): Promise; unlink(path: string): Promise; deleteSessionWithArtifacts(sessionPath: string): Promise; @@ -229,53 +241,102 @@ export class FileSessionStorage implements SessionStorage { await Bun.write(path, content, { createPath: true }); } - async writeTextAtomic(fpath: string, content: string): Promise { + async writeTextAtomic(fpath: string, content: string, options?: WriteTextAtomicOptions): Promise { 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 { + /** + * 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()) 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. + try { + this.renameSync(backupPath, targetPath); + } catch (restoreErr) { + logger.warn("Failed to restore backup after commitGuard rejection", { + sessionFile: targetPath, + backupPath, + error: toError(restoreErr).message, + }); + } + 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 +349,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", { @@ -621,7 +682,8 @@ export class MemorySessionStorage implements SessionStorage { return Promise.resolve(); } - writeTextAtomic(path: string, content: string): Promise { + writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise { + if (options?.commitGuard && !options.commitGuard()) return Promise.resolve(); this.writeTextSync(path, content); return Promise.resolve(); } diff --git a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts index 367503732..fd96ffabb 100644 --- a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -1,7 +1,11 @@ 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 } from "@oh-my-pi/pi-coding-agent/session/session-storage"; +import { + MemorySessionStorage, + type SessionStorageWriter, + type WriteTextAtomicOptions, +} from "@oh-my-pi/pi-coding-agent/session/session-storage"; interface DetachableWriter extends SessionStorageWriter { detach(): void; @@ -11,6 +15,8 @@ class DetachingRewriteStorage extends MemorySessionStorage { readonly detachedLines: string[] = []; readonly rewriteStarted = Promise.withResolvers(); readonly allowRewrite = Promise.withResolvers(); + pausedRewrites = 0; + guardRejections = 0; readonly #writers = new Set(); openWriter(path: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }): SessionStorageWriter { @@ -50,9 +56,14 @@ class DetachingRewriteStorage extends MemorySessionStorage { return writer; } - async writeTextAtomic(path: string, content: string): Promise { + override async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise { + 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); } @@ -170,4 +181,65 @@ describe("SessionManager atomic rewrite race", () => { ).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([]); + }); }); diff --git a/packages/coding-agent/test/session-manager-close-race.test.ts b/packages/coding-agent/test/session-manager-close-race.test.ts index 474425122..b05af58c9 100644 --- a/packages/coding-agent/test/session-manager-close-race.test.ts +++ b/packages/coding-agent/test/session-manager-close-race.test.ts @@ -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 { return this.#inner.writeText(p, content); } - writeTextAtomic(p: string, content: string): Promise { - return this.#inner.writeTextAtomic(p, content); + writeTextAtomic(p: string, content: string, options?: WriteTextAtomicOptions): Promise { + return this.#inner.writeTextAtomic(p, content, options); } rename(p: string, nextPath: string): Promise { return this.#inner.rename(p, nextPath); diff --git a/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts b/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts index 4934806ac..5a580433f 100644 --- a/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts +++ b/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts @@ -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 { + 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 { - 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 { - 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); } } diff --git a/packages/coding-agent/test/session-manager/title-source-persistence.test.ts b/packages/coding-agent/test/session-manager/title-source-persistence.test.ts index 85cae4d49..3d189401d 100644 --- a/packages/coding-agent/test/session-manager/title-source-persistence.test.ts +++ b/packages/coding-agent/test/session-manager/title-source-persistence.test.ts @@ -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 { + override async writeTextAtomic(filePath: string, content: string, options?: WriteTextAtomicOptions): Promise { this.atomicWrites++; - await super.writeTextAtomic(filePath, content); + await super.writeTextAtomic(filePath, content, options); } resetCounts(): void { From 24c6b3a9f92e93bf76cc863635d51b0c11b7bb41 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 19:07:36 +0000 Subject: [PATCH 3/8] fix(session): fenced writer close-yield inside atomic rewrite SessionManager.#rewriteAtomically now enables #atomicRewriteActive before #closeWriterHandle() and keeps it set until the rewrite task exits, so a sync append landing in the close-yield window is fenced and cannot open a fresh writer that the pending writeTextAtomic would then detach from the current JSONL path. Same pattern applied to the #persistTitleChangeEntry atomic fallback. Added a regression that pauses the fake storage's writer.close() gate, appends a message and a custom entry during the pause, and asserts (1) no new writer opens (writerOpens counter unchanged) and (2) the fenced entries land on the current JSONL path after the rewrite completes. Fixes #4338 --- .../src/session/session-manager.ts | 58 ++++---- ...ession-manager-atomic-rewrite-race.test.ts | 128 ++++++++++++++++++ 2 files changed, 160 insertions(+), 26 deletions(-) diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index 02fa84ec9..4886c25dc 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -559,12 +559,13 @@ export class SessionManager { /** * Rewrite the whole file atomically (temp-write + rename, EPERM-safe) on the - * disk chain. The body is serialized after the writer is closed. Appends that - * arrive while the storage backend is replacing the path are fenced out of the - * append hot path and force the rewrite task to loop with a fresh body before - * it resolves. Passes a `commitGuard` to the storage layer so a synchronous - * rewrite (`flushSync` → `#rewriteSynchronously`) that supersedes this task - * cannot be overwritten by the stale body serialized before it ran. + * 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 { if (!this.#persist || !this.#sessionFile) return; @@ -572,25 +573,25 @@ export class SessionManager { const startEpoch = this.#diskEpoch; await this.#scheduleDiskWork( async () => { - do { - this.#atomicRewriteDirty = false; - await this.#closeWriterHandle(); - const sessionFile = this.#sessionFile; - if (!sessionFile) return; - if (this.#diskEpoch !== startEpoch) return; - this.#atomicRewriteActive = true; - try { + this.#atomicRewriteActive = true; + try { + do { + this.#atomicRewriteDirty = false; + await this.#closeWriterHandle(); + const sessionFile = this.#sessionFile; + if (!sessionFile) return; + if (this.#diskEpoch !== startEpoch) return; await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { commitGuard: () => this.#diskEpoch === startEpoch, }); - } finally { - this.#atomicRewriteActive = false; - } - if (this.#diskEpoch !== startEpoch) return; - } while (this.#atomicRewriteDirty); - this.#fileIsCurrent = true; - this.#rewriteRequired = false; - this.#hasTitleSlot = true; + if (this.#diskEpoch !== startEpoch) return; + } while (this.#atomicRewriteDirty); + this.#fileIsCurrent = true; + this.#rewriteRequired = false; + this.#hasTitleSlot = true; + } finally { + this.#atomicRewriteActive = false; + } }, { epoch: startEpoch }, ); @@ -668,10 +669,15 @@ export class SessionManager { await this.#storage.updateSessionTitle(sessionFile, update); if (this.#diskEpoch === epoch) this.#fileIsCurrent = true; } catch { - await this.#closeWriterHandle(); - await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { - commitGuard: () => this.#diskEpoch === epoch, - }); + this.#atomicRewriteActive = true; + try { + await this.#closeWriterHandle(); + await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { + commitGuard: () => this.#diskEpoch === epoch, + }); + } finally { + this.#atomicRewriteActive = false; + } if (this.#diskEpoch !== epoch) return; this.#clearDiskError(); this.#fileIsCurrent = true; diff --git a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts index fd96ffabb..b7688d3fc 100644 --- a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -69,6 +69,73 @@ class DetachingRewriteStorage extends MemorySessionStorage { } } +class CloseGatedRewriteStorage extends MemorySessionStorage { + readonly closeStarted = Promise.withResolvers(); + readonly allowClose = Promise.withResolvers(); + readonly writeStarted = Promise.withResolvers(); + readonly allowWrite = Promise.withResolvers(); + readonly detachedLines: string[] = []; + writerOpens = 0; + guardRejections = 0; + readonly #detachables = new Set(); + + 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 { + if (detached) { + detachedLines.push(line); + return; + } + await inner.append(line); + }, + async flush(): Promise { + await inner.flush(); + }, + isOpen(): boolean { + return inner.isOpen(); + }, + async close(): Promise { + 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 { + 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(); @@ -243,3 +310,64 @@ describe("SessionManager atomic rewrite race", () => { 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([]); + }); +}); From 23d9f7c89828fa0ad4c6a9a297f23527ed73f8d4 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 19:11:25 +0000 Subject: [PATCH 4/8] fix(session): discarded temp on EPERM guard-reject branches FileSessionStorage.#replaceSessionFileAfterEpermSync now unlinks the staged temp file when commitGuard returns false in both fallback branches: the ENOENT-vanished-target path and the post-move-aside path (where the moved-aside backup is also restored). Honors the writeTextAtomic contract that a guard-rejected stage is discarded. Regressions cover all three guard-reject exits: the direct rename pre-check, the ENOENT branch inside the EPERM fallback, and the move-aside branch that also restores the backup. Each asserts no orphan .tmp remains in the session dir. Fixes #4338 --- .../src/session/session-storage.ts | 9 +- .../rewrite-rename-eperm.test.ts | 87 +++++++++++++++++++ 2 files changed, 94 insertions(+), 2 deletions(-) diff --git a/packages/coding-agent/src/session/session-storage.ts b/packages/coding-agent/src/session/session-storage.ts index f7e2bcc60..485aa28f2 100644 --- a/packages/coding-agent/src/session/session-storage.ts +++ b/packages/coding-agent/src/session/session-storage.ts @@ -311,7 +311,10 @@ export class FileSessionStorage implements SessionStorage { this.renameSync(targetPath, backupPath); } catch (moveAsideError) { if (isEnoent(moveAsideError)) { - if (commitGuard && !commitGuard()) return; + if (commitGuard && !commitGuard()) { + this.#discardTemp(tempPath, targetPath); + return; + } this.renameSync(tempPath, targetPath); return; } @@ -320,7 +323,8 @@ export class FileSessionStorage implements SessionStorage { 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. + // 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) { @@ -330,6 +334,7 @@ export class FileSessionStorage implements SessionStorage { error: toError(restoreErr).message, }); } + this.#discardTemp(tempPath, targetPath); return; } try { diff --git a/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts b/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts index 5a580433f..e1c670616 100644 --- a/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts +++ b/packages/coding-agent/test/session-manager/rewrite-rename-eperm.test.ts @@ -136,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 { + 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 .jsonl..bak back to the primary path when the primary is missing", async () => { const storage = new MemorySessionStorage(); From f614ec1537604d404e30018f16466691218b95d5 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 19:17:45 +0000 Subject: [PATCH 5/8] fix(session): honored commitGuard at indexed publish time MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit IndexedSessionStorage.writeTextAtomic no longer delegates directly to writeText, which yielded on #awaitPath between the guard check and the backend publish. The new impl consults the guard three times — up front, again after #awaitPath resolves, and finally inside the enqueued task immediately before #backend.writeFull — so a flushSync that bumps #diskEpoch while the atomic rewrite is suspended cannot land stale content on Redis/SQL backends. When the enqueue-time guard rejects, the optimistic index update is restored only when nothing has advanced it past our mtime, so a concurrent writer's state is preserved. Added a PausableWriteFullBackend regression: the first writeTextAtomic parks inside backend.writeFull holding the per-path tail; the second queues with a guard that flips after the first is released. The backend records only the first content, confirming the guard is honored at publish time. Fixes #4338 --- .../src/session/indexed-session-storage.ts | 48 +++++++++-- .../coding-agent/test/session-storage.test.ts | 80 +++++++++++++++++++ 2 files changed, 120 insertions(+), 8 deletions(-) diff --git a/packages/coding-agent/src/session/indexed-session-storage.ts b/packages/coding-agent/src/session/indexed-session-storage.ts index cab099d87..b620765b8 100644 --- a/packages/coding-agent/src/session/indexed-session-storage.ts +++ b/packages/coding-agent/src/session/indexed-session-storage.ts @@ -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,13 +239,40 @@ export class IndexedSessionStorage implements SessionStorage { } } - writeTextAtomic( - path: string, - content: string, - options?: import("./session-storage").WriteTextAtomicOptions, - ): Promise { - if (options?.commitGuard && !options.commitGuard()) return Promise.resolve(); - return this.writeText(path, content); + async writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise { + 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 { diff --git a/packages/coding-agent/test/session-storage.test.ts b/packages/coding-agent/test/session-storage.test.ts index dd086e2e2..0cb8221f2 100644 --- a/packages/coding-agent/test/session-storage.test.ts +++ b/packages/coding-agent/test/session-storage.test.ts @@ -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(); + readonly firstWriteRelease = Promise.withResolvers(); + #firstReleased = false; + + init(): Promise { + return Promise.resolve(); + } + loadIndex(): Promise> { + return Promise.resolve([]); + } + readFull(): Promise { + return Promise.resolve(null); + } + readSlices(): Promise<[string, string]> { + return Promise.resolve(["", ""]); + } + async writeFull(_path: string, content: string, mtimeMs: number): Promise { + if (!this.#firstReleased) { + this.#firstReleased = true; + this.firstWriteStarted.resolve(); + await this.firstWriteRelease.promise; + } + this.writeFullCalls.push({ content, mtimeMs }); + } + append(): Promise { + return Promise.resolve(); + } + updateSessionTitle(): Promise { + return Promise.resolve(); + } + truncate(): Promise { + return Promise.resolve(); + } + remove(): Promise { + return Promise.resolve(); + } + move(): Promise { + 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"]); + }); +}); From babb731cf51d8572ec2271194753e79b735ec781 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 20:32:27 +0000 Subject: [PATCH 6/8] fix(session): looped title fallback + drained backend on close MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionManager.#persistTitleChangeEntry's catch fallback previously did a single-shot atomic rewrite: any prompt/tool appended while it awaited was fenced with #atomicRewriteDirty=true but never re-serialized. Extracted the fenced-rewrite do-while loop into #runFencedAtomicRewrite and used it from both #rewriteAtomically and #persistTitleChangeEntry, so fenced entries during either path are captured before the task resolves. Added SessionStorage.drain(): for FileSessionStorage and MemorySessionStorage it is a no-op; IndexedSessionStorage already had one and now conforms to the interface. SessionManager.flush() and close() await it so a graceful shutdown does not exit while a fire-and-forget writeTextSync publish (queued by flushSync on an indexed backend) is still on the wire — reducing the residual publish-window race for Redis/SQL where the backend cannot be aborted mid-flight. Regression covers the title fallback loop: TitleFallbackPausingStorage forces updateSessionTitle to throw, pauses the fallback's writeTextAtomic, appends a message and a custom entry during the pause, and asserts (a) both fenced entries land on the current JSONL, (b) the final title is applied, and (c) writeTextAtomicCalls >= 2 proving the loop iterated. Fixes #4338 --- .../src/session/session-manager.ts | 64 ++++++++------ .../src/session/session-storage.ts | 19 ++++ ...ession-manager-atomic-rewrite-race.test.ts | 87 +++++++++++++++++++ .../test/session-manager-close-race.test.ts | 3 + 4 files changed, 148 insertions(+), 25 deletions(-) diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index 4886c25dc..d9e1be313 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -573,30 +573,45 @@ export class SessionManager { const startEpoch = this.#diskEpoch; await this.#scheduleDiskWork( async () => { - this.#atomicRewriteActive = true; - try { - do { - this.#atomicRewriteDirty = false; - await this.#closeWriterHandle(); - const sessionFile = this.#sessionFile; - if (!sessionFile) return; - if (this.#diskEpoch !== startEpoch) return; - await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { - commitGuard: () => this.#diskEpoch === startEpoch, - }); - if (this.#diskEpoch !== startEpoch) return; - } while (this.#atomicRewriteDirty); + if (await this.#runFencedAtomicRewrite(startEpoch)) { this.#fileIsCurrent = true; this.#rewriteRequired = false; this.#hasTitleSlot = true; - } finally { - this.#atomicRewriteActive = false; } }, { 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 { + this.#atomicRewriteActive = true; + try { + do { + this.#atomicRewriteDirty = false; + await this.#closeWriterHandle(); + const sessionFile = this.#sessionFile; + 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 { + this.#atomicRewriteActive = false; + } + } + #appendToSessionFile(entry: SessionEntry): void { if (!this.#persist || !this.#sessionFile) return; if (this.#diskFailure) throw this.#diskFailure; @@ -669,16 +684,7 @@ export class SessionManager { await this.#storage.updateSessionTitle(sessionFile, update); if (this.#diskEpoch === epoch) this.#fileIsCurrent = true; } catch { - this.#atomicRewriteActive = true; - try { - await this.#closeWriterHandle(); - await this.#storage.writeTextAtomic(sessionFile, this.#fileBody(), { - commitGuard: () => this.#diskEpoch === epoch, - }); - } finally { - this.#atomicRewriteActive = false; - } - if (this.#diskEpoch !== epoch) return; + if (!(await this.#runFencedAtomicRewrite(epoch))) return; this.#clearDiskError(); this.#fileIsCurrent = true; this.#rewriteRequired = false; @@ -1087,6 +1093,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; } @@ -1116,6 +1126,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; } diff --git a/packages/coding-agent/src/session/session-storage.ts b/packages/coding-agent/src/session/session-storage.ts index 485aa28f2..a2db80cf2 100644 --- a/packages/coding-agent/src/session/session-storage.ts +++ b/packages/coding-agent/src/session/session-storage.ts @@ -65,6 +65,15 @@ export interface SessionStorage { unlink(path: string): Promise; deleteSessionWithArtifacts(sessionPath: string): Promise; 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; } // FinalizationRegistry to clean up leaked file descriptors @@ -378,6 +387,12 @@ export class FileSessionStorage implements SessionStorage { return fs.promises.unlink(path); } + drain(): Promise { + // 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); } @@ -709,6 +724,10 @@ export class MemorySessionStorage implements SessionStorage { return Promise.resolve(); } + drain(): Promise { + return Promise.resolve(); + } + openWriter(path: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }): SessionStorageWriter { return new MemorySessionStorageWriter(this, path, options); } diff --git a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts index b7688d3fc..20e71f37b 100644 --- a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -6,6 +6,7 @@ import { 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; @@ -371,3 +372,89 @@ describe("SessionManager atomic rewrite fence spans writer.close()", () => { expect(storage.detachedLines).toEqual([]); }); }); + +class TitleFallbackPausingStorage extends MemorySessionStorage { + readonly writeStarted = Promise.withResolvers(); + readonly allowWrite = Promise.withResolvers(); + writeTextAtomicCalls = 0; + failNextUpdateTitle = false; + + override async updateSessionTitle(path: string, update: SessionTitleUpdate): Promise { + 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 { + 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); + }); +}); diff --git a/packages/coding-agent/test/session-manager-close-race.test.ts b/packages/coding-agent/test/session-manager-close-race.test.ts index b05af58c9..69fbe05c6 100644 --- a/packages/coding-agent/test/session-manager-close-race.test.ts +++ b/packages/coding-agent/test/session-manager-close-race.test.ts @@ -117,6 +117,9 @@ class CloseHoldingStorage implements SessionStorage { deleteSessionWithArtifacts(sessionPath: string): Promise { return this.#inner.deleteSessionWithArtifacts(sessionPath); } + drain(): Promise { + return this.#inner.drain(); + } } /** Drive microtasks while releasing every parked close until `promise` settles. */ From 549b4c13a8d7f0c79d6801861617ffbdc82034a1 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 20:54:24 +0000 Subject: [PATCH 7/8] fix(session): relaxed fence once flushSync superseded the atomic Replaced the boolean #atomicRewriteActive flag with #atomicRewriteFenceEpoch: number | null. The fence branch in #appendToSessionFile now applies only while the pending atomic rewrite's epoch still matches #diskEpoch. Once flushSync -> #rewriteSynchronously bumps the epoch, the in-flight writeTextAtomic is guaranteed to abandon via its commitGuard, so subsequent sync appends can (and must) take the hot path against the freshly-published body instead of being stranded in memory when close() returns without another rewrite. New regression: pauses writeTextAtomic mid-flight, appends a fenced custom entry, calls flushSync (which captures it into the durable body), then appends a message + custom entry after the epoch bump. Reads the current JSONL BEFORE releasing the paused atomic and asserts both post-flushSync entries are already on disk; then releases the atomic (commitGuard rejects) and closes the session and asserts nothing is lost. Fixes #4338 --- .../src/session/session-manager.ts | 21 ++++-- ...ession-manager-atomic-rewrite-race.test.ts | 73 +++++++++++++++++++ 2 files changed, 89 insertions(+), 5 deletions(-) diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index d9e1be313..d7756430f 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -383,8 +383,15 @@ export class SessionManager { #diskFailureLogged = false; /** Bumped on every sync rewrite / chain reset so stale queued tasks become no-ops. */ #diskEpoch = 0; - /** True while an atomic full-file replacement can detach append handles opened to the old path. */ - #atomicRewriteActive = false; + /** + * 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; @@ -593,7 +600,7 @@ export class SessionManager { * their post-publish state updates. */ async #runFencedAtomicRewrite(epoch: number): Promise { - this.#atomicRewriteActive = true; + this.#atomicRewriteFenceEpoch = epoch; try { do { this.#atomicRewriteDirty = false; @@ -608,7 +615,7 @@ export class SessionManager { } while (this.#atomicRewriteDirty); return true; } finally { - this.#atomicRewriteActive = false; + this.#atomicRewriteFenceEpoch = null; } } @@ -627,7 +634,11 @@ export class SessionManager { // 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. - if (this.#atomicRewriteActive) { + // 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; diff --git a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts index 20e71f37b..ea9384795 100644 --- a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -458,3 +458,76 @@ describe("SessionManager title-change fallback fenced-append durability", () => 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([]); + }); +}); From 01ab7e26d77ee04c8c56d909311d782507970478 Mon Sep 17 00:00:00 2001 From: roboomp Date: Thu, 2 Jul 2026 21:20:53 +0000 Subject: [PATCH 8/8] fix(session): kept newer fence owner on stale rewrite unwind MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SessionManager.#runFencedAtomicRewrite's finally now only clears #atomicRewriteFenceEpoch when it still matches the unwinding task's epoch. When flushSync supersedes an in-flight rewrite (bumping #diskEpoch and resetting #diskTail), a fresh atomic task scheduled at the new epoch can take ownership of the fence before the stale rewrite finally settles; the previous unconditional clear stranded the newer rewrite's bookkeeping so subsequent sync appends took the hot writer path and were then detached by the newer publish. Regression: SequencedRewriteStorage pauses the first N writeTextAtomic calls on per-call gates. Test schedules a stale rewrite, forces flushSync to bump the epoch via a fenced append, schedules a newer rewrite that parks at pauses[1], releases the stale gate (stale unwinds and guard-rejects), then appends a custom entry — asserts writerOpens does not grow (fence preserved) and the fenced entry lands in the newer publish's body. Without the fix, writerOpens grows from 1 to 2. Fixes #4338 --- .../src/session/session-manager.ts | 8 +- ...ession-manager-atomic-rewrite-race.test.ts | 148 ++++++++++++++++++ 2 files changed, 155 insertions(+), 1 deletion(-) diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index d7756430f..896255836 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -615,7 +615,13 @@ export class SessionManager { } while (this.#atomicRewriteDirty); return true; } finally { - this.#atomicRewriteFenceEpoch = null; + // 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; } } diff --git a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts index ea9384795..41b8dc695 100644 --- a/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -531,3 +531,151 @@ describe("SessionManager fence relaxes when flushSync supersedes the atomic rewr expect(storage.detachedLines).toEqual([]); }); }); + +interface PauseHandle { + started: PromiseWithResolvers; + allow: PromiseWithResolvers; +} + +class SequencedRewriteStorage extends MemorySessionStorage { + readonly detachedLines: string[] = []; + readonly pauses: PauseHandle[] = []; + guardRejections = 0; + writerOpens = 0; + pauseCount = 0; + #calls = 0; + readonly #writers = new Set(); + + 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 { + if (detached) { + detachedLines.push(line); + return; + } + await inner.append(line); + }, + async flush(): Promise { + await inner.flush(); + }, + isOpen(): boolean { + return inner.isOpen(); + }, + async close(): Promise { + 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 { + const index = this.#calls++; + if (index < this.pauseCount) { + const pause: PauseHandle = { + started: Promise.withResolvers(), + allow: Promise.withResolvers(), + }; + 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([]); + }); +});