diff --git a/packages/agent/src/harness/env/nodejs.ts b/packages/agent/src/harness/env/nodejs.ts index 05c667309..1225efb92 100644 --- a/packages/agent/src/harness/env/nodejs.ts +++ b/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> { + const source = resolvePath(this.cwd, sourcePath); + const destination = resolvePath(this.cwd, destinationPath); + const aborted = abortResult(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> { const resolved = resolvePath(this.cwd, path); try { diff --git a/packages/agent/src/harness/session/jsonl/repo.ts b/packages/agent/src/harness/session/jsonl/repo.ts index b2fd6487d..5cde6f6c3 100644 --- a/packages/agent/src/harness/session/jsonl/repo.ts +++ b/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> { - 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> { - return this.openDirect(metadata); + return new Session(await this.loadStorage(metadata)); } list(): Promise; @@ -67,44 +68,15 @@ export class JsonlSessionRepo source: JsonlSessionMetadata, options: ForkOptions & JsonlSessionCreateOptions, ): Promise> { - 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> { + private async loadStorage(metadata: JsonlSessionMetadata): Promise { 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; - 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 { diff --git a/packages/agent/src/harness/session/jsonl/storage.ts b/packages/agent/src/harness/session/jsonl/storage.ts index 236dfddec..d9dca47af 100644 --- a/packages/agent/src/harness/session/jsonl/storage.ts +++ b/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, +): Promise { + 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 { private readonly fs: JsonlSessionRepoFileSystem; @@ -30,6 +56,16 @@ export class JsonlSessionStorage implements SessionStorage this.metadata = structuredClone(metadata); } + static async create( + fs: JsonlSessionRepoFileSystem, + path: string, + header: JsonlV4Header, + ): Promise { + 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 { 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 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 return storage; } + async fork(path: string, header: JsonlV4Header, options: ForkOptions): Promise { + 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 { await this.tail; } @@ -107,27 +157,6 @@ export class JsonlSessionStorage implements SessionStorage }); } - appendCopiedEntry(source: TEntry): Promise { - 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 { - 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(newRecord: NewRecord): Promise { return this.enqueue(async () => { this.state.requireLane(newRecord.lane); diff --git a/packages/agent/src/harness/session/jsonl/types.ts b/packages/agent/src/harness/session/jsonl/types.ts index f1d39a906..0483d9e55 100644 --- a/packages/agent/src/harness/session/jsonl/types.ts +++ b/packages/agent/src/harness/session/jsonl/types.ts @@ -8,6 +8,7 @@ export type JsonlSessionRepoFileSystem = Pick< | "readTextFile" | "writeFile" | "appendFile" + | "renameFile" | "fileInfo" | "listDir" | "exists" diff --git a/packages/agent/src/harness/session/memory.ts b/packages/agent/src/harness/session/memory.ts index 5ec394737..598d8dcf4 100644 --- a/packages/agent/src/harness/session/memory.ts +++ b/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; } diff --git a/packages/agent/src/harness/session/state.ts b/packages/agent/src/harness/session/state.ts index 0a3047281..edc16f85b 100644 --- a/packages/agent/src/harness/session/state.ts +++ b/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, diff --git a/packages/agent/src/harness/types.ts b/packages/agent/src/harness/types.ts index 2e1062ef9..7fb03f3ed 100644 --- a/packages/agent/src/harness/types.ts +++ b/packages/agent/src/harness/types.ts @@ -249,6 +249,8 @@ export interface FileSystem { writeFile(path: string, content: string | Uint8Array, abortSignal?: AbortSignal): Promise>; /** Create or append to a file, creating parent directories when supported. */ appendFile(path: string, content: string | Uint8Array, abortSignal?: AbortSignal): Promise>; + /** Atomically rename a file, replacing the destination when it exists. Does not copy across filesystems. */ + renameFile(sourcePath: string, destinationPath: string, abortSignal?: AbortSignal): Promise>; /** Return metadata for the addressed path without following symlinks. */ fileInfo(path: string, abortSignal?: AbortSignal): Promise>; /** List direct children of a directory without following symlinks. */ diff --git a/packages/agent/test/harness/nodejs-env.test.ts b/packages/agent/test/harness/nodejs-env.test.ts index 6ff80b20e..09b706516 100644 --- a/packages/agent/test/harness/nodejs-env.test.ts +++ b/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) { diff --git a/packages/agent/test/harness/session/jsonl-storage.test.ts b/packages/agent/test/harness/session/jsonl-storage.test.ts index d282731c0..6b971dd6e 100644 --- a/packages/agent/test/harness/session/jsonl-storage.test.ts +++ b/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 }]); + }); }); diff --git a/packages/agent/test/harness/session/jsonl.test.ts b/packages/agent/test/harness/session/jsonl.test.ts index f0b97af5a..80971a0e0 100644 --- a/packages/agent/test/harness/session/jsonl.test.ts +++ b/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[]): 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); + }); });