authorgravatar for julius0216@outlook.comJulius Marminge <julius0216@outlook.com> 2026-02-17 11:31:22-08:00
committergravatar for jmarminge@gmail.comJulius Marminge <jmarminge@gmail.com> 2026-02-18 21:59:17-08:00
logd2e6cfe10929eea55cbc70d6ffad8f7077b4d858
tree316a2c5dff34d0e70fcd86582de8b0222b7b42f8
parent351736baa119e8a9e71f1811edf37ef0615a0541

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

4 files changed, 609 insertions(+), 26 deletions(-)

apps/server/src/filesystemCheckpointStore.test.ts created+132
...@@ -0,0 +1,132 @@
1import { mkdtemp, readFile, rm, writeFile } from "node:fs/promises";
2import fs from "node:fs";
3import os from "node:os";
4import path from "node:path";
5
6import { describe, expect, it } from "vitest";
7
8import { checkpointRefForThreadTurn, FilesystemCheckpointStore } from "./filesystemCheckpointStore";
9import { runProcess } from "./processRunner";
10
11async 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
22function exists(filePath: string): boolean {
23 return fs.existsSync(filePath);
24}
25
26async 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
38describe("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 created+202
...@@ -0,0 +1,202 @@
1import { mkdtemp, rm, writeFile } from "node:fs/promises";
2import os from "node:os";
3import path from "node:path";
4import { randomUUID } from "node:crypto";
5
6import { runProcess } from "./processRunner";
7
8const CHECKPOINT_REFS_PREFIX = "refs/t3/checkpoints";
9const DEFAULT_TIMEOUT_MS = 30_000;
10const DEFAULT_MAX_OUTPUT_BYTES = 1_000_000;
11
12function quoteGitCommand(args: readonly string[]): string {
13 return `git ${args.join(" ")}`;
14}
15
16function checkpointRefThreadSegment(threadId: string): string {
17 return Buffer.from(threadId, "utf8").toString("base64url");
18}
19
20export 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
27interface GitOptions {
28 env?: NodeJS.ProcessEnv;
29 stdin?: string;
30 allowNonZeroExit?: boolean;
31}
32
33export 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,4 +1,4 @@
1import { describe, expect, it } from "vitest";1import { describe, expect, it, vi } from "vitest";
22
3import { ProviderManager } from "./providerManager";3import { ProviderManager } from "./providerManager";
44
...@@ -191,4 +191,123 @@ describe("ProviderManager", () => {...@@ -191,4 +191,123 @@ describe("ProviderManager", () => {
191191
192 manager.dispose();192 manager.dispose();
193 });193 });
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 });
194});313});
apps/server/src/providerManager.ts+155-25
...@@ -1,6 +1,7 @@...@@ -1,6 +1,7 @@
1import { EventEmitter } from "node:events";1import { EventEmitter } from "node:events";
2import fs from "node:fs";2import fs from "node:fs";
3import path from "node:path";3import path from "node:path";
4import { randomUUID } from "node:crypto";
45
5import {6import {
6 type ProviderCheckpoint,7 type ProviderCheckpoint,
...@@ -26,6 +27,7 @@ import {...@@ -26,6 +27,7 @@ import {
26} from "@t3tools/contracts";27} from "@t3tools/contracts";
27import type { CodexThreadTurnSnapshot } from "./codexAppServerManager";28import type { CodexThreadTurnSnapshot } from "./codexAppServerManager";
28import { CodexAppServerManager } from "./codexAppServerManager";29import { CodexAppServerManager } from "./codexAppServerManager";
30import { FilesystemCheckpointStore } from "./filesystemCheckpointStore";
2931
30export interface ProviderManagerEvents {32export interface ProviderManagerEvents {
31 event: [event: ProviderEvent];33 event: [event: ProviderEvent];
...@@ -157,10 +159,13 @@ function buildCheckpoints(turns: CodexThreadTurnSnapshot[]): ProviderCheckpoint[...@@ -157,10 +159,13 @@ function buildCheckpoints(turns: CodexThreadTurnSnapshot[]): ProviderCheckpoint[
157159
158export class ProviderManager extends EventEmitter<ProviderManagerEvents> {160export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
159 private readonly codex = new CodexAppServerManager();161 private readonly codex = new CodexAppServerManager();
162 private readonly filesystemCheckpointStore = new FilesystemCheckpointStore();
160 private readonly threadLogsDir: string;163 private readonly threadLogsDir: string;
161 private readonly threadLogStreams = new Map<string, fs.WriteStream>();164 private readonly threadLogStreams = new Map<string, fs.WriteStream>();
162 private readonly sessionThreadIds = new Map<string, string>();165 private readonly sessionThreadIds = new Map<string, string>();
163 private readonly pendingEventsBySession = new Map<string, ProviderEvent[]>();166 private readonly pendingEventsBySession = new Map<string, ProviderEvent[]>();
167 private readonly sessionCheckpointCwds = new Map<string, string>();
168 private readonly filesystemLocks = new Map<string, Promise<void>>();
164 private disposed = false;169 private disposed = false;
165 private readonly onCodexEvent = (event: ProviderEvent) => {170 private readonly onCodexEvent = (event: ProviderEvent) => {
166 if (this.disposed) {171 if (this.disposed) {
...@@ -169,6 +174,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {...@@ -169,6 +174,7 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
169174
170 this.routeEventToThreadLog(event);175 this.routeEventToThreadLog(event);
171 this.emit("event", event);176 this.emit("event", event);
177 this.maybeCaptureFilesystemCheckpoint(event);
172 };178 };
173179
174 constructor() {180 constructor() {
...@@ -192,6 +198,11 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {...@@ -192,6 +198,11 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
192 this.sessionThreadIds.set(session.sessionId, session.threadId);198 this.sessionThreadIds.set(session.sessionId, session.threadId);
193 this.flushPendingSessionEvents(session.sessionId, session.threadId);199 this.flushPendingSessionEvents(session.sessionId, session.threadId);
194 }200 }
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 });
195 return session;206 return session;
196 }207 }
197208
...@@ -227,6 +238,8 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {...@@ -227,6 +238,8 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
227 this.codex.stopSession(input.sessionId);238 this.codex.stopSession(input.sessionId);
228 this.sessionThreadIds.delete(input.sessionId);239 this.sessionThreadIds.delete(input.sessionId);
229 this.pendingEventsBySession.delete(input.sessionId);240 this.pendingEventsBySession.delete(input.sessionId);
241 this.sessionCheckpointCwds.delete(input.sessionId);
242 this.filesystemLocks.delete(input.sessionId);
230 }243 }
231244
232 listSessions(): ProviderSession[] {245 listSessions(): ProviderSession[] {
...@@ -254,33 +267,68 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {...@@ -254,33 +267,68 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
254 throw new Error(`Unknown provider session: ${input.sessionId}`);267 throw new Error(`Unknown provider session: ${input.sessionId}`);
255 }268 }
256269
257 const beforeSnapshot = await this.codex.readThread(input.sessionId);270 return this.withFilesystemLock(input.sessionId, async () => {
258 const currentTurnCount = beforeSnapshot.turns.length;271 const beforeSnapshot = await this.codex.readThread(input.sessionId);
259 if (input.turnCount > currentTurnCount) {272 const currentTurnCount = beforeSnapshot.turns.length;
260 throw new Error(273 if (input.turnCount > currentTurnCount) {
261 `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,274 throw new Error(
262 );275 `Checkpoint turn count ${input.turnCount} exceeds current turn count ${currentTurnCount}.`,
263 }276 );
277 }
264278
265 const requestedRollbackTurns = currentTurnCount - input.turnCount;279 const checkpointCwd = this.sessionCheckpointCwds.get(input.sessionId);
266 const afterSnapshot =280 if (checkpointCwd) {
267 requestedRollbackTurns > 0281 const hasCheckpoint = await this.filesystemCheckpointStore.hasCheckpoint({
268 ? await this.codex.rollbackThread(input.sessionId, requestedRollbackTurns)282 cwd: checkpointCwd,
269 : beforeSnapshot;283 threadId: beforeSnapshot.threadId,
270 const checkpoints = buildCheckpoints(afterSnapshot.turns);284 turnCount: input.turnCount,
271 const currentCheckpoint =285 });
272 checkpoints.find((checkpoint) => checkpoint.isCurrent) ??286 if (!hasCheckpoint) {
273 checkpoints[checkpoints.length - 1] ??287 throw new Error(
274 checkpoints[0];288 `Filesystem checkpoint is unavailable for turn ${input.turnCount} in thread ${beforeSnapshot.threadId}.`,
275 const rolledBackTurns = Math.max(0, currentTurnCount - afterSnapshot.turns.length);289 );
290 }
291 }
276292
277 return {293 const requestedRollbackTurns = currentTurnCount - input.turnCount;
278 threadId: afterSnapshot.threadId,294 const afterSnapshot =
279 turnCount: currentCheckpoint?.turnCount ?? 0,295 requestedRollbackTurns > 0
280 messageCount: currentCheckpoint?.messageCount ?? 0,296 ? await this.codex.rollbackThread(input.sessionId, requestedRollbackTurns)
281 rolledBackTurns,297 : beforeSnapshot;
282 checkpoints,298
283 };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 });
284 }332 }
285333
286 stopAll(): void {334 stopAll(): void {
...@@ -300,6 +348,88 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {...@@ -300,6 +348,88 @@ export class ProviderManager extends EventEmitter<ProviderManagerEvents> {
300 this.threadLogStreams.clear();348 this.threadLogStreams.clear();
301 this.sessionThreadIds.clear();349 this.sessionThreadIds.clear();
302 this.pendingEventsBySession.clear();350 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 });
303 }433 }
304434
305 private routeEventToThreadLog(event: ProviderEvent): void {435 private routeEventToThreadLog(event: ProviderEvent): void {