1import {
2 CheckpointId,
3 CheckpointRef,
4 CheckpointScopeId,
5 NodeId,
6 OrchestrationV2Checkpoint,
7 OrchestrationV2CheckpointScope,
8 ProviderThreadId,
9 RunId,
10 ThreadId,
11} from "@t3tools/contracts";
12import * as Context from "effect/Context";
13import * as Crypto from "effect/Crypto";
14import * as DateTime from "effect/DateTime";
15import * as Effect from "effect/Effect";
16import * as Base64Url from "effect/encoding/Base64Url";
17import * as Hex from "effect/encoding/Hex";
18import * as KeyedLock from "@t3tools/shared/KeyedLock";
19import * as Layer from "effect/Layer";
20import * as Schema from "effect/Schema";
21
22import { isGitImport, parseTurnDiffFilesFromNumstat } from "../checkpointing/Diffs.ts";
23import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
24import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
25
26const CHECKPOINT_REFS_PREFIX = "refs/t3/orchestration-v2/checkpoints";
27const ROOT_CHECKPOINT_SCOPE_NAME = "root";
28
29export class CheckpointRootScopePrepareError extends Schema.TaggedError<CheckpointRootScopePrepareError>()(
30 "CheckpointRootScopePrepareError",
31 {
32 threadId: ThreadId,
33 runId: RunId,
34 cause: Schema.Defect(),
35 },
36) {
37 override get message(): string {
38 return `Failed to prepare root checkpoint scope for run ${this.runId}.`;
39 }
40}
41
42export class CheckpointScopeEnsureError extends Schema.TaggedError<CheckpointScopeEnsureError>()(
43 "CheckpointScopeEnsureError",
44 {
45 scopeId: CheckpointScopeId,
46 cause: Schema.Defect(),
47 },
48) {
49 override get message(): string {
50 return `Failed to ensure checkpoint scope ${this.scopeId}.`;
51 }
52}
53
54export class CheckpointBaselineCaptureError extends Schema.TaggedError<CheckpointBaselineCaptureError>()(
55 "CheckpointBaselineCaptureError",
56 {
57 scopeId: CheckpointScopeId,
58 ordinalWithinScope: Schema.Number,
59 cause: Schema.Defect(),
60 },
61) {
62 override get message(): string {
63 return `Failed to capture checkpoint baseline ${this.ordinalWithinScope} for scope ${this.scopeId}.`;
64 }
65}
66
67export class CheckpointCaptureError extends Schema.TaggedError<CheckpointCaptureError>()(
68 "CheckpointCaptureError",
69 {
70 scopeId: CheckpointScopeId,
71 parentCheckpointId: Schema.optional(CheckpointId),
72 cause: Schema.Defect(),
73 },
74) {
75 override get message(): string {
76 return `Failed to capture checkpoint for scope ${this.scopeId}.`;
77 }
78}
79
80export class CheckpointRestoreError extends Schema.TaggedError<CheckpointRestoreError>()(
81 "CheckpointRestoreError",
82 {
83 scopeId: CheckpointScopeId,
84 checkpointId: CheckpointId,
85 cause: Schema.Defect(),
86 },
87) {
88 override get message(): string {
89 return `Failed to restore checkpoint ${this.checkpointId} for scope ${this.scopeId}.`;
90 }
91}
92
93export class CheckpointDeleteStaleRefsError extends Schema.TaggedError<CheckpointDeleteStaleRefsError>()(
94 "CheckpointDeleteStaleRefsError",
95 {
96 scopeId: CheckpointScopeId,
97 checkpointIds: Schema.Array(CheckpointId),
98 cause: Schema.Defect(),
99 },
100) {
101 override get message(): string {
102 return `Failed to delete stale checkpoint refs for scope ${this.scopeId}.`;
103 }
104}
105
106export const CheckpointServiceV2Error = Schema.Union([
107 CheckpointRootScopePrepareError,
108 CheckpointScopeEnsureError,
109 CheckpointBaselineCaptureError,
110 CheckpointCaptureError,
111 CheckpointRestoreError,
112 CheckpointDeleteStaleRefsError,
113]);
114export type CheckpointServiceV2Error = typeof CheckpointServiceV2Error.Type;
115
116const isCheckpointRestoreError = Schema.is(CheckpointRestoreError);
117
118export interface CheckpointServiceV2Shape {
119 readonly prepareRootRunScope: (input: {
120 readonly threadId: ThreadId;
121 readonly runId: RunId;
122 readonly rootNodeId: NodeId;
123 readonly providerThreadId: ProviderThreadId;
124 readonly cwd: string;
125 readonly createdAt: DateTime.Utc;
126 }) => Effect.Effect<OrchestrationV2CheckpointScope, CheckpointServiceV2Error>;
127 readonly ensureScope: (
128 scope: OrchestrationV2CheckpointScope,
129 ) => Effect.Effect<OrchestrationV2CheckpointScope, CheckpointServiceV2Error>;
130 readonly captureBaseline: (input: {
131 readonly scope: OrchestrationV2CheckpointScope;
132 readonly ordinalWithinScope: number;
133 }) => Effect.Effect<void, CheckpointServiceV2Error>;
134 readonly materializeBaselineCheckpoint: (input: {
135 readonly scope: OrchestrationV2CheckpointScope;
136 readonly ordinalWithinScope: number;
137 }) => Effect.Effect<OrchestrationV2Checkpoint, CheckpointServiceV2Error>;
138 readonly capture: (input: {
139 readonly scope: OrchestrationV2CheckpointScope;
140 readonly runId: RunId | null;
141 readonly nodeId: NodeId;
142 readonly ordinalWithinScope: number;
143 readonly appRunOrdinal: number | null;
144 readonly capturedAt: DateTime.Utc;
145 }) => Effect.Effect<OrchestrationV2Checkpoint, CheckpointServiceV2Error>;
146 readonly restore: (input: {
147 readonly scope: OrchestrationV2CheckpointScope;
148 readonly checkpoint: OrchestrationV2Checkpoint;
149 }) => Effect.Effect<void, CheckpointServiceV2Error>;
150 readonly deleteStaleRefs: (input: {
151 readonly scope: OrchestrationV2CheckpointScope;
152 readonly checkpoints: ReadonlyArray<OrchestrationV2Checkpoint>;
153 }) => Effect.Effect<void, CheckpointServiceV2Error>;
154}
155
156export class CheckpointServiceV2 extends Context.Service<
157 CheckpointServiceV2,
158 CheckpointServiceV2Shape
159>()("t3/orchestration-v2/CheckpointService/CheckpointServiceV2") {}
160
161export const checkpointRefForScopeOrdinal = Effect.fn("checkpointRefForScopeOrdinal")(
162 function* (input: { readonly scopeId: CheckpointScopeId; readonly ordinalWithinScope: number }) {
163 const crypto = yield* Crypto.Crypto;
164 const digest = yield* crypto
165 .digest("SHA-256", new TextEncoder().encode(input.scopeId))
166 .pipe(Effect.orDie);
167 const scopeKey = Hex.encode(digest).slice(0, 32);
168 return CheckpointRef.make(
169 `${CHECKPOINT_REFS_PREFIX}/${Base64Url.encode(scopeKey)}/ordinal/${input.ordinalWithinScope}`,
170 );
171 },
172);
173
174function checkpointIdForScopeOrdinal(
175 idAllocator: IdAllocator.IdAllocatorV2["Service"],
176 input: {
177 readonly scopeId: CheckpointScopeId;
178 readonly ordinalWithinScope: number;
179 },
180) {
181 return idAllocator.allocate.checkpoint({
182 checkpointScopeId: input.scopeId,
183 name: String(input.ordinalWithinScope),
184 });
185}
186
187function makeRootRunScope(input: {
188 readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
189 readonly threadId: ThreadId;
190 readonly runId: RunId;
191 readonly rootNodeId: NodeId;
192 readonly providerThreadId: ProviderThreadId;
193 readonly cwd: string;
194 readonly createdAt: DateTime.Utc;
195}) {
196 return Effect.gen(function* () {
197 const scopeId = yield* input.idAllocator.allocate.checkpointScope({
198 threadId: input.threadId,
199 name: ROOT_CHECKPOINT_SCOPE_NAME,
200 });
201 return {
202 id: scopeId,
203 threadId: input.threadId,
204 runId: input.runId,
205 nodeId: input.rootNodeId,
206 parentScopeId: null,
207 providerThreadId: input.providerThreadId,
208 kind: "root_run",
209 ordinalWithinParent: 0,
210 advancesAppRunCount: true,
211 cwd: input.cwd,
212 createdAt: input.createdAt,
213 } satisfies OrchestrationV2CheckpointScope;
214 });
215}
216
217function makeCheckpoint(input: {
218 readonly id: CheckpointId;
219 readonly scope: OrchestrationV2CheckpointScope;
220 readonly runId: RunId | null;
221 readonly nodeId: NodeId;
222 readonly parentCheckpointId: CheckpointId | null;
223 readonly ordinalWithinScope: number;
224 readonly appRunOrdinal: number | null;
225 readonly ref: CheckpointRef;
226 readonly status: OrchestrationV2Checkpoint["status"];
227 readonly files: OrchestrationV2Checkpoint["files"];
228 readonly capturedAt: DateTime.Utc;
229}): OrchestrationV2Checkpoint {
230 return {
231 id: input.id,
232 threadId: input.scope.threadId,
233 scopeId: input.scope.id,
234 runId: input.runId,
235 nodeId: input.nodeId,
236 parentCheckpointId: input.parentCheckpointId,
237 ordinalWithinScope: input.ordinalWithinScope,
238 appRunOrdinal: input.appRunOrdinal,
239 ref: input.ref,
240 status: input.status,
241 files: input.files,
242 capturedAt: input.capturedAt,
243 };
244}
245
246export const layer: Layer.Layer<
247 CheckpointServiceV2,
248 never,
249 CheckpointStore.CheckpointStore | Crypto.Crypto | IdAllocator.IdAllocatorV2
250> = Layer.effect(
251 CheckpointServiceV2,
252 Effect.gen(function* () {
253 const checkpointStore = yield* CheckpointStore.CheckpointStore;
254 const crypto = yield* Crypto.Crypto;
255 const checkpointRefFor = (input: Parameters<typeof checkpointRefForScopeOrdinal>[0]) =>
256 checkpointRefForScopeOrdinal(input).pipe(Effect.provideService(Crypto.Crypto, crypto));
257 const idAllocator = yield* IdAllocator.IdAllocatorV2;
258 const workspaceLocks = yield* KeyedLock.make<string>();
259 const withWorkspaceLock = <A, E, R>(cwd: string, effect: Effect.Effect<A, E, R>) =>
260 workspaceLocks.withLock(cwd, effect);
261
262 const isGitCheckpointable = (cwd: string) =>
263 checkpointStore.isGitRepository(cwd).pipe(Effect.orElseSucceed(() => false));
264
265 const ensureScope: CheckpointServiceV2Shape["ensureScope"] = (scope) => Effect.succeed(scope);
266
267 const captureBaseline: CheckpointServiceV2Shape["captureBaseline"] = (input) =>
268 withWorkspaceLock(
269 input.scope.cwd,
270 Effect.gen(function* () {
271 if (!(yield* isGitCheckpointable(input.scope.cwd))) {
272 return;
273 }
274
275 const checkpointRef = yield* checkpointRefFor({
276 scopeId: input.scope.id,
277 ordinalWithinScope: input.ordinalWithinScope,
278 });
279 const exists = yield* checkpointStore.hasCheckpointRef({
280 cwd: input.scope.cwd,
281 checkpointRef,
282 });
283 if (exists) {
284 return;
285 }
286
287 yield* checkpointStore.captureCheckpoint({
288 cwd: input.scope.cwd,
289 checkpointRef,
290 });
291 }),
292 ).pipe(
293 Effect.mapError(
294 (cause) =>
295 new CheckpointBaselineCaptureError({
296 scopeId: input.scope.id,
297 ordinalWithinScope: input.ordinalWithinScope,
298 cause,
299 }),
300 ),
301 );
302
303 const materializeBaselineCheckpoint: CheckpointServiceV2Shape["materializeBaselineCheckpoint"] =
304 (input) =>
305 withWorkspaceLock(
306 input.scope.cwd,
307 Effect.gen(function* () {
308 const checkpointRef = yield* checkpointRefFor({
309 scopeId: input.scope.id,
310 ordinalWithinScope: input.ordinalWithinScope,
311 });
312 const checkpointId = yield* checkpointIdForScopeOrdinal(idAllocator, {
313 scopeId: input.scope.id,
314 ordinalWithinScope: input.ordinalWithinScope,
315 });
316 const checkpointable = yield* isGitCheckpointable(input.scope.cwd);
317 const available = checkpointable
318 ? yield* checkpointStore
319 .hasCheckpointRef({
320 cwd: input.scope.cwd,
321 checkpointRef,
322 })
323 .pipe(
324 Effect.catch((cause) =>
325 Effect.logWarning("orchestration V2 baseline ref lookup failed", {
326 scopeId: input.scope.id,
327 checkpointRef,
328 cause: String(cause),
329 }).pipe(Effect.as(false)),
330 ),
331 )
332 : false;
333 return makeCheckpoint({
334 id: checkpointId,
335 scope: input.scope,
336 runId: null,
337 nodeId: input.scope.nodeId,
338 parentCheckpointId: null,
339 ordinalWithinScope: input.ordinalWithinScope,
340 appRunOrdinal: null,
341 ref: checkpointRef,
342 status: available ? "ready" : "missing",
343 files: [],
344 capturedAt: input.scope.createdAt,
345 });
346 }),
347 ).pipe(
348 Effect.mapError(
349 (cause) =>
350 new CheckpointCaptureError({
351 scopeId: input.scope.id,
352 cause,
353 }),
354 ),
355 );
356
357 const capture: CheckpointServiceV2Shape["capture"] = (input) =>
358 withWorkspaceLock(
359 input.scope.cwd,
360 Effect.gen(function* () {
361 const checkpointId = yield* checkpointIdForScopeOrdinal(idAllocator, {
362 scopeId: input.scope.id,
363 ordinalWithinScope: input.ordinalWithinScope,
364 });
365 const parentCheckpointId =
366 input.ordinalWithinScope > 0
367 ? yield* checkpointIdForScopeOrdinal(idAllocator, {
368 scopeId: input.scope.id,
369 ordinalWithinScope: input.ordinalWithinScope - 1,
370 })
371 : null;
372 const checkpointRef = yield* checkpointRefFor({
373 scopeId: input.scope.id,
374 ordinalWithinScope: input.ordinalWithinScope,
375 });
376 const previousCheckpointRef = yield* checkpointRefFor({
377 scopeId: input.scope.id,
378 ordinalWithinScope: Math.max(0, input.ordinalWithinScope - 1),
379 });
380
381 if (!(yield* isGitCheckpointable(input.scope.cwd))) {
382 return makeCheckpoint({
383 id: checkpointId,
384 scope: input.scope,
385 runId: input.runId,
386 nodeId: input.nodeId,
387 parentCheckpointId,
388 ordinalWithinScope: input.ordinalWithinScope,
389 appRunOrdinal: input.appRunOrdinal,
390 ref: checkpointRef,
391 status: "missing",
392 files: [],
393 capturedAt: input.capturedAt,
394 });
395 }
396
397 const captured = yield* checkpointStore
398 .captureCheckpoint({
399 cwd: input.scope.cwd,
400 checkpointRef,
401 })
402 .pipe(
403 Effect.as(true),
404 Effect.catch((cause) =>
405 Effect.logWarning("orchestration V2 checkpoint capture failed", {
406 scopeId: input.scope.id,
407 checkpointRef,
408 cause: String(cause),
409 }).pipe(Effect.as(false)),
410 ),
411 );
412
413 if (!captured) {
414 return makeCheckpoint({
415 id: checkpointId,
416 scope: input.scope,
417 runId: input.runId,
418 nodeId: input.nodeId,
419 parentCheckpointId,
420 ordinalWithinScope: input.ordinalWithinScope,
421 appRunOrdinal: input.appRunOrdinal,
422 ref: checkpointRef,
423 status: "error",
424 files: [],
425 capturedAt: input.capturedAt,
426 });
427 }
428
429 const previousExists = yield* checkpointStore
430 .hasCheckpointRef({
431 cwd: input.scope.cwd,
432 checkpointRef: previousCheckpointRef,
433 })
434 .pipe(
435 Effect.catch((cause) =>
436 Effect.logWarning("orchestration V2 previous checkpoint ref lookup failed", {
437 scopeId: input.scope.id,
438 checkpointRef: previousCheckpointRef,
439 cause: String(cause),
440 }).pipe(Effect.as(false)),
441 ),
442 );
443 const refs = {
444 cwd: input.scope.cwd,
445 fromCheckpointRef: previousCheckpointRef,
446 toCheckpointRef: checkpointRef,
447 };
448 // A pull or rebase can change thousands of files the turn did not write.
449 // Keep only the files the turn's own work touched.
450 const files = previousExists
451 ? yield* Effect.all([
452 checkpointStore.diffCheckpoints({
453 ...refs,
454 fallbackFromToHead: false,
455 ignoreWhitespace: false,
456 format: "numstat",
457 }),
458 checkpointStore.listAuthoredPaths(refs).pipe(
459 Effect.catch((cause) =>
460 Effect.logWarning("orchestration V2 checkpoint authored paths failed", {
461 scopeId: input.scope.id,
462 checkpointRef,
463 cause: String(cause),
464 }).pipe(Effect.as(null)),
465 ),
466 ),
467 ]).pipe(
468 Effect.map(([diff, authoredPaths]) =>
469 parseTurnDiffFilesFromNumstat(diff)
470 .filter((file) => !isGitImport(file, authoredPaths))
471 .map((file) => ({
472 path: file.path,
473 kind: "modified",
474 additions: file.additions,
475 deletions: file.deletions,
476 })),
477 ),
478 Effect.catch((cause) =>
479 Effect.logWarning("orchestration V2 checkpoint diff summary failed", {
480 scopeId: input.scope.id,
481 checkpointRef,
482 cause: String(cause),
483 }).pipe(Effect.as([])),
484 ),
485 )
486 : [];
487
488 return makeCheckpoint({
489 id: checkpointId,
490 scope: input.scope,
491 runId: input.runId,
492 nodeId: input.nodeId,
493 parentCheckpointId,
494 ordinalWithinScope: input.ordinalWithinScope,
495 appRunOrdinal: input.appRunOrdinal,
496 ref: checkpointRef,
497 status: "ready",
498 files,
499 capturedAt: input.capturedAt,
500 });
501 }),
502 ).pipe(
503 Effect.mapError(
504 (cause) =>
505 new CheckpointCaptureError({
506 scopeId: input.scope.id,
507 cause,
508 }),
509 ),
510 );
511
512 const restore: CheckpointServiceV2Shape["restore"] = (input) =>
513 withWorkspaceLock(
514 input.scope.cwd,
515 Effect.gen(function* () {
516 if (input.checkpoint.status !== "ready") {
517 return yield* new CheckpointRestoreError({
518 scopeId: input.scope.id,
519 checkpointId: input.checkpoint.id,
520 cause: `Checkpoint status is ${input.checkpoint.status}.`,
521 });
522 }
523
524 const restored = yield* checkpointStore.restoreCheckpoint({
525 cwd: input.scope.cwd,
526 checkpointRef: input.checkpoint.ref,
527 fallbackToHead: false,
528 });
529 if (!restored) {
530 return yield* new CheckpointRestoreError({
531 scopeId: input.scope.id,
532 checkpointId: input.checkpoint.id,
533 cause: "Checkpoint ref is unavailable.",
534 });
535 }
536 }),
537 ).pipe(
538 Effect.mapError((cause) =>
539 isCheckpointRestoreError(cause)
540 ? cause
541 : new CheckpointRestoreError({
542 scopeId: input.scope.id,
543 checkpointId: input.checkpoint.id,
544 cause,
545 }),
546 ),
547 );
548
549 const deleteStaleRefs: CheckpointServiceV2Shape["deleteStaleRefs"] = (input) =>
550 withWorkspaceLock(
551 input.scope.cwd,
552 checkpointStore.deleteCheckpointRefs({
553 cwd: input.scope.cwd,
554 checkpointRefs: input.checkpoints.map((checkpoint) => checkpoint.ref),
555 }),
556 ).pipe(
557 Effect.mapError(
558 (cause) =>
559 new CheckpointDeleteStaleRefsError({
560 scopeId: input.scope.id,
561 checkpointIds: input.checkpoints.map((checkpoint) => checkpoint.id),
562 cause,
563 }),
564 ),
565 );
566
567 return CheckpointServiceV2.of({
568 prepareRootRunScope: (input) =>
569 makeRootRunScope({ ...input, idAllocator }).pipe(
570 Effect.mapError(
571 (cause) =>
572 new CheckpointRootScopePrepareError({
573 threadId: input.threadId,
574 runId: input.runId,
575 cause,
576 }),
577 ),
578 ),
579 ensureScope,
580 captureBaseline,
581 materializeBaselineCheckpoint,
582 capture,
583 restore,
584 deleteStaleRefs,
585 } satisfies CheckpointServiceV2Shape);
586 }),
587);