History

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

Julius Marminge committed 8 months ago

Commit e4482211 parent 57ca2544 files · +609 −26Browse files

Unified · Split

4 files
  1. apps/server/src/filesystemCheckpointStore.test.ts +132 −0
  2. apps/server/src/filesystemCheckpointStore.ts +202 −0
  3. apps/server/src/providerManager.test.ts +120 −1
  4. apps/server/src/providerManager.ts +155 −25
apps/server/src/filesystemCheckpointStore.test.ts+132−0
@@ -0,0 +1,132 @@
1+import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
2+import fs from "node:fs";
3+import os from "node:os";
4+import path from "node:path";
5+
6+import { describe, expect, it } from "vitest";
7+
8+import { checkpointRefForThreadTurn, FilesystemCheckpointStore } from "./filesystemCheckpointStore";
9+import { runProcess } from "./processRunner";
10+
11+async function runGit(cwd: string, args: readonly string[]): Promise<string> {
12+ const result = await runProcess("git", args, {
13+ cwd,
14+ allowNonZeroExit: true,
15+ });
16+ if (result.code !== 0) {
17+ throw new Error(result.stderr.trim() || `git ${args.join(" ")} failed`);
18+ }
19+ return result.stdout.trim();
20+}
21+
22+function exists(filePath: string): boolean {
23+ return fs.existsSync(filePath);
24+}
25+
26+async function createRepo(): Promise<string> {
27+ const cwd = await mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoints-test-"));
28+ await runGit(cwd, ["init", "--initial-branch=main"]);
29+ await runGit(cwd, ["config", "user.email", "test@example.com"]);
30+ await runGit(cwd, ["config", "user.name", "Test User"]);
31+ await writeFile(path.join(cwd, "README.md"), "v1\n");
32+ await writeFile(path.join(cwd, "tracked.txt"), "tracked-v1\n");
33+ await runGit(cwd, ["add", "."]);
34+ await runGit(cwd, ["commit", "-m", "Initial"]);
35+ return cwd;
36+}
37+
38+describe("FilesystemCheckpointStore", () => {
39+ it("captures refs without mutating branch history and restores workspace state", async () => {
40+ const cwd = await createRepo();
41+ const store = new FilesystemCheckpointStore();
42+ const threadId = "thr_1";
43+
44+ try {
45+ const headBefore = await runGit(cwd, ["rev-parse", "HEAD"]);
46+
47+ await store.captureCheckpoint({ cwd, threadId, turnCount: 0 });
48+ const rootRef = checkpointRefForThreadTurn(threadId, 0);
49+ const rootCommit = await runGit(cwd, ["rev-parse", "--verify", `${rootRef}^{commit}`]);
50+ expect(rootCommit.length).toBeGreaterThan(0);
51+ expect(await runGit(cwd, ["rev-parse", "HEAD"])).toBe(headBefore);
52+
53+ await writeFile(path.join(cwd, "README.md"), "v2\n");
54+ await rm(path.join(cwd, "tracked.txt"));
55+ await writeFile(path.join(cwd, "notes.md"), "notes\n");
56+ await store.captureCheckpoint({ cwd, threadId, turnCount: 1 });
57+
58+ await writeFile(path.join(cwd, "README.md"), "v3\n");
59+ await writeFile(path.join(cwd, "scratch.txt"), "scratch\n");
60+ await writeFile(path.join(cwd, "tracked.txt"), "tracked-new\n");
61+
62+ const restoredRoot = await store.restoreCheckpoint({ cwd, threadId, turnCount: 0 });
63+ expect(restoredRoot).toBe(true);
64+ expect(await readFile(path.join(cwd, "README.md"), "utf8")).toBe("v1\n");
65+ expect(await readFile(path.join(cwd, "tracked.txt"), "utf8")).toBe("tracked-v1\n");
66+ expect(exists(path.join(cwd, "notes.md"))).toBe(false);
67+ expect(exists(path.join(cwd, "scratch.txt"))).toBe(false);
68+
69+ const restoredTurn1 = await store.restoreCheckpoint({ cwd, threadId, turnCount: 1 });
70+ expect(restoredTurn1).toBe(true);
71+ expect(await readFile(path.join(cwd, "README.md"), "utf8")).toBe("v2\n");
72+ expect(exists(path.join(cwd, "tracked.txt"))).toBe(false);
73+ expect(await readFile(path.join(cwd, "notes.md"), "utf8")).toBe("notes\n");
74+ expect(exists(path.join(cwd, "scratch.txt"))).toBe(false);
75+ } finally {
76+ await rm(cwd, { recursive: true, force: true });
77+ }
78+ });
79+
80+ it("returns false when restoring a missing checkpoint", async () => {
81+ const cwd = await createRepo();
82+ const store = new FilesystemCheckpointStore();
83+
84+ try {
85+ const restored = await store.restoreCheckpoint({
86+ cwd,
87+ threadId: "thr_2",
88+ turnCount: 999,
89+ });
90+ expect(restored).toBe(false);
91+ } finally {
92+ await rm(cwd, { recursive: true, force: true });
93+ }
94+ });
95+
96+ it("prunes refs newer than the retained turn", async () => {
97+ const cwd = await createRepo();
98+ const store = new FilesystemCheckpointStore();
99+ const threadId = "thr_3";
100+
101+ try {
102+ await store.captureCheckpoint({ cwd, threadId, turnCount: 0 });
103+ await writeFile(path.join(cwd, "README.md"), "turn1\n");
104+ await store.captureCheckpoint({ cwd, threadId, turnCount: 1 });
105+ await writeFile(path.join(cwd, "README.md"), "turn2\n");
106+ await store.captureCheckpoint({ cwd, threadId, turnCount: 2 });
107+
108+ await store.pruneAfterTurn({
109+ cwd,
110+ threadId,
111+ maxTurnCount: 1,
112+ });
113+
114+ expect(await store.hasCheckpoint({ cwd, threadId, turnCount: 0 })).toBe(true);
115+ expect(await store.hasCheckpoint({ cwd, threadId, turnCount: 1 })).toBe(true);
116+ expect(await store.hasCheckpoint({ cwd, threadId, turnCount: 2 })).toBe(false);
117+ } finally {
118+ await rm(cwd, { recursive: true, force: true });
119+ }
120+ });
121+
122+ it("detects non-repo folders", async () => {
123+ const cwd = await mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoints-nonrepo-"));
124+ const store = new FilesystemCheckpointStore();
125+
126+ try {
127+ expect(await store.isGitRepository(cwd)).toBe(false);
128+ } finally {
129+ await rm(cwd, { recursive: true, force: true });
130+ }
131+ });
132+});
apps/server/src/filesystemCheckpointStore.ts+202−0
@@ -0,0 +1,202 @@
1+import { mkdtemp, rm, writeFile } from "node:fs/promises";
2+import os from "node:os";
3+import path from "node:path";
4+import { randomUUID } from "node:crypto";
5+
6+import { runProcess } from "./processRunner";
7+
8+const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints";
9+const DEFAULT_TIMEOUT_MS = 30_000;
10+const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000;
11+
12+function quoteGitCommand(args: readonly string[]): string {
13+ return `git ${args.join(" ")}`;
14+}
15+
16+function checkpointRefThreadSegment(threadId: string): string {
17+ return Buffer.from(threadId, "utf8").toString("base64url");
18+}
19+
20+export function checkpointRefForThreadTurn(threadId: string, turnCount: number): string {
21+ if (!Number.isInteger(turnCount) || turnCount < 0) {
22+ throw new Error(`Invalid checkpoint turn count: ${turnCount}`);
23+ }
24+ return `${CHECKPOINT_REFS_PREFIX}/${checkpointRefThreadSegment(threadId)}/turn/${turnCount}`;
25+}
26+
27+interface GitOptions {
28+ env?: NodeJS.ProcessEnv;
29+ stdin?: string;
30+ allowNonZeroExit?: boolean;
31+}
32+
33+export class FilesystemCheckpointStore {
34+ async isGitRepository(cwd: string): Promise<boolean> {
35+ const result = await this.runGit(cwd, ["rev-parse", "--is-inside-work-tree"], {
36+ allowNonZeroExit: true,
37+ });
38+ return result.code === 0 && result.stdout.trim() === "true";
39+ }
40+
41+ async captureCheckpoint(input: { cwd: string; threadId: string; turnCount: number }): Promise<void> {
42+ const { cwd, threadId, turnCount } = input;
43+ const ref = checkpointRefForThreadTurn(threadId, turnCount);
44+
45+ const tempDir = await mkdtemp(path.join(os.tmpdir(), "t3-fs-checkpoint-"));
46+ const tempIndexPath = path.join(tempDir, `index-${randomUUID()}`);
47+ const commitEnv: NodeJS.ProcessEnv = {
48+ ...process.env,
49+ GIT_INDEX_FILE: tempIndexPath,
50+ GIT_AUTHOR_NAME: "T3 Code",
51+ GIT_AUTHOR_EMAIL: "codex@users.noreply.github.com",
52+ GIT_COMMITTER_NAME: "T3 Code",
53+ GIT_COMMITTER_EMAIL: "codex@users.noreply.github.com",
54+ };
55+
56+ try {
57+ await writeFile(tempIndexPath, "");
58+
59+ const hasHead = await this.hasHeadCommit(cwd);
60+ if (hasHead) {
61+ await this.runGit(cwd, ["read-tree", "HEAD"], { env: commitEnv });
62+ }
63+
64+ await this.runGit(cwd, ["add", "-A", "--", "."], { env: commitEnv });
65+ const writeTreeResult = await this.runGit(cwd, ["write-tree"], { env: commitEnv });
66+ const treeOid = writeTreeResult.stdout.trim();
67+ if (!treeOid) {
68+ throw new Error("git write-tree returned an empty tree oid.");
69+ }
70+
71+ const message = `t3 checkpoint thread=${threadId} turn=${turnCount}`;
72+ const commitTreeResult = await this.runGit(cwd, ["commit-tree", treeOid], {
73+ env: commitEnv,
74+ stdin: `${message}\n`,
75+ });
76+ const commitOid = commitTreeResult.stdout.trim();
77+ if (!commitOid) {
78+ throw new Error("git commit-tree returned an empty commit oid.");
79+ }
80+
81+ await this.runGit(cwd, ["update-ref", ref, commitOid]);
82+ } finally {
83+ await rm(tempDir, { recursive: true, force: true });
84+ }
85+ }
86+
87+ async hasCheckpoint(input: { cwd: string; threadId: string; turnCount: number }): Promise<boolean> {
88+ const { cwd, threadId, turnCount } = input;
89+ const ref = checkpointRefForThreadTurn(threadId, turnCount);
90+ const commit = await this.resolveCheckpointCommit(cwd, ref);
91+ return commit !== null;
92+ }
93+
94+ async restoreCheckpoint(input: { cwd: string; threadId: string; turnCount: number }): Promise<boolean> {
95+ const { cwd, threadId, turnCount } = input;
96+ const ref = checkpointRefForThreadTurn(threadId, turnCount);
97+ const commitOid = await this.resolveCheckpointCommit(cwd, ref);
98+ if (!commitOid) {
99+ return false;
100+ }
101+
102+ await this.runGit(cwd, ["restore", "--source", commitOid, "--worktree", "--staged", "--", "."]);
103+ await this.runGit(cwd, ["clean", "-fd", "--", "."]);
104+ await this.runGit(cwd, ["reset", "--quiet", "--", "."], { allowNonZeroExit: true });
105+ return true;
106+ }
107+
108+ async pruneAfterTurn(input: { cwd: string; threadId: string; maxTurnCount: number }): Promise<void> {
109+ const { cwd, threadId, maxTurnCount } = input;
110+ if (!Number.isInteger(maxTurnCount) || maxTurnCount < 0) {
111+ throw new Error(`Invalid max turn count: ${maxTurnCount}`);
112+ }
113+
114+ const threadRefPrefix = `${CHECKPOINT_REFS_PREFIX}/${checkpointRefThreadSegment(threadId)}/turn/`;
115+ const result = await this.runGit(cwd, ["for-each-ref", "--format=%(refname)", threadRefPrefix], {
116+ allowNonZeroExit: true,
117+ });
118+ if (result.code !== 0) {
119+ return;
120+ }
121+
122+ const refsToDelete = result.stdout
123+ .split(/\r?\n/g)
124+ .map((line) => line.trim())
125+ .filter((line) => line.length > 0)
126+ .filter((refName) => {
127+ const turn = this.readRefTurnCount(refName);
128+ return turn !== null && turn > maxTurnCount;
129+ });
130+
131+ await Promise.all(
132+ refsToDelete.map((refName) =>
133+ this.runGit(cwd, ["update-ref", "-d", refName], { allowNonZeroExit: true }),
134+ ),
135+ );
136+ }
137+
138+ private async hasHeadCommit(cwd: string): Promise<boolean> {
139+ const result = await this.runGit(cwd, ["rev-parse", "--verify", "HEAD"], {
140+ allowNonZeroExit: true,
141+ });
142+ return result.code === 0;
143+ }
144+
145+ private async resolveCheckpointCommit(cwd: string, ref: string): Promise<string | null> {
146+ const result = await this.runGit(cwd, ["rev-parse", "--verify", "--quiet", `${ref}^{commit}`], {
147+ allowNonZeroExit: true,
148+ });
149+ if (result.code !== 0) {
150+ return null;
151+ }
152+ const commit = result.stdout.trim();
153+ return commit.length > 0 ? commit : null;
154+ }
155+
156+ private readRefTurnCount(ref: string): number | null {
157+ const segments = ref.split("/");
158+ const rawTurn = segments.at(-1);
159+ if (!rawTurn) {
160+ return null;
161+ }
162+ const turn = Number.parseInt(rawTurn, 10);
163+ return Number.isInteger(turn) && turn >= 0 ? turn : null;
164+ }
165+
166+ private async runGit(
167+ cwd: string,
168+ args: readonly string[],
169+ options: GitOptions = {},
170+ ): Promise<Awaited<ReturnType<typeof runProcess>>> {
171+ let result: Awaited<ReturnType<typeof runProcess>>;
172+ try {
173+ result = await runProcess("git", args, {
174+ cwd,
175+ env: options.env,
176+ stdin: options.stdin,
177+ timeoutMs: DEFAULT_TIMEOUT_MS,
178+ allowNonZeroExit: true,
179+ maxBufferBytes: DEFAULT_MAX_OUTPUT_BYTES,
180+ outputMode: "truncate",
181+ });
182+ } catch (error) {
183+ const message = error instanceof Error ? error.message : String(error);
184+ throw new Error(`Failed to run ${quoteGitCommand(args)}: ${message}`, { cause: error });
185+ }
186+
187+ if (result.timedOut) {
188+ throw new Error(`${quoteGitCommand(args)} timed out.`);
189+ }
190+
191+ if (!options.allowNonZeroExit && result.code !== 0) {
192+ const stderr = result.stderr.trim();
193+ throw new Error(
194+ stderr.length > 0
195+ ? `${quoteGitCommand(args)} failed: ${stderr}`
196+ : `${quoteGitCommand(args)} failed with code ${result.code ?? "null"}.`,
197+ );
198+ }
199+
200+ return result;
201+ }
202+}
apps/server/src/providerManager.test.ts+120−1
@@ -1,4 +1,4 @@
1−import { describe, expect, it } from "vitest";
1+import { describe, expect, it, vi } from "vitest";
22
33 import { ProviderManager } from "./providerManager";
44
@@ -191,4 +191,123 @@ describe("ProviderManager", () => { · 186 unmodified lines
191191
192192 manager.dispose();
193193 });
194+
195+ it("restores filesystem checkpoint when reverting and checkpointing is enabled", async () => {
196+ const manager = new ProviderManager();
197+ const internals = manager as unknown as {
198+ codex: {
199+ hasSession: (sessionId: string) => boolean;
200+ readThread: (sessionId: string) => Promise<{
201+ threadId: string;
202+ turns: Array<{ id: string; items: unknown[] }>;
203+ }>;
204+ rollbackThread: (sessionId: string, numTurns: number) => Promise<{
205+ threadId: string;
206+ turns: Array<{ id: string; items: unknown[] }>;
207+ }>;
208+ };
209+ filesystemCheckpointStore: {
210+ hasCheckpoint: (input: { cwd: string; threadId: string; turnCount: number }) => Promise<boolean>;
211+ restoreCheckpoint: (input: {
212+ cwd: string;
213+ threadId: string;
214+ turnCount: number;
215+ }) => Promise<boolean>;
216+ pruneAfterTurn: (input: { cwd: string; threadId: string; maxTurnCount: number }) => Promise<void>;
217+ };
218+ sessionCheckpointCwds: Map<string, string>;
219+ };
220+
221+ internals.codex.hasSession = () => true;
222+ internals.codex.readThread = async () => ({
223+ threadId: "thr_1",
224+ turns: [
225+ { id: "turn_1", items: [] },
226+ { id: "turn_2", items: [] },
227+ ],
228+ });
229+ internals.codex.rollbackThread = async () => ({
230+ threadId: "thr_1",
231+ turns: [{ id: "turn_1", items: [] }],
232+ });
233+ const hasCheckpoint = vi.fn(async () => true);
234+ const restoreCheckpoint = vi.fn(async () => true);
235+ const pruneAfterTurn = vi.fn(async () => undefined);
236+ internals.filesystemCheckpointStore.hasCheckpoint = hasCheckpoint;
237+ internals.filesystemCheckpointStore.restoreCheckpoint = restoreCheckpoint;
238+ internals.filesystemCheckpointStore.pruneAfterTurn = pruneAfterTurn;
239+ internals.sessionCheckpointCwds.set("sess_1", "/repo");
240+
241+ const result = await manager.revertToCheckpoint({
242+ sessionId: "sess_1",
243+ turnCount: 1,
244+ });
245+
246+ expect(result.turnCount).toBe(1);
247+ expect(hasCheckpoint).toHaveBeenCalledWith({
248+ cwd: "/repo",
249+ threadId: "thr_1",
250+ turnCount: 1,
251+ });
252+ expect(restoreCheckpoint).toHaveBeenCalledWith({
253+ cwd: "/repo",
254+ threadId: "thr_1",
255+ turnCount: 1,
256+ });
257+ expect(pruneAfterTurn).toHaveBeenCalledWith({
258+ cwd: "/repo",
259+ threadId: "thr_1",
260+ maxTurnCount: 1,
261+ });
262+
263+ manager.dispose();
264+ });
265+
266+ it("fails before rollback when filesystem checkpoint is missing", async () => {
267+ const manager = new ProviderManager();
268+ const internals = manager as unknown as {
269+ codex: {
270+ hasSession: (sessionId: string) => boolean;
271+ readThread: (sessionId: string) => Promise<{
272+ threadId: string;
273+ turns: Array<{ id: string; items: unknown[] }>;
274+ }>;
275+ rollbackThread: (sessionId: string, numTurns: number) => Promise<{
276+ threadId: string;
277+ turns: Array<{ id: string; items: unknown[] }>;
278+ }>;
279+ };
280+ filesystemCheckpointStore: {
281+ hasCheckpoint: (input: { cwd: string; threadId: string; turnCount: number }) => Promise<boolean>;
282+ };
283+ sessionCheckpointCwds: Map<string, string>;
284+ };
285+
286+ const rollbackSpy = vi.fn(async () => ({
287+ threadId: "thr_1",
288+ turns: [{ id: "turn_1", items: [] }],
289+ }));
290+
291+ internals.codex.hasSession = () => true;
292+ internals.codex.readThread = async () => ({
293+ threadId: "thr_1",
294+ turns: [
295+ { id: "turn_1", items: [] },
296+ { id: "turn_2", items: [] },
297+ ],
298+ });
299+ internals.codex.rollbackThread = rollbackSpy;
300+ internals.filesystemCheckpointStore.hasCheckpoint = async () => false;
301+ internals.sessionCheckpointCwds.set("sess_1", "/repo");
302+
303+ await expect(
304+ manager.revertToCheckpoint({
305+ sessionId: "sess_1",
306+ turnCount: 1,
307+ }),
308+ ).rejects.toThrow("Filesystem checkpoint is unavailable for turn 1 in thread thr_1.");
309+ expect(rollbackSpy).not.toHaveBeenCalled();
310+
311+ manager.dispose();
312+ });
194313 });
apps/server/src/providerManager.ts+155−25
@@ -1,6 +1,7 @@
11 import { EventEmitter } from "node:events";
22 import fs from "node:fs";
33 import path from "node:path";
4+import { randomUUID } from "node:crypto";
45
56 import {
67 type ProviderCheckpoint,
@@ -26,6 +27,7 @@ import { · 19 unmodified lines
2627 } from "@t3tools/contracts";
2728 import type { CodexThreadTurnSnapshot } from "./codexAppServerManager";
2829 import { CodexAppServerManager } from "./codexAppServerManager";
30+import { FilesystemCheckpointStore } from "./filesystemCheckpointStore";
2931
3032 export interface ProviderManagerEvents {
3133 event: [event: ProviderEvent];
@@ -157,10 +159,13 @@ function buildCheckpoints(turns: CodexThreadTurnSnapshot[]): ProviderCheckpoint[ · 125 unmodified lines
157159
158160 export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
159161 private readonly codex = new CodexAppServerManager();
162+ private readonly filesystemCheckpointStore = new FilesystemCheckpointStore();
160163 private readonly threadLogsDir: string;
161164 private readonly threadLogStreams = new Map<string, fs.WriteStream>();
162165 private readonly sessionThreadIds = new Map<string, string>();
163166 private readonly pendingEventsBySession = new Map<string, ProviderEvent[]>();
167+ private readonly sessionCheckpointCwds = new Map<string, string>();
168+ private readonly filesystemLocks = new Map<string, Promise<void>>();
164169 private disposed = false;
165170 private readonly onCodexEvent = (event: ProviderEvent) => {
166171 if (this.disposed) {
@@ -169,6 +174,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 2 unmodified lines
169174
170175 this.routeEventToThreadLog(event);
171176 this.emit("event", event);
177+ this.maybeCaptureFilesystemCheckpoint(event);
172178 };
173179
174180 constructor() {
@@ -192,6 +198,11 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 17 unmodified lines
192198 this.sessionThreadIds.set(session.sessionId, session.threadId);
193199 this.flushPendingSessionEvents(session.sessionId, session.threadId);
194200 }
201+ await this.initializeFilesystemCheckpointing(session).catch((error) => {
202+ const message =
203+ error instanceof Error ? error.message : "Failed to initialize filesystem checkpoints.";
204+ this.emitFilesystemCheckpointError(session.sessionId, message, session.threadId);
205+ });
195206 return session;
196207 }
197208
@@ -227,6 +238,8 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 29 unmodified lines
227238 this.codex.stopSession(input.sessionId);
228239 this.sessionThreadIds.delete(input.sessionId);
229240 this.pendingEventsBySession.delete(input.sessionId);
241+ this.sessionCheckpointCwds.delete(input.sessionId);
242+ this.filesystemLocks.delete(input.sessionId);
230243 }
231244
232245 listSessions(): ProviderSession[] {
@@ -254,33 +267,68 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 21 unmodified lines
254267 throw new Error(`Unknown provider session: ${input.sessionId}`);
255268 }
256269
257− const beforeSnapshot = await this.codex.readThread(input.sessionId);
258− const currentTurnCount = beforeSnapshot.turns.length;
259− if (input.turnCount > currentTurnCount) {
260− throw new Error(
261− `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
262− );
263− }
270+ return this.withFilesystemLock(input.sessionId, async () => {
271+ const beforeSnapshot = await this.codex.readThread(input.sessionId);
272+ const currentTurnCount = beforeSnapshot.turns.length;
273+ if (input.turnCount > currentTurnCount) {
274+ throw new Error(
275+ `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
276+ );
277+ }
264278
265− const requestedRollbackTurns = currentTurnCount - input.turnCount;
266− const afterSnapshot =
267− requestedRollbackTurns > 0
268− ? await this.codex.rollbackThread(input.sessionId, requestedRollbackTurns)
269− : beforeSnapshot;
270− const checkpoints = buildCheckpoints(afterSnapshot.turns);
271− const currentCheckpoint =
272− checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
273− checkpoints[checkpoints.length - 1] ??
274− checkpoints[0];
275− const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length);
279+ const checkpointCwd = this.sessionCheckpointCwds.get(input.sessionId);
280+ if (checkpointCwd) {
281+ const hasCheckpoint = await this.filesystemCheckpointStore.hasCheckpoint({
282+ cwd: checkpointCwd,
283+ threadId: beforeSnapshot.threadId,
284+ turnCount: input.turnCount,
285+ });
286+ if (!hasCheckpoint) {
287+ throw new Error(
288+ `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
289+ );
290+ }
291+ }
276292
277− return {
278− threadId: afterSnapshot.threadId,
279− turnCount: currentCheckpoint?.turnCount ?? 0,
280− messageCount: currentCheckpoint?.messageCount ?? 0,
281− rolledBackTurns,
282− checkpoints,
283− };
293+ const requestedRollbackTurns = currentTurnCount - input.turnCount;
294+ const afterSnapshot =
295+ requestedRollbackTurns > 0
296+ ? await this.codex.rollbackThread(input.sessionId, requestedRollbackTurns)
297+ : beforeSnapshot;
298+
299+ if (checkpointCwd) {
300+ const restored = await this.filesystemCheckpointStore.restoreCheckpoint({
301+ cwd: checkpointCwd,
302+ threadId: beforeSnapshot.threadId,
303+ turnCount: input.turnCount,
304+ });
305+ if (!restored) {
306+ throw new Error(
307+ `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
308+ );
309+ }
310+ await this.filesystemCheckpointStore.pruneAfterTurn({
311+ cwd: checkpointCwd,
312+ threadId: afterSnapshot.threadId,
313+ maxTurnCount: afterSnapshot.turns.length,
314+ });
315+ }
316+
317+ const checkpoints = buildCheckpoints(afterSnapshot.turns);
318+ const currentCheckpoint =
319+ checkpoints.find((checkpoint) => checkpoint.isCurrent) ??
320+ checkpoints[checkpoints.length - 1] ??
321+ checkpoints[0];
322+ const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length);
323+
324+ return {
325+ threadId: afterSnapshot.threadId,
326+ turnCount: currentCheckpoint?.turnCount ?? 0,
327+ messageCount: currentCheckpoint?.messageCount ?? 0,
328+ rolledBackTurns,
329+ checkpoints,
330+ };
331+ });
284332 }
285333
286334 stopAll(): void {
@@ -300,6 +348,88 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> { · 13 unmodified lines
300348 this.threadLogStreams.clear();
301349 this.sessionThreadIds.clear();
302350 this.pendingEventsBySession.clear();
351+ this.sessionCheckpointCwds.clear();
352+ this.filesystemLocks.clear();
353+ }
354+
355+ private async initializeFilesystemCheckpointing(session: ProviderSession): Promise<void> {
356+ const cwd = session.cwd ?? process.cwd();
357+
358+ await this.withFilesystemLock(session.sessionId, async () => {
359+ const supportsGit = await this.filesystemCheckpointStore.isGitRepository(cwd);
360+ if (!supportsGit) {
361+ this.sessionCheckpointCwds.delete(session.sessionId);
362+ return;
363+ }
364+
365+ const snapshot = await this.codex.readThread(session.sessionId);
366+ await this.filesystemCheckpointStore.captureCheckpoint({
367+ cwd,
368+ threadId: snapshot.threadId,
369+ turnCount: snapshot.turns.length,
370+ });
371+ this.sessionCheckpointCwds.set(session.sessionId, cwd);
372+ });
373+ }
374+
375+ private maybeCaptureFilesystemCheckpoint(event: ProviderEvent): void {
376+ if (event.kind !== "notification" || event.method !== "turn/completed") {
377+ return;
378+ }
379+
380+ const checkpointCwd = this.sessionCheckpointCwds.get(event.sessionId);
381+ if (!checkpointCwd) {
382+ return;
383+ }
384+
385+ void this.withFilesystemLock(event.sessionId, async () => {
386+ if (!this.codex.hasSession(event.sessionId)) {
387+ return;
388+ }
389+
390+ const snapshot = await this.codex.readThread(event.sessionId);
391+ await this.filesystemCheckpointStore.captureCheckpoint({
392+ cwd: checkpointCwd,
393+ threadId: snapshot.threadId,
394+ turnCount: snapshot.turns.length,
395+ });
396+ }).catch((error) => {
397+ const message = error instanceof Error ? error.message : "Failed to capture checkpoint.";
398+ this.emitFilesystemCheckpointError(event.sessionId, message, event.threadId);
399+ });
400+ }
401+
402+ private emitFilesystemCheckpointError(
403+ sessionId: string,
404+ message: string,
405+ threadId?: string,
406+ ): void {
407+ this.emit("event", {
408+ id: randomUUID(),
409+ kind: "error",
410+ provider: "codex",
411+ sessionId,
412+ createdAt: new Date().toISOString(),
413+ method: "checkpoint/filesystemError",
414+ message,
415+ ...(threadId ? { threadId } : {}),
416+ });
417+ }
418+
419+ private withFilesystemLock<T>(sessionId: string, fn: () => Promise<T>): Promise<T> {
420+ const previous = this.filesystemLocks.get(sessionId) ?? Promise.resolve();
421+ const next = previous.catch(() => undefined).then(fn);
422+ const completion = next.then(
423+ () => undefined,
424+ () => undefined,
425+ );
426+ this.filesystemLocks.set(sessionId, completion);
427+ return next.finally(() => {
428+ const tracked = this.filesystemLocks.get(sessionId);
429+ if (tracked === completion) {
430+ this.filesystemLocks.delete(sessionId);
431+ }
432+ });
303433 }
304434
305435 private routeEventToThreadLog(event: ProviderEvent): void {