diff --git a/packages/coding-agent/CHANGELOG.md b/packages/coding-agent/CHANGELOG.md index 9c7b76c18..1c2853da8 100644 --- a/packages/coding-agent/CHANGELOG.md +++ b/packages/coding-agent/CHANGELOG.md @@ -25,6 +25,7 @@ ### Fixed +- Fixed session transcript entries being lost on process crash: file-backed JSONL writers now write each completed entry to the OS page cache on append instead of microtask-batching; concurrent appends during an in-place atomic rewrite supersede the paused publish with a synchronous full-body rewrite so fenced events are durable before return; concurrent appends during `/move` write a full body to the live relocation path (source pre-rename, destination post-rename) so a crash mid-move no longer drops completed events; and the first write failure latches `#diskFailure` synchronously via `appendSync` so a later flush/close reports it instead of silently succeeding after a discarded rejected Promise ([#7444](https://github.com/can1357/oh-my-pi/pull/7444) by [@olegpulatov](https://github.com/olegpulatov)). - Fixed template argument substitution (`substituteArgs`) executing recursive placeholder expansion when positional argument values contain literal `$@` or `$ARGUMENTS` tokens. - Fixed focused-agent status bar dimming darkening Powerline end caps. - Fixed the browser relay creating duplicate "omp" tab groups: the bridge now keeps at most one group RPC in flight (a queued drain replaces fire-and-forget per-tab requests), so concurrent requests can no longer race the extension's non-atomic query→create→set-title sequence in the same window. Also fixed an extension reconnect (relay daemon restart, service-worker recycle) being misread as the user dragging every tab out of the omp group — grouping state is reset when the extension socket closes, so tabs regroup on the next hello instead of being permanently opted out. diff --git a/packages/coding-agent/src/session/indexed-session-storage.ts b/packages/coding-agent/src/session/indexed-session-storage.ts index 4786f4229..e25c648d4 100644 --- a/packages/coding-agent/src/session/indexed-session-storage.ts +++ b/packages/coding-agent/src/session/indexed-session-storage.ts @@ -511,6 +511,15 @@ class IndexedSessionStorageWriter implements SessionStorageWriter { return next; } + appendSync(line: string): void { + if (this.#closed) throw new Error("Writer closed"); + if (this.#error) throw this.#error; + // Local index is updated immediately; remote publish stays ordered on the + // path queue. Callers that need remote durability still await append()/flush(). + const mtimeMs = this.#storage._appendForWriter(this.#path, line); + void this.#trackPromise(this.#storage._queueAppend(this.#path, line, mtimeMs, () => this.#error)); + } + async append(line: string): Promise { if (this.#closed) throw new Error("Writer closed"); if (this.#error) throw this.#error; diff --git a/packages/coding-agent/src/session/session-manager.ts b/packages/coding-agent/src/session/session-manager.ts index b8ae83e91..1c630f026 100644 --- a/packages/coding-agent/src/session/session-manager.ts +++ b/packages/coding-agent/src/session/session-manager.ts @@ -405,17 +405,22 @@ export class SessionPersistenceIndeterminateError extends AggregateError { * tree by `(id, parentId)`, and the mutable leaf pointer selects which path is * active for future appends and for LLM context construction. * - * Durability is software-crash safe but not power-loss safe: appends are normally - * handed to the OS synchronously in-body but never `fsync`'d. While an in-place - * atomic rewrite publishes, a synchronous `flushSync` still persists fenced - * appends by superseding the pending publish against the same (stable) path. - * A {@link moveTo} relocation is the one exception: the session file is being - * renamed to a new path, so appends landing in the rename window are held in - * memory and persisted only by the move's trailing rewrite. A `flushSync` in - * that narrow window cannot durably place them — writing the vacated source - * would recreate the orphan the move eliminates, and a queued backend publish - * (fire-and-forget on `IndexedSessionStorage`) could land after the rename all - * the same. This gap is inherent to closing the writer for a Windows-safe rename. + * Durability is software-crash safe but not power-loss safe: completed entries + * (user/assistant/toolResult messages, tool_execution_start markers, custom + * entries) are handed to the OS synchronously in-body on append and never + * `fsync`'d. In-flight streaming text is intentionally not durable until + * `message_end` persists the finished message. + * + * While an in-place atomic rewrite is publishing, a concurrent completed append + * supersedes that publish with a synchronous full-body rewrite so the entry is + * software-crash durable before the append returns; the abandoned atomic's + * `commitGuard` then refuses to clobber the fresher body. + * + * During {@link moveTo}, appends write a full body to the live relocation path + * (source until rename, destination once the rename has landed) so a crash mid- + * move still preserves completed entries without recreating a vacated source. + * A trailing atomic rewrite still rewrites the header cwd after the path is + * repointed. */ export class SessionManager { #cwd: string; @@ -484,15 +489,12 @@ export class SessionManager { /** Set by synchronous appends that land while an atomic replacement is active. */ #atomicRewriteDirty = false; /** - * True while {@link moveTo} is renaming the session file but has not yet - * repointed `#sessionFile`. Both the synchronous append hot path - * ({@link #appendToSessionFile}) and synchronous rewrites - * ({@link #rewriteSynchronously}, reached via `flushSync` on Ctrl+C) would - * otherwise touch the pre-rename path — opening a fresh writer or writing the - * full body — recreating the very orphan the move eliminates. Both defer - * while this is set; the move's trailing atomic rewrite persists them. + * Active {@link moveTo} relocation. Concurrent completed appends write a + * full body to the live path: source while it still exists, destination + * once rename has landed (source gone). Never recreates a vacated source. + * `null` outside an active relocation. */ - #sessionFileRelocating = false; + #sessionFileRelocating: { source: string; dest: string } | null = null; /** Atomic entry batch currently staged for a full-file commit. */ #atomicEntryBatch: AtomicEntryBatch | undefined; @@ -763,30 +765,56 @@ export class SessionManager { return this.#forceFileCreation || this.#fileIsCurrent || this.#historyContainsAssistantMessage(); } + /** + * Live path for concurrent completed appends during {@link moveTo}. + * Prefers destination once rename has landed (source gone); otherwise + * source. Never invents a path that does not already exist. + */ + #liveRelocationWritePath(): string | null { + const relocating = this.#sessionFileRelocating; + if (!relocating) return null; + if (this.#storage.existsSync(relocating.dest)) return relocating.dest; + if (this.#storage.existsSync(relocating.source)) return relocating.source; + // Rename in flight with neither path visible (rare cross-device edge): + // fall back to destination so we do not recreate a vacated source. + return relocating.dest; + } + /** * Synchronously rewrite the whole file (header + entries) and keep no open * writer; the next append re-opens one. `writeTextSync` returns with the * bytes in the kernel page cache, so the file is software-crash durable. + * + * During {@link moveTo}, writes to the live relocation path (source pre- + * rename, destination post-rename) rather than always `#sessionFile`, so + * concurrent completed entries are durable without recreating a vacated source. */ #rewriteSynchronously(): void { - if (!this.#persist || !this.#sessionFile || !this.#shouldHaveSessionFile()) return; - // A move is renaming the file out from under `#sessionFile`. Writing the - // full body now would target either the vacated source (recreating the - // orphan the move eliminates) or a path a queued backend publish may clobber - // post-rename, so defer to moveTo's trailing atomic rewrite. This narrows - // flushSync durability for entries in the rename window (see class doc). - if (this.#sessionFileRelocating) return; + if (!this.#persist || !this.#shouldHaveSessionFile()) return; + const targetPath = this.#liveRelocationWritePath() ?? this.#sessionFile; + if (!targetPath) return; try { const body = this.#fileBody(); this.#diskEpoch++; this.#diskTail = Promise.resolve(); this.#closeWriterEventually(); - this.#storage.writeTextSync(this.#sessionFile, body); - this.#fileIsCurrent = true; - this.#materializeBreadcrumb(); - this.#rewriteRequired = false; - this.#hasTitleSlot = true; + this.#storage.writeTextSync(targetPath, body); + // Only mark the manager current when writing the active session path. + // Mid-move writes update the live relocation path; `#sessionFile` is + // still the pre-repoint source until moveTo repoints it. + if (!this.#sessionFileRelocating || targetPath === this.#sessionFile) { + this.#fileIsCurrent = true; + this.#materializeBreadcrumb(); + this.#rewriteRequired = false; + this.#hasTitleSlot = true; + } else { + // Destination body is current on disk; in-memory still needs a + // header-cwd rewrite after repoint, but entries are durable. + this.#fileIsCurrent = false; + this.#rewriteRequired = true; + this.#hasTitleSlot = true; + } } catch (err) { this.#noteDiskFailure(err); } @@ -872,22 +900,21 @@ export class SessionManager { return; } - // Atomic replacement window: the old path may be moved aside underneath - // any newly-opened append handle (Windows EPERM fallback), or a `moveTo` - // may be renaming the session file to a new directory. Do not open a - // writer here; the active rewrite loops and serializes a fresh full body, - // and the move's trailing atomic rewrite folds in these deferred entries. - // A superseding synchronous rewrite bumps `#diskEpoch`, at which point - // the pending atomic publish is guaranteed to abandon via its - // `commitGuard`, so appends can (and must) take the hot path so they - // don't strand in memory while `close()` returns without a rewrite. - if ( - this.#sessionFileRelocating || - (this.#atomicRewriteFenceEpoch !== null && this.#atomicRewriteFenceEpoch === this.#diskEpoch) - ) { - this.#fileIsCurrent = false; - this.#rewriteRequired = true; + // Atomic replacement / move window: do not open a fresh append writer that + // a Windows EPERM replace could detach from the current JSONL path. + // - moveTo: write a full body to the live relocation path (source pre- + // rename, destination post-rename) so completed entries are durable + // without recreating a vacated source. + // - in-place atomic fence: supersede the pending publish with a + // synchronous full-body rewrite; bumping `#diskEpoch` abandons the + // in-flight atomic via its `commitGuard`. + if (this.#sessionFileRelocating) { + this.#rewriteSynchronously(); + return; + } + if (this.#atomicRewriteFenceEpoch !== null && this.#atomicRewriteFenceEpoch === this.#diskEpoch) { this.#atomicRewriteDirty = true; + this.#rewriteSynchronously(); return; } // Cold/divergent: not on disk yet, or in-memory entries diverged from the @@ -897,15 +924,23 @@ export class SessionManager { return; } - // Hot path: queue the entry directly on the writer, outside the async disk - // chain. File writers batch same-turn appends until their microtask - // boundary; flush(), flushSync(), and close() drain that batch explicitly. + // Hot path: write the entry directly on the writer, outside the async disk + // chain. Prefer appendSync so write failures latch `#diskFailure` before + // this call returns (not via a discarded rejected Promise after a later + // microtask). Callers stay non-throwing here — the core turn loop invokes + // appendMessage/appendCustomEntry without try/catch; flushSync/close and + // subsequent appends still throw the latched error. File writers apply + // each line to the OS page cache before return. // A mid-close writer leaves `#writer` undefined, so `#appendWriter` simply // opens a fresh append handle and the entry still lands. try { - void this.#appendWriter() - .append(this.#lineFor(entry)) - .catch(err => this.#noteDiskFailure(err)); + const writer = this.#appendWriter(); + const line = this.#lineFor(entry); + if (writer.appendSync) { + writer.appendSync(line); + } else { + void writer.append(line).catch(err => this.#noteDiskFailure(err)); + } } catch (err) { this.#noteDiskFailure(err); } @@ -921,12 +956,11 @@ export class SessionManager { } // Title changes use their own asynchronous append path rather than - // #appendToSessionFile. Defer them under the same move fence so they cannot - // recreate the vacated source path before #sessionFile is repointed. + // #appendToSessionFile. During move, write the full body (including the + // title entry) to the live relocation path so a crash mid-move still + // keeps the title change; the trailing rewrite still updates header cwd. if (this.#sessionFileRelocating) { - this.#fileIsCurrent = false; - this.#rewriteRequired = true; - this.#atomicRewriteDirty = true; + this.#rewriteSynchronously(); return; } @@ -1362,14 +1396,15 @@ export class SessionManager { : computeDefaultSessionDir(resolvedCwd, this.#storage)); let sessionFileExisted = false; - // Fence synchronous appends and rewrites away from the session file for the - // duration of the relocation (see `#sessionFileRelocating`). This flag — - // not a `#diskEpoch` bump — is used deliberately: bumping the epoch here - // would cancel any disk task already queued at the current epoch (e.g. a - // header-only `ensureOnDisk()` materializing rewrite) before the drain - // below runs it, silently dropping an explicitly materialized session. + // Track source+dest for concurrent completed appends during relocation + // (see `#sessionFileRelocating`). Existence of either path decides the + // live write target — not a `#diskEpoch` bump, which would cancel any + // disk task already queued at the current epoch (e.g. a header-only + // `ensureOnDisk()` materializing rewrite) before the drain below runs it. if (this.#persist && this.#sessionFile) { - this.#sessionFileRelocating = true; + const source = this.#sessionFile; + const dest = path.join(nextSessionDir, path.basename(source)); + this.#sessionFileRelocating = { source, dest }; } try { @@ -1433,8 +1468,8 @@ export class SessionManager { this.#sessionFile = newSessionFile; this.#artifactManager = null; this.#artifactManagerSessionFile = null; - // Path is repointed; a sync rewrite may now safely target the new file. - this.#sessionFileRelocating = false; + // Path is repointed; hot-path appends may use `#sessionFile` again. + this.#sessionFileRelocating = null; } this.#cwd = resolvedCwd; @@ -1459,7 +1494,7 @@ export class SessionManager { if (this.#sessionFile) this.#rememberBreadcrumb(resolvedCwd, this.#sessionFile); } finally { - this.#sessionFileRelocating = false; + this.#sessionFileRelocating = null; } } diff --git a/packages/coding-agent/src/session/session-storage.ts b/packages/coding-agent/src/session/session-storage.ts index c35773ad4..a10e8dc7c 100644 --- a/packages/coding-agent/src/session/session-storage.ts +++ b/packages/coding-agent/src/session/session-storage.ts @@ -14,13 +14,25 @@ export interface SessionStorageStat { export interface SessionStorageWriter { /** - * Append one newline-terminated line. File storage batches same-turn appends - * until its microtask boundary; memory storage updates in-body; indexed - * backends queue in call order. + * Append one newline-terminated line. + * + * File and memory storage apply the line synchronously before the returned + * promise settles, so a software crash after `append` returns (or after a + * fire-and-forget call begins) still sees the entry on disk / in body. No + * `fsync` — power loss may still drop the last page. Indexed backends update + * the local index immediately and queue the remote publish in call order. * * `line` MUST include the trailing newline. */ append(line: string): Promise; + /** + * Synchronous append when the backend can apply the line before return. + * File and memory implement this so {@link SessionManager} can latch the + * first write failure before the appending call returns (surfaced by a later + * flushSync/close/next append — the turn loop does not throw from append). + * Indexed backends update the local index immediately and queue remote I/O. + */ + appendSync?(line: string): void; /** Resolve once all queued appends complete. No fsync. */ flush(): Promise; /** Drain synchronously flushable queued work when the backend supports it. No fsync. */ @@ -93,8 +105,6 @@ class FileSessionStorageWriter implements SessionStorageWriter { #closed = false; #error: Error | undefined; #onError: ((err: Error) => void) | undefined; - #pending = ""; - #flushScheduled = false; constructor(fpath: string, options?: { flags?: "a" | "w"; onError?: (err: Error) => void }) { this.#onError = options?.onError; @@ -129,34 +139,29 @@ class FileSessionStorageWriter implements SessionStorageWriter { } } - #flushPendingNow(): void { - this.#flushScheduled = false; - if (this.#pending.length === 0) return; - const pending = this.#pending; - this.#pending = ""; + appendSync(line: string): void { + if (this.#closed) throw new Error("Writer closed"); + if (this.#error) throw this.#error; + // Write in-body so software crash after the call still sees the entry. + // Microtask batching used to leave completed transcript lines only in + // memory until the next event-loop turn; process crash then lost every + // post-checkpoint event. flush/flushSync remain no-op drains (no fsync). try { - this.#writeNow(pending); + this.#writeNow(line); } catch (err) { - this.#recordError(err); + throw this.#recordError(err); } } async append(line: string): Promise { - if (this.#closed) throw new Error("Writer closed"); - if (this.#error) throw this.#error; - this.#pending += line; - if (this.#flushScheduled) return; - this.#flushScheduled = true; - queueMicrotask(() => this.#flushPendingNow()); + this.appendSync(line); } async flush(): Promise { - this.#flushPendingNow(); if (this.#error) throw this.#error; } flushSync(): void { - this.#flushPendingNow(); if (this.#error) throw this.#error; } @@ -166,7 +171,6 @@ class FileSessionStorageWriter implements SessionStorageWriter { async close(): Promise { if (this.#closed) return; - this.#flushPendingNow(); this.#closed = true; // Unregister from finalization - we're closing properly writerRegistry.unregister(this); @@ -487,7 +491,7 @@ class MemorySessionStorageWriter implements SessionStorageWriter { return error; } - async append(line: string): Promise { + appendSync(line: string): void { if (this.#closed) throw new Error("Writer closed"); if (this.#error) throw this.#error; try { @@ -498,6 +502,10 @@ class MemorySessionStorageWriter implements SessionStorageWriter { } } + async append(line: string): Promise { + this.appendSync(line); + } + async flush(): Promise { if (this.#error) throw this.#error; } 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 8c8bf6732..a82c94e9e 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 @@ -364,10 +364,16 @@ describe("SessionManager atomic rewrite fence spans writer.close()", () => { sessionManager.appendMessage({ role: "user", content: "during close", timestamp: Date.now() }); sessionManager.appendCustomEntry("during_close_custom", { reason: "guard" }); - // Pre-fix, #appendToSessionFile would take the hot path and call - // storage.openWriter here; the writer would then be caught by the pending - // writeTextAtomic detachment. Fence keeps writerOpens flat. - expect(storage.writerOpens).toBe(opensBeforeRewrite); + // First fenced append supersedes the in-flight atomic with a synchronous + // full-body rewrite (software-crash durable before return). That bumps + // `#diskEpoch`, so a second append may open a hot-path writer against the + // already-published body; the abandoned atomic's commitGuard must still + // refuse to clobber it, and nothing may land on a detached handle. + const sessionFileMid = sessionManager.getSessionFile(); + if (!sessionFileMid) throw new Error("Expected session file"); + const midContent = await storage.readText(sessionFileMid); + expect(midContent).toContain("during close"); + expect(midContent).toContain('"customType":"during_close_custom"'); storage.allowClose.resolve(); storage.allowWrite.resolve(); @@ -379,6 +385,9 @@ describe("SessionManager atomic rewrite fence spans writer.close()", () => { const content = await storage.readText(sessionFile); expect(content).toContain("during close"); expect(content).toContain('"customType":"during_close_custom"'); + // Superseding rewrite may finish before the paused atomic reaches its + // commitGuard; either way the fenced entries must remain and no append + // may land on a detached handle. expect(storage.detachedLines).toEqual([]); }); }); @@ -443,9 +452,10 @@ describe("SessionManager title-change fallback fenced-append durability", () => const rename = sessionManager.setSessionName("second title", "user", "test"); await storage.writeStarted.promise; - // Fenced appends during the paused fallback rewrite: pre-fix these - // would be marked dirty and dropped from the serialized body because - // the fallback never looped on that flag. + // Fenced appends during the paused fallback rewrite supersede the atomic + // with a synchronous full-body rewrite, so they are on disk before the + // paused publish resumes. The abandoned atomic's body must not clobber + // them when released. sessionManager.appendMessage({ role: "user", content: "during title fallback", @@ -453,6 +463,12 @@ describe("SessionManager title-change fallback fenced-append durability", () => }); sessionManager.appendCustomEntry("during_title_fallback_custom", { reason: "test" }); + const sessionFileMid = sessionManager.getSessionFile(); + if (!sessionFileMid) throw new Error("Expected session file"); + const midContent = await storage.readText(sessionFileMid); + expect(midContent).toContain("during title fallback"); + expect(midContent).toContain('"customType":"during_title_fallback_custom"'); + storage.allowWrite.resolve(); await rename; await sessionManager.flush(); @@ -463,9 +479,10 @@ describe("SessionManager title-change fallback fenced-append durability", () => expect(content).toContain('"title":"second title"'); expect(content).toContain("during title fallback"); expect(content).toContain('"customType":"during_title_fallback_custom"'); - // Loop must have executed at least twice: first pass paused, dirty from - // the fenced appends triggers a second pass that includes them. - expect(storage.writeTextAtomicCalls).toBeGreaterThanOrEqual(2); + // At least the failed title path's atomic fallback ran once; fenced + // appends may have superseded it via writeTextSync without a second + // atomic pass. + expect(storage.writeTextAtomicCalls).toBeGreaterThanOrEqual(1); }); }); diff --git a/packages/coding-agent/test/session-manager-immediate-persist.test.ts b/packages/coding-agent/test/session-manager-immediate-persist.test.ts index c26c2ed24..045ac116c 100644 --- a/packages/coding-agent/test/session-manager-immediate-persist.test.ts +++ b/packages/coding-agent/test/session-manager-immediate-persist.test.ts @@ -1,9 +1,10 @@ -import { afterEach, describe, expect, it } from "bun:test"; +import { afterEach, describe, expect, it, spyOn } from "bun:test"; import * as fs from "node:fs"; import * as path from "node:path"; import { getBundledModel } from "@oh-my-pi/pi-catalog/models"; import { SessionManager } from "@oh-my-pi/pi-coding-agent/session/session-manager"; -import { TempDir } from "@oh-my-pi/pi-utils"; +import { MemorySessionStorage, type WriteTextAtomicOptions } from "@oh-my-pi/pi-coding-agent/session/session-storage"; +import { parseJsonlLenient, TempDir } from "@oh-my-pi/pi-utils"; const tempDirs: TempDir[] = []; @@ -27,11 +28,11 @@ function assistantMessage(text: string) { provider: model.provider, model: model.id, usage: { - input: 0, - output: 0, + input: 1, + output: 1, cacheRead: 0, cacheWrite: 0, - totalTokens: 0, + totalTokens: 2, cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 }, }, stopReason: "stop" as const, @@ -39,6 +40,17 @@ function assistantMessage(text: string) { }; } +function toolResultMessage(toolCallId: string, toolName: string, text: string) { + return { + role: "toolResult" as const, + toolCallId, + toolName, + content: [{ type: "text" as const, text }], + isError: false, + timestamp: Date.now(), + }; +} + function readJsonl(file: string): Array> { return fs .readFileSync(file, "utf8") @@ -52,18 +64,27 @@ function readJsonl(file: string): Array> { function messageRole(entry: Record): string | undefined { const message = entry.message; if (!message || typeof message !== "object") return undefined; - const role = (message as { role?: unknown }).role; - return typeof role === "string" ? role : undefined; + if (!("role" in message) || typeof message.role !== "string") return undefined; + return message.role; } function messageContent(entry: Record): unknown { const message = entry.message; if (!message || typeof message !== "object") return undefined; - return (message as { content?: unknown }).content; + if (!("content" in message)) return undefined; + return message.content; } -describe("SessionManager batched JSONL persistence", () => { - it("writes later entries at the microtask or synchronous flush boundary", async () => { +function entryKind(entry: Record): string { + if (entry.type === "message") return messageRole(entry) ?? "message"; + if (entry.type === "custom") { + return `custom:${typeof entry.customType === "string" ? entry.customType : ""}`; + } + return typeof entry.type === "string" ? entry.type : "unknown"; +} + +describe("SessionManager JSONL software-crash durability", () => { + it("makes completed entries visible on disk without a microtask or flush", () => { const cwd = makeTempDir("@pi-immediate-cwd-"); const sessionDir = path.join(cwd, "sessions"); const manager = SessionManager.create(cwd, sessionDir); @@ -73,6 +94,7 @@ describe("SessionManager batched JSONL persistence", () => { manager.appendMessage({ role: "user", content: "queued before assistant", timestamp: Date.now() }); expect(fs.existsSync(sessionFile)).toBe(false); + // First assistant materializes the file via the synchronous rewrite path. manager.appendMessage(assistantMessage("hello")); expect(fs.existsSync(sessionFile)).toBe(true); @@ -81,24 +103,99 @@ describe("SessionManager batched JSONL persistence", () => { expect(messageRole(entries[1] ?? {})).toBe("user"); expect(messageRole(entries[2] ?? {})).toBe("assistant"); + // Hot-path appends must land in the OS page cache before the call returns. + // A software crash (SIGKILL / OOM / hard abort) after append and before the + // next event-loop turn must not drop them. manager.appendMessage({ role: "user", content: "written immediately", timestamp: Date.now() }); - - entries = readJsonl(sessionFile); - expect(entries).toHaveLength(3); - - await Promise.resolve(); entries = readJsonl(sessionFile); expect(entries).toHaveLength(4); expect(messageRole(entries[3] ?? {})).toBe("user"); expect(messageContent(entries[3] ?? {})).toBe("written immediately"); - manager.appendMessage({ role: "user", content: "flushed synchronously", timestamp: Date.now() }); - manager.flushSync(); + manager.appendMessage(assistantMessage("second turn")); + manager.appendCustomEntry("tool_execution_start", { + toolCallId: "call-1", + toolName: "bash", + args: { command: "echo hi" }, + }); + manager.appendMessage(toolResultMessage("call-1", "bash", "hi")); entries = readJsonl(sessionFile); - expect(entries).toHaveLength(5); - expect(messageRole(entries[4] ?? {})).toBe("user"); - expect(messageContent(entries[4] ?? {})).toBe("flushed synchronously"); + expect(entries.map(entryKind)).toEqual([ + "session", + "user", + "assistant", + "user", + "assistant", + "custom:tool_execution_start", + "toolResult", + ]); + }); + + it("reopens post-checkpoint user/assistant/tool events after a crash-equivalent snapshot", async () => { + const cwd = makeTempDir("@pi-crash-reopen-cwd-"); + const sessionDir = path.join(cwd, "sessions"); + const manager = SessionManager.create(cwd, sessionDir); + const sessionFile = manager.getSessionFile(); + if (!sessionFile) throw new Error("Expected a persisted session file path"); + + // Durable checkpoint: materialize, then snapshot bytes. + manager.appendMessage({ role: "user", content: "checkpoint user", timestamp: Date.now() }); + manager.appendMessage(assistantMessage("checkpoint assistant")); + const checkpointBytes = fs.readFileSync(sessionFile); + const checkpointKinds = parseJsonlLenient>(checkpointBytes.toString("utf8")) + .filter(entry => entry.type !== "title") + .map(entryKind); + expect(checkpointKinds).toEqual(["session", "user", "assistant"]); + + // Completed events after the checkpoint — no flushSync, no await, no close. + manager.appendMessage({ role: "user", content: "post-checkpoint user", timestamp: Date.now() }); + manager.appendMessage(assistantMessage("post-checkpoint assistant")); + manager.appendCustomEntry("tool_execution_start", { + toolCallId: "call-post", + toolName: "read", + args: { path: "README.md" }, + }); + manager.appendMessage(toolResultMessage("call-post", "read", "# readme")); + + // Crash equivalent: only bytes already on disk survive. Copy without yielding + // to microtasks and reopen from that snapshot in a fresh manager. + const crashPath = path.join(cwd, "crashed-session.jsonl"); + fs.writeFileSync(crashPath, fs.readFileSync(sessionFile)); + + const reopened = await SessionManager.open(crashPath); + const reopenedKinds = reopened.getEntries().map(entry => { + if (entry.type === "message") return entry.message.role; + if (entry.type === "custom") return `custom:${entry.customType}`; + return entry.type; + }); + expect(reopenedKinds).toEqual([ + "user", + "assistant", + "user", + "assistant", + "custom:tool_execution_start", + "toolResult", + ]); + + const postUser = reopened + .getEntries() + .find( + entry => + entry.type === "message" && + entry.message.role === "user" && + entry.message.content === "post-checkpoint user", + ); + expect(postUser).toBeDefined(); + const toolResult = reopened + .getEntries() + .find( + entry => + entry.type === "message" && + entry.message.role === "toolResult" && + entry.message.toolCallId === "call-post", + ); + expect(toolResult).toBeDefined(); }); it("keeps pre-assistant sessions out of history during shutdown", async () => { @@ -137,4 +234,126 @@ describe("SessionManager batched JSONL persistence", () => { expect(messageRole(entries[1] ?? {})).toBe("user"); expect(messageContent(entries[1] ?? {})).toBe("persist me"); }); + + it("makes fenced appends durable while an atomic rewrite is paused", async () => { + // H1: in-place atomic fence used to mark entries dirty and return without + // writing. A crash before the paused publish resumed lost every fenced + // completed event. Superseding with #rewriteSynchronously must put them + // on disk before append returns, and the abandoned atomic must not clobber. + class PausedRewriteStorage extends MemorySessionStorage { + readonly rewriteStarted = Promise.withResolvers(); + readonly allowRewrite = Promise.withResolvers(); + guardRejections = 0; + + override async writeTextAtomic( + filePath: string, + content: string, + options?: WriteTextAtomicOptions, + ): Promise { + this.rewriteStarted.resolve(); + await this.allowRewrite.promise; + if (options?.commitGuard && !options.commitGuard()) { + this.guardRejections++; + return; + } + this.writeTextSync(filePath, content); + } + } + + const storage = new PausedRewriteStorage(); + const manager = SessionManager.create("/cwd", "/sessions", storage); + manager.appendMessage(assistantMessage("seed")); + await manager.flush(); + manager.appendMessage({ role: "user", content: "checkpoint", timestamp: Date.now() }); + await manager.flush(); + + const sessionFile = manager.getSessionFile(); + if (!sessionFile) throw new Error("Expected session file"); + + const rewrite = manager.rewriteEntries(); + await storage.rewriteStarted.promise; + + manager.appendMessage({ role: "user", content: "fenced-user", timestamp: Date.now() }); + manager.appendMessage(assistantMessage("fenced-assistant")); + manager.appendCustomEntry("tool_execution_start", { + toolCallId: "fenced-call", + toolName: "bash", + }); + manager.appendMessage(toolResultMessage("fenced-call", "bash", "hi")); + + // Crash-equivalent while the atomic is still paused: only storage bytes survive. + const crashKinds = parseJsonlLenient>(await storage.readText(sessionFile)) + .filter(entry => entry.type !== "title") + .map(entryKind); + expect(crashKinds).toEqual([ + "session", + "assistant", + "user", + "user", + "assistant", + "custom:tool_execution_start", + "toolResult", + ]); + + storage.allowRewrite.resolve(); + await rewrite; + await manager.flush(); + + const afterKinds = parseJsonlLenient>(await storage.readText(sessionFile)) + .filter(entry => entry.type !== "title") + .map(entryKind); + // Fenced events remain after the paused atomic settles (commitGuard may + // reject a stale body, or a superseding rewrite already owns the file). + expect(afterKinds).toEqual(crashKinds); + }); + + it("latches the first hot-path write failure before append returns", () => { + // H2: async append + discarded Promise meant ENOSPC/EIO only latched after a + // later flush/append. appendSync must latch `#diskFailure` on the first call + // without throwing through unguarded turn-loop callers. + const cwd = makeTempDir("@pi-write-fail-cwd-"); + const sessionDir = path.join(cwd, "sessions"); + const manager = SessionManager.create(cwd, sessionDir); + const sessionFile = manager.getSessionFile(); + if (!sessionFile) throw new Error("Expected session file"); + + manager.appendMessage(assistantMessage("seed")); + manager.appendMessage({ role: "user", content: "ok-user", timestamp: Date.now() }); + + let failWrites = true; + const origWrite = fs.writeSync.bind(fs) as typeof fs.writeSync; + const writeSpy = spyOn(fs, "writeSync").mockImplementation(((...args: Parameters) => { + if (failWrites) { + const err = new Error("ENOSPC: no space left on device") as NodeJS.ErrnoException; + err.code = "ENOSPC"; + throw err; + } + return origWrite(...args); + }) as typeof fs.writeSync); + + try { + let threw = false; + try { + manager.appendMessage({ role: "user", content: "should-fail-user", timestamp: Date.now() }); + } catch { + threw = true; + } + // Unguarded turn-loop contract: appendMessage itself must not throw. + expect(threw).toBe(false); + + // Failure is latched before return — flushSync / next append surface it. + expect(() => manager.flushSync()).toThrow("ENOSPC"); + expect(() => manager.appendMessage({ role: "user", content: "next-user", timestamp: Date.now() })).toThrow( + "ENOSPC", + ); + + const users = parseJsonlLenient>(fs.readFileSync(sessionFile, "utf8")) + .filter(entry => entry.type === "message" && messageRole(entry) === "user") + .map(entry => messageContent(entry)); + expect(users).toEqual(["ok-user"]); + } finally { + failWrites = false; + writeSpy.mockRestore(); + } + }); }); diff --git a/packages/coding-agent/test/session-manager/move-to.test.ts b/packages/coding-agent/test/session-manager/move-to.test.ts index 9434926e7..c45b7b8fc 100644 --- a/packages/coding-agent/test/session-manager/move-to.test.ts +++ b/packages/coding-agent/test/session-manager/move-to.test.ts @@ -387,4 +387,100 @@ describe("SessionManager.moveTo", () => { const targetSessions = await SessionManager.list(cwdB); expect(targetSessions.some(item => item.path === movedFile)).toBe(true); }); + + it("keeps post-rename fenced appends durable before trailing rewrite", async () => { + // Crash window: session file has been renamed to dest, `#sessionFile` is + // still the source path, and the trailing atomic rewrite has not run. + // Completed entries appended in this window must land on dest (not recreate + // source) and survive a crash-equivalent snapshot + reopen. + const session = SessionManager.create(cwdA); + session.appendMessage({ role: "user", content: "before move", timestamp: 1 }); + session.appendMessage(makeAssistantMessage()); + await session.flush(); + + const oldFile = session.getSessionFile(); + if (!oldFile) throw new Error("Expected session file"); + + const renameFinished = Promise.withResolvers<{ dest: string }>(); + const allowMoveToResume = Promise.withResolvers(); + const rename = fs.promises.rename.bind(fs.promises); + const renameSpy = spyOn(fs.promises, "rename").mockImplementation(async (source, target) => { + await rename(source, target); + if (path.resolve(source.toString()) !== path.resolve(oldFile)) return; + renameFinished.resolve({ dest: path.resolve(target.toString()) }); + await allowMoveToResume.promise; + }); + + let dest = ""; + try { + const move = session.moveTo(cwdB); + ({ dest } = await renameFinished.promise); + + session.appendMessage({ role: "user", content: "during move crash window", timestamp: 2 }); + session.appendCustomEntry("tool_execution_start", { + toolCallId: "move-call", + toolName: "bash", + }); + session.appendMessage({ + role: "toolResult", + toolCallId: "move-call", + toolName: "bash", + content: [{ type: "text", text: "ok" }], + isError: false, + timestamp: 3, + }); + + // Crash-equivalent: only dest bytes exist; source must stay absent. + expect(fs.existsSync(oldFile)).toBe(false); + expect(fs.existsSync(dest)).toBe(true); + const crashBytes = fs.readFileSync(dest, "utf8"); + expect(crashBytes).toContain("during move crash window"); + expect(crashBytes).toContain('"customType":"tool_execution_start"'); + expect(crashBytes).toContain("move-call"); + + const crashPath = path.join(testAgentDir, "crashed-move.jsonl"); + fs.writeFileSync(crashPath, crashBytes); + const reopened = await SessionManager.open(crashPath); + const reopenedEntries = reopened.getEntries(); + expect( + reopenedEntries.some( + entry => + entry.type === "message" && + entry.message.role === "user" && + entry.message.content === "during move crash window", + ), + ).toBe(true); + expect( + reopenedEntries.some( + entry => + entry.type === "message" && + entry.message.role === "toolResult" && + entry.message.toolCallId === "move-call", + ), + ).toBe(true); + expect( + reopenedEntries.some(entry => entry.type === "custom" && entry.customType === "tool_execution_start"), + ).toBe(true); + + allowMoveToResume.resolve(); + await move; + await session.flush(); + } finally { + allowMoveToResume.resolve(); + renameSpy.mockRestore(); + } + + expect(fs.existsSync(oldFile)).toBe(false); + const movedFile = session.getSessionFile(); + if (!movedFile) throw new Error("Expected moved session file"); + const entries = await loadEntriesFromFile(movedFile); + expect( + entries.some( + entry => + entry.type === "message" && + entry.message.role === "user" && + entry.message.content === "during move crash window", + ), + ).toBe(true); + }); }); diff --git a/packages/coding-agent/test/session-storage.test.ts b/packages/coding-agent/test/session-storage.test.ts index 27e5ed38b..3488837bc 100644 --- a/packages/coding-agent/test/session-storage.test.ts +++ b/packages/coding-agent/test/session-storage.test.ts @@ -101,8 +101,23 @@ describe("FileSessionStorage writer", () => { await fsp.rm(tempDir, { recursive: true, force: true }); }); + it("makes each append visible on disk without awaiting a microtask", () => { + const sessionPath = path.join(tempDir, "immediate.jsonl"); + const writer = storage.openWriter(sessionPath, { flags: "w" }); + // Contract: visibility must not depend on awaiting the returned Promise + // (or a microtask drain). appendSync / the sync body of append writes + // before return; awaiting alone would pass on the old microtask writer. + const appendSync = writer.appendSync?.bind(writer); + if (!appendSync) throw new Error("File writer must expose appendSync"); + appendSync("one\n"); + expect(fs.readFileSync(sessionPath, "utf8")).toBe("one\n"); + appendSync("two\n"); + expect(fs.readFileSync(sessionPath, "utf8")).toBe("one\ntwo\n"); + void writer.close(); + }); + it("preserves append order through flush and close", async () => { - const sessionPath = path.join(tempDir, "batched.jsonl"); + const sessionPath = path.join(tempDir, "ordered.jsonl"); const writer = storage.openWriter(sessionPath, { flags: "w" }); await writer.append("one\n"); await writer.append("two\n"); @@ -122,14 +137,17 @@ describe("FileSessionStorage writer", () => { expect(fs.readFileSync(sessionPath, "utf8")).toBe("one\ntwo\n"); }); - it("rejects close when draining a queued append fails", async () => { - const sessionPath = path.join(tempDir, "close-error.jsonl"); + it("rejects appendSync and append when the underlying write fails", async () => { + const sessionPath = path.join(tempDir, "append-error.jsonl"); const writer = storage.openWriter(sessionPath, { flags: "w" }); - void writer.append("one\n"); vi.spyOn(fs, "writeSync").mockImplementation(() => { throw new Error("disk full"); }); + const appendSync = writer.appendSync?.bind(writer); + if (!appendSync) throw new Error("File writer must expose appendSync"); + expect(() => appendSync("one\n")).toThrow("disk full"); + await expect(writer.append("two\n")).rejects.toThrow("disk full"); await expect(writer.close()).rejects.toThrow("disk full"); }); });