1import {
2 CheckpointScopeId,
3 CommandId,
4 type OrchestrationV2Checkpoint,
5 type OrchestrationV2ExecutionNode,
6 type OrchestrationV2ProviderThread,
7 type OrchestrationV2Run,
8 type OrchestrationV2TurnItem,
9 RunId,
10 ThreadId,
11} from "@t3tools/contracts";
12import * as Context from "effect/Context";
13import * as DateTime from "effect/DateTime";
14import * as Effect from "effect/Effect";
15import * as Layer from "effect/Layer";
16import * as Schema from "effect/Schema";
17
18import * as CheckpointService from "./CheckpointService.ts";
19import * as EventSink from "./EventSink.ts";
20import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
21import * as ProjectionStore from "./ProjectionStore.ts";
22
23export class CheckpointCaptureExecutionError extends Schema.TaggedError<CheckpointCaptureExecutionError>()(
24 "CheckpointCaptureExecutionError",
25 {
26 threadId: ThreadId,
27 runId: RunId,
28 scopeId: CheckpointScopeId,
29 cause: Schema.optional(Schema.Defect()),
30 },
31) {}
32
33const isCheckpointCaptureExecutionError = Schema.is(CheckpointCaptureExecutionError);
34
35export interface CheckpointCaptureServiceV2Shape {
36 readonly execute: (input: {
37 readonly threadId: ThreadId;
38 readonly runId: RunId;
39 readonly scopeId: CheckpointScopeId;
40 }) => Effect.Effect<void, CheckpointCaptureExecutionError>;
41}
42
43export class CheckpointCaptureServiceV2 extends Context.Service<
44 CheckpointCaptureServiceV2,
45 CheckpointCaptureServiceV2Shape
46>()("t3/orchestration-v2/CheckpointCaptureService/CheckpointCaptureServiceV2") {}
47
48export const layer: Layer.Layer<
49 CheckpointCaptureServiceV2,
50 never,
51 | CheckpointService.CheckpointServiceV2
52 | EventSink.EventSinkV2
53 | IdAllocator.IdAllocatorV2
54 | ProjectionStore.ProjectionStoreV2
55> = Layer.effect(
56 CheckpointCaptureServiceV2,
57 Effect.gen(function* () {
58 const checkpoints = yield* CheckpointService.CheckpointServiceV2;
59 const eventSink = yield* EventSink.EventSinkV2;
60 const ids = yield* IdAllocator.IdAllocatorV2;
61 const projections = yield* ProjectionStore.ProjectionStoreV2;
62
63 const execute = Effect.fn("orchestrationV2.checkpointCapture.execute")(function* (input: {
64 readonly threadId: ThreadId;
65 readonly runId: RunId;
66 readonly scopeId: CheckpointScopeId;
67 }) {
68 const { run, rootNode, scope, providerThread, preservedCheckpointOrdinals, sideChat } =
69 yield* projections.getCheckpointCaptureContext(input.threadId, input);
70 // Stopped runs remain terminal; their checkpoint is the next message's rollback point.
71 const stopped = run?.status === "interrupted" || run?.status === "cancelled";
72
73 // A checkpoint on a settled run proves this at-least-once effect already committed.
74 if (
75 run !== undefined &&
76 run.checkpointId !== null &&
77 (run.status === "completed" || stopped)
78 ) {
79 return;
80 }
81 // Rollback uses the same lane and can precede a capture retry; preserve the restored workspace and discarded run.
82 if (run?.status === "rolled_back") {
83 return;
84 }
85
86 if (
87 run === undefined ||
88 (run.status !== "waiting" && !stopped) ||
89 rootNode === undefined ||
90 scope === undefined ||
91 rootNode.checkpointScopeId !== scope.id ||
92 providerThread === undefined
93 ) {
94 return yield* new CheckpointCaptureExecutionError({
95 threadId: input.threadId,
96 runId: input.runId,
97 scopeId: input.scopeId,
98 cause: "The persisted checkpoint capture target is incomplete or no longer waiting.",
99 });
100 }
101
102 const capturedAt = yield* DateTime.now;
103 const baselineOrdinalWithinScope = Math.max(0, run.ordinal - 1);
104 const isPreserved = (ordinalWithinScope: number) =>
105 preservedCheckpointOrdinals.includes(ordinalWithinScope);
106 // The baselines this capture materializes are attributed to the run materializing them. The
107 // thread's scope may already be a later run's: one queued meanwhile, or sent after Stop.
108 const materializeBaseline = (ordinalWithinScope: number) =>
109 checkpoints.materializeBaselineCheckpoint({
110 scope,
111 ordinalWithinScope,
112 nodeId: rootNode.id,
113 capturedAt: run.requestedAt,
114 forceMissing: sideChat,
115 });
116 const threadStartCheckpoint =
117 baselineOrdinalWithinScope === 0 || isPreserved(0) ? null : yield* materializeBaseline(0);
118 const baselineCheckpoint = isPreserved(baselineOrdinalWithinScope)
119 ? null
120 : yield* materializeBaseline(baselineOrdinalWithinScope);
121 const checkpoint = yield* checkpoints.capture({
122 scope,
123 runId: run.id,
124 nodeId: rootNode.id,
125 ordinalWithinScope: run.ordinal,
126 appRunOrdinal: run.ordinal,
127 capturedAt,
128 forceMissing: sideChat,
129 });
130 // Omit delegatedCompletion to preserve cohort writes made after capture loaded the run.
131 const { delegatedCompletion: _delegatedCompletion, ...runWithoutDelegatedCompletion } = run;
132 const commandId = CommandId.make(`command:effect:checkpoint.capture:${run.id}`);
133 yield* eventSink.commitCommand({
134 commandId,
135 threadId: input.threadId,
136 commandType: "checkpoint.capture",
137 acceptedAt: capturedAt,
138 effects: [],
139 events: [
140 ...(threadStartCheckpoint === null
141 ? []
142 : [
143 {
144 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
145 type: "checkpoint.captured" as const,
146 threadId: input.threadId,
147 nodeId: threadStartCheckpoint.nodeId,
148 driver: providerThread.driver,
149 providerInstanceId: run.providerInstanceId,
150 occurredAt: capturedAt,
151 payload: threadStartCheckpoint,
152 },
153 ]),
154 ...(baselineCheckpoint === null
155 ? []
156 : [
157 {
158 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
159 type: "checkpoint.captured" as const,
160 threadId: input.threadId,
161 nodeId: baselineCheckpoint.nodeId,
162 driver: providerThread.driver,
163 providerInstanceId: run.providerInstanceId,
164 occurredAt: capturedAt,
165 payload: baselineCheckpoint,
166 },
167 ]),
168 {
169 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
170 type: "checkpoint.captured",
171 threadId: input.threadId,
172 runId: run.id,
173 nodeId: rootNode.id,
174 driver: providerThread.driver,
175 providerInstanceId: run.providerInstanceId,
176 occurredAt: capturedAt,
177 payload: checkpoint,
178 },
179 {
180 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
181 type: "turn-item.updated",
182 threadId: input.threadId,
183 runId: run.id,
184 nodeId: rootNode.id,
185 driver: providerThread.driver,
186 providerInstanceId: run.providerInstanceId,
187 occurredAt: capturedAt,
188 payload: makeCheckpointTurnItem({
189 idAllocator: ids,
190 run,
191 rootNode,
192 providerThread,
193 checkpoint,
194 completedAt: capturedAt,
195 }),
196 },
197 {
198 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
199 type: "run.updated",
200 threadId: input.threadId,
201 runId: run.id,
202 nodeId: rootNode.id,
203 providerInstanceId: run.providerInstanceId,
204 occurredAt: capturedAt,
205 payload: stopped
206 ? { ...runWithoutDelegatedCompletion, checkpointId: checkpoint.id }
207 : {
208 ...runWithoutDelegatedCompletion,
209 status: "completed",
210 completedAt: capturedAt,
211 checkpointId: checkpoint.id,
212 },
213 },
214 ...(stopped
215 ? []
216 : [
217 {
218 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
219 type: "node.updated" as const,
220 threadId: input.threadId,
221 runId: run.id,
222 nodeId: rootNode.id,
223 providerInstanceId: run.providerInstanceId,
224 occurredAt: capturedAt,
225 payload: {
226 ...rootNode,
227 status: "completed" as const,
228 completedAt: capturedAt,
229 checkpointScopeId: scope.id,
230 },
231 },
232 ]),
233 ],
234 });
235 });
236
237 return CheckpointCaptureServiceV2.of({
238 execute: Effect.fn("CheckpointCaptureServiceV2.execute")(function (input) {
239 return execute(input).pipe(
240 Effect.mapError((cause) =>
241 isCheckpointCaptureExecutionError(cause)
242 ? cause
243 : new CheckpointCaptureExecutionError({ ...input, cause }),
244 ),
245 );
246 }),
247 });
248 }),
249);
250
251function makeCheckpointTurnItem(input: {
252 readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
253 readonly run: OrchestrationV2Run;
254 readonly rootNode: OrchestrationV2ExecutionNode;
255 readonly providerThread: OrchestrationV2ProviderThread;
256 readonly checkpoint: OrchestrationV2Checkpoint;
257 readonly completedAt: DateTime.Utc;
258}): OrchestrationV2TurnItem {
259 return {
260 id: input.idAllocator.derive.turnItemFromProviderItem({
261 driver: input.providerThread.driver,
262 nativeItemId: `checkpoint:${input.checkpoint.id}`,
263 }),
264 threadId: input.run.threadId,
265 runId: input.run.id,
266 nodeId: input.rootNode.id,
267 providerThreadId: input.providerThread.id,
268 providerTurnId: input.rootNode.providerTurnId,
269 nativeItemRef: null,
270 parentItemId: null,
271 ordinal: input.run.ordinal * 100 + 99,
272 status: "completed",
273 title: null,
274 startedAt: input.completedAt,
275 completedAt: input.completedAt,
276 updatedAt: input.completedAt,
277 type: "checkpoint",
278 checkpointId: input.checkpoint.id,
279 scopeId: input.checkpoint.scopeId,
280 files: input.checkpoint.files,
281 };
282}