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