diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 2116ba31d..100a90cd0 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -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 ` 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 diff --git a/packages/coding-agent/src/session/indexed-session-storage.ts b/packages/coding-agent/src/session/indexed-session-storage.ts index 4ae9fb9df..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,8 +239,40 @@ export class IndexedSessionStorage implements SessionStorage { } } - writeTextAtomic(path: string, content: string): Promise { - 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/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index 7637e0448..896255836 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -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 { 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 { + 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; } diff --git a/packages/coding-agent/src/session/session-storage.ts b/packages/coding-agent/src/session/session-storage.ts index 5afef3ea4..a2db80cf2 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,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; - 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; 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 @@ -229,53 +250,107 @@ 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()) { + 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 { + // 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 { + writeTextAtomic(path: string, content: string, options?: WriteTextAtomicOptions): Promise { + 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 { + 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 new file mode 100644 index 000000000..41b8dc695 --- /dev/null +++ b/packages/coding-agent/test/session-manager-atomic-rewrite-race.test.ts @@ -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(); + readonly allowRewrite = Promise.withResolvers(); + pausedRewrites = 0; + guardRejections = 0; + 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; + } + + 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); + } +} + +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(); + 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(); + 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); + }); +}); + +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; + 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([]); + }); +}); 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..69fbe05c6 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); @@ -116,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. */ 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..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 @@ -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); } } @@ -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 { + 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(); 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 { 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"]); + }); +});