1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it, vi } from "@effect/vitest";
3import {
4 CheckpointId,
5 CheckpointRef,
6 EventId,
7 MessageId,
8 NodeId,
9 type OrchestrationV2DomainEvent,
10 ProviderDriverKind,
11 ProviderTurnId,
12 RunAttemptId,
13 RunId,
14 CheckpointScopeId,
15 ProjectId,
16 ProviderInstanceId,
17 ProviderSessionId,
18 ProviderThreadId,
19 ThreadId,
20} from "@t3tools/contracts";
21import * as Effect from "effect/Effect";
22import * as DateTime from "effect/DateTime";
23import * as SqlClient from "effect/sql/SqlClient";
24import * as Deferred from "effect/Deferred";
25import * as Fiber from "effect/Fiber";
26import * as FileSystem from "effect/FileSystem";
27import * as Path from "effect/Path";
28import * as Option from "effect/Option";
29import * as Layer from "effect/Layer";
30
31import * as SqlitePersistence from "../persistence/Sqlite.ts";
32import * as ServerConfig from "../config.ts";
33import * as ServerSettings from "../serverSettings.ts";
34import * as GitWorkflow from "../git/GitWorkflowService.ts";
35import { withTurnStartReservation, withWorkspaceLease } from "../workspace/workspaceLease.ts";
36import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";
37import { resolveCodexRollbackTurnCount } from "./Adapters/CodexAdapterV2.ts";
38import * as CheckpointService from "./CheckpointService.ts";
39import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
40import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
41import * as EventSink from "./EventSink.ts";
42import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
43import * as ProjectionStore from "./ProjectionStore.ts";
44import * as ProjectStore from "./ProjectStore.ts";
45import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
46import * as ProviderSessionManager from "./ProviderSessionManager.ts";
47import * as RuntimePolicy from "./RuntimePolicy.ts";
48
49const unrelatedProject = Option.some({
50 workspaceRoot: "/nonexistent/t3-rollback-project",
51} as never);
52const layerCheckpointRollbackService = CheckpointRollbackService.layer.pipe(
53 Layer.provide(
54 Layer.mergeAll(
55 NodeServices.layer,
56 Layer.mock(ServerConfig.ServerConfig)({ worktreesDir: "/synthetic-worktrees" } as never),
57 ServerSettings.layerTest(),
58 Layer.mock(GitWorkflow.GitWorkflowService)({}),
59 Layer.mock(ProjectStore.ProjectStoreV2)({ get: () => Effect.succeed(unrelatedProject) }),
60 ),
61 ),
62);
63
64it.effect.each(["initial", "waiting"])(
65 "rejects a non-ready checkpoint from %s before opening a session or restoring files",
66 (state) => {
67 const threadId = ThreadId.make("thread:rollback-non-ready");
68 const providerThreadId = ProviderThreadId.make("provider-thread:rollback-non-ready");
69 const providerSessionId = ProviderSessionId.make("provider-session:rollback-non-ready");
70 const checkpointId = CheckpointId.make("checkpoint:rollback-non-ready");
71 const scopeId = CheckpointScopeId.make("checkpoint-scope:rollback-non-ready");
72 const providerInstanceId = ProviderInstanceId.make("provider_rollback_non_ready");
73 const restore = vi.fn(() => Effect.die("checkpoint restore must not run"));
74 const open = vi.fn(() => Effect.die("provider session open must not run"));
75 const resolveRuntimePolicy = vi.fn(() => Effect.die("runtime policy resolution must not run"));
76 const readEntered = Deferred.makeUnsafe<void>();
77 const release = Deferred.makeUnsafe<void>();
78 let projection = {
79 thread: {
80 worktreePath: process.cwd(),
81 activeProviderThreadId: providerThreadId,
82 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
83 },
84 providerThread: { id: providerThreadId, providerSessionId, providerInstanceId },
85 checkpoint: {
86 id: checkpointId,
87 scopeId,
88 status: state === "waiting" ? "ready" : "stale",
89 appRunOrdinal: 0,
90 },
91 scope: { id: scopeId, cwd: process.cwd() },
92 } as unknown as ProjectionStore.ProjectionCheckpointRollbackContext;
93 const getThread = vi.fn<ProjectionStore.ProjectionStoreV2Shape["getThread"]>(() =>
94 Deferred.succeed(readEntered, undefined).pipe(Effect.as(projection.thread)),
95 );
96 const getCheckpointRollbackContext = vi.fn(() => Effect.succeed(projection));
97 const testLayer = layerCheckpointRollbackService.pipe(
98 Layer.provide(
99 Layer.mergeAll(
100 ThreadCommandExecutor.layer,
101 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
102 Layer.mock(EventSink.EventSinkV2)({}),
103 IdAllocator.layer,
104 Layer.mock(ProjectionStore.ProjectionStoreV2)({
105 getThread,
106 getCheckpointRollbackContext,
107 getShellSnapshot: () =>
108 Effect.succeed({
109 schemaVersion: 1,
110 snapshotSequence: 0,
111 threads: [],
112 archivedThreads: [],
113 }),
114 }),
115 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open }),
116 Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy }),
117 ),
118 ),
119 );
120
121 return Effect.gen(function* () {
122 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
123 const holder =
124 state === "waiting"
125 ? yield* withWorkspaceLease(process.cwd(), Deferred.await(release)).pipe(
126 Effect.forkChild({ startImmediately: true }),
127 )
128 : undefined;
129 const rollback = yield* service
130 .execute({
131 threadId,
132 providerThreadId,
133 checkpointId,
134 scopeId,
135 })
136 .pipe(Effect.flip, Effect.forkChild({ startImmediately: true }));
137 if (state === "waiting") {
138 yield* Deferred.await(readEntered);
139 projection = {
140 ...projection,
141 checkpoint: { ...projection.checkpoint!, status: "stale" },
142 };
143 yield* Deferred.succeed(release, undefined);
144 if (holder !== undefined) yield* Fiber.join(holder);
145 }
146 const error = yield* Fiber.join(rollback);
147
148 assert.deepEqual(getThread.mock.calls, [[threadId]]);
149 assert.deepEqual(getCheckpointRollbackContext.mock.calls, [
150 [
151 threadId,
152 {
153 threadId,
154 providerThreadId,
155 checkpointId,
156 scopeId,
157 },
158 ],
159 ]);
160 assert.equal(error.reason, "rollback-target-invalid");
161 assert.equal(
162 error.message,
163 `Rollback target ${checkpointId} for provider thread ${providerThreadId} on thread ${threadId} is incomplete or invalid.`,
164 );
165 assert.equal(error.cause, undefined);
166 assert.equal(resolveRuntimePolicy.mock.calls.length, 0);
167 assert.equal(open.mock.calls.length, 0);
168 assert.equal(restore.mock.calls.length, 0);
169 }).pipe(Effect.provide(testLayer));
170 },
171);
172
173it.effect.each([
174 { order: "cleanup-first", restoreFiles: true },
175 { order: "rollback-first", restoreFiles: true },
176 { order: "cleanup-first", restoreFiles: false, unmanaged: true },
177 { order: "cleanup-first", restoreFiles: true, leavesFolder: true },
178 { order: "cleanup-first", restoreFiles: false, leavesFolder: true },
179 { order: "cleanup-first", restoreFiles: true, leavesFolder: true, shared: true },
180 { order: "cleanup-first", restoreFiles: true, leavesFolder: true, createFails: true },
181])("locks rollback with %s", (row) =>
182 Effect.gen(function* () {
183 const { order, restoreFiles, unmanaged, leavesFolder, shared, createFails } = row;
184 const fileSystem = yield* FileSystem.FileSystem;
185 const path = yield* Path.Path;
186 const workspaceRoot = yield* fileSystem.realPath(
187 yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-rollback-" }),
188 );
189 const worktreesDir = path.join(workspaceRoot, "worktrees");
190 const worktreePath = unmanaged
191 ? path.join(workspaceRoot, "elsewhere", "feature")
192 : path.join(worktreesDir, "repo", "feature");
193 yield* fileSystem.makeDirectory(path.join(worktreePath, ".git"), { recursive: true });
194 const threadId = ThreadId.make("rollback-worktree");
195 const projectId = ProjectId.make("rollback-worktree-project");
196 const providerThreadId = ProviderThreadId.make("rollback-worktree-provider-thread");
197 const providerSessionId = ProviderSessionId.make("rollback-worktree-provider-session");
198 const providerInstanceId = ProviderInstanceId.make("rollback-worktree-provider-instance");
199 const checkpointId = CheckpointId.make("rollback-worktree-checkpoint");
200 const scopeId = CheckpointScopeId.make("rollback-worktree-scope");
201 const branch = "feature/rollback";
202 const release = Deferred.makeUnsafe<void>();
203 const openEntered = Deferred.makeUnsafe<void>();
204 const calls: string[] = [];
205 let cleanupEntered = false;
206 const providerThread = { id: providerThreadId, providerSessionId, providerInstanceId };
207 const projection = {
208 thread: {
209 id: threadId,
210 projectId,
211 worktreePath,
212 branch,
213 activeProviderThreadId: providerThreadId,
214 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
215 },
216 providerThread,
217 providerSession: undefined,
218 checkpoint: { id: checkpointId, scopeId, status: "ready", appRunOrdinal: 0 },
219 scope: { id: scopeId, cwd: worktreePath },
220 runsToRollback: [],
221 targetAttempt: undefined,
222 rolledBackAttemptIds: [],
223 staleCheckpoints: [],
224 rootNodes: [],
225 providerTurns: [],
226 } as unknown as ProjectionStore.ProjectionCheckpointRollbackContext;
227 const testLayer = CheckpointRollbackService.layer.pipe(
228 Layer.provide(
229 Layer.mergeAll(
230 ThreadCommandExecutor.layer,
231 NodeServices.layer,
232 Layer.mock(ServerConfig.ServerConfig)({ worktreesDir } as never),
233 ServerSettings.layerTest(),
234 Layer.mock(ProjectStore.ProjectStoreV2)({
235 get: (id) => {
236 assert.equal(id, projectId);
237 return Effect.succeedSome({ workspaceRoot } as never);
238 },
239 }),
240 Layer.mock(GitWorkflow.GitWorkflowService)({
241 pruneWorktrees: (input) =>
242 Effect.sync(() => {
243 assert.deepEqual(input, { cwd: workspaceRoot });
244 calls.push("prune");
245 }),
246 createWorktree: (input) =>
247 Effect.gen(function* () {
248 assert.deepEqual(input, {
249 cwd: workspaceRoot,
250 refName: branch,
251 path: worktreePath,
252 });
253 calls.push("create");
254 if (createFails) return yield* Effect.die("the branch is gone");
255 yield* fileSystem
256 .makeDirectory(path.join(worktreePath, ".git"), { recursive: true })
257 .pipe(Effect.orDie);
258 return {} as never;
259 }),
260 }),
261 Layer.mock(CheckpointService.CheckpointServiceV2)({
262 restore: () =>
263 Effect.gen(function* () {
264 assert.isTrue(
265 yield* fileSystem.exists(path.join(worktreePath, ".git")).pipe(Effect.orDie),
266 );
267 calls.push("files");
268 }),
269 }),
270 Layer.mock(EventSink.EventSinkV2)({
271 write: () =>
272 Effect.sync(() => {
273 calls.push("projection");
274 return [];
275 }),
276 }),
277 IdAllocator.layer,
278 Layer.mock(ProjectionStore.ProjectionStoreV2)({
279 getThread: () => Effect.succeed(projection.thread),
280 getCheckpointRollbackContext: () => Effect.succeed(projection),
281 getShellSnapshot: () =>
282 Effect.succeed({
283 schemaVersion: 1,
284 snapshotSequence: 0,
285 threads: shared
286 ? [
287 {
288 id: ThreadId.make("rollback-worktree-sharer"),
289 projectId,
290 worktreePath,
291 deletedAt: null,
292 lineage: {},
293 },
294 ]
295 : [],
296 archivedThreads: [],
297 } as never),
298 getCheckpointContext: () => Effect.succeed({ checkpointScopes: [] } as never),
299 getThreadProviderContext: () => Effect.succeed({ providerSessions: [] } as never),
300 }),
301 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
302 open: (input) =>
303 Effect.gen(function* () {
304 assert.equal(input.runtimePolicy.cwd, worktreePath);
305 if (!(yield* fileSystem.exists(worktreePath).pipe(Effect.orDie))) {
306 return yield* new ProviderWorkspaceMissingError({
307 threadId,
308 cwd: worktreePath,
309 });
310 }
311 calls.push("session");
312 yield* Deferred.succeed(openEntered, undefined);
313 if (order === "rollback-first") yield* Deferred.await(release);
314 return {} as never;
315 }),
316 }),
317 Layer.mock(RuntimePolicy.RuntimePolicyV2)({
318 resolve: ({ thread }) => {
319 assert.equal(thread.worktreePath, worktreePath);
320 return Effect.succeed({ cwd: thread.worktreePath } as never);
321 },
322 }),
323 ),
324 ),
325 );
326 yield* Effect.gen(function* () {
327 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
328 const cleanupEffect = withWorkspaceLease(
329 worktreePath,
330 Effect.gen(function* () {
331 cleanupEntered = true;
332 if (order === "cleanup-first") yield* Deferred.await(release);
333 yield* fileSystem.remove(worktreePath, { recursive: true, force: true });
334 // A failed `git worktree remove` can leave the folder without its checkout.
335 if (leavesFolder) {
336 yield* fileSystem.makeDirectory(worktreePath);
337 yield* fileSystem.writeFileString(path.join(worktreePath, "unsaved.txt"), "keep");
338 }
339 calls.push("cleanup");
340 }),
341 );
342 const cleanup =
343 order === "cleanup-first"
344 ? yield* cleanupEffect.pipe(Effect.forkChild({ startImmediately: true }))
345 : undefined;
346 const rollback = yield* service
347 .execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles })
348 .pipe(Effect.forkChild({ startImmediately: true }));
349 if (order === "cleanup-first") assert.deepEqual(calls, []);
350 if (order === "rollback-first") yield* Deferred.await(openEntered);
351 const laterCleanup =
352 order === "rollback-first"
353 ? yield* cleanupEffect.pipe(Effect.forkChild({ startImmediately: true }))
354 : undefined;
355 if (order === "rollback-first") assert.isFalse(cleanupEntered);
356 yield* Deferred.succeed(release, undefined);
357 if (cleanup !== undefined) yield* Fiber.join(cleanup);
358 if (unmanaged || shared) {
359 const error = yield* Fiber.join(rollback).pipe(Effect.flip);
360 assert.equal(error._tag, "CheckpointRollbackExecutionError");
361 assert.deepEqual(calls, ["cleanup"]);
362 if (shared) {
363 assert.equal(error.reason, "shared-workspace");
364 assert.isTrue(yield* fileSystem.exists(path.join(worktreePath, "unsaved.txt")));
365 }
366 return;
367 }
368 if (createFails) {
369 yield* Fiber.join(rollback).pipe(Effect.exit);
370 assert.deepEqual(calls, ["cleanup", "prune", "create"]);
371 assert.equal(
372 yield* fileSystem.readFileString(path.join(worktreePath, "unsaved.txt")),
373 "keep",
374 );
375 assert.deepEqual(yield* fileSystem.readDirectory(path.dirname(worktreePath)), ["feature"]);
376 return;
377 }
378 yield* Fiber.join(rollback);
379 if (laterCleanup !== undefined) yield* Fiber.join(laterCleanup);
380 const rollbackCalls = restoreFiles
381 ? ["session", "files", "projection"]
382 : ["session", "projection"];
383 const recreates = !leavesFolder || restoreFiles;
384 assert.deepEqual(
385 calls,
386 order === "cleanup-first"
387 ? ["cleanup", ...(recreates ? ["prune", "create"] : []), ...rollbackCalls]
388 : [...rollbackCalls, "cleanup"],
389 );
390 if (!recreates) {
391 assert.isTrue(yield* fileSystem.exists(path.join(worktreePath, "unsaved.txt")));
392 }
393 }).pipe(Effect.provide(testLayer));
394 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
395);
396
397it.effect.each([{ restoreFiles: true }, { restoreFiles: false }])(
398 "holds turn starts inside the worktree during rollback with restoreFiles $restoreFiles",
399 ({ restoreFiles }) =>
400 Effect.gen(function* () {
401 const fileSystem = yield* FileSystem.FileSystem;
402 const path = yield* Path.Path;
403 const workspaceRoot = yield* fileSystem.realPath(
404 yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-rollback-reservation-" }),
405 );
406 const worktreesDir = path.join(workspaceRoot, "worktrees");
407 const worktreePath = path.join(worktreesDir, "repo", "feature");
408 yield* fileSystem.makeDirectory(path.join(worktreePath, ".git"), { recursive: true });
409 const threadId = ThreadId.make("rollback-reservation");
410 const providerThreadId = ProviderThreadId.make("rollback-reservation-provider-thread");
411 const providerInstanceId = ProviderInstanceId.make("rollback-reservation-provider");
412 const checkpointId = CheckpointId.make("rollback-reservation-checkpoint");
413 const scopeId = CheckpointScopeId.make("rollback-reservation-scope");
414 const release = Deferred.makeUnsafe<void>();
415 const openEntered = Deferred.makeUnsafe<void>();
416 const restoreEntered = Deferred.makeUnsafe<void>();
417 const releaseRestore = Deferred.makeUnsafe<void>();
418 const calls: string[] = [];
419 const projection = {
420 thread: {
421 id: threadId,
422 projectId: ProjectId.make("rollback-reservation-project"),
423 worktreePath,
424 branch: "feature/rollback-reservation",
425 activeProviderThreadId: providerThreadId,
426 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
427 },
428 providerThread: {
429 id: providerThreadId,
430 providerSessionId: ProviderSessionId.make("rollback-reservation-session"),
431 providerInstanceId,
432 },
433 providerSession: undefined,
434 checkpoint: { id: checkpointId, scopeId, status: "ready", appRunOrdinal: 0 },
435 scope: { id: scopeId, cwd: worktreePath },
436 runsToRollback: [],
437 targetAttempt: undefined,
438 rolledBackAttemptIds: [],
439 staleCheckpoints: [],
440 rootNodes: [],
441 providerTurns: [],
442 } as unknown as ProjectionStore.ProjectionCheckpointRollbackContext;
443 const testLayer = CheckpointRollbackService.layer.pipe(
444 Layer.provide(
445 Layer.mergeAll(
446 ThreadCommandExecutor.layer,
447 NodeServices.layer,
448 Layer.mock(ServerConfig.ServerConfig)({ worktreesDir } as never),
449 ServerSettings.layerTest(),
450 Layer.mock(ProjectStore.ProjectStoreV2)({
451 get: () => Effect.succeedSome({ workspaceRoot } as never),
452 }),
453 Layer.mock(GitWorkflow.GitWorkflowService)({}),
454 Layer.mock(CheckpointService.CheckpointServiceV2)({
455 restore: () =>
456 Deferred.succeed(restoreEntered, undefined).pipe(
457 Effect.andThen(Deferred.await(releaseRestore)),
458 Effect.andThen(Effect.sync(() => calls.push("files"))),
459 ),
460 }),
461 Layer.mock(EventSink.EventSinkV2)({ write: () => Effect.succeed([]) }),
462 IdAllocator.layer,
463 Layer.mock(ProjectionStore.ProjectionStoreV2)({
464 getThread: () => Effect.succeed(projection.thread),
465 getCheckpointRollbackContext: () => Effect.succeed(projection),
466 getShellSnapshot: () =>
467 Effect.succeed({
468 schemaVersion: 1,
469 snapshotSequence: 0,
470 threads: [],
471 archivedThreads: [],
472 }),
473 }),
474 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
475 open: () =>
476 Deferred.succeed(openEntered, undefined).pipe(
477 Effect.andThen(Deferred.await(release)),
478 Effect.as({} as never),
479 ),
480 }),
481 Layer.mock(RuntimePolicy.RuntimePolicyV2)({
482 resolve: () => Effect.succeed({ cwd: worktreePath } as never),
483 }),
484 ),
485 ),
486 );
487 yield* Effect.gen(function* () {
488 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
489 const rollback = yield* service
490 .execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles })
491 .pipe(Effect.forkChild({ startImmediately: true }));
492 yield* Deferred.await(openEntered);
493 // A synchronous realPath lets the forked start reach the reservation
494 // before the next assertion, so "not entered" means "waiting".
495 const startTurn = withTurnStartReservation(
496 FileSystem.makeNoop({ realPath: (target) => Effect.succeed(target) }),
497 path,
498 path.join(worktreePath, "packages", "app"),
499 Effect.sync(() => calls.push("turn start")),
500 ).pipe(Effect.forkChild({ startImmediately: true }));
501 const turnStart = yield* startTurn;
502 if (restoreFiles) {
503 assert.deepEqual(calls, []);
504 yield* Deferred.succeed(release, undefined);
505 // The reservation must still be held while files are being restored,
506 // not only while the provider rollback is in flight.
507 yield* Deferred.await(restoreEntered);
508 const duringRestore = yield* startTurn;
509 assert.deepEqual(calls, []);
510 yield* Deferred.succeed(releaseRestore, undefined);
511 yield* Fiber.join(duringRestore);
512 } else {
513 // Nothing is rewritten, so the turn start must not wait for the provider.
514 yield* Fiber.join(turnStart);
515 yield* Deferred.succeed(release, undefined);
516 }
517 yield* Fiber.join(rollback);
518 yield* Fiber.join(turnStart);
519 assert.deepEqual(
520 calls,
521 restoreFiles ? ["files", "turn start", "turn start"] : ["turn start"],
522 );
523 }).pipe(Effect.provide(testLayer));
524 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
525);
526
527const seedRollbackHistory = Effect.gen(function* () {
528 const store = yield* ProjectionStore.ProjectionStoreV2;
529 const now = yield* DateTime.now;
530 const threadId = ThreadId.make("rollback-history");
531 const providerThreadId = ProviderThreadId.make("rollback-history-provider");
532 const providerSessionId = ProviderSessionId.make("rollback-history-session");
533 const providerInstanceId = ProviderInstanceId.make("codex");
534 const driver = ProviderDriverKind.make("codex");
535 const modelSelection = { instanceId: providerInstanceId, model: "test" };
536 const scopeId = CheckpointScopeId.make("rollback-history-root");
537 const otherScopeId = CheckpointScopeId.make("rollback-history-child");
538 const runId = (ordinal: number) => RunId.make(`rollback-history-run-${ordinal}`);
539 const nodeId = (ordinal: number) => NodeId.make(`rollback-history-node-${ordinal}`);
540 const attemptId = (ordinal: number) => RunAttemptId.make(`rollback-history-attempt-${ordinal}`);
541 const turnId = (ordinal: number) => ProviderTurnId.make(`rollback-history-turn-${ordinal}`);
542 const checkpointId = (ordinal: number) =>
543 CheckpointId.make(`rollback-history-checkpoint-${ordinal}`);
544 type EventInput<Event = OrchestrationV2DomainEvent> = Event extends OrchestrationV2DomainEvent
545 ? Omit<Event, "id" | "occurredAt" | "threadId">
546 : never;
547 let sequence = 0;
548 const apply = (event: EventInput) =>
549 store.apply({
550 ...event,
551 id: EventId.make(`rollback-history-event-${sequence++}`),
552 threadId,
553 occurredAt: now,
554 } as OrchestrationV2DomainEvent);
555 yield* apply({
556 type: "thread.created",
557 payload: {
558 id: threadId,
559 createdBy: "user",
560 creationSource: "web",
561 projectId: ProjectId.make("rollback-history-project"),
562 title: "Rollback history",
563 providerInstanceId,
564 modelSelection,
565 runtimeMode: "full-access",
566 interactionMode: "default",
567 branch: null,
568 worktreePath: process.cwd(),
569 activeProviderThreadId: providerThreadId,
570 lineage: { parentThreadId: null, relationshipToParent: null, rootThreadId: threadId },
571 forkedFrom: null,
572 createdAt: now,
573 updatedAt: now,
574 archivedAt: null,
575 settledOverride: null,
576 settledAt: null,
577 lastVisitedAt: null,
578 deletedAt: null,
579 },
580 });
581 const providerThread = {
582 id: providerThreadId,
583 driver,
584 providerInstanceId,
585 providerSessionId,
586 appThreadId: threadId,
587 ownerNodeId: null,
588 nativeThreadRef: null,
589 nativeConversationHeadRef: null,
590 status: "idle" as const,
591 firstRunOrdinal: 1,
592 lastRunOrdinal: 6,
593 handoffIds: [],
594 forkedFrom: null,
595 createdAt: now,
596 updatedAt: now,
597 };
598 yield* apply({ type: "provider-thread.updated", payload: providerThread });
599 for (const id of [scopeId, otherScopeId]) {
600 yield* apply({
601 type: "checkpoint-scope.created",
602 payload: {
603 id,
604 threadId,
605 runId: runId(1),
606 nodeId: nodeId(1),
607 parentScopeId: id === scopeId ? null : scopeId,
608 providerThreadId,
609 kind: id === scopeId ? "root_run" : "subagent",
610 ordinalWithinParent: 0,
611 advancesAppRunCount: id === scopeId,
612 cwd: process.cwd(),
613 createdAt: now,
614 },
615 });
616 }
617 const statuses = [
618 "completed",
619 "completed",
620 "completed",
621 "interrupted",
622 "failed",
623 "cancelled",
624 "rolled_back",
625 ] as const;
626 for (const [index, status] of statuses.entries()) {
627 const ordinal = index + 1;
628 yield* apply({
629 type: "run.created",
630 payload: {
631 id: runId(ordinal),
632 threadId,
633 ordinal,
634 providerInstanceId,
635 modelSelection,
636 providerThreadId,
637 userMessageId: MessageId.make(`rollback-history-message-${ordinal}`),
638 rootNodeId: nodeId(ordinal),
639 activeAttemptId: attemptId(ordinal),
640 status,
641 requestedAt: now,
642 startedAt: now,
643 completedAt: now,
644 checkpointId: checkpointId(ordinal),
645 contextHandoffId: null,
646 },
647 });
648 yield* apply({
649 type: "node.updated",
650 payload: {
651 id: nodeId(ordinal),
652 threadId,
653 runId: runId(ordinal),
654 parentNodeId: null,
655 rootNodeId: nodeId(ordinal),
656 kind: "root_turn",
657 status,
658 countsForRun: true,
659 providerThreadId,
660 providerTurnId: turnId(ordinal),
661 nativeItemRef: null,
662 runtimeRequestId: null,
663 checkpointScopeId: scopeId,
664 startedAt: now,
665 completedAt: now,
666 },
667 });
668 // Rollback must exclude every attempt of discarded runs, including an inactive retry.
669 for (const attemptOrdinal of ordinal === 7 ? [6.5, 7] : [ordinal]) {
670 yield* apply({
671 type: "run-attempt.created",
672 runId: runId(ordinal),
673 payload: {
674 id: attemptId(attemptOrdinal),
675 runId: runId(ordinal),
676 attemptOrdinal: attemptOrdinal === ordinal ? 2 : 1,
677 rootNodeId: nodeId(ordinal),
678 providerInstanceId,
679 providerThreadId,
680 providerTurnId: turnId(attemptOrdinal),
681 reason: "initial",
682 status: "completed",
683 startedAt: now,
684 completedAt: now,
685 },
686 });
687 yield* apply({
688 type: "provider-turn.updated",
689 payload: {
690 id: turnId(attemptOrdinal),
691 providerThreadId,
692 nodeId: nodeId(ordinal),
693 runAttemptId: attemptId(attemptOrdinal),
694 nativeTurnRef: { driver, nativeId: `native-turn-${attemptOrdinal}`, strength: "strong" },
695 ordinal: attemptOrdinal * 2,
696 status: status === "rolled_back" ? "completed" : status,
697 startedAt: now,
698 completedAt: now,
699 },
700 });
701 }
702 }
703 // A continuation shares the target attempt; its first provider turn is no longer the boundary.
704 yield* apply({
705 type: "provider-turn.updated",
706 payload: {
707 id: turnId(2.5),
708 providerThreadId,
709 nodeId: nodeId(2),
710 runAttemptId: attemptId(2),
711 nativeTurnRef: { driver, nativeId: "native-turn-2-continuation", strength: "strong" },
712 ordinal: 5,
713 status: "completed",
714 startedAt: now,
715 completedAt: now,
716 },
717 });
718 for (let ordinal = 0; ordinal <= 8; ordinal++) {
719 yield* apply({
720 type: "checkpoint.captured",
721 payload: {
722 id: checkpointId(ordinal),
723 threadId,
724 scopeId: ordinal === 8 ? otherScopeId : scopeId,
725 runId: ordinal === 0 ? null : runId(Math.min(ordinal, 7)),
726 nodeId: nodeId(Math.max(1, Math.min(ordinal, 7))),
727 parentCheckpointId: null,
728 ordinalWithinScope: ordinal,
729 appRunOrdinal: ordinal === 0 ? null : ordinal,
730 ref: CheckpointRef.make(`refs/t3/rollback-history/${ordinal}`),
731 status: ordinal === 7 ? "stale" : "ready",
732 files: [],
733 capturedAt: now,
734 },
735 });
736 }
737
738 const rollbackCounts: number[] = [];
739 const restored: CheckpointId[] = [];
740 const deletedRefs: CheckpointId[] = [];
741 const committed: OrchestrationV2DomainEvent[] = [];
742 const serviceLayer = layerCheckpointRollbackService.pipe(
743 Layer.provide(
744 Layer.mergeAll(
745 ThreadCommandExecutor.layer,
746 IdAllocator.layer,
747 Layer.mock(CheckpointService.CheckpointServiceV2)({
748 restore: ({ checkpoint }) =>
749 Effect.sync(() => {
750 restored.push(checkpoint.id);
751 }),
752 deleteStaleRefs: ({ checkpoints }) =>
753 Effect.sync(() => {
754 deletedRefs.push(...checkpoints.map((checkpoint) => checkpoint.id));
755 }),
756 }),
757 Layer.mock(EventSink.EventSinkV2)({
758 write: ({ events }) =>
759 Effect.gen(function* () {
760 for (const event of events) yield* store.apply(event);
761 committed.push(...events);
762 return [];
763 }).pipe(
764 Effect.mapError(
765 (cause) =>
766 new EventSink.EventSinkWriteError({
767 eventCount: events.length,
768 cause,
769 }),
770 ),
771 ),
772 }),
773 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
774 open: () =>
775 Effect.succeed({
776 rollbackThread: (input: ProviderAdapter.ProviderAdapterV2RollbackThreadInput) =>
777 Effect.gen(function* () {
778 rollbackCounts.push(yield* resolveCodexRollbackTurnCount(input));
779 return { providerThread: input.providerThread };
780 }),
781 } as never),
782 }),
783 Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: () => Effect.succeed({} as never) }),
784 ),
785 ),
786 );
787 const rollback = (ordinal: number) =>
788 Effect.gen(function* () {
789 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
790 yield* service.execute({
791 threadId,
792 providerThreadId,
793 scopeId,
794 checkpointId: checkpointId(ordinal),
795 });
796 }).pipe(Effect.provide(serviceLayer));
797
798 return {
799 store,
800 now,
801 threadId,
802 runId,
803 nodeId,
804 checkpointId,
805 rollback,
806 rollbackCounts,
807 restored,
808 deletedRefs,
809 committed,
810 };
811});
812
813it.layer(ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)))(
814 "rollback history reads",
815 (it) => {
816 it.effect("rolls back later terminal runs despite obsolete unrelated SQLite payloads", () =>
817 Effect.gen(function* () {
818 const history = yield* seedRollbackHistory;
819 const sql = yield* SqlClient.SqlClient;
820 const { threadId, runId, nodeId, checkpointId } = history;
821 yield* sql`
822 INSERT INTO orchestration_v2_projection_turn_items (
823 turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
824 parent_item_id, ordinal, type, status, updated_at, payload_json
825 ) VALUES ('rollback-obsolete-transcript', ${threadId}, ${runId(1)}, ${nodeId(1)},
826 NULL, NULL, NULL, 1, 'assistant_message', 'completed', ${DateTime.formatIso(history.now)},
827 '{"obsolete":"transcript"}')`;
828 yield* sql`UPDATE orchestration_v2_projection_runs
829 SET payload_json = '{"obsolete":"run"}' WHERE run_id = ${runId(1)}`;
830 yield* sql`UPDATE orchestration_v2_projection_nodes
831 SET payload_json = '{"obsolete":"node"}' WHERE node_id = ${nodeId(1)}`;
832 yield* sql`UPDATE orchestration_v2_projection_checkpoints
833 SET payload_json = json_set(payload_json, '$.files', 'obsolete file summary')
834 WHERE checkpoint_id IN (${checkpointId(0)}, ${checkpointId(1)}, ${checkpointId(7)}, ${checkpointId(8)})`;
835 assert.equal(
836 (yield* Effect.exit(history.store.getThreadProjection(threadId)))._tag,
837 "Failure",
838 );
839
840 yield* history.rollback(2);
841
842 assert.deepEqual(history.rollbackCounts, [4]);
843 assert.deepEqual(history.restored, [checkpointId(2)]);
844 assert.sameMembers(history.deletedRefs, [3, 4, 5, 6].map(checkpointId));
845 const runs = yield* sql<{ ordinal: number; status: string }>`
846 SELECT ordinal, status FROM orchestration_v2_projection_runs
847 WHERE thread_id = ${threadId} ORDER BY ordinal`;
848 assert.deepEqual(
849 runs.map((run) => run.status),
850 [
851 "completed",
852 "completed",
853 "rolled_back",
854 "rolled_back",
855 "rolled_back",
856 "rolled_back",
857 "rolled_back",
858 ],
859 );
860 const nodes = yield* sql<{ status: string }>`
861 SELECT status FROM orchestration_v2_projection_nodes
862 WHERE thread_id = ${threadId} ORDER BY node_id`;
863 assert.deepEqual(
864 nodes.map((node) => node.status),
865 runs.map((run) => run.status),
866 );
867 const checkpoints = yield* sql<{ status: string }>`
868 SELECT status FROM orchestration_v2_projection_checkpoints
869 WHERE thread_id = ${threadId} ORDER BY ordinal_within_scope`;
870 assert.deepEqual(
871 checkpoints.map((checkpoint) => checkpoint.status),
872 ["ready", "ready", "ready", "stale", "stale", "stale", "stale", "stale", "ready"],
873 );
874 }),
875 );
876
877 it.effect(
878 "rewinds stopped runs repeatedly without counting discarded provider turns again",
879 () =>
880 Effect.gen(function* () {
881 const history = yield* seedRollbackHistory;
882 yield* history.rollback(5);
883 yield* history.rollback(2);
884 yield* history.rollback(0);
885 yield* history.rollback(0);
886
887 assert.deepEqual(history.rollbackCounts, [1, 3, 3]);
888 assert.sameMembers(history.deletedRefs, [1, 2, 3, 4, 5, 6].map(history.checkpointId));
889 assert.sameMembers(
890 history.committed
891 .filter((event) => event.type === "run.updated")
892 .map((event) => event.payload.id),
893 [1, 2, 3, 4, 5, 6].map(history.runId),
894 );
895 const records = yield* history.store.getThreadRecords(history.threadId, [
896 "runs",
897 "nodes",
898 "checkpoints",
899 ]);
900 assert.isTrue(records.runs.every((run) => run.status === "rolled_back"));
901 assert.isTrue(records.nodes.every((node) => node.status === "rolled_back"));
902 assert.sameMembers(
903 records.checkpoints
904 .filter((checkpoint) => checkpoint.status === "ready")
905 .map((checkpoint) => checkpoint.id),
906 [history.checkpointId(0), history.checkpointId(8)],
907 );
908 }),
909 );
910 },
911);