1import * as NodeCrypto from "@effect/platform-node/NodeCrypto";
2import { assert, it } from "@effect/vitest";
3import {
4 CheckpointId,
5 CheckpointRef,
6 CheckpointScopeId,
7 EventId,
8 MessageId,
9 NodeId,
10 type OrchestrationV2DomainEvent,
11 type OrchestrationV2Run,
12 ProjectId,
13 ProviderDriverKind,
14 ProviderInstanceId,
15 ProviderThreadId,
16 RunId,
17 ThreadId,
18} from "@t3tools/contracts";
19import * as DateTime from "effect/DateTime";
20import * as Effect from "effect/Effect";
21import * as Layer from "effect/Layer";
22import * as Ref from "effect/Ref";
23import * as SqlClient from "effect/sql/SqlClient";
24
25import * as SqlitePersistence from "../persistence/Sqlite.ts";
26import * as CheckpointStore from "../checkpointing/CheckpointStore.ts";
27import { VcsProcessTimeoutError } from "@t3tools/contracts";
28import * as CheckpointService from "./CheckpointService.ts";
29import * as CheckpointCaptureService from "./CheckpointCaptureService.ts";
30import * as EventSink from "./EventSink.ts";
31import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
32import * as ProjectionStore from "./ProjectionStore.ts";
33
34const layerProjectionStoreTest = Layer.mergeAll(
35 ProjectionStore.layer.pipe(Layer.provideMerge(SqlitePersistence.layerMemory)),
36 SqlitePersistence.layerMemory,
37);
38
39const threadId = ThreadId.make("thread:checkpoint-capture-delegated");
40const projectId = ProjectId.make("project:checkpoint-capture-delegated");
41const runId = RunId.make("run:checkpoint-capture-delegated");
42const scopeId = CheckpointScopeId.make("scope:checkpoint-capture-delegated");
43const rootNodeId = NodeId.make("node:checkpoint-capture-delegated-root");
44const taskId = NodeId.make("node:checkpoint-capture-task");
45const deliveryMessageId = MessageId.make("message:checkpoint-capture-delivery");
46const providerThreadId = ProviderThreadId.make("provider-thread:checkpoint-capture-delegated");
47const providerInstanceId = ProviderInstanceId.make("codex");
48const driver = ProviderDriverKind.make("codex");
49const modelSelection = {
50 instanceId: providerInstanceId,
51 model: "gpt-5.4",
52} as const;
53
54it.layer(layerProjectionStoreTest)("CheckpointCaptureServiceV2", (it) => {
55 it.effect.each([false, true])(
56 "captures without decoding history or losing newer delegated completion, ref lookup fails=%s",
57 (refLookupFails) =>
58 Effect.gen(function* () {
59 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
60 const now = yield* DateTime.now;
61 const later = DateTime.add(now, { seconds: 1 });
62
63 const staleDelegatedCompletion = {
64 disposition: "open" as const,
65 nextGeneration: 1,
66 delivery: null,
67 };
68 const newerCohort = {
69 disposition: "open" as const,
70 nextGeneration: 2,
71 delivery: {
72 generation: 1,
73 messageId: deliveryMessageId,
74 taskIds: [taskId],
75 },
76 };
77 const staleRun: OrchestrationV2Run = {
78 id: runId,
79 threadId,
80 ordinal: 1,
81 providerInstanceId,
82 modelSelection,
83 providerThreadId,
84 userMessageId: MessageId.make("message:checkpoint-capture-user"),
85 rootNodeId,
86 activeAttemptId: null,
87 status: "waiting",
88 requestedAt: now,
89 startedAt: now,
90 completedAt: null,
91 checkpointId: null,
92 contextHandoffId: null,
93 delegatedCompletion: staleDelegatedCompletion,
94 };
95 const rootNode = {
96 id: rootNodeId,
97 threadId,
98 runId,
99 parentNodeId: null,
100 rootNodeId,
101 kind: "root_turn" as const,
102 status: "waiting" as const,
103 countsForRun: true,
104 providerThreadId,
105 providerTurnId: null,
106 nativeItemRef: null,
107 runtimeRequestId: null,
108 checkpointScopeId: scopeId,
109 startedAt: now,
110 completedAt: null,
111 };
112 const scope = {
113 id: scopeId,
114 threadId,
115 runId,
116 nodeId: rootNodeId,
117 parentScopeId: null,
118 providerThreadId,
119 kind: "root_run" as const,
120 ordinalWithinParent: 0,
121 advancesAppRunCount: true,
122 cwd: "/repo",
123 createdAt: now,
124 };
125 const providerThread = {
126 id: providerThreadId,
127 driver,
128 providerInstanceId,
129 providerSessionId: null,
130 appThreadId: threadId,
131 ownerNodeId: rootNodeId,
132 nativeThreadRef: null,
133 nativeConversationHeadRef: null,
134 status: "idle" as const,
135 firstRunOrdinal: 1,
136 lastRunOrdinal: 1,
137 handoffIds: [],
138 forkedFrom: null,
139 createdAt: now,
140 updatedAt: now,
141 };
142 const readyBaseline = {
143 id: CheckpointId.make("checkpoint:baseline-0"),
144 threadId,
145 scopeId,
146 runId: null,
147 nodeId: rootNodeId,
148 parentCheckpointId: null,
149 ordinalWithinScope: 0,
150 appRunOrdinal: null,
151 ref: CheckpointRef.make("checkpoint-ref:baseline-0"),
152 status: "ready" as const,
153 files: [],
154 capturedAt: now,
155 };
156 const captured = {
157 id: CheckpointId.make("checkpoint:captured-1"),
158 threadId,
159 scopeId,
160 runId,
161 nodeId: rootNodeId,
162 parentCheckpointId: readyBaseline.id,
163 ordinalWithinScope: 1,
164 appRunOrdinal: 1,
165 ref: CheckpointRef.make("checkpoint-ref:captured-1"),
166 status: "ready" as const,
167 files: [],
168 capturedAt: now,
169 };
170
171 yield* projectionStore.apply({
172 id: EventId.make("event:checkpoint-capture:thread"),
173 type: "thread.created",
174 threadId,
175 occurredAt: now,
176 payload: {
177 createdBy: "user",
178 creationSource: "web",
179 id: threadId,
180 projectId,
181 title: "Checkpoint capture delegated completion",
182 providerInstanceId,
183 modelSelection,
184 runtimeMode: "full-access",
185 interactionMode: "default",
186 branch: null,
187 worktreePath: null,
188 activeProviderThreadId: null,
189 lineage: {
190 parentThreadId: null,
191 relationshipToParent: null,
192 rootThreadId: threadId,
193 },
194 forkedFrom: null,
195 createdAt: now,
196 updatedAt: now,
197 archivedAt: null,
198 settledOverride: null,
199 settledAt: null,
200 lastVisitedAt: null,
201 deletedAt: null,
202 },
203 });
204 yield* projectionStore.apply({
205 id: EventId.make("event:checkpoint-capture:run-stale"),
206 type: "run.updated",
207 threadId,
208 runId,
209 nodeId: rootNodeId,
210 providerInstanceId,
211 occurredAt: now,
212 payload: staleRun,
213 });
214 yield* projectionStore.apply({
215 id: EventId.make("event:checkpoint-capture:node"),
216 type: "node.updated",
217 threadId,
218 runId,
219 nodeId: rootNodeId,
220 providerInstanceId,
221 occurredAt: now,
222 payload: rootNode,
223 });
224 yield* projectionStore.apply({
225 id: EventId.make("event:checkpoint-capture:provider-thread"),
226 type: "provider-thread.updated",
227 threadId,
228 nodeId: rootNodeId,
229 driver,
230 providerInstanceId,
231 occurredAt: now,
232 payload: providerThread,
233 });
234 yield* projectionStore.apply({
235 id: EventId.make("event:checkpoint-capture:scope"),
236 type: "checkpoint-scope.created",
237 threadId,
238 runId,
239 nodeId: rootNodeId,
240 occurredAt: now,
241 payload: scope,
242 });
243 yield* projectionStore.apply({
244 id: EventId.make("event:checkpoint-capture:baseline"),
245 type: "checkpoint.captured",
246 threadId,
247 nodeId: rootNodeId,
248 driver,
249 providerInstanceId,
250 occurredAt: now,
251 payload: readyBaseline,
252 });
253 const sql = yield* SqlClient.SqlClient;
254 yield* sql`INSERT OR REPLACE INTO orchestration_v2_projection_turn_items
255 (turn_item_id, thread_id, run_id, node_id, provider_thread_id, provider_turn_id,
256 parent_item_id, ordinal, type, status, updated_at, payload_json)
257 VALUES ('obsolete-history', ${threadId}, ${runId}, ${rootNodeId}, NULL, NULL,
258 NULL, 1, 'assistant_message', 'completed', ${DateTime.formatIso(now)}, '{"obsolete":true}')`;
259 yield* sql`UPDATE orchestration_v2_projection_checkpoints
260 SET payload_json = json_set(payload_json, '$.files', 'obsolete file summary')
261 WHERE checkpoint_id = ${readyBaseline.id}`;
262 assert.equal(
263 (yield* Effect.exit(projectionStore.getThreadProjection(threadId)))._tag,
264 "Failure",
265 );
266
267 const committed = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
268 const layerCapture = CheckpointCaptureService.layer.pipe(
269 Layer.provide(
270 Layer.mergeAll(
271 IdAllocator.layer,
272 refLookupFails
273 ? CheckpointService.layer.pipe(
274 Layer.provide(
275 Layer.mergeAll(
276 IdAllocator.layer,
277 NodeCrypto.layer,
278 Layer.mock(CheckpointStore.CheckpointStore)({
279 listAuthoredPaths: () => Effect.succeed(null),
280 isGitRepository: () => Effect.succeed(true),
281 captureCheckpoint: () => Effect.void,
282 hasCheckpointRef: () =>
283 Effect.fail(
284 new VcsProcessTimeoutError({
285 operation: "test.hasCheckpointRef",
286 command: "git",
287 cwd: "/repo",
288 timeoutMs: 30000,
289 }),
290 ),
291 }),
292 ),
293 ),
294 )
295 : Layer.mock(CheckpointService.CheckpointServiceV2)({
296 materializeBaselineCheckpoint: () =>
297 Effect.die(
298 "baseline materialization must be skipped when ordinal 0 is ready",
299 ),
300 capture: () => Effect.succeed(captured),
301 }),
302 Layer.mock(EventSink.EventSinkV2)({
303 commitCommand: (input) =>
304 Ref.set(committed, input.events).pipe(
305 Effect.as({
306 commandId: input.commandId,
307 committed: true,
308 sequence: 1,
309 events: input.events,
310 effects: [],
311 } as never),
312 ),
313 }),
314 ),
315 ),
316 );
317
318 yield* Effect.gen(function* () {
319 const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2;
320 const incomplete = yield* service
321 .execute({ threadId, runId, scopeId: CheckpointScopeId.make("missing-scope") })
322 .pipe(Effect.flip);
323 assert.instanceOf(incomplete, CheckpointCaptureService.CheckpointCaptureExecutionError);
324 yield* service.execute({ threadId, runId, scopeId });
325
326 const events = yield* Ref.get(committed);
327 const runUpdated = events.find((event) => event.type === "run.updated");
328 assert.isDefined(runUpdated);
329 if (runUpdated?.type !== "run.updated") {
330 return;
331 }
332 assert.equal(runUpdated.payload.status, "completed");
333 const capturedEvent = events.find((event) => event.type === "checkpoint.captured");
334 assert.equal(
335 runUpdated.payload.checkpointId,
336 refLookupFails ? capturedEvent?.payload.id : captured.id,
337 );
338 if (refLookupFails && capturedEvent?.type === "checkpoint.captured") {
339 assert.equal(capturedEvent.payload.status, "ready");
340 assert.deepEqual(capturedEvent.payload.files, []);
341 }
342 assert.isUndefined(
343 runUpdated.payload.delegatedCompletion,
344 "checkpoint capture must omit delegatedCompletion so ProjectionStore can keep a newer cohort",
345 );
346
347 yield* projectionStore.apply({
348 id: EventId.make("event:checkpoint-capture:newer-cohort"),
349 type: "run.updated",
350 threadId,
351 runId,
352 nodeId: rootNodeId,
353 providerInstanceId,
354 occurredAt: later,
355 payload: {
356 ...staleRun,
357 delegatedCompletion: newerCohort,
358 },
359 });
360
361 yield* projectionStore.apply(runUpdated);
362
363 const projectedRun = (yield* projectionStore.getCheckpointCaptureContext(threadId, {
364 runId,
365 scopeId,
366 })).run;
367 assert.isDefined(projectedRun);
368 assert.equal(projectedRun?.status, "completed");
369 assert.equal(projectedRun?.checkpointId, runUpdated.payload.checkpointId);
370 assert.deepEqual(projectedRun?.delegatedCompletion, newerCohort);
371 assert.equal(projectedRun?.delegatedCompletion?.delivery?.messageId, deliveryMessageId);
372 assert.deepEqual(projectedRun?.delegatedCompletion?.delivery?.taskIds, [taskId]);
373 yield* Ref.set(committed, []);
374 yield* service.execute({ threadId, runId, scopeId });
375 assert.deepEqual(yield* Ref.get(committed), []);
376 }).pipe(Effect.provide(layerCapture));
377 }),
378 );
379
380 it.effect("does not capture a stopped run that a rollback already discarded", () =>
381 Effect.gen(function* () {
382 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
383 const now = yield* DateTime.now;
384 const stoppedThreadId = ThreadId.make("thread:checkpoint-capture-rolled-back");
385 const stoppedRunId = RunId.make("run:checkpoint-capture-rolled-back");
386 yield* projectionStore.apply({
387 id: EventId.make("event:checkpoint-capture-rolled-back:thread"),
388 type: "thread.created",
389 threadId: stoppedThreadId,
390 occurredAt: now,
391 payload: {
392 createdBy: "user",
393 creationSource: "web",
394 id: stoppedThreadId,
395 projectId,
396 title: "Checkpoint capture after rollback",
397 providerInstanceId,
398 modelSelection,
399 runtimeMode: "full-access",
400 interactionMode: "default",
401 branch: null,
402 worktreePath: null,
403 activeProviderThreadId: null,
404 lineage: {
405 parentThreadId: null,
406 relationshipToParent: null,
407 rootThreadId: stoppedThreadId,
408 },
409 forkedFrom: null,
410 createdAt: now,
411 updatedAt: now,
412 archivedAt: null,
413 settledOverride: null,
414 settledAt: null,
415 lastVisitedAt: null,
416 deletedAt: null,
417 },
418 });
419 const stoppedRun: OrchestrationV2Run = {
420 id: stoppedRunId,
421 threadId: stoppedThreadId,
422 ordinal: 2,
423 providerInstanceId,
424 modelSelection,
425 providerThreadId,
426 userMessageId: MessageId.make("message:checkpoint-capture-rolled-back"),
427 rootNodeId,
428 activeAttemptId: null,
429 status: "interrupted",
430 requestedAt: now,
431 startedAt: now,
432 completedAt: now,
433 checkpointId: null,
434 contextHandoffId: null,
435 };
436 yield* projectionStore.apply({
437 id: EventId.make("event:checkpoint-capture-rolled-back:run"),
438 type: "run.updated",
439 threadId: stoppedThreadId,
440 runId: stoppedRunId,
441 providerInstanceId,
442 occurredAt: now,
443 payload: { ...stoppedRun, status: "rolled_back" },
444 });
445
446 const committed = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
447 const layerCapture = CheckpointCaptureService.layer.pipe(
448 Layer.provide(
449 Layer.mergeAll(
450 IdAllocator.layer,
451 Layer.mock(CheckpointService.CheckpointServiceV2)({
452 materializeBaselineCheckpoint: () => Effect.die("a discarded run has no baseline"),
453 capture: () => Effect.die("a discarded run must not be captured"),
454 }),
455 Layer.mock(EventSink.EventSinkV2)({
456 commitCommand: (input) =>
457 Ref.set(committed, input.events).pipe(Effect.as({ committed: true } as never)),
458 }),
459 ),
460 ),
461 );
462
463 yield* CheckpointCaptureService.CheckpointCaptureServiceV2.pipe(
464 Effect.flatMap((service) =>
465 service.execute({ threadId: stoppedThreadId, runId: stoppedRunId, scopeId }),
466 ),
467 Effect.provide(layerCapture),
468 );
469
470 assert.deepEqual(yield* Ref.get(committed), []);
471 const projected = yield* projectionStore.getCheckpointCaptureContext(stoppedThreadId, {
472 runId: stoppedRunId,
473 scopeId,
474 });
475 assert.equal(projected.run?.status, "rolled_back");
476 assert.isNull(projected.run?.checkpointId ?? null);
477 }),
478 );
479
480 // A stopped run is captured after it ends; by then the next root run, dispatched or queued, may
481 // have re-created the thread's one root scope with its own node and time.
482 it.effect.each([1, 2])(
483 "attributes the baselines a late capture materializes to the stopped run %i",
484 (ordinal) =>
485 Effect.gen(function* () {
486 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
487 const requestedAt = yield* DateTime.now;
488 const stoppedAt = DateTime.add(requestedAt, { seconds: 1 });
489 const rescopedAt = DateTime.add(requestedAt, { seconds: 2 });
490 const lateThreadId = ThreadId.make(`thread:checkpoint-capture-late-${ordinal}`);
491 const lateRunId = RunId.make(`run:checkpoint-capture-late-${ordinal}`);
492 const lateRootNodeId = NodeId.make(`node:checkpoint-capture-late-${ordinal}-root`);
493 const lateScopeId = CheckpointScopeId.make(`scope:checkpoint-capture-late-${ordinal}`);
494 const lateProviderThreadId = ProviderThreadId.make(
495 `provider-thread:checkpoint-capture-late-${ordinal}`,
496 );
497 yield* projectionStore.apply({
498 id: EventId.make(`event:checkpoint-capture-late-${ordinal}:thread`),
499 type: "thread.created",
500 threadId: lateThreadId,
501 occurredAt: requestedAt,
502 payload: {
503 createdBy: "user",
504 creationSource: "web",
505 id: lateThreadId,
506 projectId,
507 title: "Checkpoint capture after the next run re-created the scope",
508 providerInstanceId,
509 modelSelection,
510 runtimeMode: "full-access",
511 interactionMode: "default",
512 branch: null,
513 worktreePath: null,
514 activeProviderThreadId: null,
515 lineage: {
516 parentThreadId: null,
517 relationshipToParent: null,
518 rootThreadId: lateThreadId,
519 },
520 forkedFrom: null,
521 createdAt: requestedAt,
522 updatedAt: requestedAt,
523 archivedAt: null,
524 settledOverride: null,
525 settledAt: null,
526 lastVisitedAt: null,
527 deletedAt: null,
528 },
529 });
530 yield* projectionStore.apply({
531 id: EventId.make(`event:checkpoint-capture-late-${ordinal}:run`),
532 type: "run.updated",
533 threadId: lateThreadId,
534 runId: lateRunId,
535 nodeId: lateRootNodeId,
536 providerInstanceId,
537 occurredAt: stoppedAt,
538 payload: {
539 id: lateRunId,
540 threadId: lateThreadId,
541 ordinal,
542 providerInstanceId,
543 modelSelection,
544 providerThreadId: lateProviderThreadId,
545 userMessageId: MessageId.make(`message:checkpoint-capture-late-${ordinal}`),
546 rootNodeId: lateRootNodeId,
547 activeAttemptId: null,
548 status: "interrupted",
549 requestedAt,
550 startedAt: requestedAt,
551 completedAt: stoppedAt,
552 checkpointId: null,
553 contextHandoffId: null,
554 },
555 });
556 yield* projectionStore.apply({
557 id: EventId.make(`event:checkpoint-capture-late-${ordinal}:node`),
558 type: "node.updated",
559 threadId: lateThreadId,
560 runId: lateRunId,
561 nodeId: lateRootNodeId,
562 providerInstanceId,
563 occurredAt: stoppedAt,
564 payload: {
565 id: lateRootNodeId,
566 threadId: lateThreadId,
567 runId: lateRunId,
568 parentNodeId: null,
569 rootNodeId: lateRootNodeId,
570 kind: "root_turn",
571 status: "interrupted",
572 countsForRun: true,
573 providerThreadId: lateProviderThreadId,
574 providerTurnId: null,
575 nativeItemRef: null,
576 runtimeRequestId: null,
577 checkpointScopeId: lateScopeId,
578 startedAt: requestedAt,
579 completedAt: stoppedAt,
580 },
581 });
582 yield* projectionStore.apply({
583 id: EventId.make(`event:checkpoint-capture-late-${ordinal}:provider-thread`),
584 type: "provider-thread.updated",
585 threadId: lateThreadId,
586 nodeId: lateRootNodeId,
587 driver,
588 providerInstanceId,
589 occurredAt: requestedAt,
590 payload: {
591 id: lateProviderThreadId,
592 driver,
593 providerInstanceId,
594 providerSessionId: null,
595 appThreadId: lateThreadId,
596 ownerNodeId: lateRootNodeId,
597 nativeThreadRef: null,
598 nativeConversationHeadRef: null,
599 status: "idle",
600 firstRunOrdinal: 1,
601 lastRunOrdinal: ordinal,
602 handoffIds: [],
603 forkedFrom: null,
604 createdAt: requestedAt,
605 updatedAt: requestedAt,
606 },
607 });
608 const nextRunId = RunId.make(`run:checkpoint-capture-late-${ordinal}-next`);
609 const nextRootNodeId = NodeId.make(`node:checkpoint-capture-late-${ordinal}-next-root`);
610 yield* projectionStore.apply({
611 id: EventId.make(`event:checkpoint-capture-late-${ordinal}:scope-of-next-run`),
612 type: "checkpoint-scope.created",
613 threadId: lateThreadId,
614 runId: nextRunId,
615 nodeId: nextRootNodeId,
616 occurredAt: rescopedAt,
617 payload: {
618 id: lateScopeId,
619 threadId: lateThreadId,
620 runId: nextRunId,
621 nodeId: nextRootNodeId,
622 parentScopeId: null,
623 providerThreadId: lateProviderThreadId,
624 kind: "root_run",
625 ordinalWithinParent: 0,
626 advancesAppRunCount: true,
627 cwd: "/repo",
628 createdAt: rescopedAt,
629 },
630 });
631
632 const committed = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
633 const layerCapture = CheckpointCaptureService.layer.pipe(
634 Layer.provide(
635 Layer.mergeAll(
636 IdAllocator.layer,
637 CheckpointService.layer.pipe(
638 Layer.provide(
639 Layer.mergeAll(
640 IdAllocator.layer,
641 NodeCrypto.layer,
642 Layer.mock(CheckpointStore.CheckpointStore)({
643 isGitRepository: () => Effect.succeed(true),
644 hasCheckpointRef: () => Effect.succeed(true),
645 captureCheckpoint: () => Effect.void,
646 diffCheckpoints: () => Effect.succeed(""),
647 listAuthoredPaths: () => Effect.succeed(null),
648 }),
649 ),
650 ),
651 ),
652 Layer.mock(EventSink.EventSinkV2)({
653 commitCommand: (input) =>
654 Ref.set(committed, input.events).pipe(Effect.as({ committed: true } as never)),
655 }),
656 ),
657 ),
658 );
659 yield* CheckpointCaptureService.CheckpointCaptureServiceV2.use((service) =>
660 service.execute({ threadId: lateThreadId, runId: lateRunId, scopeId: lateScopeId }),
661 ).pipe(Effect.provide(layerCapture));
662
663 const baselines = (yield* Ref.get(committed)).flatMap((event) =>
664 event.type === "checkpoint.captured" && event.payload.runId === null ? [event] : [],
665 );
666 assert.deepEqual(
667 baselines.map((event) => event.payload.ordinalWithinScope),
668 ordinal === 1 ? [0] : [0, 1],
669 );
670 for (const baseline of baselines) {
671 assert.equal(baseline.nodeId, lateRootNodeId);
672 assert.equal(baseline.payload.nodeId, lateRootNodeId);
673 assert.deepEqual(baseline.payload.capturedAt, requestedAt);
674 }
675 }),
676 );
677
678 it.effect("records the checkpoint of a cancelled run and keeps it cancelled", () =>
679 Effect.gen(function* () {
680 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
681 const now = yield* DateTime.now;
682 const cancelledAt = DateTime.add(now, { seconds: 1 });
683 const cancelledThreadId = ThreadId.make("thread:checkpoint-capture-cancelled");
684 const cancelledRunId = RunId.make("run:checkpoint-capture-cancelled");
685 const cancelledScopeId = CheckpointScopeId.make("scope:checkpoint-capture-cancelled");
686 const cancelledRootNodeId = NodeId.make("node:checkpoint-capture-cancelled-root");
687 const cancelledProviderThreadId = ProviderThreadId.make(
688 "provider-thread:checkpoint-capture-cancelled",
689 );
690 yield* projectionStore.apply({
691 id: EventId.make("event:checkpoint-capture-cancelled:thread"),
692 type: "thread.created",
693 threadId: cancelledThreadId,
694 occurredAt: now,
695 payload: {
696 createdBy: "user",
697 creationSource: "web",
698 id: cancelledThreadId,
699 projectId,
700 title: "Checkpoint capture after cancel",
701 providerInstanceId,
702 modelSelection,
703 runtimeMode: "full-access",
704 interactionMode: "default",
705 branch: null,
706 worktreePath: null,
707 activeProviderThreadId: null,
708 lineage: {
709 parentThreadId: null,
710 relationshipToParent: null,
711 rootThreadId: cancelledThreadId,
712 },
713 forkedFrom: null,
714 createdAt: now,
715 updatedAt: now,
716 archivedAt: null,
717 settledOverride: null,
718 settledAt: null,
719 lastVisitedAt: null,
720 deletedAt: null,
721 },
722 });
723 const runningRun: OrchestrationV2Run = {
724 id: cancelledRunId,
725 threadId: cancelledThreadId,
726 ordinal: 1,
727 providerInstanceId,
728 modelSelection,
729 providerThreadId: cancelledProviderThreadId,
730 userMessageId: MessageId.make("message:checkpoint-capture-cancelled"),
731 rootNodeId: cancelledRootNodeId,
732 activeAttemptId: null,
733 status: "running",
734 requestedAt: now,
735 startedAt: now,
736 completedAt: null,
737 checkpointId: null,
738 contextHandoffId: null,
739 };
740 const runningRootNode = {
741 id: cancelledRootNodeId,
742 threadId: cancelledThreadId,
743 runId: cancelledRunId,
744 parentNodeId: null,
745 rootNodeId: cancelledRootNodeId,
746 kind: "root_turn" as const,
747 status: "running" as const,
748 countsForRun: true,
749 providerThreadId: cancelledProviderThreadId,
750 providerTurnId: null,
751 nativeItemRef: null,
752 runtimeRequestId: null,
753 checkpointScopeId: cancelledScopeId,
754 startedAt: now,
755 completedAt: null,
756 };
757 const scope = {
758 id: cancelledScopeId,
759 threadId: cancelledThreadId,
760 runId: cancelledRunId,
761 nodeId: cancelledRootNodeId,
762 parentScopeId: null,
763 providerThreadId: cancelledProviderThreadId,
764 kind: "root_run" as const,
765 ordinalWithinParent: 0,
766 advancesAppRunCount: true,
767 cwd: "/repo",
768 createdAt: now,
769 };
770 yield* projectionStore.apply({
771 id: EventId.make("event:checkpoint-capture-cancelled:provider-thread"),
772 type: "provider-thread.updated",
773 threadId: cancelledThreadId,
774 nodeId: cancelledRootNodeId,
775 driver,
776 providerInstanceId,
777 occurredAt: now,
778 payload: {
779 id: cancelledProviderThreadId,
780 driver,
781 providerInstanceId,
782 providerSessionId: null,
783 appThreadId: cancelledThreadId,
784 ownerNodeId: cancelledRootNodeId,
785 nativeThreadRef: null,
786 nativeConversationHeadRef: null,
787 status: "idle",
788 firstRunOrdinal: 1,
789 lastRunOrdinal: 1,
790 handoffIds: [],
791 forkedFrom: null,
792 createdAt: now,
793 updatedAt: now,
794 },
795 });
796 yield* projectionStore.apply({
797 id: EventId.make("event:checkpoint-capture-cancelled:scope"),
798 type: "checkpoint-scope.created",
799 threadId: cancelledThreadId,
800 runId: cancelledRunId,
801 nodeId: cancelledRootNodeId,
802 occurredAt: now,
803 payload: scope,
804 });
805 yield* projectionStore.apply({
806 id: EventId.make("event:checkpoint-capture-cancelled:baseline"),
807 type: "checkpoint.captured",
808 threadId: cancelledThreadId,
809 nodeId: cancelledRootNodeId,
810 driver,
811 providerInstanceId,
812 occurredAt: now,
813 payload: {
814 id: CheckpointId.make("checkpoint:cancelled-baseline-0"),
815 threadId: cancelledThreadId,
816 scopeId: cancelledScopeId,
817 runId: null,
818 nodeId: cancelledRootNodeId,
819 parentCheckpointId: null,
820 ordinalWithinScope: 0,
821 appRunOrdinal: null,
822 ref: CheckpointRef.make("checkpoint-ref:cancelled-baseline-0"),
823 status: "ready",
824 files: [],
825 capturedAt: now,
826 },
827 });
828 for (const [status, completedAt] of [
829 ["running", null],
830 ["cancelled", cancelledAt],
831 ] as const) {
832 yield* projectionStore.apply({
833 id: EventId.make(`event:checkpoint-capture-cancelled:run-${status}`),
834 type: "run.updated",
835 threadId: cancelledThreadId,
836 runId: cancelledRunId,
837 nodeId: cancelledRootNodeId,
838 providerInstanceId,
839 occurredAt: completedAt ?? now,
840 payload: { ...runningRun, status, completedAt },
841 });
842 yield* projectionStore.apply({
843 id: EventId.make(`event:checkpoint-capture-cancelled:node-${status}`),
844 type: "node.updated",
845 threadId: cancelledThreadId,
846 runId: cancelledRunId,
847 nodeId: cancelledRootNodeId,
848 providerInstanceId,
849 occurredAt: completedAt ?? now,
850 payload: { ...runningRootNode, status, completedAt },
851 });
852 }
853
854 const captured = {
855 id: CheckpointId.make("checkpoint:cancelled-captured-1"),
856 threadId: cancelledThreadId,
857 scopeId: cancelledScopeId,
858 runId: cancelledRunId,
859 nodeId: cancelledRootNodeId,
860 parentCheckpointId: CheckpointId.make("checkpoint:cancelled-baseline-0"),
861 ordinalWithinScope: 1,
862 appRunOrdinal: 1,
863 ref: CheckpointRef.make("checkpoint-ref:cancelled-captured-1"),
864 status: "ready" as const,
865 files: [],
866 capturedAt: cancelledAt,
867 };
868 const commits = yield* Ref.make(0);
869 const layerCapture = CheckpointCaptureService.layer.pipe(
870 Layer.provide(
871 Layer.mergeAll(
872 IdAllocator.layer,
873 Layer.mock(CheckpointService.CheckpointServiceV2)({
874 materializeBaselineCheckpoint: () =>
875 Effect.die("baseline materialization must be skipped when ordinal 0 is ready"),
876 capture: () => Effect.succeed(captured),
877 }),
878 Layer.mock(EventSink.EventSinkV2)({
879 commitCommand: (input) =>
880 Effect.forEach(input.events, (event) => projectionStore.apply(event)).pipe(
881 Effect.andThen(Ref.update(commits, (count) => count + 1)),
882 Effect.as({ committed: true } as never),
883 Effect.orDie,
884 ),
885 }),
886 ),
887 ),
888 );
889
890 yield* Effect.gen(function* () {
891 const service = yield* CheckpointCaptureService.CheckpointCaptureServiceV2;
892 yield* service.execute({
893 threadId: cancelledThreadId,
894 runId: cancelledRunId,
895 scopeId: cancelledScopeId,
896 });
897 yield* service.execute({
898 threadId: cancelledThreadId,
899 runId: cancelledRunId,
900 scopeId: cancelledScopeId,
901 });
902 }).pipe(Effect.provide(layerCapture));
903
904 assert.equal(yield* Ref.get(commits), 1);
905 const projected = yield* projectionStore.getCheckpointCaptureContext(cancelledThreadId, {
906 runId: cancelledRunId,
907 scopeId: cancelledScopeId,
908 });
909 assert.equal(projected.run?.status, "cancelled");
910 assert.equal(projected.run?.checkpointId, captured.id);
911 const completedAt = projected.run?.completedAt;
912 assert.equal(
913 completedAt ? DateTime.formatIso(completedAt) : completedAt,
914 DateTime.formatIso(cancelledAt),
915 );
916 assert.equal(projected.rootNode?.status, "cancelled");
917 assert.deepEqual([...projected.preservedCheckpointOrdinals].toSorted(), [0, 1]);
918
919 // A rewind leaves the run's checkpoint stale; the next capture must rebaseline it.
920 yield* projectionStore.apply({
921 id: EventId.make("event:checkpoint-capture-cancelled:captured-stale"),
922 type: "checkpoint.captured",
923 threadId: cancelledThreadId,
924 runId: cancelledRunId,
925 nodeId: cancelledRootNodeId,
926 providerInstanceId,
927 occurredAt: cancelledAt,
928 payload: { ...captured, status: "stale" },
929 });
930 const afterRewind = yield* projectionStore.getCheckpointCaptureContext(cancelledThreadId, {
931 runId: cancelledRunId,
932 scopeId: cancelledScopeId,
933 });
934 assert.deepEqual(afterRewind.preservedCheckpointOrdinals, [0]);
935 }),
936 );
937
938 it.effect("completes a side chat run with a missing checkpoint and never touches git", () =>
939 Effect.gen(function* () {
940 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
941 const now = yield* DateTime.now;
942 const sideChatThreadId = ThreadId.make("thread:checkpoint-capture-side-chat");
943 const sideChatRunId = RunId.make("run:checkpoint-capture-side-chat");
944 const sideChatScopeId = CheckpointScopeId.make("scope:checkpoint-capture-side-chat");
945 const sideChatRootNodeId = NodeId.make("node:checkpoint-capture-side-chat-root");
946 const sideChatProviderThreadId = ProviderThreadId.make(
947 "provider-thread:checkpoint-capture-side-chat",
948 );
949 const event = { threadId: sideChatThreadId, occurredAt: now };
950 const events: ReadonlyArray<OrchestrationV2DomainEvent> = [
951 {
952 ...event,
953 id: EventId.make("event:checkpoint-capture-side-chat:thread"),
954 type: "thread.created",
955 payload: {
956 createdBy: "user",
957 creationSource: "web",
958 id: sideChatThreadId,
959 projectId,
960 title: "Side chat",
961 providerInstanceId,
962 modelSelection,
963 runtimeMode: "approval-required",
964 interactionMode: "default",
965 branch: null,
966 worktreePath: "/repo",
967 activeProviderThreadId: null,
968 lineage: {
969 parentThreadId: threadId,
970 relationshipToParent: "fork",
971 rootThreadId: threadId,
972 sideChat: true,
973 },
974 forkedFrom: null,
975 createdAt: now,
976 updatedAt: now,
977 archivedAt: null,
978 settledOverride: null,
979 settledAt: null,
980 lastVisitedAt: null,
981 deletedAt: null,
982 },
983 },
984 {
985 ...event,
986 id: EventId.make("event:checkpoint-capture-side-chat:provider-thread"),
987 type: "provider-thread.updated",
988 driver,
989 providerInstanceId,
990 payload: {
991 id: sideChatProviderThreadId,
992 driver,
993 providerInstanceId,
994 providerSessionId: null,
995 appThreadId: sideChatThreadId,
996 ownerNodeId: null,
997 nativeThreadRef: null,
998 nativeConversationHeadRef: null,
999 status: "idle",
1000 firstRunOrdinal: 1,
1001 lastRunOrdinal: 1,
1002 handoffIds: [],
1003 forkedFrom: null,
1004 createdAt: now,
1005 updatedAt: now,
1006 },
1007 },
1008 {
1009 ...event,
1010 id: EventId.make("event:checkpoint-capture-side-chat:run"),
1011 type: "run.updated",
1012 runId: sideChatRunId,
1013 payload: {
1014 id: sideChatRunId,
1015 threadId: sideChatThreadId,
1016 ordinal: 1,
1017 providerInstanceId,
1018 modelSelection,
1019 providerThreadId: sideChatProviderThreadId,
1020 userMessageId: MessageId.make("message:checkpoint-capture-side-chat"),
1021 rootNodeId: sideChatRootNodeId,
1022 activeAttemptId: null,
1023 status: "waiting",
1024 requestedAt: now,
1025 startedAt: now,
1026 completedAt: null,
1027 checkpointId: null,
1028 contextHandoffId: null,
1029 },
1030 },
1031 {
1032 ...event,
1033 id: EventId.make("event:checkpoint-capture-side-chat:node"),
1034 type: "node.updated",
1035 runId: sideChatRunId,
1036 nodeId: sideChatRootNodeId,
1037 payload: {
1038 id: sideChatRootNodeId,
1039 threadId: sideChatThreadId,
1040 runId: sideChatRunId,
1041 parentNodeId: null,
1042 rootNodeId: sideChatRootNodeId,
1043 kind: "root_turn",
1044 status: "waiting",
1045 countsForRun: true,
1046 providerThreadId: sideChatProviderThreadId,
1047 providerTurnId: null,
1048 nativeItemRef: null,
1049 runtimeRequestId: null,
1050 checkpointScopeId: sideChatScopeId,
1051 startedAt: now,
1052 completedAt: null,
1053 },
1054 },
1055 {
1056 ...event,
1057 id: EventId.make("event:checkpoint-capture-side-chat:scope"),
1058 type: "checkpoint-scope.created",
1059 runId: sideChatRunId,
1060 nodeId: sideChatRootNodeId,
1061 payload: {
1062 id: sideChatScopeId,
1063 threadId: sideChatThreadId,
1064 runId: sideChatRunId,
1065 nodeId: sideChatRootNodeId,
1066 parentScopeId: null,
1067 providerThreadId: sideChatProviderThreadId,
1068 kind: "root_run",
1069 ordinalWithinParent: 0,
1070 advancesAppRunCount: true,
1071 cwd: "/repo",
1072 createdAt: now,
1073 },
1074 },
1075 ];
1076 yield* Effect.forEach(events, (candidate) => projectionStore.apply(candidate));
1077 const gitUnavailable = () => Effect.die("side chats must not snapshot the shared worktree");
1078 const layerCapture = CheckpointCaptureService.layer.pipe(
1079 Layer.provide(
1080 Layer.mergeAll(
1081 IdAllocator.layer,
1082 CheckpointService.layer.pipe(
1083 Layer.provide(
1084 Layer.mergeAll(
1085 IdAllocator.layer,
1086 NodeCrypto.layer,
1087 Layer.mock(CheckpointStore.CheckpointStore)({
1088 isGitRepository: gitUnavailable,
1089 captureCheckpoint: gitUnavailable,
1090 hasCheckpointRef: gitUnavailable,
1091 }),
1092 ),
1093 ),
1094 ),
1095 Layer.mock(EventSink.EventSinkV2)({
1096 commitCommand: (input) =>
1097 Effect.forEach(input.events, (committed) => projectionStore.apply(committed)).pipe(
1098 Effect.as({ committed: true } as never),
1099 Effect.orDie,
1100 ),
1101 }),
1102 ),
1103 ),
1104 );
1105
1106 yield* CheckpointCaptureService.CheckpointCaptureServiceV2.use((service) =>
1107 service.execute({
1108 threadId: sideChatThreadId,
1109 runId: sideChatRunId,
1110 scopeId: sideChatScopeId,
1111 }),
1112 ).pipe(Effect.provide(layerCapture));
1113
1114 const { checkpoints, runs } = yield* projectionStore.getThreadRecords(sideChatThreadId, [
1115 "runs",
1116 "checkpoints",
1117 ]);
1118 assert.equal(runs[0]?.status, "completed");
1119 const checkpoint = checkpoints.find((candidate) => candidate.id === runs[0]?.checkpointId);
1120 assert.equal(checkpoint?.status, "missing");
1121 }),
1122 );
1123});