1import {
2 CheckpointId,
3 CheckpointScopeId,
4 latestProviderTurnForAttempt,
5 type OrchestrationV2DomainEvent,
6 ProviderThreadId,
7 ThreadId,
8 VcsProcessExitError,
9} from "@t3tools/contracts";
10import * as Context from "effect/Context";
11import * as DateTime from "effect/DateTime";
12import * as Effect from "effect/Effect";
13import * as Exit from "effect/Exit";
14import * as FileSystem from "effect/FileSystem";
15import * as Layer from "effect/Layer";
16import * as Option from "effect/Option";
17import * as Path from "effect/Path";
18import * as Schema from "effect/Schema";
19
20import * as GitWorkflowService from "../git/GitWorkflowService.ts";
21import type * as ServerConfig from "../config.ts";
22import type * as ServerSettings from "../serverSettings.ts";
23import { makeIsManagedWorktreePath } from "../worktreesDirectory.ts";
24import {
25 withCheckpointRestoreReservation,
26 withWorkspaceLease,
27} from "../workspace/workspaceLease.ts";
28import {
29 isCheckpointRestoreIsolated,
30 SHARED_WORKSPACE_RESTORE_MESSAGE,
31} from "./CheckpointRestoreSafety.ts";
32import { CheckpointRestoreError, CheckpointServiceV2 } from "./CheckpointService.ts";
33import { EventSinkV2 } from "./EventSink.ts";
34import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
35import { ProjectionStoreV2 } from "./ProjectionStore.ts";
36import { GIT_INDEX_LOCKED_DETAIL } from "../vcs/VcsProcess.ts";
37import * as ProjectStore from "./ProjectStore.ts";
38import type * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
39import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts";
40import { RuntimePolicyAboveCeilingError, RuntimePolicyV2 } from "./RuntimePolicy.ts";
41import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
42
43export const ROLLBACK_FAILED_MESSAGE =
44 "The provider could not roll back this conversation. Try again; if it keeps failing, check the provider and server logs.";
45
46export class CheckpointRollbackExecutionError extends Schema.TaggedError<CheckpointRollbackExecutionError>()(
47 "CheckpointRollbackExecutionError",
48 {
49 reason: Schema.Literals([
50 "rollback-target-invalid",
51 "active-provider-changed",
52 "provider-turn-unavailable",
53 "unexpected-failure",
54 "shared-workspace",
55 ]),
56 threadId: ThreadId,
57 providerThreadId: ProviderThreadId,
58 checkpointId: CheckpointId,
59 cause: Schema.optional(Schema.Defect()),
60 },
61) {
62 override get message(): string {
63 switch (this.reason) {
64 case "rollback-target-invalid":
65 return `Rollback target ${this.checkpointId} for provider thread ${this.providerThreadId} on thread ${this.threadId} is incomplete or invalid.`;
66 case "active-provider-changed":
67 return `Active provider changed before rollback target ${this.checkpointId} could execute on thread ${this.threadId}.`;
68 case "provider-turn-unavailable":
69 return `Provider turn for rollback target ${this.checkpointId} is unavailable on provider thread ${this.providerThreadId}.`;
70 case "shared-workspace":
71 return SHARED_WORKSPACE_RESTORE_MESSAGE;
72 case "unexpected-failure":
73 return ROLLBACK_FAILED_MESSAGE;
74 }
75 }
76}
77
78const isCheckpointRollbackExecutionError = Schema.is(CheckpointRollbackExecutionError);
79const isVcsProcessExitError = Schema.is(VcsProcessExitError);
80const isCheckpointRestoreError = Schema.is(CheckpointRestoreError);
81const isRuntimePolicyAboveCeilingError = Schema.is(RuntimePolicyAboveCeilingError);
82
83/**
84 * The message clients see when a rollback fails. Only a held Git index lock
85 * gets its own explanation; other reasons carry internal ids or raw Git
86 * output, so they fall back to the generic message.
87 */
88export function rollbackFailureClientMessage(error: CheckpointRollbackExecutionError): string {
89 if (isRuntimePolicyAboveCeilingError(error.cause)) return error.cause.message;
90 const restoreCause = isCheckpointRestoreError(error.cause) ? error.cause.cause : undefined;
91 return isVcsProcessExitError(restoreCause) && restoreCause.detail === GIT_INDEX_LOCKED_DETAIL
92 ? GIT_INDEX_LOCKED_DETAIL
93 : ROLLBACK_FAILED_MESSAGE;
94}
95
96export interface CheckpointRollbackServiceV2Shape {
97 readonly execute: (input: {
98 readonly threadId: ThreadId;
99 readonly providerThreadId: ProviderThreadId;
100 readonly checkpointId: CheckpointId;
101 readonly scopeId: CheckpointScopeId;
102 readonly restoreFiles?: boolean;
103 }) => Effect.Effect<void, CheckpointRollbackExecutionError>;
104}
105
106export class CheckpointRollbackServiceV2 extends Context.Service<
107 CheckpointRollbackServiceV2,
108 CheckpointRollbackServiceV2Shape
109>()("t3/orchestration-v2/CheckpointRollbackService/CheckpointRollbackServiceV2") {}
110
111export const layer: Layer.Layer<
112 CheckpointRollbackServiceV2,
113 never,
114 | CheckpointServiceV2
115 | GitWorkflowService.GitWorkflowService
116 | EventSinkV2
117 | IdAllocator.IdAllocatorV2
118 | ProjectionStoreV2
119 | ProviderSessionManagerV2
120 | RuntimePolicyV2
121 | FileSystem.FileSystem
122 | Path.Path
123 | ProjectStore.ProjectStoreV2
124 | ServerConfig.ServerConfig
125 | ServerSettings.ServerSettingsService
126 | ThreadCommandExecutor.ThreadCommandExecutor
127> = Layer.effect(
128 CheckpointRollbackServiceV2,
129 Effect.gen(function* () {
130 const checkpoints = yield* CheckpointServiceV2;
131 const gitWorkflow = yield* GitWorkflowService.GitWorkflowService;
132 const eventSink = yield* EventSinkV2;
133 const ids = yield* IdAllocator.IdAllocatorV2;
134 const projections = yield* ProjectionStoreV2;
135 const sessions = yield* ProviderSessionManagerV2;
136 const runtimePolicy = yield* RuntimePolicyV2;
137 const fileSystem = yield* FileSystem.FileSystem;
138 const projects = yield* ProjectStore.ProjectStoreV2;
139 const path = yield* Path.Path;
140 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
141 const isManagedWorktreePath = yield* makeIsManagedWorktreePath;
142
143 const execute = Effect.fn("orchestrationV2.checkpointRollback.execute")(function* (input: {
144 readonly threadId: ThreadId;
145 readonly providerThreadId: ProviderThreadId;
146 readonly checkpointId: CheckpointId;
147 readonly scopeId: CheckpointScopeId;
148 readonly restoreFiles?: boolean;
149 }) {
150 const context = yield* projections.getCheckpointRollbackContext(input.threadId, input);
151 const { providerThread, checkpoint, scope, targetAttempt, runsToRollback, staleCheckpoints } =
152 context;
153 if (
154 providerThread === undefined ||
155 providerThread.providerSessionId === null ||
156 checkpoint === undefined ||
157 scope === undefined ||
158 checkpoint.scopeId !== scope.id ||
159 // Missing snapshots support conversation rewind only.
160 (checkpoint.status !== "ready" &&
161 !(input.restoreFiles === false && checkpoint.status === "missing"))
162 ) {
163 return yield* new CheckpointRollbackExecutionError({
164 reason: "rollback-target-invalid",
165 threadId: input.threadId,
166 providerThreadId: input.providerThreadId,
167 checkpointId: input.checkpointId,
168 });
169 }
170 if (
171 providerThread.id !== context.thread.activeProviderThreadId ||
172 providerThread.providerInstanceId !== context.thread.modelSelection.instanceId
173 ) {
174 return yield* new CheckpointRollbackExecutionError({
175 reason: "active-provider-changed",
176 threadId: input.threadId,
177 providerThreadId: input.providerThreadId,
178 checkpointId: input.checkpointId,
179 });
180 }
181
182 const targetOrdinal = checkpoint.appRunOrdinal ?? 0;
183 const rollbackTarget: ProviderAdapter.ProviderAdapterV2RollbackTarget =
184 targetOrdinal === 0
185 ? {
186 type: "thread_start",
187 checkpointId: checkpoint.id,
188 appRunOrdinal: 0,
189 }
190 : yield* Effect.gen(function* () {
191 const targetTurn =
192 latestProviderTurnForAttempt(context.providerTurns, targetAttempt?.id) ??
193 context.providerTurns.find((turn) => turn.id === targetAttempt?.providerTurnId);
194 if (
195 targetAttempt === undefined ||
196 targetTurn === undefined ||
197 targetTurn.providerThreadId !== providerThread.id
198 ) {
199 return yield* new CheckpointRollbackExecutionError({
200 reason: "provider-turn-unavailable",
201 threadId: input.threadId,
202 providerThreadId: input.providerThreadId,
203 checkpointId: input.checkpointId,
204 });
205 }
206 const { attempt } = yield* projections.getProviderControlContext(input.threadId, {
207 providerThreadId: providerThread.id,
208 providerTurnId: targetTurn.id,
209 attemptId: targetAttempt.id,
210 });
211 const message =
212 attempt === undefined
213 ? undefined
214 : yield* projections.getRunMessage(input.threadId, attempt.runId);
215 return {
216 type: "provider_turn" as const,
217 checkpointId: checkpoint.id,
218 appRunOrdinal: targetOrdinal,
219 providerTurn: targetTurn,
220 ...(message === undefined ? {} : { message }),
221 };
222 });
223
224 const restoreFiles = input.restoreFiles !== false;
225 const isolated = isCheckpointRestoreIsolated(context.thread, scope, {
226 fileSystem,
227 projections,
228 projects,
229 path,
230 });
231 const sharedWorkspace = () =>
232 new CheckpointRollbackExecutionError({
233 reason: "shared-workspace",
234 threadId: input.threadId,
235 providerThreadId: input.providerThreadId,
236 checkpointId: input.checkpointId,
237 });
238
239 const { worktreePath, branch } = context.thread;
240 // Recreated only where the server would have made it, as turn start does.
241 if (worktreePath != null && branch != null) {
242 // A failed removal can leave the folder without its checkout; only a file restore replaces it.
243 const leftover =
244 restoreFiles &&
245 (yield* fileSystem.exists(worktreePath)) &&
246 !(yield* fileSystem.exists(path.join(worktreePath, ".git")));
247 if (
248 (leftover || !(yield* fileSystem.exists(worktreePath))) &&
249 (yield* isManagedWorktreePath(worktreePath))
250 ) {
251 const project = yield* projects.get(context.thread.projectId);
252 if (Option.isSome(project)) {
253 const recreate = gitWorkflow.pruneWorktrees({ cwd: project.value.workspaceRoot }).pipe(
254 Effect.andThen(
255 gitWorkflow.createWorktree({
256 cwd: project.value.workspaceRoot,
257 refName: branch,
258 path: worktreePath,
259 }),
260 ),
261 );
262 if (leftover) {
263 if (!(yield* isolated)) return yield* sharedWorkspace();
264 // Set aside, not removed: a checkout that fails puts the folder and its files back.
265 yield* Effect.acquireUseRelease(
266 Effect.gen(function* () {
267 const aside = yield* fileSystem.makeTempDirectory({
268 directory: path.dirname(worktreePath),
269 prefix: ".t3-replaced-",
270 });
271 yield* fileSystem.rename(worktreePath, path.join(aside, "folder"));
272 return aside;
273 }),
274 () => recreate,
275 (aside, exit) =>
276 (Exit.isSuccess(exit)
277 ? Effect.void
278 : fileSystem
279 .remove(worktreePath, { recursive: true, force: true })
280 .pipe(
281 Effect.andThen(
282 fileSystem.rename(path.join(aside, "folder"), worktreePath),
283 ),
284 )
285 ).pipe(
286 Effect.andThen(fileSystem.remove(aside, { recursive: true, force: true })),
287 Effect.catchCause((cause) =>
288 Effect.logWarning("Edit from here left a worktree folder set aside", {
289 worktreePath,
290 setAside: path.join(aside, "folder"),
291 cause,
292 }),
293 ),
294 ),
295 );
296 } else {
297 yield* recreate;
298 }
299 }
300 }
301 }
302
303 if (restoreFiles && !(yield* isolated)) return yield* sharedWorkspace();
304
305 const modelSelection = context.thread.modelSelection;
306 const resolvedRuntimePolicy = yield* runtimePolicy.resolve({
307 thread: context.thread,
308 modelSelection,
309 });
310 const existingSession = context.providerSession;
311 const session = yield* sessions.open({
312 threadId: input.threadId,
313 providerSessionId: providerThread.providerSessionId,
314 modelSelection,
315 runtimePolicy: resolvedRuntimePolicy,
316 ...(existingSession === undefined ? {} : { resumeFromSession: existingSession }),
317 ...(providerThread.nativeThreadRef?.nativeId == null
318 ? {}
319 : { initialNativeThreadId: providerThread.nativeThreadRef.nativeId }),
320 ...(providerThread.nativeMetadata?.itemIdentityVersion === undefined
321 ? {}
322 : {
323 initialProviderItemIdentityVersion: providerThread.nativeMetadata.itemIdentityVersion,
324 }),
325 });
326
327 // Rolled-back turns remain in audit history but no longer exist in the provider conversation.
328 const rolledBackAttemptIds = new Set(context.rolledBackAttemptIds);
329 const providerThreadTurns = context.providerTurns.filter(
330 (turn) =>
331 turn.providerThreadId === providerThread.id &&
332 (turn.runAttemptId === null || !rolledBackAttemptIds.has(turn.runAttemptId)),
333 );
334
335 const snapshot =
336 runsToRollback.length === 0
337 ? { providerThread }
338 : yield* session.rollbackThread({
339 providerThread,
340 target: rollbackTarget,
341 providerThreadTurns,
342 });
343 if (input.restoreFiles !== false) yield* checkpoints.restore({ scope, checkpoint });
344 if (staleCheckpoints.length > 0) {
345 yield* checkpoints.deleteStaleRefs({ scope, checkpoints: staleCheckpoints });
346 }
347
348 // Lock against full-row thread writes that could restore the old native session after rollback.
349 yield* threadCommands.withLock(
350 input.threadId,
351 Effect.gen(function* () {
352 const now = yield* DateTime.now;
353 const makeEvent = <Event extends OrchestrationV2DomainEvent>(event: Omit<Event, "id">) =>
354 Effect.map(
355 ids.allocate.event({ threadId: event.threadId }),
356 (id) =>
357 ({
358 ...event,
359 id,
360 }) as Event,
361 );
362 const events: Array<OrchestrationV2DomainEvent> = [];
363 events.push(
364 yield* makeEvent({
365 type: "provider-thread.updated",
366 threadId: input.threadId,
367 driver: providerThread.driver,
368 providerInstanceId: providerThread.providerInstanceId,
369 occurredAt: now,
370 payload: {
371 ...snapshot.providerThread,
372 lastRunOrdinal: targetOrdinal === 0 ? null : targetOrdinal,
373 updatedAt: now,
374 },
375 }),
376 );
377 for (const staleCheckpoint of staleCheckpoints) {
378 events.push(
379 yield* makeEvent({
380 type: "checkpoint.captured",
381 threadId: input.threadId,
382 ...(staleCheckpoint.runId === null ? {} : { runId: staleCheckpoint.runId }),
383 nodeId: staleCheckpoint.nodeId,
384 providerInstanceId: providerThread.providerInstanceId,
385 occurredAt: now,
386 payload: { ...staleCheckpoint, status: "stale" },
387 }),
388 );
389 }
390 for (const run of runsToRollback) {
391 const rootNode = context.rootNodes.find((candidate) => candidate.id === run.rootNodeId);
392 events.push(
393 yield* makeEvent({
394 type: "run.updated",
395 threadId: input.threadId,
396 runId: run.id,
397 ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
398 providerInstanceId: run.providerInstanceId,
399 occurredAt: now,
400 payload: { ...run, status: "rolled_back", completedAt: now },
401 }),
402 );
403 if (rootNode !== undefined) {
404 events.push(
405 yield* makeEvent({
406 type: "node.updated",
407 threadId: input.threadId,
408 runId: run.id,
409 nodeId: rootNode.id,
410 providerInstanceId: run.providerInstanceId,
411 occurredAt: now,
412 payload: { ...rootNode, status: "rolled_back", completedAt: now },
413 }),
414 );
415 }
416 }
417 yield* eventSink.write({ events });
418 }),
419 );
420 });
421
422 return CheckpointRollbackServiceV2.of({
423 // The inner execute owns the rollback span; this wrapper selects the workspace lease.
424 execute: Effect.fnUntraced(
425 function* (input) {
426 // Storage cleanup removes worktrees under the same lease.
427 const thread = yield* projections.getThread(input.threadId);
428 if (thread.worktreePath === null) return yield* execute(input);
429 const worktreePath = path.resolve(thread.worktreePath);
430 const rollback = withWorkspaceLease(worktreePath, execute(input));
431 // Shared-workspace restores are rejected and conversation-only rewinds
432 // touch no files; reserving for them would stall turn starts on the
433 // slow provider rollback for nothing.
434 yield* input.restoreFiles === false
435 ? rollback
436 : withCheckpointRestoreReservation(fileSystem, path, worktreePath, rollback);
437 },
438 (effect, input) =>
439 effect.pipe(
440 Effect.mapError((cause) =>
441 isCheckpointRollbackExecutionError(cause)
442 ? cause
443 : new CheckpointRollbackExecutionError({
444 reason: "unexpected-failure",
445 threadId: input.threadId,
446 providerThreadId: input.providerThreadId,
447 checkpointId: input.checkpointId,
448 cause,
449 }),
450 ),
451 ),
452 ),
453 });
454 }),
455);