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, readyCheckpointOrdinals } =
69 yield* projections.getCheckpointCaptureContext(input.threadId, input);
70 // A stopped run is already terminal. Its checkpoint is the rollback point
71 // for the message after it, so capture leaves its status alone.
72 const stopped = run?.status === "interrupted" || run?.status === "cancelled";
73
74 // The effect is at-least-once. A settled run with a checkpoint proves
75 // that an earlier execution committed its result.
76 if (
77 run !== undefined &&
78 run.checkpointId !== null &&
79 (run.status === "completed" || stopped)
80 ) {
81 return;
82 }
83 // Rollback shares this effect lane, so it can only land before a capture
84 // runs, e.g. while a failed capture waits to retry. The workspace now
85 // holds the rollback target, and the run must stay discarded.
86 if (run?.status === "rolled_back") {
87 return;
88 }
89
90 if (
91 run === undefined ||
92 (run.status !== "waiting" && !stopped) ||
93 rootNode === undefined ||
94 scope === undefined ||
95 rootNode.checkpointScopeId !== scope.id ||
96 providerThread === undefined
97 ) {
98 return yield* new CheckpointCaptureExecutionError({
99 threadId: input.threadId,
100 runId: input.runId,
101 scopeId: input.scopeId,
102 cause: "The persisted checkpoint capture target is incomplete or no longer waiting.",
103 });
104 }
105
106 const capturedAt = yield* DateTime.now;
107 const baselineOrdinalWithinScope = Math.max(0, run.ordinal - 1);
108 const hasReadyCheckpoint = (ordinalWithinScope: number) =>
109 readyCheckpointOrdinals.includes(ordinalWithinScope);
110 const threadStartCheckpoint =
111 baselineOrdinalWithinScope === 0 || hasReadyCheckpoint(0)
112 ? null
113 : yield* checkpoints.materializeBaselineCheckpoint({
114 scope,
115 ordinalWithinScope: 0,
116 });
117 const baselineCheckpoint = hasReadyCheckpoint(baselineOrdinalWithinScope)
118 ? null
119 : yield* checkpoints.materializeBaselineCheckpoint({
120 scope,
121 ordinalWithinScope: baselineOrdinalWithinScope,
122 });
123 const checkpoint = yield* checkpoints.capture({
124 scope,
125 runId: run.id,
126 nodeId: rootNode.id,
127 ordinalWithinScope: run.ordinal,
128 appRunOrdinal: run.ordinal,
129 capturedAt,
130 });
131 // Match RunExecutionService: capture loaded the waiting run before
132 // materializing baselines. Omit delegatedCompletion so a newer cohort
133 // write during capture is not overwritten by this stale snapshot
134 // (ProjectionStore preserves the field when absent from the payload).
135 const { delegatedCompletion: _delegatedCompletion, ...runWithoutDelegatedCompletion } = run;
136 const commandId = CommandId.make(`command:effect:checkpoint.capture:${run.id}`);
137 yield* eventSink.commitCommand({
138 commandId,
139 threadId: input.threadId,
140 commandType: "checkpoint.capture",
141 acceptedAt: capturedAt,
142 effects: [],
143 events: [
144 ...(threadStartCheckpoint === null
145 ? []
146 : [
147 {
148 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
149 type: "checkpoint.captured" as const,
150 threadId: input.threadId,
151 nodeId: threadStartCheckpoint.nodeId,
152 driver: providerThread.driver,
153 providerInstanceId: run.providerInstanceId,
154 occurredAt: capturedAt,
155 payload: threadStartCheckpoint,
156 },
157 ]),
158 ...(baselineCheckpoint === null
159 ? []
160 : [
161 {
162 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
163 type: "checkpoint.captured" as const,
164 threadId: input.threadId,
165 nodeId: baselineCheckpoint.nodeId,
166 driver: providerThread.driver,
167 providerInstanceId: run.providerInstanceId,
168 occurredAt: capturedAt,
169 payload: baselineCheckpoint,
170 },
171 ]),
172 {
173 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
174 type: "checkpoint.captured",
175 threadId: input.threadId,
176 runId: run.id,
177 nodeId: rootNode.id,
178 driver: providerThread.driver,
179 providerInstanceId: run.providerInstanceId,
180 occurredAt: capturedAt,
181 payload: checkpoint,
182 },
183 {
184 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
185 type: "turn-item.updated",
186 threadId: input.threadId,
187 runId: run.id,
188 nodeId: rootNode.id,
189 driver: providerThread.driver,
190 providerInstanceId: run.providerInstanceId,
191 occurredAt: capturedAt,
192 payload: makeCheckpointTurnItem({
193 idAllocator: ids,
194 run,
195 rootNode,
196 providerThread,
197 checkpoint,
198 completedAt: capturedAt,
199 }),
200 },
201 {
202 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
203 type: "run.updated",
204 threadId: input.threadId,
205 runId: run.id,
206 nodeId: rootNode.id,
207 providerInstanceId: run.providerInstanceId,
208 occurredAt: capturedAt,
209 payload: stopped
210 ? { ...runWithoutDelegatedCompletion, checkpointId: checkpoint.id }
211 : {
212 ...runWithoutDelegatedCompletion,
213 status: "completed",
214 completedAt: capturedAt,
215 checkpointId: checkpoint.id,
216 },
217 },
218 ...(stopped
219 ? []
220 : [
221 {
222 id: yield* ids.allocate.event({ threadId: input.threadId, commandId }),
223 type: "node.updated" as const,
224 threadId: input.threadId,
225 runId: run.id,
226 nodeId: rootNode.id,
227 providerInstanceId: run.providerInstanceId,
228 occurredAt: capturedAt,
229 payload: {
230 ...rootNode,
231 status: "completed" as const,
232 completedAt: capturedAt,
233 checkpointScopeId: scope.id,
234 },
235 },
236 ]),
237 ],
238 });
239 });
240
241 return CheckpointCaptureServiceV2.of({
242 execute: (input) =>
243 execute(input).pipe(
244 Effect.mapError((cause) =>
245 isCheckpointCaptureExecutionError(cause)
246 ? cause
247 : new CheckpointCaptureExecutionError({ ...input, cause }),
248 ),
249 ),
250 });
251 }),
252);
253
254function makeCheckpointTurnItem(input: {
255 readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
256 readonly run: OrchestrationV2Run;
257 readonly rootNode: OrchestrationV2ExecutionNode;
258 readonly providerThread: OrchestrationV2ProviderThread;
259 readonly checkpoint: OrchestrationV2Checkpoint;
260 readonly completedAt: DateTime.Utc;
261}): OrchestrationV2TurnItem {
262 return {
263 id: input.idAllocator.derive.turnItemFromProviderItem({
264 driver: input.providerThread.driver,
265 nativeItemId: `checkpoint:${input.checkpoint.id}`,
266 }),
267 threadId: input.run.threadId,
268 runId: input.run.id,
269 nodeId: input.rootNode.id,
270 providerThreadId: input.providerThread.id,
271 providerTurnId: input.rootNode.providerTurnId,
272 nativeItemRef: null,
273 parentItemId: null,
274 ordinal: input.run.ordinal * 100 + 99,
275 status: "completed",
276 title: null,
277 startedAt: input.completedAt,
278 completedAt: input.completedAt,
279 updatedAt: input.completedAt,
280 type: "checkpoint",
281 checkpointId: input.checkpoint.id,
282 scopeId: input.checkpoint.scopeId,
283 files: input.checkpoint.files,
284 };
285}