From d2e6cfe10929eea55cbc70d6ffad8f7077b4d858 Mon Sep 17 00:00:00 2001 From: Julius Marminge Date: Tue, 17 Feb 2026 11:31:22 -0800 Subject: [PATCH] Add git-backed filesystem checkpoints for provider session rollback - add `FilesystemCheckpointStore` to capture, restore, and prune per-turn git refs - wire checkpoint init/capture into `ProviderManager` with per-session locking and error events - require checkpoint presence before rollback and restore workspace state after revert - add tests for checkpoint store behavior and provider rollback integration --- .../src/filesystemCheckpointStore.test.ts | 132 ++++++++++++ apps/server/src/filesystemCheckpointStore.ts | 202 ++++++++++++++++++ apps/server/src/providerManager.test.ts | 121 ++++++++++- apps/server/src/providerManager.ts | 180 +++++++++++++--- 4 files changed, 609 insertions(+), 26 deletions(-) create mode 100644 apps/server/src/filesystemCheckpointStore.test.ts create mode 100644 apps/server/src/filesystemCheckpointStore.ts diff --git a/apps/server/src/filesystemCheckpointStore.test.ts b/apps/server/src/filesystemCheckpointStore.test.ts new file mode 100644 index 0000000000000000000000000000000000000000..37a1f7696d0a3aa66429af5a5c8e4a0702348c8b --- /dev/null +++ b/apps/server/src/filesystemCheckpointStore.test.ts @@ -0,0 +1,132 @@ +import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises"; +import fs from "node:fs"; +import os from "node:os"; +import path from "node:path"; + +import { describe, expect, it } from "vitest"; + +import { checkpointRefForThreadTurn, FilesystemCheckpointStore } from "./filesystemCheckpointStore"; +import { runProcess } from "./processRunner"; + +async function runGit(cwd: string, args: readonly string[]): Promise { + const result = await runProcess("git", args, { + cwd, + allowNonZeroExit: true, + }); + if (result.code !== 0) { + throw new Error(result.stderr.trim() || `git ${args.join(" ")} failed`); + } + return result.stdout.trim(); +} + +function exists(filePath: string): boolean { + return fs.existsSync(filePath); +} + +async function createRepo(): Promise { + const cwd = await mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoints-test-")); + await runGit(cwd, ["init", "--initial-branch=main"]); + await runGit(cwd, ["config", "user.email", "test@example.com"]); + await runGit(cwd, ["config", "user.name", "Test User"]); + await writeFile(path.join(cwd, "README.md"), "v1\n"); + await writeFile(path.join(cwd, "tracked.txt"), "tracked-v1\n"); + await runGit(cwd, ["add", "."]); + await runGit(cwd, ["commit", "-m", "Initial"]); + return cwd; +} + +describe("FilesystemCheckpointStore", () => { + it("captures refs without mutating branch history and restores workspace state", async () => { + const cwd = await createRepo(); + const store = new FilesystemCheckpointStore(); + const threadId = "thr_1"; + + try { + const headBefore = await runGit(cwd, ["rev-parse", "HEAD"]); + + await store.captureCheckpoint({ cwd, threadId, turnCount: 0 }); + const rootRef = checkpointRefForThreadTurn(threadId, 0); + const rootCommit = await runGit(cwd, ["rev-parse", "--verify", `${rootRef}^{commit}`]); + expect(rootCommit.length).toBeGreaterThan(0); + expect(await runGit(cwd, ["rev-parse", "HEAD"])).toBe(headBefore); + + await writeFile(path.join(cwd, "README.md"), "v2\n"); + await rm(path.join(cwd, "tracked.txt")); + await writeFile(path.join(cwd, "notes.md"), "notes\n"); + await store.captureCheckpoint({ cwd, threadId, turnCount: 1 }); + + await writeFile(path.join(cwd, "README.md"), "v3\n"); + await writeFile(path.join(cwd, "scratch.txt"), "scratch\n"); + await writeFile(path.join(cwd, "tracked.txt"), "tracked-new\n"); + + const restoredRoot = await store.restoreCheckpoint({ cwd, threadId, turnCount: 0 }); + expect(restoredRoot).toBe(true); + expect(await readFile(path.join(cwd, "README.md"), "utf8")).toBe("v1\n"); + expect(await readFile(path.join(cwd, "tracked.txt"), "utf8")).toBe("tracked-v1\n"); + expect(exists(path.join(cwd, "notes.md"))).toBe(false); + expect(exists(path.join(cwd, "scratch.txt"))).toBe(false); + + const restoredTurn1 = await store.restoreCheckpoint({ cwd, threadId, turnCount: 1 }); + expect(restoredTurn1).toBe(true); + expect(await readFile(path.join(cwd, "README.md"), "utf8")).toBe("v2\n"); + expect(exists(path.join(cwd, "tracked.txt"))).toBe(false); + expect(await readFile(path.join(cwd, "notes.md"), "utf8")).toBe("notes\n"); + expect(exists(path.join(cwd, "scratch.txt"))).toBe(false); + } finally { + await rm(cwd, { recursive: true, force: true }); + } + }); + + it("returns false when restoring a missing checkpoint", async () => { + const cwd = await createRepo(); + const store = new FilesystemCheckpointStore(); + + try { + const restored = await store.restoreCheckpoint({ + cwd, + threadId: "thr_2", + turnCount: 999, + }); + expect(restored).toBe(false); + } finally { + await rm(cwd, { recursive: true, force: true }); + } + }); + + it("prunes refs newer than the retained turn", async () => { + const cwd = await createRepo(); + const store = new FilesystemCheckpointStore(); + const threadId = "thr_3"; + + try { + await store.captureCheckpoint({ cwd, threadId, turnCount: 0 }); + await writeFile(path.join(cwd, "README.md"), "turn1\n"); + await store.captureCheckpoint({ cwd, threadId, turnCount: 1 }); + await writeFile(path.join(cwd, "README.md"), "turn2\n"); + await store.captureCheckpoint({ cwd, threadId, turnCount: 2 }); + + await store.pruneAfterTurn({ + cwd, + threadId, + maxTurnCount: 1, + }); + + expect(await store.hasCheckpoint({ cwd, threadId, turnCount: 0 })).toBe(true); + expect(await store.hasCheckpoint({ cwd, threadId, turnCount: 1 })).toBe(true); + expect(await store.hasCheckpoint({ cwd, threadId, turnCount: 2 })).toBe(false); + } finally { + await rm(cwd, { recursive: true, force: true }); + } + }); + + it("detects non-repo folders", async () => { + const cwd = await mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoints-nonrepo-")); + const store = new FilesystemCheckpointStore(); + + try { + expect(await store.isGitRepository(cwd)).toBe(false); + } finally { + await rm(cwd, { recursive: true, force: true }); + } + }); +}); diff --git a/apps/server/src/filesystemCheckpointStore.ts b/apps/server/src/filesystemCheckpointStore.ts new file mode 100644 index 0000000000000000000000000000000000000000..99ab6567ad877ff509f9ec53291e1d2514a7adfe --- /dev/null +++ b/apps/server/src/filesystemCheckpointStore.ts @@ -0,0 +1,202 @@ +import { mkdtemp, rm, writeFile } from "node:fs/promises"; +import os from "node:os"; +import path from "node:path"; +import { randomUUID } from "node:crypto"; + +import { runProcess } from "./processRunner"; + +const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints"; +const DEFAULT_TIMEOUT_MS = 30_000; +const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000; + +function quoteGitCommand(args: readonly string[]): string { + return `git ${args.join(" ")}`; +} + +function checkpointRefThreadSegment(threadId: string): string { + return Buffer.from(threadId, "utf8").toString("base64url"); +} + +export function checkpointRefForThreadTurn(threadId: string, turnCount: number): string { + if (!Number.isInteger(turnCount) || turnCount < 0) { + throw new Error(`Invalid checkpoint turn count: ${turnCount}`); + } + return `${CHECKPOINT_REFS_PREFIX}/${checkpointRefThreadSegment(threadId)}/turn/${turnCount}`; +} + +interface GitOptions { + env?: NodeJS.ProcessEnv; + stdin?: string; + allowNonZeroExit?: boolean; +} + +export class FilesystemCheckpointStore { + async isGitRepository(cwd: string): Promise { + const result = await this.runGit(cwd, ["rev-parse", "--is-inside-work-tree"], { + allowNonZeroExit: true, + }); + return result.code === 0 && result.stdout.trim() === "true"; + } + + async captureCheckpoint(input: { cwd: string; threadId: string; turnCount: number }): Promise { + const { cwd, threadId, turnCount } = input; + const ref = checkpointRefForThreadTurn(threadId, turnCount); + + const tempDir = await mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoint-")); + const tempIndexPath = path.join(tempDir, `index-${randomUUID()}`); + const commitEnv: NodeJS.ProcessEnv = { + ...process.env, + GIT_INDEX_FILE: tempIndexPath, + GIT_AUTHOR_NAME: "T3 Code", + GIT_AUTHOR_EMAIL: "codex@users.noreply.github.com", + GIT_COMMITTER_NAME: "T3 Code", + GIT_COMMITTER_EMAIL: "codex@users.noreply.github.com", + }; + + try { + await writeFile(tempIndexPath, ""); + + const hasHead = await this.hasHeadCommit(cwd); + if (hasHead) { + await this.runGit(cwd, ["read-tree", "HEAD"], { env: commitEnv }); + } + + await this.runGit(cwd, ["add", "-A", "--", "."], { env: commitEnv }); + const writeTreeResult = await this.runGit(cwd, ["write-tree"], { env: commitEnv }); + const treeOid = writeTreeResult.stdout.trim(); + if (!treeOid) { + throw new Error("git write-tree returned an empty tree oid."); + } + + const message = `t3 checkpoint thread=${threadId} turn=${turnCount}`; + const commitTreeResult = await this.runGit(cwd, ["commit-tree", treeOid], { + env: commitEnv, + stdin: `${message}\n`, + }); + const commitOid = commitTreeResult.stdout.trim(); + if (!commitOid) { + throw new Error("git commit-tree returned an empty commit oid."); + } + + await this.runGit(cwd, ["update-ref", ref, commitOid]); + } finally { + await rm(tempDir, { recursive: true, force: true }); + } + } + + async hasCheckpoint(input: { cwd: string; threadId: string; turnCount: number }): Promise { + const { cwd, threadId, turnCount } = input; + const ref = checkpointRefForThreadTurn(threadId, turnCount); + const commit = await this.resolveCheckpointCommit(cwd, ref); + return commit !== null; + } + + async restoreCheckpoint(input: { cwd: string; threadId: string; turnCount: number }): Promise { + const { cwd, threadId, turnCount } = input; + const ref = checkpointRefForThreadTurn(threadId, turnCount); + const commitOid = await this.resolveCheckpointCommit(cwd, ref); + if (!commitOid) { + return false; + } + + await this.runGit(cwd, ["restore", "--source", commitOid, "--worktree", "--staged", "--", "."]); + await this.runGit(cwd, ["clean", "-fd", "--", "."]); + await this.runGit(cwd, ["reset", "--quiet", "--", "."], { allowNonZeroExit: true }); + return true; + } + + async pruneAfterTurn(input: { cwd: string; threadId: string; maxTurnCount: number }): Promise { + const { cwd, threadId, maxTurnCount } = input; + if (!Number.isInteger(maxTurnCount) || maxTurnCount < 0) { + throw new Error(`Invalid max turn count: ${maxTurnCount}`); + } + + const threadRefPrefix = `${CHECKPOINT_REFS_PREFIX}/${checkpointRefThreadSegment(threadId)}/turn/`; + const result = await this.runGit(cwd, ["for-each-ref", "--format=%(refname)", threadRefPrefix], { + allowNonZeroExit: true, + }); + if (result.code !== 0) { + return; + } + + const refsToDelete = result.stdout + .split(/\r?\n/g) + .map((line) => line.trim()) + .filter((line) => line.length > 0) + .filter((refName) => { + const turn = this.readRefTurnCount(refName); + return turn !== null && turn > maxTurnCount; + }); + + await Promise.all( + refsToDelete.map((refName) => + this.runGit(cwd, ["update-ref", "-d", refName], { allowNonZeroExit: true }), + ), + ); + } + + private async hasHeadCommit(cwd: string): Promise { + const result = await this.runGit(cwd, ["rev-parse", "--verify", "HEAD"], { + allowNonZeroExit: true, + }); + return result.code === 0; + } + + private async resolveCheckpointCommit(cwd: string, ref: string): Promise { + const result = await this.runGit(cwd, ["rev-parse", "--verify", "--quiet", `${ref}^{commit}`], { + allowNonZeroExit: true, + }); + if (result.code !== 0) { + return null; + } + const commit = result.stdout.trim(); + return commit.length > 0 ? commit : null; + } + + private readRefTurnCount(ref: string): number | null { + const segments = ref.split("/"); + const rawTurn = segments.at(-1); + if (!rawTurn) { + return null; + } + const turn = Number.parseInt(rawTurn, 10); + return Number.isInteger(turn) && turn >= 0 ? turn : null; + } + + private async runGit( + cwd: string, + args: readonly string[], + options: GitOptions = {}, + ): Promise>> { + let result: Awaited>; + try { + result = await runProcess("git", args, { + cwd, + env: options.env, + stdin: options.stdin, + timeoutMs: DEFAULT_TIMEOUT_MS, + allowNonZeroExit: true, + maxBufferBytes: DEFAULT_MAX_OUTPUT_BYTES, + outputMode: "truncate", + }); + } catch (error) { + const message = error instanceof Error ? error.message : String(error); + throw new Error(`Failed to run ${quoteGitCommand(args)}: ${message}`, { cause: error }); + } + + if (result.timedOut) { + throw new Error(`${quoteGitCommand(args)} timed out.`); + } + + if (!options.allowNonZeroExit && result.code !== 0) { + const stderr = result.stderr.trim(); + throw new Error( + stderr.length > 0 + ? `${quoteGitCommand(args)} failed: ${stderr}` + : `${quoteGitCommand(args)} failed with code ${result.code ?? "null"}.`, + ); + } + + return result; + } +} diff --git a/apps/server/src/providerManager.test.ts b/apps/server/src/providerManager.test.ts index 7d7e7ad76fdef7321ecabfbda7a715bed79b6e1d..794e2183d21e039b3bfc27333f42623be521ff5c 100644 --- a/apps/server/src/providerManager.test.ts +++ b/apps/server/src/providerManager.test.ts @@ -1,4 +1,4 @@ -import { describe, expect, it } from "vitest"; +import { describe, expect, it, vi } from "vitest"; import { ProviderManager } from "./providerManager"; @@ -191,4 +191,123 @@ describe("ProviderManager", () => { manager.dispose(); }); + + it("restores filesystem checkpoint when reverting and checkpointing is enabled", async () => { + const manager = new ProviderManager(); + const internals = manager as unknown as { + codex: { + hasSession: (sessionId: string) => boolean; + readThread: (sessionId: string) => Promise<{ + threadId: string; + turns: Array<{ id: string; items: unknown[] }>; + }>; + rollbackThread: (sessionId: string, numTurns: number) => Promise<{ + threadId: string; + turns: Array<{ id: string; items: unknown[] }>; + }>; + }; + filesystemCheckpointStore: { + hasCheckpoint: (input: { cwd: string; threadId: string; turnCount: number }) => Promise; + restoreCheckpoint: (input: { + cwd: string; + threadId: string; + turnCount: number; + }) => Promise; + pruneAfterTurn: (input: { cwd: string; threadId: string; maxTurnCount: number }) => Promise; + }; + sessionCheckpointCwds: Map; + }; + + internals.codex.hasSession = () => true; + internals.codex.readThread = async () => ({ + threadId: "thr_1", + turns: [ + { id: "turn_1", items: [] }, + { id: "turn_2", items: [] }, + ], + }); + internals.codex.rollbackThread = async () => ({ + threadId: "thr_1", + turns: [{ id: "turn_1", items: [] }], + }); + const hasCheckpoint = vi.fn(async () => true); + const restoreCheckpoint = vi.fn(async () => true); + const pruneAfterTurn = vi.fn(async () => undefined); + internals.filesystemCheckpointStore.hasCheckpoint = hasCheckpoint; + internals.filesystemCheckpointStore.restoreCheckpoint = restoreCheckpoint; + internals.filesystemCheckpointStore.pruneAfterTurn = pruneAfterTurn; + internals.sessionCheckpointCwds.set("sess_1", "/repo"); + + const result = await manager.revertToCheckpoint({ + sessionId: "sess_1", + turnCount: 1, + }); + + expect(result.turnCount).toBe(1); + expect(hasCheckpoint).toHaveBeenCalledWith({ + cwd: "/repo", + threadId: "thr_1", + turnCount: 1, + }); + expect(restoreCheckpoint).toHaveBeenCalledWith({ + cwd: "/repo", + threadId: "thr_1", + turnCount: 1, + }); + expect(pruneAfterTurn).toHaveBeenCalledWith({ + cwd: "/repo", + threadId: "thr_1", + maxTurnCount: 1, + }); + + manager.dispose(); + }); + + it("fails before rollback when filesystem checkpoint is missing", async () => { + const manager = new ProviderManager(); + const internals = manager as unknown as { + codex: { + hasSession: (sessionId: string) => boolean; + readThread: (sessionId: string) => Promise<{ + threadId: string; + turns: Array<{ id: string; items: unknown[] }>; + }>; + rollbackThread: (sessionId: string, numTurns: number) => Promise<{ + threadId: string; + turns: Array<{ id: string; items: unknown[] }>; + }>; + }; + filesystemCheckpointStore: { + hasCheckpoint: (input: { cwd: string; threadId: string; turnCount: number }) => Promise; + }; + sessionCheckpointCwds: Map; + }; + + const rollbackSpy = vi.fn(async () => ({ + threadId: "thr_1", + turns: [{ id: "turn_1", items: [] }], + })); + + internals.codex.hasSession = () => true; + internals.codex.readThread = async () => ({ + threadId: "thr_1", + turns: [ + { id: "turn_1", items: [] }, + { id: "turn_2", items: [] }, + ], + }); + internals.codex.rollbackThread = rollbackSpy; + internals.filesystemCheckpointStore.hasCheckpoint = async () => false; + internals.sessionCheckpointCwds.set("sess_1", "/repo"); + + await expect( + manager.revertToCheckpoint({ + sessionId: "sess_1", + turnCount: 1, + }), + ).rejects.toThrow("Filesystem checkpoint is unavailable for turn 1 in thread thr_1."); + expect(rollbackSpy).not.toHaveBeenCalled(); + + manager.dispose(); + }); }); diff --git a/apps/server/src/providerManager.ts b/apps/server/src/providerManager.ts index 58977b29b9e73bbbd4fcc0ccf0e6e4b75a75b79c..1038ed41d4f2852e2e3015708c0d347736a1c494 100644 --- a/apps/server/src/providerManager.ts +++ b/apps/server/src/providerManager.ts @@ -1,6 +1,7 @@ import { EventEmitter } from "node:events"; import fs from "node:fs"; import path from "node:path"; +import { randomUUID } from "node:crypto"; import { type ProviderCheckpoint, @@ -26,6 +27,7 @@ import { } from "@t3tools/contracts"; import type { CodexThreadTurnSnapshot } from "./codexAppServerManager"; import { CodexAppServerManager } from "./codexAppServerManager"; +import { FilesystemCheckpointStore } from "./filesystemCheckpointStore"; export interface ProviderManagerEvents { event: [event: ProviderEvent]; @@ -157,10 +159,13 @@ function buildCheckpoints(turns: CodexThreadTurnSnapshot[]): ProviderCheckpoint[ export class ProviderManager extends EventEmitter { private readonly codex = new CodexAppServerManager(); + private readonly filesystemCheckpointStore = new FilesystemCheckpointStore(); private readonly threadLogsDir: string; private readonly threadLogStreams = new Map(); private readonly sessionThreadIds = new Map(); private readonly pendingEventsBySession = new Map(); + private readonly sessionCheckpointCwds = new Map(); + private readonly filesystemLocks = new Map>(); private disposed = false; private readonly onCodexEvent = (event: ProviderEvent) => { if (this.disposed) { @@ -169,6 +174,7 @@ export class ProviderManager extends EventEmitter { this.routeEventToThreadLog(event); this.emit("event", event); + this.maybeCaptureFilesystemCheckpoint(event); }; constructor() { @@ -192,6 +198,11 @@ export class ProviderManager extends EventEmitter { this.sessionThreadIds.set(session.sessionId, session.threadId); this.flushPendingSessionEvents(session.sessionId, session.threadId); } + await this.initializeFilesystemCheckpointing(session).catch((error) => { + const message = + error instanceof Error ? error.message : "Failed to initialize filesystem checkpoints."; + this.emitFilesystemCheckpointError(session.sessionId, message, session.threadId); + }); return session; } @@ -227,6 +238,8 @@ export class ProviderManager extends EventEmitter { this.codex.stopSession(input.sessionId); this.sessionThreadIds.delete(input.sessionId); this.pendingEventsBySession.delete(input.sessionId); + this.sessionCheckpointCwds.delete(input.sessionId); + this.filesystemLocks.delete(input.sessionId); } listSessions(): ProviderSession[] { @@ -254,33 +267,68 @@ export class ProviderManager extends EventEmitter { throw new Error(`Unknown provider session: ${input.sessionId}`); } - const beforeSnapshot = await this.codex.readThread(input.sessionId); - const currentTurnCount = beforeSnapshot.turns.length; - if (input.turnCount > currentTurnCount) { - throw new Error( - `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`, - ); - } + return this.withFilesystemLock(input.sessionId, async () => { + const beforeSnapshot = await this.codex.readThread(input.sessionId); + const currentTurnCount = beforeSnapshot.turns.length; + if (input.turnCount > currentTurnCount) { + throw new Error( + `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`, + ); + } - const requestedRollbackTurns = currentTurnCount - input.turnCount; - const afterSnapshot = - requestedRollbackTurns > 0 - ? await this.codex.rollbackThread(input.sessionId, requestedRollbackTurns) - : beforeSnapshot; - const checkpoints = buildCheckpoints(afterSnapshot.turns); - const currentCheckpoint = - checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? - checkpoints[checkpoints.length - 1] ?? - checkpoints[0]; - const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length); + const checkpointCwd = this.sessionCheckpointCwds.get(input.sessionId); + if (checkpointCwd) { + const hasCheckpoint = await this.filesystemCheckpointStore.hasCheckpoint({ + cwd: checkpointCwd, + threadId: beforeSnapshot.threadId, + turnCount: input.turnCount, + }); + if (!hasCheckpoint) { + throw new Error( + `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`, + ); + } + } - return { - threadId: afterSnapshot.threadId, - turnCount: currentCheckpoint?.turnCount ?? 0, - messageCount: currentCheckpoint?.messageCount ?? 0, - rolledBackTurns, - checkpoints, - }; + const requestedRollbackTurns = currentTurnCount - input.turnCount; + const afterSnapshot = + requestedRollbackTurns > 0 + ? await this.codex.rollbackThread(input.sessionId, requestedRollbackTurns) + : beforeSnapshot; + + if (checkpointCwd) { + const restored = await this.filesystemCheckpointStore.restoreCheckpoint({ + cwd: checkpointCwd, + threadId: beforeSnapshot.threadId, + turnCount: input.turnCount, + }); + if (!restored) { + throw new Error( + `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`, + ); + } + await this.filesystemCheckpointStore.pruneAfterTurn({ + cwd: checkpointCwd, + threadId: afterSnapshot.threadId, + maxTurnCount: afterSnapshot.turns.length, + }); + } + + const checkpoints = buildCheckpoints(afterSnapshot.turns); + const currentCheckpoint = + checkpoints.find((checkpoint) => checkpoint.isCurrent) ?? + checkpoints[checkpoints.length - 1] ?? + checkpoints[0]; + const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length); + + return { + threadId: afterSnapshot.threadId, + turnCount: currentCheckpoint?.turnCount ?? 0, + messageCount: currentCheckpoint?.messageCount ?? 0, + rolledBackTurns, + checkpoints, + }; + }); } stopAll(): void { @@ -300,6 +348,88 @@ export class ProviderManager extends EventEmitter { this.threadLogStreams.clear(); this.sessionThreadIds.clear(); this.pendingEventsBySession.clear(); + this.sessionCheckpointCwds.clear(); + this.filesystemLocks.clear(); + } + + private async initializeFilesystemCheckpointing(session: ProviderSession): Promise { + const cwd = session.cwd ?? process.cwd(); + + await this.withFilesystemLock(session.sessionId, async () => { + const supportsGit = await this.filesystemCheckpointStore.isGitRepository(cwd); + if (!supportsGit) { + this.sessionCheckpointCwds.delete(session.sessionId); + return; + } + + const snapshot = await this.codex.readThread(session.sessionId); + await this.filesystemCheckpointStore.captureCheckpoint({ + cwd, + threadId: snapshot.threadId, + turnCount: snapshot.turns.length, + }); + this.sessionCheckpointCwds.set(session.sessionId, cwd); + }); + } + + private maybeCaptureFilesystemCheckpoint(event: ProviderEvent): void { + if (event.kind !== "notification" || event.method !== "turn/completed") { + return; + } + + const checkpointCwd = this.sessionCheckpointCwds.get(event.sessionId); + if (!checkpointCwd) { + return; + } + + void this.withFilesystemLock(event.sessionId, async () => { + if (!this.codex.hasSession(event.sessionId)) { + return; + } + + const snapshot = await this.codex.readThread(event.sessionId); + await this.filesystemCheckpointStore.captureCheckpoint({ + cwd: checkpointCwd, + threadId: snapshot.threadId, + turnCount: snapshot.turns.length, + }); + }).catch((error) => { + const message = error instanceof Error ? error.message : "Failed to capture checkpoint."; + this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId); + }); + } + + private emitFilesystemCheckpointError( + sessionId: string, + message: string, + threadId?: string, + ): void { + this.emit("event", { + id: randomUUID(), + kind: "error", + provider: "codex", + sessionId, + createdAt: new Date().toISOString(), + method: "checkpoint/filesystemError", + message, + ...(threadId ? { threadId } : {}), + }); + } + + private withFilesystemLock(sessionId: string, fn: () => Promise): Promise { + const previous = this.filesystemLocks.get(sessionId) ?? Promise.resolve(); + const next = previous.catch(() => undefined).then(fn); + const completion = next.then( + () => undefined, + () => undefined, + ); + this.filesystemLocks.set(sessionId, completion); + return next.finally(() => { + const tracked = this.filesystemLocks.get(sessionId); + if (tracked === completion) { + this.filesystemLocks.delete(sessionId); + } + }); } private routeEventToThreadLog(event: ProviderEvent): void { -- 2.54.0