fix(coding-agent): build forks atomically and keep a header write's prefix

changes/unreleased/4305.md

@@ -0,0 +1,27 @@
+<!-- changes-log: packages/coding-agent/src/core/changes.md -->
+## 2026-10-26 - Fork files are built atomically and a header catch-up keeps its prefix (senpi#4295)
+
+### What changed
+
+- `packages/coding-agent/src/core/session-manager.ts`: `forkFrom` builds the fork through `replaceFileAtomically` (temp file, fsync, rename) instead of an exclusive header write followed by one `appendFileSync` per entry. The fresh timestamp and session id make collisions unlikely; the atomic rename replaces an existing target rather than rejecting it with `EEXIST`. When the post-close catch-up append of `_writeHeaderAsync` fails, the file is kept and cut back to the bytes known whole (`_keepCaughtUpPrefix`), and the missing entries are written, one line each, before the next append.
+- `session-write-state.ts` makes buffering, an in-flight header writer, and a flushed file mutually exclusive states. Discard, reload, new-session, branch and full-rewrite transitions replace the entire state, so no stale append cursor or torn-tail cause survives a new file. The header writer owns its queue across compaction instead of relying on the resident array's identity. The flushed state's `tailFailure` carries the failure's `SessionRecoveryCause`; a torn final line found on open whose backup the disk refused rides the buffered state as `deferredTail` until the load completes, then is set aside before the first append.
+- `session-write-recovery.ts`: `truncateToLastCompleteLine(path, cause, knownCompleteBytes?)` keeps the set-aside backup and recovery report for a torn tail, and accepts the exact durable byte boundary when a header catch-up's cut-back failed. Retrying cuts the whole failed batch, including any complete lines it landed, before replaying it; those bytes are written again from memory, so that cut keeps no backup and reports no recovery.
+- The transcript journal counts the file as persisted only when no catch-up entries are still owed (`_fileHoldsMemory`).
+- The atomic replacement helper removes its temporary file on a caught failure. A process crash before rename can still leave a hidden `.tmp` file; it is not a session and is not swept automatically, since another live writer may own it.
+
+### Why
+
+- A full disk part-way through `forkFrom` left a half-copied fork in the session list. A failed catch-up deleted the whole new file, including the header and entries already on disk. Independently tracked flush flags and pending entries could replay entries after discard/recreation, while compaction could strand a header-owned file behind repeated `EEXIST`.
+
+### Why an extension could not handle it
+
+- Both writes run inside `SessionManager`.
+
+### Expected merge conflict zones
+
+- MEDIUM: `packages/coding-agent/src/core/session-manager.ts`, `_writeHeaderAsync`, `_beginSessionFile` and the flushed append in `_persist`: the torn-tail cause and the deferred open-time tail live in `writeState`, not in separate fields.
+- LOW: `forkFrom` file creation; `truncateToLastCompleteLine`'s known-boundary branch.
+
+<!-- changes-log: packages/coding-agent/CHANGELOG.md -->
+### Fixed
+- A fork interrupted by a full disk leaves no partial session file behind, and a session whose first save was interrupted after its header was written keeps what was written and saves the rest next ([#4295](https://github.com/code-yeongyu/senpi/issues/4295))

packages/coding-agent/src/core/session-write-recovery.ts

@@ -57,30 +57,41 @@ export function backupTornBytes(path: string, bytes: Buffer, cause: SessionRecov
  * that failed part-way (ENOSPC) can leave a partial line, and the next append would be glued onto it, losing both
  * entries on reload. A file that already ends in a newline is left untouched and returns undefined; a refused copy
  * throws before anything is cut.
+ *
+ * A failed header catch-up instead supplies its exact durable boundary (`knownCompleteBytes`): every byte past it
+ * belongs to entries the writer still holds and writes again before its next append, whole lines included, so they
+ * are cut with no backup and no recovery to report.
  */
-export function truncateToLastCompleteLine(path: string, cause: SessionRecoveryCause): SessionRecovery | undefined {
+export function truncateToLastCompleteLine(
+	path: string,
+	cause: SessionRecoveryCause,
+	knownCompleteBytes?: number,
+): SessionRecovery | undefined {
 	const fd = openSync(path, "r+");
 	try {
 		const size = fstatSync(fd).size;
-		const chunk = Buffer.alloc(Math.min(SCAN_CHUNK_BYTES, size));
+		let completeLength = knownCompleteBytes;
+		let recovery: SessionRecovery | undefined;
+		const chunk = Buffer.alloc(completeLength === undefined ? Math.min(SCAN_CHUNK_BYTES, size) : 0);
 		const tail: Buffer[] = [];
-		let end = size;
+		let end = completeLength === undefined ? size : 0;
 		while (end > 0) {
 			const start = Math.max(0, end - chunk.length);
 			const read = readSync(fd, chunk, 0, end - start, start);
 			const newline = chunk.subarray(0, read).lastIndexOf(NEWLINE);
 			if (newline >= 0) {
-				const completeLength = start + newline + 1;
+				completeLength = start + newline + 1;
 				if (completeLength >= size) return undefined;
 				tail.unshift(Buffer.from(chunk.subarray(newline + 1, read)));
-				const recovery = backupTornBytes(path, Buffer.concat(tail), cause);
-				ftruncateSync(fd, completeLength);
-				return recovery;
+				recovery = backupTornBytes(path, Buffer.concat(tail), cause);
+				break;
 			}
 			tail.unshift(Buffer.from(chunk.subarray(0, read)));
 			end = start;
 		}
-		return undefined;
+		if (completeLength === undefined || completeLength >= size) return undefined;
+		ftruncateSync(fd, completeLength);
+		return recovery;
 	} finally {
 		closeSync(fd);
 	}

packages/coding-agent/src/core/session-write-state.ts

@@ -0,0 +1,24 @@
+import type { FileEntry } from "./session-manager.ts";
+import type { SessionRecoveryCause } from "./session-write-recovery.ts";
+
+/** A header writer owns its queue even when compaction replaces the resident mirror. */
+export interface HeaderWriteState {
+	readonly kind: "writing";
+	readonly file: string;
+	readonly entries: FileEntry[];
+}
+
+/**
+ * Buffering cannot retain a previous file's append cursor, ownership, or torn-tail cause. The one tail it carries is
+ * the opened file's own: a torn final line whose backup the disk refused (`deferredTail`), set aside before the
+ * first append once the load completes.
+ */
+export type SessionWriteState =
+	| { readonly kind: "buffered"; readonly deferredTail?: SessionRecoveryCause }
+	| HeaderWriteState
+	| {
+			readonly kind: "flushed";
+			readonly ownedHeaderFile?: string;
+			pending?: { readonly entries: FileEntry[]; from: number };
+			tailFailure?: { readonly cause: SessionRecoveryCause; readonly completeBytes?: number };
+	  };

packages/coding-agent/test/session-manager-header-write.test.ts

@@ -8,12 +8,28 @@ import { SessionManager } from "../src/core/session-manager.ts";
 const hooks = vi.hoisted(() => ({
 	beforeClose: undefined as (() => void) | undefined,
 	writeFails: undefined as Error | undefined,
+	appendFails: undefined as Error | undefined,
+	cutFails: false,
+	keepWholeLine: false,
 }));
 
 vi.mock("fs/promises", async (importOriginal) => {
 	const real = await importOriginal<typeof import("fs/promises")>();
 	return {
 		...real,
+		appendFile: async (...args: Parameters<typeof real.appendFile>) => {
+			const failure = hooks.appendFails;
+			if (failure === undefined || typeof args[1] !== "string") return real.appendFile(...args);
+			hooks.appendFails = undefined;
+			const keep = hooks.keepWholeLine ? args[1].indexOf("\n") + 18 : Math.floor(args[1].length / 2);
+			await real.appendFile(args[0], args[1].slice(0, keep));
+			throw failure;
+		},
+		truncate: async (...args: Parameters<typeof real.truncate>) => {
+			if (!hooks.cutFails) return real.truncate(...args);
+			hooks.cutFails = false;
+			throw Object.assign(new Error("EIO"), { code: "EIO" });
+		},
 		open: async (...args: Parameters<typeof real.open>) => {
 			const handle = await real.open(...args);
 			const writeFile = handle.writeFile.bind(handle);
@@ -39,6 +55,9 @@ const roots: string[] = [];
 afterEach(() => {
 	hooks.beforeClose = undefined;
 	hooks.writeFails = undefined;
+	hooks.appendFails = undefined;
+	hooks.cutFails = false;
+	hooks.keepWholeLine = false;
 	for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true });
 });
 
@@ -85,6 +104,61 @@ function idsOnDisk(file: string): string[] {
 }
 
 describe("persistHeaderNow's asynchronous header write", () => {
+	it("retries an exact failed cut-back before replaying a batch that landed whole lines too", async () => {
+		const { manager, file } = freshSession();
+		manager.appendCustomEntry("before");
+		hooks.beforeClose = () => {
+			hooks.beforeClose = undefined;
+			manager.appendCustomEntry("during-one");
+			manager.appendCustomEntry("during-two");
+			hooks.appendFails = Object.assign(new Error("ENOSPC"), { code: "ENOSPC" });
+			hooks.keepWholeLine = true;
+			hooks.cutFails = true;
+		};
+		await expect(manager.persistHeaderNow()).rejects.toMatchObject({ code: "ENOSPC" });
+		expect(readFileSync(file, "utf8")).toContain('"customType":"during-one"');
+		manager.appendCustomEntry("after");
+		expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id));
+	});
+
+	it.each([false, true])("keeps header ownership across compaction, catch-up failure=%s", async (fails) => {
+		const { manager, file } = freshSession();
+		const first = manager.appendCustomEntry("before");
+		let compaction = "";
+		hooks.beforeClose = () => {
+			hooks.beforeClose = undefined;
+			compaction = manager.appendCompaction("summary", first, 1);
+			if (fails) hooks.appendFails = Object.assign(new Error("ENOSPC"), { code: "ENOSPC" });
+		};
+		const writing = manager.persistHeaderNow();
+		if (fails) await expect(writing).rejects.toMatchObject({ code: "ENOSPC" });
+		else await writing;
+		expect(manager.isTranscriptFlushed()).toBe(true);
+		const next = manager.appendCustomEntry("after");
+		expect(idsOnDisk(file)).toEqual([first, compaction, next]);
+		expect(idsOnDisk(file)).toEqual(
+			SessionManager.open(file)
+				.getEntries()
+				.map((entry) => entry.id),
+		);
+	});
+
+	it("does not replay a failed catch-up after discarding and recreating its startup-only file", async () => {
+		const { manager, file } = freshSession();
+		manager.appendCustomEntry("before", {}, { startupBookkeeping: true });
+		hooks.beforeClose = () => {
+			hooks.beforeClose = undefined;
+			manager.appendCustomEntry("during-close", {}, { startupBookkeeping: true });
+			hooks.appendFails = Object.assign(new Error("ENOSPC"), { code: "ENOSPC" });
+		};
+		await expect(manager.persistHeaderNow()).rejects.toMatchObject({ code: "ENOSPC" });
+		expect(await manager.discardHeaderOnlyFile()).toBe(true);
+		expect(existsSync(file)).toBe(false);
+		manager.appendMessage({ role: "user", content: "first flush", timestamp: 1 });
+		manager.appendMessage(assistant("next append"));
+		expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id));
+	});
+
 	it("writes the entries persisted while its file handle closes before the transcript counts as flushed", async () => {
 		const { manager, file } = freshSession();
 		manager.appendCustomEntry("before", { n: 0 });
@@ -140,4 +214,33 @@ describe("persistHeaderNow's asynchronous header write", () => {
 		expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id));
 		expect(manager.getEntries()).toHaveLength(4);
 	});
+
+	it("keeps the header and entries already on disk when only the catch-up append fails, and writes the rest next", async () => {
+		// Given a header write whose catch-up append (entries persisted while its handle closed) runs out of space
+		const { manager, file } = freshSession();
+		manager.appendCustomEntry("before", { n: 0 });
+		hooks.beforeClose = () => {
+			hooks.beforeClose = undefined;
+			manager.appendCustomEntry("during-close", { n: 1 });
+			hooks.appendFails = Object.assign(new Error("ENOSPC: no space left on device, write"), { code: "ENOSPC" });
+		};
+
+		// When the header write settles and, once there is space, the next entry is appended
+		await expect(manager.persistHeaderNow()).rejects.toMatchObject({ code: "ENOSPC" });
+		const kept = existsSync(file)
+			? linesOnDisk(file).map((line) => (line.type === "session" ? "session" : line.customType))
+			: [];
+		manager.appendCustomEntry("after", { n: 2 });
+
+		// Then the written prefix stayed on disk, and the file then holds every entry memory holds, once, in order
+		expect(kept).toEqual(["session", "before"]);
+		expect(manager.isTranscriptFlushed()).toBe(true);
+		expect(linesOnDisk(file).map((line) => (line.type === "session" ? "session" : line.customType))).toEqual([
+			"session",
+			"before",
+			"during-close",
+			"after",
+		]);
+		expect(idsOnDisk(file)).toEqual(manager.getEntries().map((entry) => entry.id));
+	});
 });

packages/coding-agent/test/session-manager/fork-full-disk.test.ts

@@ -0,0 +1,131 @@
+import type { AssistantMessage } from "@earendil-works/pi-ai";
+import type * as FsModule from "fs";
+import { mkdtempSync, readdirSync, rmSync } from "fs";
+import { tmpdir } from "os";
+import { join } from "path";
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { SessionManager } from "../../src/core/session-manager.ts";
+
+/** A full disk part-way through creating a fork or a branch must leave no fork file in the session list (senpi#4295). */
+
+const fault = vi.hoisted(() => ({ budget: undefined as number | undefined }));
+
+function noSpace(): Error {
+	return Object.assign(new Error("ENOSPC: no space left on device, write"), { code: "ENOSPC" });
+}
+
+vi.mock("fs", async (importOriginal) => {
+	const actual = await importOriginal<typeof FsModule>();
+	/** Lands what fits in the remaining budget, then fails the write the way a full volume does. */
+	function limited<T extends unknown[]>(write: (target: T[0], data: string, ...rest: unknown[]) => void) {
+		return (target: T[0], data: unknown, ...rest: unknown[]) => {
+			if (fault.budget === undefined || typeof data !== "string") return write(target, data as string, ...rest);
+			if (data.length <= fault.budget) {
+				fault.budget -= data.length;
+				return write(target, data, ...rest);
+			}
+			const landed = data.slice(0, fault.budget);
+			fault.budget = undefined;
+			if (landed.length > 0) write(target, landed, ...rest);
+			throw noSpace();
+		};
+	}
+	return {
+		...actual,
+		appendFileSync: limited((target, data, ...rest) =>
+			actual.appendFileSync(target as FsModule.PathOrFileDescriptor, data, ...(rest as [])),
+		),
+		writeFileSync: limited((target, data, ...rest) =>
+			actual.writeFileSync(target as FsModule.PathOrFileDescriptor, data, ...(rest as [])),
+		),
+	};
+});
+
+const assistant = (text: string): AssistantMessage => ({
+	role: "assistant",
+	content: [{ type: "text", text }],
+	api: "anthropic-messages",
+	provider: "anthropic",
+	model: "test",
+	usage: {
+		input: 1,
+		output: 1,
+		cacheRead: 0,
+		cacheWrite: 0,
+		totalTokens: 2,
+		cost: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, total: 0 },
+	},
+	stopReason: "stop",
+	timestamp: 2,
+});
+
+const roots: string[] = [];
+
+afterEach(() => {
+	fault.budget = undefined;
+	for (const root of roots.splice(0)) rmSync(root, { recursive: true, force: true });
+});
+
+function savedSession(): { readonly root: string; readonly session: SessionManager; readonly file: string } {
+	const root = mkdtempSync(join(tmpdir(), "senpi-fork-full-"));
+	roots.push(root);
+	const session = SessionManager.create(root, join(root, "sessions"));
+	for (let turn = 0; turn < 3; turn++) {
+		session.appendMessage({ role: "user", content: `question ${turn} ${"x".repeat(200)}`, timestamp: 1 });
+		session.appendMessage(assistant(`answer ${turn} ${"y".repeat(200)}`));
+	}
+	const file = session.getSessionFile();
+	if (!file) throw new Error("test setup: persisted session has no file");
+	return { root, session, file };
+}
+
+function captureError(run: () => unknown): unknown {
+	try {
+		run();
+	} catch (error) {
+		return error;
+	}
+	return undefined;
+}
+
+describe("creating a session copy on a full disk", () => {
+	it("leaves no fork file behind when forkFrom runs out of space part-way", () => {
+		// Given a saved session and a target session directory
+		const { root, file } = savedSession();
+		const forkDir = join(root, "forks");
+		const failures: unknown[] = [];
+
+		// When the disk fills up after each possible amount of the fork has been written
+		for (const budget of [0, 150, 600, 1200]) {
+			fault.budget = budget;
+			failures.push(captureError(() => SessionManager.forkFrom(file, root, forkDir)));
+			fault.budget = undefined;
+		}
+
+		// Then every attempt reported the full disk and no partial fork (or temp file) is left in the directory
+		expect(failures.map((failure) => (failure as { code?: string } | undefined)?.code)).toEqual([
+			"ENOSPC",
+			"ENOSPC",
+			"ENOSPC",
+			"ENOSPC",
+		]);
+		expect(readdirSync(forkDir)).toEqual([]);
+	});
+
+	it("leaves no branch file behind when createBranchedSession runs out of space part-way", () => {
+		// Given a saved session
+		const { root, session } = savedSession();
+		const leaf = session.getLeafId();
+		if (!leaf) throw new Error("test setup: no leaf");
+		const before = readdirSync(join(root, "sessions")).sort();
+
+		// When the branch copy runs out of space part-way
+		fault.budget = 300;
+		const failure = captureError(() => session.createBranchedSession(leaf));
+		fault.budget = undefined;
+
+		// Then the failure surfaced and the session directory holds exactly what it held before
+		expect(failure).toMatchObject({ code: "ENOSPC" });
+		expect(readdirSync(join(root, "sessions")).sort()).toEqual(before);
+	});
+});