harness-v2 jsonl session atomic writes for forks & torn-tail truncation (#7707)

* fix(agent): make JSONL publication crash-safe

* fix(agent): move JSONL file creation from repo to storage

* fix(agent): move JSONL forking from repo to storage

* fix(agent): move fork mutation creation into session state

packages/agent/src/harness/env/nodejs.ts

@@ -10,6 +10,7 @@ import {
 	readdir,
 	readFile,
 	realpath,
+	rename,
 	rm,
 	writeFile,
 } from "node:fs/promises";
@@ -93,12 +94,14 @@ function isNodeError(error: unknown): error is NodeJS.ErrnoException {
 	return error instanceof Error && "code" in error;
 }
 
-function toFileError(error: unknown, path?: string): FileError {
+function toFileError(error: unknown, fallbackPath?: string): FileError {
 	if (error instanceof FileError) return error;
 	const cause = toError(error);
-	if (isNodeError(error)) {
-		const message = error.message;
-		switch (error.code) {
+	const nodeError = isNodeError(error) ? error : undefined;
+	const path = typeof nodeError?.path === "string" ? nodeError.path : fallbackPath;
+	if (nodeError) {
+		const message = nodeError.message;
+		switch (nodeError.code) {
 			case "ABORT_ERR":
 				return new FileError("aborted", message, path, cause);
 			case "ENOENT":
@@ -579,6 +582,23 @@ export class NodeExecutionEnv implements ExecutionEnv {
 		}
 	}
 
+	async renameFile(
+		sourcePath: string,
+		destinationPath: string,
+		abortSignal?: AbortSignal,
+	): Promise<Result<void, FileError>> {
+		const source = resolvePath(this.cwd, sourcePath);
+		const destination = resolvePath(this.cwd, destinationPath);
+		const aborted = abortResult<void>(abortSignal, destination);
+		if (aborted) return aborted;
+		try {
+			await rename(source, destination);
+			return ok(undefined);
+		} catch (error) {
+			return err(toFileError(error, source));
+		}
+	}
+
 	async fileInfo(path: string): Promise<Result<FileInfo, FileError>> {
 		const resolved = resolvePath(this.cwd, path);
 		try {

packages/agent/src/harness/session/jsonl/repo.ts

@@ -1,7 +1,7 @@
 import { uuidv7 } from "@earendil-works/pi-ai";
 import { assertJsonSerializable, Session } from "../session.ts";
-import { type Entry, type ForkOptions, type LanePointer, SessionError, type SessionRepo } from "../types.ts";
-import { encodeHeader, metadataFromHeader, parseHeader } from "./codec.ts";
+import { type ForkOptions, SessionError, type SessionRepo } from "../types.ts";
+import { metadataFromHeader, parseHeader } from "./codec.ts";
 import { fileResult, invalidFile } from "./errors.ts";
 import { JsonlSessionStorage } from "./storage.ts";
 import type {
@@ -46,11 +46,12 @@ export class JsonlSessionRepo
 	}
 
 	async create(options: JsonlSessionCreateOptions): Promise<Session<JsonlSessionMetadata>> {
-		return (await this.createDirect(options)).session;
+		const { header, path } = await this.prepareCreate(options);
+		return new Session(await JsonlSessionStorage.create(this.fs, path, header));
 	}
 
 	async open(metadata: JsonlSessionMetadata): Promise<Session<JsonlSessionMetadata>> {
-		return this.openDirect(metadata);
+		return new Session(await this.loadStorage(metadata));
 	}
 
 	list(): Promise<JsonlSessionMetadata[]>;
@@ -67,44 +68,15 @@ export class JsonlSessionRepo
 		source: JsonlSessionMetadata,
 		options: ForkOptions & JsonlSessionCreateOptions,
 	): Promise<Session<JsonlSessionMetadata>> {
-		const sourceSession = await this.openDirect(source);
-		let copiedEntries: Entry[];
-		let forkLanes: LanePointer[];
-		if (options.scope === "tree") {
-			copiedEntries = await sourceSession.findEntries({ order: "oldestFirst" });
-			forkLanes = await sourceSession.getLanes();
-		} else {
-			const selectedEntryId = options.entryId ?? (await sourceSession.getLeafId());
-			let targetId: string | null = null;
-			if (selectedEntryId !== null) {
-				const entry = await sourceSession.getEntry(selectedEntryId);
-				if (!entry || entry.type !== "message") {
-					throw new SessionError("invalid_fork_target", `Fork target is not a message entry: ${selectedEntryId}`);
-				}
-				const position = options.position ?? (options.entryId === undefined ? "at" : "before");
-				targetId = position === "at" ? entry.id : entry.parentId;
-			}
-			copiedEntries =
-				targetId === null ? [] : await sourceSession.findEntriesOnBranch({ start: targetId, order: "oldestFirst" });
-			forkLanes = [{ lane: "main", leafId: targetId }];
-		}
-
-		const { session: target, storage: targetStorage } = await this.createDirect({
+		const sourceStorage = await this.loadStorage(source);
+		const { header, path } = await this.prepareCreate({
 			...options,
 			parentSessionId: options.parentSessionId ?? source.id,
 		});
-		for (const entry of copiedEntries) await targetStorage.appendCopiedEntry(entry);
-		for (const pointer of forkLanes) await targetStorage.appendForkLane(pointer.lane, pointer.leafId);
-		const name = await sourceSession.getName();
-		if (name !== undefined) await target.setName(name);
-		for (const entry of copiedEntries) {
-			const label = await sourceSession.getLabel(entry.id);
-			if (label !== undefined) await target.setLabel(entry.id, label);
-		}
-		return target;
+		return new Session(await sourceStorage.fork(path, header, options));
 	}
 
-	private async openDirect(metadata: JsonlSessionMetadata): Promise<Session<JsonlSessionMetadata>> {
+	private async loadStorage(metadata: JsonlSessionMetadata): Promise<JsonlSessionStorage> {
 		if (!fileResult(await this.fs.exists(metadata.path), `Failed to check session ${metadata.path}`)) {
 			throw new SessionError("not_found", `Session not found: ${metadata.id}`);
 		}
@@ -113,12 +85,12 @@ export class JsonlSessionRepo
 		if (loadedMetadata.id !== metadata.id) {
 			throw new SessionError("invalid_entry", `Session id does not match header: ${metadata.id}`);
 		}
-		return new Session(storage);
+		return storage;
 	}
 
-	private async createDirect(options: JsonlSessionCreateOptions): Promise<{
-		session: Session<JsonlSessionMetadata>;
-		storage: JsonlSessionStorage;
+	private async prepareCreate(options: JsonlSessionCreateOptions): Promise<{
+		header: JsonlV4Header;
+		path: string;
 	}> {
 		const id = options.id ?? uuidv7();
 		validateSessionId(id);
@@ -144,10 +116,7 @@ export class JsonlSessionRepo
 			metadata: options.metadata,
 		};
 		fileResult(await this.fs.createDir(sessionDirectory, { recursive: true }), `Failed to create sessions directory`);
-		fileResult(await this.fs.writeFile(path, encodeHeader(header)), `Failed to create session ${path}`);
-		const fileInfo = fileResult(await this.fs.fileInfo(path), `Failed to read session metadata ${path}`);
-		const storage = new JsonlSessionStorage(this.fs, metadataFromHeader(header, path, fileInfo.mtimeMs));
-		return { session: new Session(storage), storage };
+		return { header, path };
 	}
 
 	private async listDirect(options: JsonlSessionListOptions): Promise<JsonlSessionMetadata[]> {

packages/agent/src/harness/session/jsonl/storage.ts

@@ -3,6 +3,7 @@ import {
 	type BranchBounds,
 	type Entry,
 	type EntryQuery,
+	type ForkOptions,
 	type LanePointer,
 	type LaneRecord,
 	type LogItem,
@@ -15,9 +16,34 @@ import {
 	type SessionStats,
 	type SessionStorage,
 } from "../types.ts";
-import { encodeMutation, metadataFromHeader, parseHeader, parseMutation } from "./codec.ts";
+import { encodeHeader, encodeMutation, metadataFromHeader, parseHeader, parseMutation } from "./codec.ts";
 import { fileResult, invalidFile } from "./errors.ts";
-import type { JsonlSessionMetadata, JsonlSessionRepoFileSystem } from "./types.ts";
+import type { JsonlSessionMetadata, JsonlSessionRepoFileSystem, JsonlV4Header } from "./types.ts";
+
+/**
+ * Build a complete sibling temporary file, then atomically rename it over the destination.
+ * The populate callback must create or overwrite `tempPath` with the complete file. The
+ * destination is untouched until the rename commits, so a process crash while populating
+ * can leave only the ignored `.tmp` file behind.
+ *
+ * Rejects when population or rename fails. On rejection, temporary-file removal is
+ * best-effort and the original error is preserved. Callers must serialize publications to
+ * the same destination because they share its deterministic `.tmp` path.
+ */
+async function publishFileAtomically(
+	fs: JsonlSessionRepoFileSystem,
+	destinationPath: string,
+	populate: (tempPath: string) => Promise<void>,
+): Promise<void> {
+	const tempPath = `${destinationPath}.tmp`;
+	try {
+		await populate(tempPath);
+		fileResult(await fs.renameFile(tempPath, destinationPath), `Failed to publish staged file ${destinationPath}`);
+	} catch (error) {
+		await fs.remove(tempPath, { force: true });
+		throw error;
+	}
+}
 
 export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata> {
 	private readonly fs: JsonlSessionRepoFileSystem;
@@ -30,6 +56,16 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
 		this.metadata = structuredClone(metadata);
 	}
 
+	static async create(
+		fs: JsonlSessionRepoFileSystem,
+		path: string,
+		header: JsonlV4Header,
+	): Promise<JsonlSessionStorage> {
+		fileResult(await fs.writeFile(path, encodeHeader(header)), `Failed to initialize session ${path}`);
+		const fileInfo = fileResult(await fs.fileInfo(path), `Failed to read session metadata ${path}`);
+		return new JsonlSessionStorage(fs, metadataFromHeader(header, path, fileInfo.mtimeMs));
+	}
+
 	static async load(fs: JsonlSessionRepoFileSystem, path: string): Promise<JsonlSessionStorage> {
 		const content = fileResult(await fs.readTextFile(path), `Failed to read session ${path}`);
 		const physicalLines = content.split("\n");
@@ -47,7 +83,9 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
 				if (index !== physicalLines.length - 1 || !(error instanceof SessionError) || error.cause === undefined)
 					throw error;
 				const validPrefix = `${physicalLines.slice(0, index).join("\n")}\n`;
-				fileResult(await fs.writeFile(path, validPrefix), `Failed to truncate torn session tail ${path}`);
+				await publishFileAtomically(fs, path, async (tempPath) => {
+					fileResult(await fs.writeFile(tempPath, validPrefix), `Failed to stage torn-tail repair ${path}`);
+				});
 				return storage;
 			}
 			storage.applyMutation(mutation, path, index + 1);
@@ -58,6 +96,18 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
 		return storage;
 	}
 
+	async fork(path: string, header: JsonlV4Header, options: ForkOptions): Promise<JsonlSessionStorage> {
+		const mutations = this.state.createForkMutations(options);
+		await publishFileAtomically(this.fs, path, async (tempPath) => {
+			const targetStorage = await JsonlSessionStorage.create(this.fs, tempPath, header);
+			for (const mutation of mutations) {
+				await targetStorage.appendMutation(mutation);
+				targetStorage.applyMutation(mutation);
+			}
+		});
+		return JsonlSessionStorage.load(this.fs, path);
+	}
+
 	async drain(): Promise<void> {
 		await this.tail;
 	}
@@ -107,27 +157,6 @@ export class JsonlSessionStorage implements SessionStorage<JsonlSessionMetadata>
 		});
 	}
 
-	appendCopiedEntry<TEntry extends Entry>(source: TEntry): Promise<TEntry> {
-		return this.enqueue(async () => {
-			this.state.validateUnusedId(source.id);
-			this.state.validateTarget(source.parentId);
-			const entry = { ...structuredClone(source), seq: this.state.nextSequence };
-			const mutation: SessionMutation = { kind: "entry", entry };
-			await this.appendMutation(mutation);
-			this.applyMutation(mutation);
-			return structuredClone(entry);
-		});
-	}
-
-	appendForkLane(lane: string, leafId: string | null): Promise<void> {
-		return this.enqueue(async () => {
-			this.state.validateTarget(leafId);
-			const mutation: SessionMutation = { kind: "lane", seq: this.state.nextSequence, lane, leafId };
-			await this.appendMutation(mutation);
-			this.applyMutation(mutation);
-		});
-	}
-
 	appendRecord<TRecord extends LaneRecord>(newRecord: NewRecord<TRecord>): Promise<TRecord> {
 		return this.enqueue(async () => {
 			this.state.requireLane(newRecord.lane);

packages/agent/src/harness/session/jsonl/types.ts

@@ -8,6 +8,7 @@ export type JsonlSessionRepoFileSystem = Pick<
 	| "readTextFile"
 	| "writeFile"
 	| "appendFile"
+	| "renameFile"
 	| "fileInfo"
 	| "listDir"
 	| "exists"

packages/agent/src/harness/session/memory.ts

@@ -32,54 +32,7 @@ export class InMemorySessionStorage implements SessionStorage {
 
 	fork(metadata: SessionMetadata, options: ForkOptions & SessionCreateOptions): InMemorySessionStorage {
 		const storage = new InMemorySessionStorage(metadata);
-		let copiedEntries: Entry[];
-		let forkLanes: LanePointer[];
-		if (options.scope === "tree") {
-			copiedEntries = this.state.findEntries({ order: "oldestFirst" });
-			forkLanes = this.state.getLanes();
-		} else {
-			const selectedEntryId = options.entryId ?? this.state.requireLane("main");
-			let targetId: string | null = null;
-			if (selectedEntryId !== null) {
-				const target = this.state.getEntry(selectedEntryId);
-				if (!target || target.type !== "message") {
-					throw new SessionError("invalid_fork_target", `Fork target is not a message entry: ${selectedEntryId}`);
-				}
-				const position = options.position ?? (options.entryId === undefined ? "at" : "before");
-				targetId = position === "at" ? target.id : target.parentId;
-			}
-			copiedEntries =
-				targetId === null ? [] : this.state.findEntriesOnBranch({ start: targetId, order: "oldestFirst" });
-			forkLanes = [{ lane: "main", leafId: targetId }];
-		}
-
-		for (const sourceEntry of copiedEntries) {
-			const entry = { ...structuredClone(sourceEntry), seq: storage.state.nextSequence };
-			storage.state.applyMutation({ kind: "entry", entry });
-		}
-		for (const pointer of forkLanes) {
-			storage.state.applyMutation({
-				kind: "lane",
-				seq: storage.state.nextSequence,
-				lane: pointer.lane,
-				leafId: pointer.leafId,
-			});
-		}
-		const name = this.state.getName();
-		if (name !== undefined) {
-			storage.state.applyMutation({ kind: "fact", seq: storage.state.nextSequence, fact: "name", name });
-		}
-		for (const entry of copiedEntries) {
-			const label = this.state.getLabel(entry.id);
-			if (label === undefined) continue;
-			storage.state.applyMutation({
-				kind: "fact",
-				seq: storage.state.nextSequence,
-				fact: "label",
-				targetId: entry.id,
-				label,
-			});
-		}
+		for (const mutation of this.state.createForkMutations(options)) storage.state.applyMutation(mutation);
 		return storage;
 	}
 

packages/agent/src/harness/session/state.ts

@@ -3,6 +3,7 @@ import {
 	type Entry,
 	type EntryOrder,
 	type EntryQuery,
+	type ForkOptions,
 	type LanePointer,
 	type LaneRecord,
 	type LogItem,
@@ -256,6 +257,47 @@ export class SessionState {
 		return this.stats;
 	}
 
+	createForkMutations(options: ForkOptions): SessionMutation[] {
+		let copiedEntries: Entry[];
+		let forkLanes: LanePointer[];
+		if (options.scope === "tree") {
+			copiedEntries = this.findEntries({ order: "oldestFirst" });
+			forkLanes = this.getLanes();
+		} else {
+			const selectedEntryId = options.entryId ?? this.requireLane("main");
+			let targetId: string | null = null;
+			if (selectedEntryId !== null) {
+				const entry = this.getEntry(selectedEntryId);
+				if (!entry || entry.type !== "message") {
+					throw new SessionError("invalid_fork_target", `Fork target is not a message entry: ${selectedEntryId}`);
+				}
+				const position = options.position ?? (options.entryId === undefined ? "at" : "before");
+				targetId = position === "at" ? entry.id : entry.parentId;
+			}
+			copiedEntries = targetId === null ? [] : this.findEntriesOnBranch({ start: targetId, order: "oldestFirst" });
+			forkLanes = [{ lane: "main", leafId: targetId }];
+		}
+
+		const mutations: SessionMutation[] = [];
+		let sequence = 1;
+		for (const sourceEntry of copiedEntries) {
+			mutations.push({ kind: "entry", entry: { ...structuredClone(sourceEntry), seq: sequence++ } });
+		}
+		for (const pointer of forkLanes) {
+			mutations.push({ kind: "lane", seq: sequence++, lane: pointer.lane, leafId: pointer.leafId });
+		}
+		if (this.name !== undefined) {
+			mutations.push({ kind: "fact", seq: sequence++, fact: "name", name: this.name });
+		}
+		for (const entry of copiedEntries) {
+			const label = this.labels.get(entry.id);
+			if (label !== undefined) {
+				mutations.push({ kind: "fact", seq: sequence++, fact: "label", targetId: entry.id, label });
+			}
+		}
+		return mutations;
+	}
+
 	private *walkToRoot(
 		start: string | null,
 		bounds?: Pick<BranchBounds, "stopAtId" | "stopAtType">,

packages/agent/src/harness/types.ts

@@ -249,6 +249,8 @@ export interface FileSystem {
 	writeFile(path: string, content: string | Uint8Array, abortSignal?: AbortSignal): Promise<Result<void, FileError>>;
 	/** Create or append to a file, creating parent directories when supported. */
 	appendFile(path: string, content: string | Uint8Array, abortSignal?: AbortSignal): Promise<Result<void, FileError>>;
+	/** Atomically rename a file, replacing the destination when it exists. Does not copy across filesystems. */
+	renameFile(sourcePath: string, destinationPath: string, abortSignal?: AbortSignal): Promise<Result<void, FileError>>;
 	/** Return metadata for the addressed path without following symlinks. */
 	fileInfo(path: string, abortSignal?: AbortSignal): Promise<Result<FileInfo, FileError>>;
 	/** List direct children of a directory without following symlinks. */

packages/agent/test/harness/nodejs-env.test.ts

@@ -193,6 +193,35 @@ describe("NodeExecutionEnv", () => {
 		expect(getOrThrow(await env.readTextFile("new/nested/file.txt"))).toBe("ab");
 	});
 
+	it("atomically renames a file and replaces the destination", async () => {
+		const root = createTempDir();
+		const env = new NodeExecutionEnv({ cwd: root });
+		getOrThrow(await env.writeFile("source.txt", "new"));
+		getOrThrow(await env.writeFile("destination.txt", "old"));
+
+		getOrThrow(await env.renameFile("source.txt", "destination.txt"));
+
+		expect(getOrThrow(await env.exists("source.txt"))).toBe(false);
+		expect(getOrThrow(await env.readTextFile("destination.txt"))).toBe("new");
+	});
+
+	it("reports the source path when rename fails because the source is missing", async () => {
+		const root = createTempDir();
+		const env = new NodeExecutionEnv({ cwd: root });
+		getOrThrow(await env.writeFile("destination.txt", "unchanged"));
+
+		const result = await env.renameFile("missing-source.txt", "destination.txt");
+
+		expect(result).toMatchObject({
+			ok: false,
+			error: {
+				code: "not_found",
+				path: join(root, "missing-source.txt"),
+			},
+		});
+		expect(getOrThrow(await env.readTextFile("destination.txt"))).toBe("unchanged");
+	});
+
 	it("creates temporary directories and files", async () => {
 		const root = createTempDir();
 		const env = new NodeExecutionEnv({ cwd: root });
@@ -234,6 +263,7 @@ describe("NodeExecutionEnv", () => {
 			env.readTextLines("file.txt", { abortSignal: signal }),
 			env.readBinaryFile("file.txt", signal),
 			env.writeFile("other.txt", "hello", signal),
+			env.renameFile("file.txt", "renamed.txt", signal),
 			env.listDir(".", signal),
 		]);
 		for (const result of results) {

packages/agent/test/harness/session/jsonl-storage.test.ts

@@ -2,7 +2,7 @@ import { mkdtempSync, readFileSync, rmSync } from "node:fs";
 import { tmpdir } from "node:os";
 import { join } from "node:path";
 import type { Usage } from "@earendil-works/pi-ai";
-import { afterEach, describe, expect, it } from "vitest";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import { NodeExecutionEnv } from "../../../src/harness/env/nodejs.ts";
 import type {
 	Entry,
@@ -13,6 +13,7 @@ import type {
 	Session,
 } from "../../../src/harness/session/index.ts";
 import { JsonlSessionRepo } from "../../../src/harness/session/index.ts";
+import { FileError } from "../../../src/harness/types.ts";
 import type { AgentMessage } from "../../../src/types.ts";
 
 const tempDirs: string[] = [];
@@ -474,4 +475,23 @@ describe("JSONL v4 per-session storage", () => {
 		expect(valid.seq).toBe(1);
 		expect((await reopen(root, restored)).getEntry("valid")).resolves.toEqual(valid);
 	});
+
+	it("does not advance state or poison the write queue after an append failure", async () => {
+		const root = createTempDir();
+		const env = new NodeExecutionEnv({ cwd: root });
+		vi.spyOn(env, "appendFile").mockResolvedValueOnce({
+			ok: false,
+			error: new FileError("unknown", "injected append failure"),
+		});
+		const repository = new JsonlSessionRepo({ fs: env, sessionsRoot: root });
+		const session = await repository.create({ id: "append-failure", cwd: root });
+
+		await expect(session.appendCustomEntry("rejected")).rejects.toMatchObject({ code: "storage" });
+		expect(await session.getLog()).toEqual([]);
+		const committed = await session.appendEntry({ type: "custom", id: "committed", customType: "note" }, "main");
+		expect(committed.seq).toBe(1);
+
+		const reopened = await createRepository(root).open(await session.getMetadata());
+		expect(await reopened.getLog()).toEqual([{ kind: "entry", seq: 1, entry: committed }]);
+	});
 });

packages/agent/test/harness/session/jsonl.test.ts

@@ -1,9 +1,19 @@
-import { appendFileSync, mkdtempSync, readFileSync, rmSync, statSync, utimesSync, writeFileSync } from "node:fs";
+import {
+	appendFileSync,
+	existsSync,
+	mkdtempSync,
+	readdirSync,
+	readFileSync,
+	rmSync,
+	statSync,
+	utimesSync,
+	writeFileSync,
+} from "node:fs";
 import { tmpdir } from "node:os";
-import { join } from "node:path";
-import { afterEach, describe, expect, it } from "vitest";
+import { dirname, join } from "node:path";
+import { afterEach, describe, expect, it, vi } from "vitest";
 import { NodeExecutionEnv } from "../../../src/harness/env/nodejs.ts";
-import { JsonlSessionRepo, type SessionRepo } from "../../../src/harness/session/index.ts";
+import { type JsonlSessionMetadata, JsonlSessionRepo, type SessionRepo } from "../../../src/harness/session/index.ts";
 import {
 	createSessionBackendConformance,
 	type SessionBackendFixture,
@@ -47,6 +57,21 @@ function expectedSessionPath(root: string, cwd: string, createdAt: number, id: s
 	return join(root, directory, `${timestamp}_${id}.jsonl`);
 }
 
+function writeRawSession(root: string, id: string, mutations: Record<string, unknown>[]): JsonlSessionMetadata {
+	const path = join(root, `${id}.jsonl`);
+	const createdAt = 1;
+	const header = { kind: "header", version: 4, id, createdAt, cwd: root };
+	writeFileSync(path, `${[header, ...mutations].map((line) => JSON.stringify(line)).join("\n")}\n`);
+	return {
+		id,
+		createdAt,
+		cwd: root,
+		path,
+		modifiedAt: statSync(path).mtimeMs,
+		sourceFormat: 4,
+	};
+}
+
 afterEach(() => {
 	while (tempDirs.length > 0) rmSync(tempDirs.pop()!, { recursive: true, force: true });
 });
@@ -259,6 +284,50 @@ describe("JSONL v4 persistence", () => {
 		expect(await reopened.findRecords()).toEqual([]);
 	});
 
+	it("does not publish a partial fork when staging fails", async () => {
+		const root = createTempDir();
+		const env = new NodeExecutionEnv({ cwd: root });
+		const repository = new JsonlSessionRepo({ fs: env, sessionsRoot: root });
+		const source = await repository.create({ id: "source", cwd: root });
+		await source.appendMessage({ role: "user", content: [{ type: "text", text: "one" }], timestamp: 1 });
+		await source.appendMessage({ role: "user", content: [{ type: "text", text: "two" }], timestamp: 2 });
+		const sourceMetadata = await source.getMetadata();
+		const appendFile = env.appendFile.bind(env);
+		vi.spyOn(env, "appendFile")
+			.mockImplementationOnce(appendFile)
+			.mockResolvedValueOnce({
+				ok: false,
+				error: new FileError("unknown", "injected staging failure"),
+			});
+
+		await expect(repository.fork(sourceMetadata, { id: "fork", cwd: root })).rejects.toMatchObject({
+			code: "storage",
+		});
+
+		expect((await repository.list()).map((metadata) => metadata.id)).toEqual(["source"]);
+		expect(readdirSync(dirname(sourceMetadata.path)).filter((name) => name.endsWith(".tmp"))).toEqual([]);
+	});
+
+	it("does not publish a fork when atomic rename fails", async () => {
+		const root = createTempDir();
+		const env = new NodeExecutionEnv({ cwd: root });
+		const repository = new JsonlSessionRepo({ fs: env, sessionsRoot: root });
+		const source = await repository.create({ id: "source", cwd: root });
+		await source.appendMessage({ role: "user", content: [{ type: "text", text: "one" }], timestamp: 1 });
+		const sourceMetadata = await source.getMetadata();
+		vi.spyOn(env, "renameFile").mockResolvedValueOnce({
+			ok: false,
+			error: new FileError("unknown", "injected rename failure"),
+		});
+
+		await expect(repository.fork(sourceMetadata, { id: "fork", cwd: root })).rejects.toMatchObject({
+			code: "storage",
+		});
+
+		expect((await repository.list()).map((metadata) => metadata.id)).toEqual(["source"]);
+		expect(readdirSync(dirname(sourceMetadata.path)).filter((name) => name.endsWith(".tmp"))).toEqual([]);
+	});
+
 	it("repairs a valid final line missing its newline", async () => {
 		const root = createTempDir();
 		const repository = createRepository(root);
@@ -432,4 +501,150 @@ describe("JSONL v4 persistence", () => {
 		const moved = await movedRepository.open(metadata);
 		expect(await moved.getLeafId()).toBe("imported");
 	});
+
+	it.each([
+		{
+			name: "a non-consecutive sequence",
+			message: "non-consecutive seq",
+			mutations: [
+				{ kind: "entry", type: "custom", id: "entry", customType: "note", parentId: null, seq: 2, timestamp: 1 },
+			],
+		},
+		{
+			name: "a duplicate entry/record id",
+			message: "duplicate id",
+			mutations: [
+				{
+					kind: "entry",
+					type: "custom",
+					id: "duplicate",
+					customType: "note",
+					parentId: null,
+					seq: 1,
+					timestamp: 1,
+				},
+				{
+					kind: "record",
+					type: "operation_started",
+					id: "duplicate",
+					lane: "main",
+					seq: 2,
+					timestamp: 2,
+					sourceLeafId: null,
+					intent: { kind: "run", originalPrompt: [], initialMessages: [] },
+				},
+			],
+		},
+		{
+			name: "an entry with a missing parent",
+			message: "missing parent",
+			mutations: [
+				{
+					kind: "entry",
+					type: "custom",
+					id: "entry",
+					customType: "note",
+					parentId: "missing",
+					seq: 1,
+					timestamp: 1,
+				},
+			],
+		},
+		{
+			name: "an entry referencing a missing lane",
+			message: "missing lane",
+			mutations: [
+				{
+					kind: "entry",
+					lane: "thread",
+					type: "custom",
+					id: "entry",
+					customType: "note",
+					parentId: null,
+					seq: 1,
+					timestamp: 1,
+				},
+			],
+		},
+		{
+			name: "a record referencing a missing lane",
+			message: "missing lane",
+			mutations: [
+				{
+					kind: "record",
+					type: "operation_started",
+					id: "run",
+					lane: "thread",
+					seq: 1,
+					timestamp: 1,
+					sourceLeafId: null,
+					intent: { kind: "run", originalPrompt: [], initialMessages: [] },
+				},
+			],
+		},
+		{
+			name: "a lane move referencing a missing entry",
+			message: "missing lane target",
+			mutations: [{ kind: "lane", lane: "thread", leafId: "missing", seq: 1 }],
+		},
+		{
+			name: "a label referencing a missing entry",
+			message: "missing label target",
+			mutations: [{ kind: "fact", fact: "label", targetId: "missing", label: "checkpoint", seq: 1 }],
+		},
+	])("rejects $name during replay", async ({ name, message, mutations }) => {
+		const root = createTempDir();
+		const metadata = writeRawSession(root, name.replace(/[^A-Za-z0-9._-]/g, "-"), mutations);
+
+		await expect(createRepository(root).open(metadata)).rejects.toMatchObject({
+			code: "invalid_entry",
+			message: expect.stringContaining(message),
+		});
+	});
+
+	it("rejects a complete malformed interior mutation without modifying the file", async () => {
+		const root = createTempDir();
+		const metadata = writeRawSession(root, "malformed-interior", [
+			{
+				kind: "record",
+				type: "operation_started",
+				id: "run",
+				lane: "main",
+				seq: 1,
+				timestamp: 1,
+				sourceLeafId: null,
+			},
+			{ kind: "fact", fact: "name", name: "after", seq: 2 },
+		]);
+		const corrupted = readFileSync(metadata.path, "utf8");
+
+		await expect(createRepository(root).open(metadata)).rejects.toMatchObject({ code: "invalid_entry" });
+		expect(readFileSync(metadata.path, "utf8")).toBe(corrupted);
+	});
+
+	it("preserves the session when staging torn-tail repair fails", async () => {
+		const root = createTempDir();
+		const repository = createRepository(root);
+		const session = await repository.create({ id: "repair-failure", cwd: root });
+		const metadata = await session.getMetadata();
+		await session.appendCustomEntry("kept");
+		appendFileSync(metadata.path, '{"kind":"entry"');
+		const original = readFileSync(metadata.path, "utf8");
+
+		const env = new NodeExecutionEnv({ cwd: root });
+		const writeFile = env.writeFile.bind(env);
+		vi.spyOn(env, "writeFile").mockImplementationOnce(async (path: string) => {
+			const damaged = await writeFile(path, "");
+			if (!damaged.ok) return damaged;
+			return {
+				ok: false,
+				error: new FileError("unknown", "repair interrupted after truncation", path),
+			};
+		});
+		const failingRepository = new JsonlSessionRepo({ fs: env, sessionsRoot: root });
+
+		await expect(failingRepository.open(metadata)).rejects.toMatchObject({ code: "storage" });
+		expect(readFileSync(metadata.path, "utf8")).toBe(original);
+		expect(existsSync(`${metadata.path}.tmp`)).toBe(false);
+	});
 });