1import * as NodeServices from "@effect/platform-node/NodeServices";
2import { assert, it, vi } from "@effect/vitest";
3import {
4 CheckpointId,
5 CheckpointScopeId,
6 type OrchestrationV2ThreadProjection,
7 ProjectId,
8 ProviderInstanceId,
9 ProviderSessionId,
10 ProviderThreadId,
11 ThreadId,
12} from "@t3tools/contracts";
13import * as Effect from "effect/Effect";
14import * as Deferred from "effect/Deferred";
15import * as Fiber from "effect/Fiber";
16import * as FileSystem from "effect/FileSystem";
17import * as Path from "effect/Path";
18import * as Option from "effect/Option";
19import { symlinksSupported } from "@t3tools/shared/testing/symlinks";
20import * as Layer from "effect/Layer";
21
22import * as GitWorkflow from "../git/GitWorkflowService.ts";
23import { withWorkspaceLease } from "../workspace/workspaceLease.ts";
24import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";
25import { resolveCodexRollbackTurnCount } from "./Adapters/CodexAdapterV2.ts";
26import { isCheckpointRestoreIsolated } from "./CheckpointRestoreSafety.ts";
27import * as CheckpointService from "./CheckpointService.ts";
28import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
29import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
30import * as KeyedLock from "@t3tools/shared/KeyedLock";
31import * as EventSink from "./EventSink.ts";
32import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
33import * as ProjectionStore from "./ProjectionStore.ts";
34import * as ProjectStore from "./ProjectStore.ts";
35import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
36import * as ProviderSessionManager from "./ProviderSessionManager.ts";
37import * as RuntimePolicy from "./RuntimePolicy.ts";
38
39// A root that does not exist never overlaps, so other owners decide isolation.
40const unrelatedProject = Option.some({
41 workspaceRoot: "/nonexistent/t3-rollback-project",
42} as never);
43const layerCheckpointRollbackService = CheckpointRollbackService.layer.pipe(
44 Layer.provide(
45 Layer.mergeAll(
46 NodeServices.layer,
47 Layer.mock(GitWorkflow.GitWorkflowService)({}),
48 Layer.mock(ProjectStore.ProjectStoreV2)({ get: () => Effect.succeed(unrelatedProject) }),
49 ),
50 ),
51);
52
53it.effect.each(["initial", "waiting"])(
54 "rejects a non-ready checkpoint from %s before opening a session or restoring files",
55 (state) => {
56 const threadId = ThreadId.make("thread:rollback-non-ready");
57 const providerThreadId = ProviderThreadId.make("provider-thread:rollback-non-ready");
58 const providerSessionId = ProviderSessionId.make("provider-session:rollback-non-ready");
59 const checkpointId = CheckpointId.make("checkpoint:rollback-non-ready");
60 const scopeId = CheckpointScopeId.make("checkpoint-scope:rollback-non-ready");
61 const providerInstanceId = ProviderInstanceId.make("provider_rollback_non_ready");
62 const restore = vi.fn(() => Effect.die("checkpoint restore must not run"));
63 const open = vi.fn(() => Effect.die("provider session open must not run"));
64 const resolveRuntimePolicy = vi.fn(() => Effect.die("runtime policy resolution must not run"));
65 const readEntered = Deferred.makeUnsafe<void>();
66 const release = Deferred.makeUnsafe<void>();
67 let readCount = 0;
68 let projection = {
69 thread: {
70 worktreePath: process.cwd(),
71 activeProviderThreadId: providerThreadId,
72 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
73 },
74 providerThreads: [{ id: providerThreadId, providerSessionId, providerInstanceId }],
75 checkpoints: [
76 {
77 id: checkpointId,
78 scopeId,
79 status: state === "waiting" ? "ready" : "stale",
80 appRunOrdinal: 0,
81 },
82 ],
83 checkpointScopes: [{ id: scopeId, cwd: process.cwd() }],
84 } as unknown as OrchestrationV2ThreadProjection;
85 const getThreadRecords = vi.fn<ProjectionStore.ProjectionStoreV2Shape["getThreadRecords"]>(
86 () => {
87 const current = projection;
88 readCount++;
89 return readCount === 1
90 ? Deferred.succeed(readEntered, undefined).pipe(Effect.as(current))
91 : Effect.succeed(current);
92 },
93 );
94 const testLayer = layerCheckpointRollbackService.pipe(
95 Layer.provide(
96 Layer.mergeAll(
97 ThreadCommandExecutor.layer,
98 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
99 Layer.mock(EventSink.EventSinkV2)({}),
100 IdAllocator.layer,
101 Layer.mock(ProjectionStore.ProjectionStoreV2)({
102 getThreadRecords,
103 getShellSnapshot: () =>
104 Effect.succeed({
105 schemaVersion: 1,
106 snapshotSequence: 0,
107 threads: [],
108 archivedThreads: [],
109 }),
110 }),
111 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open }),
112 Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy }),
113 ),
114 ),
115 );
116
117 return Effect.gen(function* () {
118 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
119 const holder =
120 state === "waiting"
121 ? yield* withWorkspaceLease(process.cwd(), Deferred.await(release)).pipe(
122 Effect.forkChild({ startImmediately: true }),
123 )
124 : undefined;
125 const rollback = yield* service
126 .execute({
127 threadId,
128 providerThreadId,
129 checkpointId,
130 scopeId,
131 })
132 .pipe(Effect.flip, Effect.forkChild({ startImmediately: true }));
133 if (state === "waiting") {
134 yield* Deferred.await(readEntered);
135 projection = {
136 ...projection,
137 checkpoints: projection.checkpoints.map((checkpoint) => ({
138 ...checkpoint,
139 status: "stale",
140 })),
141 };
142 yield* Deferred.succeed(release, undefined);
143 if (holder !== undefined) yield* Fiber.join(holder);
144 }
145 const error = yield* Fiber.join(rollback);
146
147 assert.equal(getThreadRecords.mock.calls.length, 2);
148 assert.deepEqual(getThreadRecords.mock.calls[0]?.[1], []);
149 assert.equal(getThreadRecords.mock.calls[1]?.[1].includes("checkpoints"), true);
150 assert.equal(error.reason, "rollback-target-invalid");
151 assert.equal(
152 error.message,
153 `Rollback target ${checkpointId} for provider thread ${providerThreadId} on thread ${threadId} is incomplete or invalid.`,
154 );
155 assert.equal(error.cause, undefined);
156 assert.equal(resolveRuntimePolicy.mock.calls.length, 0);
157 assert.equal(open.mock.calls.length, 0);
158 assert.equal(restore.mock.calls.length, 0);
159 }).pipe(Effect.provide(testLayer));
160 },
161);
162
163it.effect("rejects a rollback when another provider thread became active", () => {
164 const threadId = ThreadId.make("thread:rollback-inactive-provider-thread");
165 const requestedProviderThreadId = ProviderThreadId.make(
166 "provider-thread:rollback-inactive-provider-thread:requested",
167 );
168 const activeProviderThreadId = ProviderThreadId.make(
169 "provider-thread:rollback-inactive-provider-thread:active",
170 );
171 const providerSessionId = ProviderSessionId.make(
172 "provider-session:rollback-inactive-provider-thread",
173 );
174 const checkpointId = CheckpointId.make("checkpoint:rollback-inactive-provider-thread");
175 const scopeId = CheckpointScopeId.make("checkpoint-scope:rollback-inactive-provider-thread");
176 const providerInstanceId = ProviderInstanceId.make("provider_rollback_inactive_provider_thread");
177 const restore = vi.fn(() => Effect.die("checkpoint restore must not run"));
178 const open = vi.fn(() => Effect.die("provider session open must not run"));
179 const resolveRuntimePolicy = vi.fn(() => Effect.die("runtime policy resolution must not run"));
180 const projection = {
181 thread: {
182 worktreePath: null,
183 activeProviderThreadId,
184 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
185 },
186 providerThreads: [
187 {
188 id: requestedProviderThreadId,
189 providerSessionId,
190 providerInstanceId,
191 },
192 ],
193 checkpoints: [{ id: checkpointId, scopeId, status: "ready" }],
194 checkpointScopes: [{ id: scopeId, cwd: process.cwd() }],
195 } as unknown as OrchestrationV2ThreadProjection;
196 const layerTest = layerCheckpointRollbackService.pipe(
197 Layer.provide(
198 Layer.mergeAll(
199 ThreadCommandExecutor.layer,
200 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
201 Layer.mock(EventSink.EventSinkV2)({}),
202 IdAllocator.layer,
203 Layer.mock(ProjectionStore.ProjectionStoreV2)({
204 getThreadRecords: () => Effect.succeed(projection),
205 getShellSnapshot: () =>
206 Effect.succeed({
207 schemaVersion: 1,
208 snapshotSequence: 0,
209 threads: [],
210 archivedThreads: [],
211 }),
212 }),
213 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open }),
214 Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy }),
215 ),
216 ),
217 );
218
219 return Effect.gen(function* () {
220 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
221 const error = yield* service
222 .execute({
223 threadId,
224 providerThreadId: requestedProviderThreadId,
225 checkpointId,
226 scopeId,
227 })
228 .pipe(Effect.flip);
229
230 assert.equal(error.reason, "active-provider-changed");
231 assert.equal(
232 error.message,
233 `Active provider changed before rollback target ${checkpointId} could execute on thread ${threadId}.`,
234 );
235 assert.equal(error.cause, undefined);
236 assert.equal(resolveRuntimePolicy.mock.calls.length, 0);
237 assert.equal(open.mock.calls.length, 0);
238 assert.equal(restore.mock.calls.length, 0);
239 }).pipe(Effect.provide(layerTest));
240});
241
242it.effect("rejects a rollback when provider selection changed before execution", () => {
243 const threadId = ThreadId.make("thread:rollback-provider-selection-changed");
244 const providerThreadId = ProviderThreadId.make(
245 "provider-thread:rollback-provider-selection-changed",
246 );
247 const providerSessionId = ProviderSessionId.make(
248 "provider-session:rollback-provider-selection-changed",
249 );
250 const checkpointId = CheckpointId.make("checkpoint:rollback-provider-selection-changed");
251 const scopeId = CheckpointScopeId.make("checkpoint-scope:rollback-provider-selection-changed");
252 const originalProviderInstanceId = ProviderInstanceId.make(
253 "provider_rollback_provider_selection_changed_original",
254 );
255 const selectedProviderInstanceId = ProviderInstanceId.make(
256 "provider_rollback_provider_selection_changed_selected",
257 );
258 const restore = vi.fn(() => Effect.die("checkpoint restore must not run"));
259 const open = vi.fn(() => Effect.die("provider session open must not run"));
260 const resolveRuntimePolicy = vi.fn(() => Effect.die("runtime policy resolution must not run"));
261 const projection = {
262 thread: {
263 worktreePath: process.cwd(),
264 activeProviderThreadId: providerThreadId,
265 modelSelection: { instanceId: selectedProviderInstanceId, model: "test-model" },
266 },
267 providerThreads: [
268 {
269 id: providerThreadId,
270 providerSessionId,
271 providerInstanceId: originalProviderInstanceId,
272 },
273 ],
274 checkpoints: [{ id: checkpointId, scopeId, status: "ready" }],
275 checkpointScopes: [{ id: scopeId, cwd: process.cwd() }],
276 } as unknown as OrchestrationV2ThreadProjection;
277 const layerTest = layerCheckpointRollbackService.pipe(
278 Layer.provide(
279 Layer.mergeAll(
280 ThreadCommandExecutor.layer,
281 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
282 Layer.mock(EventSink.EventSinkV2)({}),
283 IdAllocator.layer,
284 Layer.mock(ProjectionStore.ProjectionStoreV2)({
285 getThreadRecords: () => Effect.succeed(projection),
286 getShellSnapshot: () =>
287 Effect.succeed({
288 schemaVersion: 1,
289 snapshotSequence: 0,
290 threads: [],
291 archivedThreads: [],
292 }),
293 }),
294 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({ open }),
295 Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: resolveRuntimePolicy }),
296 ),
297 ),
298 );
299
300 return Effect.gen(function* () {
301 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
302 const error = yield* service
303 .execute({
304 threadId,
305 providerThreadId,
306 checkpointId,
307 scopeId,
308 })
309 .pipe(Effect.flip);
310
311 assert.equal(error.reason, "active-provider-changed");
312 assert.equal(
313 error.message,
314 `Active provider changed before rollback target ${checkpointId} could execute on thread ${threadId}.`,
315 );
316 assert.equal(error.cause, undefined);
317 assert.equal(resolveRuntimePolicy.mock.calls.length, 0);
318 assert.equal(open.mock.calls.length, 0);
319 assert.equal(restore.mock.calls.length, 0);
320 }).pipe(Effect.provide(layerTest));
321});
322
323it.effect("rejects an unavailable rollback target before opening a session", () => {
324 const threadId = ThreadId.make("thread:rollback-provider-turn-unavailable");
325 const providerThreadId = ProviderThreadId.make(
326 "provider-thread:rollback-provider-turn-unavailable",
327 );
328 const providerSessionId = ProviderSessionId.make(
329 "provider-session:rollback-provider-turn-unavailable",
330 );
331 const checkpointId = CheckpointId.make("checkpoint:rollback-provider-turn-unavailable");
332 const scopeId = CheckpointScopeId.make("checkpoint-scope:rollback-provider-turn-unavailable");
333 const providerInstanceId = ProviderInstanceId.make("provider_rollback_provider_turn_unavailable");
334 const restore = vi.fn(() => Effect.die("checkpoint restore must not run"));
335 const open = vi.fn(() => Effect.die("provider session open must not run"));
336 const projection = {
337 thread: {
338 worktreePath: "/tmp/missing-invalid-rollback-worktree",
339 branch: "feature/rollback",
340 activeProviderThreadId: providerThreadId,
341 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
342 },
343 providerThreads: [{ id: providerThreadId, providerSessionId, providerInstanceId }],
344 providerSessions: [],
345 checkpoints: [{ id: checkpointId, scopeId, status: "ready", appRunOrdinal: 1 }],
346 checkpointScopes: [{ id: scopeId, cwd: process.cwd() }],
347 runs: [],
348 attempts: [],
349 providerTurns: [],
350 } as unknown as OrchestrationV2ThreadProjection;
351 const layerTest = layerCheckpointRollbackService.pipe(
352 Layer.provide(
353 Layer.mergeAll(
354 ThreadCommandExecutor.layer,
355 Layer.mock(CheckpointService.CheckpointServiceV2)({ restore }),
356 Layer.mock(EventSink.EventSinkV2)({}),
357 IdAllocator.layer,
358 Layer.mock(ProjectionStore.ProjectionStoreV2)({
359 getThreadRecords: () => Effect.succeed(projection),
360 getShellSnapshot: () =>
361 Effect.succeed({
362 schemaVersion: 1,
363 snapshotSequence: 0,
364 threads: [],
365 archivedThreads: [],
366 }),
367 }),
368 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
369 open,
370 }),
371 Layer.mock(RuntimePolicy.RuntimePolicyV2)({
372 resolve: () => Effect.succeed({} as never),
373 }),
374 ),
375 ),
376 );
377
378 return Effect.gen(function* () {
379 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
380 const error = yield* service
381 .execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles: false })
382 .pipe(Effect.flip);
383 assert.equal(open.mock.calls.length, 0);
384 assert.equal(error.reason, "provider-turn-unavailable");
385 assert.equal(
386 error.message,
387 `Provider turn for rollback target ${checkpointId} is unavailable on provider thread ${providerThreadId}.`,
388 );
389 assert.equal(error.cause, undefined);
390 assert.equal(restore.mock.calls.length, 0);
391 }).pipe(Effect.provide(layerTest));
392});
393
394it.effect.each([
395 { order: "normal", restoreFiles: true },
396 { order: "cleanup-first", restoreFiles: true },
397 { order: "cleanup-first", restoreFiles: false },
398 { order: "rollback-first", restoreFiles: true },
399])("locks rollback with %s", ({ order, restoreFiles }) =>
400 Effect.gen(function* () {
401 const fileSystem = yield* FileSystem.FileSystem;
402 const path = yield* Path.Path;
403 const workspaceRoot = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-rollback-" });
404 const worktreePath = path.join(workspaceRoot, "worktree");
405 yield* fileSystem.makeDirectory(worktreePath);
406 const threadId = ThreadId.make("rollback-worktree");
407 const projectId = ProjectId.make("rollback-worktree-project");
408 const providerThreadId = ProviderThreadId.make("rollback-worktree-provider-thread");
409 const providerSessionId = ProviderSessionId.make("rollback-worktree-provider-session");
410 const providerInstanceId = ProviderInstanceId.make("rollback-worktree-provider-instance");
411 const checkpointId = CheckpointId.make("rollback-worktree-checkpoint");
412 const scopeId = CheckpointScopeId.make("rollback-worktree-scope");
413 const branch = "feature/rollback";
414 const release = Deferred.makeUnsafe<void>();
415 const openEntered = Deferred.makeUnsafe<void>();
416 const calls: string[] = [];
417 let cleanupEntered = false;
418 const providerThread = { id: providerThreadId, providerSessionId, providerInstanceId };
419 const projection = {
420 thread: {
421 id: threadId,
422 projectId,
423 worktreePath,
424 branch,
425 activeProviderThreadId: providerThreadId,
426 modelSelection: { instanceId: providerInstanceId, model: "test-model" },
427 },
428 providerThreads: [providerThread],
429 providerSessions: [],
430 checkpoints: [{ id: checkpointId, scopeId, status: "ready", appRunOrdinal: 0 }],
431 checkpointScopes: [{ id: scopeId, cwd: worktreePath }],
432 runs: [],
433 attempts: [],
434 nodes: [],
435 providerTurns: [],
436 } as unknown as OrchestrationV2ThreadProjection;
437 const testLayer = CheckpointRollbackService.layer.pipe(
438 Layer.provide(
439 Layer.mergeAll(
440 ThreadCommandExecutor.layer,
441 NodeServices.layer,
442 Layer.mock(ProjectStore.ProjectStoreV2)({
443 get: (id) => {
444 assert.equal(id, projectId);
445 return Effect.succeed(Option.some({ workspaceRoot } as never));
446 },
447 }),
448 Layer.mock(GitWorkflow.GitWorkflowService)({
449 pruneWorktrees: (input) =>
450 Effect.sync(() => {
451 assert.deepEqual(input, { cwd: workspaceRoot });
452 calls.push("prune");
453 }),
454 createWorktree: (input) =>
455 Effect.gen(function* () {
456 assert.deepEqual(input, {
457 cwd: workspaceRoot,
458 refName: branch,
459 path: worktreePath,
460 });
461 calls.push("create");
462 yield* fileSystem.makeDirectory(worktreePath).pipe(Effect.orDie);
463 return {} as never;
464 }),
465 }),
466 Layer.mock(CheckpointService.CheckpointServiceV2)({
467 restore: () =>
468 Effect.gen(function* () {
469 assert.isTrue(yield* fileSystem.exists(worktreePath).pipe(Effect.orDie));
470 calls.push("files");
471 }),
472 }),
473 Layer.mock(EventSink.EventSinkV2)({
474 write: () =>
475 Effect.sync(() => {
476 calls.push("projection");
477 return [];
478 }),
479 }),
480 IdAllocator.layer,
481 Layer.mock(ProjectionStore.ProjectionStoreV2)({
482 getThreadRecords: () => Effect.succeed(projection),
483 getShellSnapshot: () =>
484 Effect.succeed({
485 schemaVersion: 1,
486 snapshotSequence: 0,
487 threads: [],
488 archivedThreads: [],
489 }),
490 }),
491 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
492 open: (input) =>
493 Effect.gen(function* () {
494 assert.equal(input.runtimePolicy.cwd, worktreePath);
495 if (!(yield* fileSystem.exists(worktreePath).pipe(Effect.orDie))) {
496 return yield* new ProviderWorkspaceMissingError({ threadId, cwd: worktreePath });
497 }
498 calls.push("session");
499 yield* Deferred.succeed(openEntered, undefined);
500 if (order === "rollback-first") yield* Deferred.await(release);
501 return {} as never;
502 }),
503 }),
504 Layer.mock(RuntimePolicy.RuntimePolicyV2)({
505 resolve: ({ thread }) => {
506 assert.equal(thread.worktreePath, worktreePath);
507 return Effect.succeed({ cwd: thread.worktreePath } as never);
508 },
509 }),
510 ),
511 ),
512 );
513 yield* Effect.gen(function* () {
514 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
515 const cleanupEffect = withWorkspaceLease(
516 worktreePath,
517 Effect.gen(function* () {
518 cleanupEntered = true;
519 if (order === "cleanup-first") yield* Deferred.await(release);
520 yield* fileSystem.remove(worktreePath, { recursive: true, force: true });
521 calls.push("cleanup");
522 }),
523 );
524 const cleanup =
525 order === "cleanup-first"
526 ? yield* cleanupEffect.pipe(Effect.forkChild({ startImmediately: true }))
527 : undefined;
528 const rollback = yield* service
529 .execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles })
530 .pipe(Effect.forkChild({ startImmediately: true }));
531 if (order === "cleanup-first") assert.deepEqual(calls, []);
532 if (order === "rollback-first") yield* Deferred.await(openEntered);
533 const laterCleanup =
534 order === "rollback-first"
535 ? yield* cleanupEffect.pipe(Effect.forkChild({ startImmediately: true }))
536 : undefined;
537 if (order === "rollback-first") assert.isFalse(cleanupEntered);
538 yield* Deferred.succeed(release, undefined);
539 if (cleanup !== undefined) yield* Fiber.join(cleanup);
540 yield* Fiber.join(rollback);
541 if (laterCleanup !== undefined) yield* Fiber.join(laterCleanup);
542 const rollbackCalls = restoreFiles
543 ? ["session", "files", "projection"]
544 : ["session", "projection"];
545 assert.deepEqual(
546 calls,
547 order === "cleanup-first"
548 ? ["cleanup", "prune", "create", ...rollbackCalls]
549 : order === "rollback-first"
550 ? [...rollbackCalls, "cleanup"]
551 : rollbackCalls,
552 );
553 }).pipe(Effect.provide(testLayer));
554 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
555);
556
557it.effect.each([
558 { restoreFiles: true, shared: "none" },
559 { restoreFiles: false, shared: "root" },
560 { restoreFiles: true, shared: "root" },
561 { restoreFiles: true, shared: "worktree" },
562 { restoreFiles: false, shared: "worktree" },
563 { restoreFiles: true, shared: "historical" },
564 { restoreFiles: false, shared: "none", targetOrdinal: 1 },
565 { restoreFiles: false, shared: "none", targetOrdinal: 1, multipleTurns: true },
566])(
567 "rewinds safely with %s",
568 ({ restoreFiles, shared, targetOrdinal = 0, multipleTurns = false }) => {
569 const threadId = ThreadId.make("rewind-files");
570 const providerThreadId = ProviderThreadId.make("rewind-provider");
571 const providerSessionId = ProviderSessionId.make("rewind-session");
572 const instanceId = ProviderInstanceId.make("rewind-instance");
573 const checkpointId = CheckpointId.make("rewind-start");
574 const scopeId = CheckpointScopeId.make("rewind-scope");
575 const calls: string[] = [];
576 let lockedThreads = Effect.succeed<ReadonlyArray<ThreadId>>([]);
577 const layerThreadCommands = Layer.effect(
578 ThreadCommandExecutor.ThreadCommandExecutor,
579 Effect.tap(KeyedLock.make<ThreadId>(), (lock) =>
580 Effect.sync(() => {
581 lockedThreads = lock.activeKeys;
582 }),
583 ),
584 );
585 const providerThread = {
586 id: providerThreadId,
587 providerSessionId,
588 providerInstanceId: instanceId,
589 };
590 const projection = {
591 thread: {
592 worktreePath: shared === "root" ? null : process.cwd(),
593 activeProviderThreadId: providerThreadId,
594 modelSelection: { instanceId, model: "test" },
595 },
596 providerThreads: [providerThread],
597 providerSessions: [],
598 // Turn 3 remains in the audit history after an earlier rollback.
599 providerTurns: [
600 ...(multipleTurns
601 ? [
602 {
603 id: "turn-0",
604 providerThreadId,
605 runAttemptId: "attempt-1",
606 ordinal: 0,
607 status: "completed",
608 },
609 ]
610 : []),
611 ...[1, 2, 3].map((ordinal) => ({
612 id: `turn-${ordinal}`,
613 providerThreadId,
614 runAttemptId: `attempt-${ordinal}`,
615 ordinal,
616 status: "completed",
617 })),
618 ],
619 nodes: [],
620 attempts: [1, 2, 3].map((ordinal) => ({ id: `attempt-${ordinal}`, runId: `run-${ordinal}` })),
621 checkpoints: [
622 { id: checkpointId, scopeId, status: "ready", appRunOrdinal: targetOrdinal || null },
623 ],
624 checkpointScopes: [{ id: scopeId, cwd: process.cwd() }],
625 runs: [1, 2, 3].map((ordinal) => ({
626 id: `run-${ordinal}`,
627 ordinal,
628 status: ordinal === 3 ? "rolled_back" : "completed",
629 rootNodeId: null,
630 activeAttemptId: `attempt-${ordinal}`,
631 })),
632 } as unknown as OrchestrationV2ThreadProjection;
633 const layerTest = layerCheckpointRollbackService.pipe(
634 Layer.provide(
635 Layer.mergeAll(
636 layerThreadCommands,
637 Layer.mock(CheckpointService.CheckpointServiceV2)({
638 restore: () =>
639 Effect.sync(() => {
640 calls.push("files");
641 }),
642 }),
643 Layer.mock(EventSink.EventSinkV2)({
644 write: ({ events }) =>
645 Effect.map(lockedThreads, (locked) => {
646 // A thread command that planned against the old provider thread
647 // must not commit after this and undo the rollback.
648 assert.include(locked, threadId);
649 assert.ok(
650 events.some(
651 (event) =>
652 event.type === "run.updated" && event.payload.status === "rolled_back",
653 ),
654 );
655 calls.push("projection");
656 return [];
657 }),
658 }),
659 IdAllocator.layer,
660 Layer.mock(ProjectionStore.ProjectionStoreV2)({
661 getThreadRecords: () => Effect.succeed(projection),
662 getThreadProviderContext: () => Effect.succeed({ providerSessions: [] } as never),
663 getCheckpointContext: () =>
664 Effect.succeed({
665 checkpointScopes: [{ cwd: process.cwd() }],
666 runs: [],
667 checkpoints: [],
668 } as never),
669 getShellSnapshot: () =>
670 Effect.succeed({
671 schemaVersion: 1,
672 snapshotSequence: 0,
673 threads: [],
674 archivedThreads:
675 shared === "worktree" || shared === "historical"
676 ? [
677 {
678 id: ThreadId.make("other-thread"),
679 deletedAt: null,
680 worktreePath: shared === "worktree" ? process.cwd() : null,
681 } as never,
682 ]
683 : [],
684 }),
685 }),
686 Layer.mock(ProviderSessionManager.ProviderSessionManagerV2)({
687 open: () =>
688 Effect.succeed({
689 rollbackThread: (input: ProviderAdapter.ProviderAdapterV2RollbackThreadInput) =>
690 Effect.gen(function* () {
691 const count = yield* resolveCodexRollbackTurnCount(input);
692 assert.equal(count, 2 - targetOrdinal);
693 calls.push("provider");
694 return { providerThread };
695 }),
696 } as never),
697 }),
698 Layer.mock(RuntimePolicy.RuntimePolicyV2)({ resolve: () => Effect.succeed({} as never) }),
699 ),
700 ),
701 );
702 return Effect.gen(function* () {
703 const service = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
704 if (restoreFiles && shared !== "none") {
705 const error = yield* Effect.flip(
706 service.execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles }),
707 );
708 assert.equal(error.reason, "shared-workspace");
709 assert.deepEqual(calls, []);
710 return;
711 }
712 yield* service.execute({ threadId, providerThreadId, checkpointId, scopeId, restoreFiles });
713 assert.deepEqual(
714 calls,
715 restoreFiles ? ["provider", "files", "projection"] : ["provider", "projection"],
716 );
717 }).pipe(Effect.provide(layerTest));
718 },
719);
720
721it.effect.skipIf(!symlinksSupported)(
722 "rejects an archived thread sharing a worktree through a symlink",
723 () =>
724 Effect.gen(function* () {
725 const fileSystem = yield* FileSystem.FileSystem;
726 const cwd = yield* fileSystem.makeTempDirectoryScoped({ prefix: "t3-restore-isolation-" });
727 const path = yield* Path.Path;
728 const alias = path.join(cwd, "alias");
729 yield* fileSystem.symlink(cwd, alias);
730 const threadId = ThreadId.make("restore-alias-current");
731 const otherId = ThreadId.make("restore-alias-archived");
732 const projections = ProjectionStore.ProjectionStoreV2.of({
733 getShellSnapshot: () =>
734 Effect.succeed({
735 schemaVersion: 1,
736 snapshotSequence: 0,
737 threads: [],
738 archivedThreads: [{ id: otherId, deletedAt: null, worktreePath: alias } as never],
739 }),
740 getThreadProviderContext: () => Effect.succeed({ providerSessions: [] } as never),
741 getCheckpointContext: () =>
742 Effect.succeed({ runs: [], checkpointScopes: [], checkpoints: [] }),
743 } as never);
744 const isolated = yield* isCheckpointRestoreIsolated(
745 { id: threadId, worktreePath: cwd },
746 { cwd },
747 {
748 fileSystem,
749 path,
750 projections,
751 projects: ProjectStore.ProjectStoreV2.of({
752 get: () => Effect.succeed(unrelatedProject),
753 } as never),
754 },
755 );
756 assert.isFalse(isolated);
757 }).pipe(Effect.scoped, Effect.provide(NodeServices.layer)),
758);