1import { assert, it, vi } from "@effect/vitest";
2import {
3 MessageId,
4 NodeId,
5 ProviderDriverKind,
6 ProviderInstanceId,
7 ProviderSessionId,
8 ProviderThreadId,
9 ProviderTurnId,
10 RunAttemptId,
11 RunId,
12 RuntimeRequestId,
13 ThreadId,
14 TurnItemId,
15 type OrchestrationV2ThreadProjection,
16} from "@t3tools/contracts";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as Ref from "effect/Ref";
20
21import * as EffectWorker from "./EffectWorker.ts";
22import * as EffectOutbox from "./EffectOutbox.ts";
23import * as EventSink from "./EventSink.ts";
24import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
25import * as ProjectionStore from "./ProjectionStore.ts";
26import * as ProviderRuntimeRecovery from "./ProviderRuntimeRecoveryService.ts";
27import * as ServerSettings from "../serverSettings.ts";
28
29it.effect("leaves durable effects for the worker after runtime reconciliation", () =>
30 Effect.gen(function* () {
31 const runs = yield* Ref.make(0);
32 const layer = ProviderRuntimeRecovery.layer.pipe(
33 Layer.provide(ServerSettings.layerTest()),
34 Layer.provide(
35 Layer.mergeAll(
36 Layer.mock(ProjectionStore.ProjectionStoreV2)({
37 getRecoveryThreadIds: () => Effect.succeed([]),
38 }),
39 Layer.mock(EventSink.EventSinkV2)({}),
40 IdAllocator.layer,
41 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
42 runRecoveryOnce: Ref.getAndUpdate(runs, (count) => count + 1).pipe(
43 Effect.map((count) => count < 2),
44 ),
45 }),
46 Layer.mock(EffectOutbox.EffectOutboxV2)({
47 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
48 }),
49 ),
50 ),
51 );
52 const summary = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService.pipe(
53 Effect.flatMap((recovery) => recovery.recover),
54 Effect.provide(layer),
55 );
56 assert.deepEqual(summary, {
57 terminalizedRuns: 0,
58 stoppedSessions: 0,
59 closedRequests: 0,
60 retiredEffects: 0,
61 requeuedEffects: 0,
62 });
63 assert.equal(yield* Ref.get(runs), 0);
64 }),
65);
66
67it.effect("reads recovery projections only for threads that need runtime recovery", () => {
68 const settledThreadIds = Array.from({ length: 1_000 }, (_, index) =>
69 ThreadId.make(`thread_recovery_settled_${index}`),
70 );
71 const recoveryThreadId = ThreadId.make("thread_recovery_candidate");
72 const projectionReads = vi.fn<(threadId: ThreadId) => void>();
73 const layer = ProviderRuntimeRecovery.layer.pipe(
74 Layer.provide(ServerSettings.layerTest()),
75 Layer.provide(
76 Layer.mergeAll(
77 Layer.mock(ProjectionStore.ProjectionStoreV2)({
78 getShellSnapshot: () =>
79 Effect.succeed({
80 schemaVersion: 2,
81 snapshotSequence: 0,
82 threads: [...settledThreadIds, recoveryThreadId].map((id) => ({ id })),
83 archivedThreads: [],
84 } as never),
85 getRecoveryThreadIds: () => Effect.succeed([recoveryThreadId]),
86 getRuntimeRecoveryProjection: (threadId) => {
87 projectionReads(threadId);
88 return Effect.succeed({
89 thread: { id: threadId },
90 runtimeRequests: [],
91 providerSessions: [],
92 providerThreads: [],
93 providerTurns: [],
94 runs: [],
95 attempts: [],
96 nodes: [],
97 subagents: [],
98 messages: [],
99 turnItems: [],
100 } as unknown as OrchestrationV2ThreadProjection);
101 },
102 }),
103 Layer.mock(EventSink.EventSinkV2)({}),
104 IdAllocator.layer,
105 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
106 runRecoveryOnce: Effect.succeed(false),
107 }),
108 Layer.mock(EffectOutbox.EffectOutboxV2)({
109 cancelUnsettled: () => Effect.succeed([]),
110 signalCancellations: () => Effect.void,
111 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
112 }),
113 ),
114 ),
115 );
116
117 return Effect.gen(function* () {
118 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).recover;
119 assert.deepEqual(
120 projectionReads.mock.calls.map(([threadId]) => threadId),
121 [recoveryThreadId],
122 );
123 }).pipe(Effect.provide(layer));
124});
125
126it.effect("expires orphaned runtime requests before command readiness", () => {
127 const threadId = ThreadId.make("thread_recovery_requests");
128 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
129 null;
130 const committed = vi.fn(
131 (input: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0]) => {
132 committedInput = input;
133 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
134 },
135 );
136 const projection = {
137 thread: { id: threadId },
138 runtimeRequests: [
139 {
140 id: RuntimeRequestId.make("request_orphaned"),
141 nodeId: NodeId.make("node_orphaned"),
142 status: "pending",
143 responseCapability: { type: "not_resumable", reason: "old process" },
144 },
145 ],
146 providerSessions: [],
147 providerThreads: [],
148 runs: [],
149 nodes: [],
150 } as unknown as OrchestrationV2ThreadProjection;
151 const layer = ProviderRuntimeRecovery.layer.pipe(
152 Layer.provide(ServerSettings.layerTest()),
153 Layer.provide(
154 Layer.mergeAll(
155 Layer.mock(ProjectionStore.ProjectionStoreV2)({
156 getRecoveryThreadIds: () => Effect.succeed([threadId]),
157 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
158 }),
159 Layer.mock(EventSink.EventSinkV2)({ commitCommand: committed }),
160 IdAllocator.layer,
161 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
162 runRecoveryOnce: Effect.succeed(false),
163 }),
164 Layer.mock(EffectOutbox.EffectOutboxV2)({
165 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
166 }),
167 ),
168 ),
169 );
170 return Effect.gen(function* () {
171 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).recover;
172 const command = committedInput;
173 assert.isNotNull(command);
174 if (command === null) return;
175 assert.equal(command?.events[0]?.type, "runtime-request.updated");
176 if (command?.events[0]?.type === "runtime-request.updated") {
177 assert.equal(command.events[0].payload.status, "expired");
178 assert.equal(command.events[0].payload.responseCapability.type, "not_resumable");
179 }
180 }).pipe(Effect.provide(layer));
181});
182
183it.effect("preserves async questions across startup and shutdown", () => {
184 const threadId = ThreadId.make("async-recovery-thread");
185 const nodeId = NodeId.make("async-recovery-node");
186 const projection = {
187 thread: { id: threadId },
188 runtimeRequests: [
189 {
190 id: RuntimeRequestId.make("async-recovery-request"),
191 nodeId,
192 status: "pending",
193 responseCapability: { type: "message" },
194 },
195 ],
196 providerSessions: [],
197 providerThreads: [],
198 runs: [],
199 nodes: [],
200 turnItems: [],
201 } as unknown as OrchestrationV2ThreadProjection;
202 const commitCommand = vi.fn(() => Effect.die("an async question needs no process-loss write"));
203 const layer = ProviderRuntimeRecovery.layer.pipe(
204 Layer.provide(ServerSettings.layerTest()),
205 Layer.provide(
206 Layer.mergeAll(
207 Layer.mock(ProjectionStore.ProjectionStoreV2)({
208 getRecoveryThreadIds: () => Effect.succeed([threadId]),
209 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
210 }),
211 Layer.mock(EventSink.EventSinkV2)({ commitCommand }),
212 IdAllocator.layer,
213 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
214 runRecoveryOnce: Effect.succeed(false),
215 }),
216 Layer.mock(EffectOutbox.EffectOutboxV2)({
217 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
218 cancelUnsettled: () => Effect.succeed([]),
219 signalCancellations: () => Effect.void,
220 }),
221 ),
222 ),
223 );
224 return Effect.gen(function* () {
225 const recovery = yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService;
226 assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0);
227 assert.equal((yield* recovery.reconcile("startup")).closedRequests, 0);
228 assert.isFalse(commitCommand.mock.calls.length > 0);
229 }).pipe(Effect.provide(layer));
230});
231
232it.effect("uses the same reconciliation path to cancel runtime requests during shutdown", () => {
233 const threadId = ThreadId.make("thread_shutdown_requests");
234 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
235 null;
236 const projection = {
237 thread: { id: threadId },
238 runtimeRequests: [
239 {
240 id: RuntimeRequestId.make("request_shutdown"),
241 nodeId: NodeId.make("node_shutdown"),
242 status: "pending",
243 responseCapability: { type: "live" },
244 },
245 ],
246 providerSessions: [],
247 providerThreads: [],
248 runs: [],
249 nodes: [],
250 } as unknown as OrchestrationV2ThreadProjection;
251 const layer = ProviderRuntimeRecovery.layer.pipe(
252 Layer.provide(ServerSettings.layerTest()),
253 Layer.provide(
254 Layer.mergeAll(
255 Layer.mock(ProjectionStore.ProjectionStoreV2)({
256 getRecoveryThreadIds: () => Effect.succeed([threadId]),
257 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
258 }),
259 Layer.mock(EventSink.EventSinkV2)({
260 commitCommand: (input) => {
261 committedInput = input;
262 return Effect.succeed({ committed: true, cancelledEffectCount: 1 } as never);
263 },
264 }),
265 IdAllocator.layer,
266 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
267 runRecoveryOnce: Effect.succeed(false),
268 }),
269 Layer.mock(EffectOutbox.EffectOutboxV2)({
270 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
271 }),
272 ),
273 ),
274 );
275
276 return Effect.gen(function* () {
277 const summary =
278 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("shutdown");
279 assert.equal(summary.closedRequests, 1);
280 assert.equal(summary.retiredEffects, 1);
281 const requestEvent = committedInput?.events[0];
282 assert.equal(requestEvent?.type, "runtime-request.updated");
283 if (requestEvent?.type === "runtime-request.updated") {
284 assert.equal(requestEvent.payload.status, "cancelled");
285 assert.equal(requestEvent.payload.responseCapability.type, "not_resumable");
286 if (requestEvent.payload.responseCapability.type === "not_resumable") {
287 assert.match(requestEvent.payload.responseCapability.reason, /shut down/);
288 }
289 }
290 }).pipe(Effect.provide(layer));
291});
292
293it.effect(
294 "preserves a replayable waiting run while cancelling its process-bound background work",
295 () => {
296 const threadId = ThreadId.make("thread_waiting_checkpoint");
297 const runId = RunId.make("run_waiting_checkpoint");
298 const providerThreadId = ProviderThreadId.make("provider_thread_waiting_checkpoint");
299 const providerInstanceId = ProviderInstanceId.make("claude");
300 const backgroundItemId = TurnItemId.make("turn_item_waiting_checkpoint");
301 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
302 null;
303 const projection = {
304 thread: { id: threadId },
305 runtimeRequests: [],
306 providerSessions: [],
307 providerThreads: [
308 {
309 id: providerThreadId,
310 driver: ProviderDriverKind.make("claude"),
311 providerInstanceId,
312 status: "idle",
313 pendingBackgroundTasks: [
314 { taskId: String(backgroundItemId), description: "Finish background task" },
315 ],
316 },
317 ],
318 runs: [{ id: runId, status: "waiting", providerInstanceId, providerThreadId }],
319 attempts: [],
320 nodes: [],
321 subagents: [],
322 messages: [],
323 turnItems: [
324 {
325 id: backgroundItemId,
326 runId,
327 nodeId: null,
328 providerThreadId,
329 type: "command_execution",
330 status: "running",
331 },
332 ],
333 } as unknown as OrchestrationV2ThreadProjection;
334 const layer = ProviderRuntimeRecovery.layer.pipe(
335 Layer.provide(ServerSettings.layerTest()),
336 Layer.provide(
337 Layer.mergeAll(
338 Layer.mock(ProjectionStore.ProjectionStoreV2)({
339 getRecoveryThreadIds: () => Effect.succeed([threadId]),
340 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
341 }),
342 Layer.mock(EventSink.EventSinkV2)({
343 commitCommand: (input) => {
344 committedInput = input;
345 return Effect.succeed({
346 committed: true,
347 cancelledEffectCount: 0,
348 } as never);
349 },
350 }),
351 IdAllocator.layer,
352 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
353 runRecoveryOnce: Effect.succeed(false),
354 }),
355 Layer.mock(EffectOutbox.EffectOutboxV2)({
356 listByCommandId: () =>
357 Effect.succeed([
358 {
359 request: { type: "checkpoint.capture", runId },
360 status: "running",
361 },
362 ] as never),
363 reconcileAfterProcessLoss: Effect.succeed({ requeued: 1, cancelled: 0 }),
364 }),
365 ),
366 ),
367 );
368
369 return Effect.gen(function* () {
370 const summary =
371 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
372 assert.equal(summary.terminalizedRuns, 0);
373 assert.equal(summary.requeuedEffects, 1);
374
375 const command = committedInput;
376 assert.isNotNull(command);
377 if (command === null) return;
378 // The run stays waiting for its checkpoint; it only records the work
379 // the restart cancelled so the next provider turn can be told.
380 assert.isFalse(command.events.some((event) => event.type === "run.updated"));
381 const recorded = command.events.flatMap((event) =>
382 event.type === "run.background-work-cancelled" ? [event.payload] : [],
383 );
384 // The roster entry is the same task as the item, so it is listed once.
385 assert.deepEqual(
386 recorded.map((entry) => ({
387 runId: entry.runId,
388 kinds: entry.restartCancelledBackgroundWork.map((work) => work.kind),
389 })),
390 [{ runId, kinds: ["shell"] }],
391 );
392 assert.isTrue(
393 command.events.some(
394 (event) =>
395 event.type === "turn-item.updated" &&
396 event.payload.id === backgroundItemId &&
397 event.payload.status === "cancelled",
398 ),
399 );
400 assert.isTrue(
401 command.events.some(
402 (event) =>
403 event.type === "provider-thread.updated" &&
404 event.payload.id === providerThreadId &&
405 event.payload.pendingBackgroundTasks?.length === 0,
406 ),
407 );
408 }).pipe(Effect.provide(layer));
409 },
410);
411
412it.effect(
413 "records cancelled work on the provider thread that lost it after a provider switch",
414 () => {
415 const threadId = ThreadId.make("thread_switch_note");
416 const claudeRunId = RunId.make("run_switch_note_claude");
417 const codexRunId = RunId.make("run_switch_note_codex");
418 const claudeThreadId = ProviderThreadId.make("provider_thread_switch_note_claude");
419 const codexThreadId = ProviderThreadId.make("provider_thread_switch_note_codex");
420 const claude = ProviderInstanceId.make("claude");
421 const codex = ProviderInstanceId.make("codex");
422 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
423 null;
424 // Claude launched a background subagent, then the thread switched to Codex.
425 const projection = {
426 thread: { id: threadId },
427 runtimeRequests: [],
428 providerSessions: [],
429 providerThreads: [
430 {
431 id: claudeThreadId,
432 ownerNodeId: null,
433 driver: ProviderDriverKind.make("claudeAgent"),
434 providerInstanceId: claude,
435 status: "idle",
436 pendingBackgroundTasks: [{ taskId: "task-claude", description: "Watch the build" }],
437 },
438 {
439 id: codexThreadId,
440 ownerNodeId: null,
441 driver: ProviderDriverKind.make("codex"),
442 providerInstanceId: codex,
443 status: "idle",
444 },
445 ],
446 providerTurns: [],
447 runs: [
448 {
449 id: claudeRunId,
450 ordinal: 1,
451 status: "completed",
452 providerInstanceId: claude,
453 providerThreadId: claudeThreadId,
454 },
455 {
456 id: codexRunId,
457 ordinal: 2,
458 status: "completed",
459 providerInstanceId: codex,
460 providerThreadId: codexThreadId,
461 },
462 ],
463 attempts: [],
464 nodes: [],
465 subagents: [],
466 messages: [],
467 turnItems: [
468 {
469 id: TurnItemId.make("turn_item_switch_note_subagent"),
470 runId: claudeRunId,
471 nodeId: null,
472 providerThreadId: claudeThreadId,
473 nativeItemRef: null,
474 type: "subagent",
475 title: "Background subagent",
476 status: "running",
477 },
478 ],
479 } as unknown as OrchestrationV2ThreadProjection;
480 const layer = ProviderRuntimeRecovery.layer.pipe(
481 Layer.provide(ServerSettings.layerTest()),
482 Layer.provide(
483 Layer.mergeAll(
484 Layer.mock(ProjectionStore.ProjectionStoreV2)({
485 getRecoveryThreadIds: () => Effect.succeed([threadId]),
486 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
487 }),
488 Layer.mock(EventSink.EventSinkV2)({
489 commitCommand: (input) => {
490 committedInput = input;
491 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
492 },
493 }),
494 IdAllocator.layer,
495 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
496 runRecoveryOnce: Effect.succeed(false),
497 }),
498 Layer.mock(EffectOutbox.EffectOutboxV2)({
499 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
500 }),
501 ),
502 ),
503 );
504
505 return Effect.gen(function* () {
506 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
507 const recorded = (committedInput?.events ?? []).flatMap((event) =>
508 event.type === "run.background-work-cancelled" ? [event.payload] : [],
509 );
510 // The Codex run is later, but only Claude's turns may be told about it.
511 assert.deepEqual(
512 recorded.map((entry) => ({
513 runId: entry.runId,
514 kinds: entry.restartCancelledBackgroundWork.map((work) => work.kind),
515 })),
516 [{ runId: claudeRunId, kinds: ["subagent", "task"] }],
517 );
518 }).pipe(Effect.provide(layer));
519 },
520);
521
522it.effect("cancels a stale waiting run when no checkpoint capture can finish it", () => {
523 const threadId = ThreadId.make("thread_stale_waiting");
524 const runId = RunId.make("run_stale_waiting");
525 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
526 null;
527 const projection = {
528 thread: { id: threadId },
529 runtimeRequests: [],
530 providerSessions: [],
531 providerThreads: [],
532 providerTurns: [],
533 runs: [
534 {
535 id: runId,
536 status: "waiting",
537 providerInstanceId: ProviderInstanceId.make("codex"),
538 },
539 ],
540 attempts: [],
541 nodes: [],
542 subagents: [],
543 messages: [],
544 turnItems: [],
545 } as unknown as OrchestrationV2ThreadProjection;
546 const layer = ProviderRuntimeRecovery.layer.pipe(
547 Layer.provide(ServerSettings.layerTest()),
548 Layer.provide(
549 Layer.mergeAll(
550 Layer.mock(ProjectionStore.ProjectionStoreV2)({
551 getRecoveryThreadIds: () => Effect.succeed([threadId]),
552 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
553 }),
554 Layer.mock(EventSink.EventSinkV2)({
555 commitCommand: (input) => {
556 committedInput = input;
557 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
558 },
559 }),
560 IdAllocator.layer,
561 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
562 runRecoveryOnce: Effect.succeed(false),
563 }),
564 Layer.mock(EffectOutbox.EffectOutboxV2)({
565 listByCommandId: () => Effect.succeed([]),
566 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
567 }),
568 ),
569 ),
570 );
571
572 return Effect.gen(function* () {
573 const summary =
574 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
575 assert.equal(summary.terminalizedRuns, 1);
576 const runEvent = committedInput?.events.find((event) => event.type === "run.updated");
577 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.status : null, "cancelled");
578 }).pipe(Effect.provide(layer));
579});
580
581it.effect("closes a secret request card's form when its run is recovered", () => {
582 const threadId = ThreadId.make("thread_secret_recovery");
583 const runId = RunId.make("run_secret_recovery");
584 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
585 null;
586 const projection = {
587 thread: { id: threadId },
588 runtimeRequests: [],
589 providerSessions: [],
590 providerThreads: [],
591 providerTurns: [],
592 runs: [{ id: runId, status: "running", providerInstanceId: ProviderInstanceId.make("codex") }],
593 attempts: [],
594 nodes: [],
595 subagents: [],
596 messages: [],
597 turnItems: [
598 {
599 id: "turn-item:secret-request:recovery",
600 threadId,
601 runId,
602 nodeId: null,
603 type: "secret_request",
604 status: "waiting",
605 secretStatus: "pending",
606 label: "GitHub token",
607 reason: "Used as GH_TOKEN.",
608 },
609 ],
610 } as unknown as OrchestrationV2ThreadProjection;
611 const layer = ProviderRuntimeRecovery.layer.pipe(
612 Layer.provide(ServerSettings.layerTest()),
613 Layer.provide(
614 Layer.mergeAll(
615 Layer.mock(ProjectionStore.ProjectionStoreV2)({
616 getRecoveryThreadIds: () => Effect.succeed([threadId]),
617 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
618 }),
619 Layer.mock(EventSink.EventSinkV2)({
620 commitCommand: (input) => {
621 committedInput = input;
622 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
623 },
624 }),
625 IdAllocator.layer,
626 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
627 runRecoveryOnce: Effect.succeed(false),
628 }),
629 Layer.mock(EffectOutbox.EffectOutboxV2)({
630 listByCommandId: () => Effect.succeed([]),
631 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
632 }),
633 ),
634 ),
635 );
636
637 return Effect.gen(function* () {
638 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
639 const itemEvent = committedInput?.events.find((event) => event.type === "turn-item.updated");
640 const card = itemEvent?.type === "turn-item.updated" ? itemEvent.payload : null;
641 assert.equal(card?.status, "cancelled");
642 assert.equal(card?.type === "secret_request" ? card.secretStatus : null, "cancelled");
643 }).pipe(Effect.provide(layer));
644});
645
646it.effect.each(["queued", "preparing"] as const)("recovers $0 work after restart", (status) => {
647 const threadId = ThreadId.make("thread_queued_restart");
648 const runId = RunId.make("run_queued_restart");
649 const attemptId = RunAttemptId.make("attempt_queued_restart");
650 const rootNodeId = NodeId.make("node_queued_restart");
651 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
652 null;
653 const projection = {
654 thread: { id: threadId },
655 runtimeRequests: [],
656 providerSessions: [],
657 providerThreads: [],
658 providerTurns: [],
659 runs: [
660 {
661 id: runId,
662 status,
663 queuePosition: 1,
664 providerInstanceId: ProviderInstanceId.make("codex"),
665 ...(status === "preparing"
666 ? {
667 workspacePreparation: {
668 type: "worktree",
669 baseRef: "parent-local-commit",
670 startFromOrigin: false,
671 },
672 }
673 : {}),
674 },
675 ],
676 attempts: [
677 {
678 id: attemptId,
679 runId,
680 rootNodeId,
681 status: "pending",
682 },
683 ],
684 nodes: [
685 {
686 id: rootNodeId,
687 runId,
688 status: "pending",
689 },
690 ],
691 subagents: [],
692 messages: [],
693 turnItems: [],
694 } as unknown as OrchestrationV2ThreadProjection;
695 const layer = ProviderRuntimeRecovery.layer.pipe(
696 Layer.provide(ServerSettings.layerTest()),
697 Layer.provide(
698 Layer.mergeAll(
699 Layer.mock(ProjectionStore.ProjectionStoreV2)({
700 getRecoveryThreadIds: () => Effect.succeed([threadId]),
701 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
702 getThreadRecords: () => Effect.succeed(projection),
703 }),
704 Layer.mock(EventSink.EventSinkV2)({
705 commitCommand: (input) => {
706 committedInput = input;
707 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
708 },
709 }),
710 IdAllocator.layer,
711 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
712 runRecoveryOnce: Effect.succeed(false),
713 }),
714 Layer.mock(EffectOutbox.EffectOutboxV2)({
715 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
716 }),
717 ),
718 ),
719 );
720
721 return Effect.gen(function* () {
722 const summary =
723 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
724 assert.equal(summary.terminalizedRuns, status === "preparing" ? 1 : 0);
725 const command = committedInput;
726 assert.isNotNull(command);
727 if (command === null) return;
728 const runEvent = command.events.find((event) => event.type === "run.updated");
729 const attemptEvent = command.events.find((event) => event.type === "run-attempt.updated");
730 const nodeEvent = command.events.find((event) => event.type === "node.updated");
731 assert.equal(
732 runEvent?.type === "run.updated" ? runEvent.payload.status : null,
733 status === "preparing" ? "cancelled" : "queued",
734 );
735 assert.isEmpty(command.effects);
736 if (status === "preparing") {
737 assert.equal(
738 attemptEvent?.type === "run-attempt.updated" ? attemptEvent.payload.status : null,
739 "cancelled",
740 );
741 assert.equal(
742 nodeEvent?.type === "node.updated" ? nodeEvent.payload.status : null,
743 "cancelled",
744 );
745 } else {
746 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queuePosition : null, 1);
747 assert.equal(runEvent?.type === "run.updated" ? runEvent.payload.queueHeld : false, true);
748 assert.isUndefined(attemptEvent);
749 assert.isUndefined(nodeEvent);
750 }
751 }).pipe(Effect.provide(layer));
752});
753
754it.effect(
755 "cancels the complete in-flight subtree and stops its persisted session without reopening it",
756 () => {
757 const threadId = ThreadId.make("thread_recovery_cancel");
758 const runId = RunId.make("run_recovery_cancel");
759 const attemptId = RunAttemptId.make("attempt_recovery_cancel");
760 const rootNodeId = NodeId.make("node_recovery_cancel");
761 const providerThreadId = ProviderThreadId.make("provider_thread_recovery_cancel");
762 const providerTurnId = ProviderTurnId.make("provider_turn_recovery_cancel");
763 const providerSessionId = ProviderSessionId.make("provider_session_recovery_cancel");
764 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
765 null;
766 const projection = {
767 thread: { id: threadId },
768 runtimeRequests: [],
769 providerSessions: [
770 {
771 id: providerSessionId,
772 driver: ProviderDriverKind.make("codex"),
773 providerInstanceId: ProviderInstanceId.make("codex"),
774 status: "ready",
775 },
776 ],
777 providerThreads: [
778 {
779 id: providerThreadId,
780 driver: ProviderDriverKind.make("codex"),
781 providerInstanceId: ProviderInstanceId.make("codex"),
782 status: "active",
783 },
784 ],
785 providerTurns: [
786 {
787 id: providerTurnId,
788 runAttemptId: attemptId,
789 nodeId: rootNodeId,
790 status: "running",
791 },
792 ],
793 runs: [
794 {
795 id: runId,
796 status: "starting",
797 providerThreadId,
798 providerInstanceId: ProviderInstanceId.make("codex"),
799 },
800 ],
801 attempts: [
802 {
803 id: attemptId,
804 runId,
805 rootNodeId,
806 status: "running",
807 },
808 ],
809 nodes: [{ id: rootNodeId, runId, status: "running" }],
810 subagents: [],
811 messages: [{ id: MessageId.make("message_recovery_cancel"), runId, streaming: true }],
812 turnItems: [
813 {
814 id: TurnItemId.make("turn_item_recovery_cancel"),
815 runId,
816 nodeId: rootNodeId,
817 status: "running",
818 },
819 ],
820 } as unknown as OrchestrationV2ThreadProjection;
821 const layer = ProviderRuntimeRecovery.layer.pipe(
822 Layer.provide(ServerSettings.layerTest()),
823 Layer.provide(
824 Layer.mergeAll(
825 Layer.mock(ProjectionStore.ProjectionStoreV2)({
826 getRecoveryThreadIds: () => Effect.succeed([threadId]),
827 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
828 }),
829 Layer.mock(EventSink.EventSinkV2)({
830 commitCommand: (input) => {
831 committedInput = input;
832 return Effect.succeed({ committed: true, cancelledEffectCount: 2 } as never);
833 },
834 }),
835 IdAllocator.layer,
836 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
837 runRecoveryOnce: Effect.succeed(false),
838 }),
839 Layer.mock(EffectOutbox.EffectOutboxV2)({
840 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
841 }),
842 ),
843 ),
844 );
845
846 return Effect.gen(function* () {
847 const summary = yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService)
848 .recover;
849 assert.equal(summary.terminalizedRuns, 1);
850 assert.equal(summary.stoppedSessions, 1);
851 assert.equal(summary.retiredEffects, 2);
852 assert.deepEqual(committedInput?.cancelUnsettledEffects?.effectTypes, [
853 "provider-turn.start",
854 "provider-turn.interrupt",
855 "provider-turn.steer",
856 "provider-turn.restart",
857 "runtime-request.respond",
858 ]);
859 const events = committedInput?.events ?? [];
860 assert.deepEqual(
861 events.map((event) => [
862 event.type,
863 "status" in event.payload ? event.payload.status : null,
864 ]),
865 [
866 ["run.updated", "cancelled"],
867 ["run-attempt.updated", "cancelled"],
868 ["node.updated", "cancelled"],
869 ["provider-turn.updated", "cancelled"],
870 ["message.updated", null],
871 ["turn-item.updated", "cancelled"],
872 ["provider-thread.updated", "idle"],
873 ["provider-session.updated", "stopped"],
874 ],
875 );
876 const messageEvent = events.find((event) => event.type === "message.updated");
877 assert.isFalse(messageEvent?.type === "message.updated" && messageEvent.payload.streaming);
878 }).pipe(Effect.provide(layer));
879 },
880);
881
882it.effect("cancels workflow members and child roots without run ids after restart", () => {
883 const parentThreadId = ThreadId.make("thread_recovery_workflow_members");
884 const childThreadId = ThreadId.make("thread_recovery_workflow_member_child");
885 const memberId = NodeId.make("node_recovery_workflow_member");
886 const childRootId = NodeId.make("node_recovery_workflow_member_root");
887 const providerInstanceId = ProviderInstanceId.make("claudeAgent");
888 const driver = ProviderDriverKind.make("claudeAgent");
889 const member = {
890 id: memberId,
891 threadId: parentThreadId,
892 runId: null,
893 origin: "provider_native",
894 status: "running",
895 driver,
896 providerInstanceId,
897 };
898 const parentProjection = {
899 thread: { id: parentThreadId, providerInstanceId },
900 runs: [],
901 attempts: [],
902 nodes: [
903 { id: memberId, threadId: parentThreadId, runId: null, kind: "subagent", status: "running" },
904 ],
905 subagents: [member],
906 runtimeRequests: [],
907 providerSessions: [],
908 providerThreads: [],
909 providerTurns: [],
910 messages: [],
911 turnItems: [],
912 } as unknown as OrchestrationV2ThreadProjection;
913 const childProjection = {
914 ...parentProjection,
915 thread: { id: childThreadId, providerInstanceId },
916 nodes: [
917 {
918 id: childRootId,
919 threadId: childThreadId,
920 runId: null,
921 kind: "root_turn",
922 status: "running",
923 },
924 ],
925 subagents: [],
926 } as unknown as OrchestrationV2ThreadProjection;
927 const committed: Array<Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0]> = [];
928 const layer = ProviderRuntimeRecovery.layer.pipe(
929 Layer.provide(ServerSettings.layerTest()),
930 Layer.provide(
931 Layer.mergeAll(
932 Layer.mock(ProjectionStore.ProjectionStoreV2)({
933 getRecoveryThreadIds: () => Effect.succeed([parentThreadId, childThreadId]),
934 getRuntimeRecoveryProjection: (threadId) =>
935 Effect.succeed(threadId === parentThreadId ? parentProjection : childProjection),
936 }),
937 Layer.mock(EventSink.EventSinkV2)({
938 commitCommand: (input) => {
939 committed.push(input);
940 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
941 },
942 }),
943 IdAllocator.layer,
944 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
945 runRecoveryOnce: Effect.succeed(false),
946 }),
947 Layer.mock(EffectOutbox.EffectOutboxV2)({
948 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
949 }),
950 ),
951 ),
952 );
953
954 return Effect.gen(function* () {
955 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
956 assert.deepEqual(
957 committed.flatMap((command) =>
958 command.events.map((event) => [
959 command.threadId,
960 event.type,
961 "status" in event.payload ? event.payload.status : null,
962 ]),
963 ),
964 [
965 [parentThreadId, "subagent.updated", "cancelled"],
966 [parentThreadId, "node.updated", "cancelled"],
967 [childThreadId, "node.updated", "cancelled"],
968 ],
969 );
970 }).pipe(Effect.provide(layer));
971});
972
973it.effect(
974 "clears persisted pendingBackgroundTasks and terminalizes stale background items on settled runs",
975 () => {
976 const threadId = ThreadId.make("thread_recovery_background");
977 const settledRunId = RunId.make("run_recovery_background_settled");
978 const activeRunId = RunId.make("run_recovery_background_active");
979 const activeAttemptId = RunAttemptId.make("attempt_recovery_background_active");
980 const activeRootNodeId = NodeId.make("node_recovery_background_active");
981 const idleProviderThreadId = ProviderThreadId.make("provider_thread_recovery_background_idle");
982 const activeProviderThreadId = ProviderThreadId.make(
983 "provider_thread_recovery_background_active",
984 );
985 const secondaryProviderThreadId = ProviderThreadId.make(
986 "provider_thread_recovery_background_secondary",
987 );
988 const providerSessionId = ProviderSessionId.make("provider_session_recovery_background");
989 const settledStaleItemId = TurnItemId.make("turn_item_recovery_background_stale");
990 const activeRunItemId = TurnItemId.make("turn_item_recovery_background_active");
991 const nullRunCommandItemId = TurnItemId.make("turn_item_recovery_background_null_run");
992 const nullRunSubagentItemId = TurnItemId.make("turn_item_recovery_background_null_subagent");
993 const claudeInstanceId = ProviderInstanceId.make("claude");
994 const secondaryInstanceId = ProviderInstanceId.make("claude-secondary");
995 const subagentInstanceId = ProviderInstanceId.make("claude-subagent");
996 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
997 null;
998 const projection = {
999 thread: { id: threadId },
1000 runtimeRequests: [],
1001 providerSessions: [
1002 {
1003 id: providerSessionId,
1004 driver: ProviderDriverKind.make("claude"),
1005 providerInstanceId: claudeInstanceId,
1006 status: "ready",
1007 },
1008 ],
1009 providerThreads: [
1010 {
1011 id: idleProviderThreadId,
1012 driver: ProviderDriverKind.make("claude"),
1013 // Index-0 is intentionally a different instance so misattribution
1014 // to providerThreads[0] fails the assertions below.
1015 providerInstanceId: claudeInstanceId,
1016 status: "idle",
1017 pendingBackgroundTasks: [{ taskId: "bg-settled", description: "sleep 30" }],
1018 },
1019 {
1020 id: activeProviderThreadId,
1021 driver: ProviderDriverKind.make("claude"),
1022 providerInstanceId: claudeInstanceId,
1023 status: "active",
1024 pendingBackgroundTasks: [{ taskId: "bg-active", description: "npm test" }],
1025 },
1026 {
1027 id: secondaryProviderThreadId,
1028 driver: ProviderDriverKind.make("claude"),
1029 providerInstanceId: secondaryInstanceId,
1030 status: "idle",
1031 pendingBackgroundTasks: [],
1032 },
1033 ],
1034 providerTurns: [],
1035 runs: [
1036 {
1037 id: settledRunId,
1038 status: "completed",
1039 providerInstanceId: claudeInstanceId,
1040 },
1041 {
1042 id: activeRunId,
1043 status: "running",
1044 providerInstanceId: claudeInstanceId,
1045 },
1046 ],
1047 attempts: [
1048 {
1049 id: activeAttemptId,
1050 runId: activeRunId,
1051 rootNodeId: activeRootNodeId,
1052 status: "running",
1053 },
1054 ],
1055 nodes: [{ id: activeRootNodeId, runId: activeRunId, status: "running" }],
1056 subagents: [],
1057 messages: [],
1058 turnItems: [
1059 {
1060 id: settledStaleItemId,
1061 runId: settledRunId,
1062 nodeId: null,
1063 providerThreadId: idleProviderThreadId,
1064 type: "command_execution",
1065 status: "running",
1066 },
1067 {
1068 id: activeRunItemId,
1069 runId: activeRunId,
1070 nodeId: activeRootNodeId,
1071 providerThreadId: activeProviderThreadId,
1072 type: "dynamic_tool",
1073 status: "running",
1074 },
1075 {
1076 // Missing run: must attribute via providerThreadId, not index 0.
1077 id: nullRunCommandItemId,
1078 runId: null,
1079 nodeId: null,
1080 providerThreadId: secondaryProviderThreadId,
1081 type: "command_execution",
1082 status: "running",
1083 },
1084 {
1085 // Missing run with a real matching provider thread whose instance
1086 // differs from the subagent's own: own providerInstanceId must win.
1087 id: nullRunSubagentItemId,
1088 runId: null,
1089 nodeId: null,
1090 providerThreadId: secondaryProviderThreadId,
1091 type: "subagent",
1092 status: "running",
1093 providerInstanceId: subagentInstanceId,
1094 },
1095 ],
1096 } as unknown as OrchestrationV2ThreadProjection;
1097 const layer = ProviderRuntimeRecovery.layer.pipe(
1098 Layer.provide(ServerSettings.layerTest()),
1099 Layer.provide(
1100 Layer.mergeAll(
1101 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1102 getRecoveryThreadIds: () => Effect.succeed([threadId]),
1103 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
1104 }),
1105 Layer.mock(EventSink.EventSinkV2)({
1106 commitCommand: (input) => {
1107 committedInput = input;
1108 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
1109 },
1110 }),
1111 IdAllocator.layer,
1112 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
1113 runRecoveryOnce: Effect.succeed(false),
1114 }),
1115 Layer.mock(EffectOutbox.EffectOutboxV2)({
1116 listByCommandId: () => Effect.succeed([]),
1117 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
1118 }),
1119 ),
1120 ),
1121 );
1122
1123 return Effect.gen(function* () {
1124 const summary =
1125 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
1126 assert.equal(summary.terminalizedRuns, 1);
1127 const events = committedInput?.events ?? [];
1128
1129 const turnItemCancels = events.filter(
1130 (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled",
1131 );
1132 // Active-run item + settled-run stale + null-run command + null-run subagent.
1133 assert.equal(turnItemCancels.length, 4);
1134 assert.deepEqual(
1135 turnItemCancels
1136 .map((event) => event.type === "turn-item.updated" && event.payload.id)
1137 .sort(),
1138 [activeRunItemId, nullRunCommandItemId, nullRunSubagentItemId, settledStaleItemId].sort(),
1139 );
1140
1141 const cancelById = (id: TurnItemId) =>
1142 turnItemCancels.find(
1143 (event) => event.type === "turn-item.updated" && event.payload.id === id,
1144 );
1145 assert.equal(cancelById(nullRunCommandItemId)?.providerInstanceId, secondaryInstanceId);
1146 // Subagent own instance wins over the matching thread's secondary instance.
1147 assert.notEqual(subagentInstanceId, secondaryInstanceId);
1148 assert.equal(cancelById(nullRunSubagentItemId)?.providerInstanceId, subagentInstanceId);
1149 // Settled-run item still prefers the run's provider instance when present.
1150 assert.equal(cancelById(settledStaleItemId)?.providerInstanceId, claudeInstanceId);
1151
1152 const providerThreadEvents = events.filter(
1153 (event) => event.type === "provider-thread.updated",
1154 );
1155 // Only threads with active status or nonempty rosters are rewritten.
1156 assert.equal(providerThreadEvents.length, 2);
1157 for (const event of providerThreadEvents) {
1158 if (event.type !== "provider-thread.updated") continue;
1159 assert.deepEqual(event.payload.pendingBackgroundTasks ?? [], []);
1160 }
1161 const idleThreadEvent = providerThreadEvents.find(
1162 (event) =>
1163 event.type === "provider-thread.updated" && event.payload.id === idleProviderThreadId,
1164 );
1165 assert.equal(
1166 idleThreadEvent?.type === "provider-thread.updated" ? idleThreadEvent.payload.status : null,
1167 "idle",
1168 );
1169 const activeThreadEvent = providerThreadEvents.find(
1170 (event) =>
1171 event.type === "provider-thread.updated" && event.payload.id === activeProviderThreadId,
1172 );
1173 assert.equal(
1174 activeThreadEvent?.type === "provider-thread.updated"
1175 ? activeThreadEvent.payload.status
1176 : null,
1177 "idle",
1178 );
1179 }).pipe(Effect.provide(layer));
1180 },
1181);
1182
1183it.effect(
1184 "terminalizes a leftover nonpersistent dynamic_tool on a settled run after process loss",
1185 () => {
1186 const threadId = ThreadId.make("thread_recovery_orphan_wait");
1187 const settledRunId = RunId.make("run_recovery_orphan_wait_settled");
1188 const providerThreadId = ProviderThreadId.make("provider_thread_recovery_orphan_wait");
1189 const orphanWaitItemId = TurnItemId.make(
1190 "turn-item:provider:codex:native-item:exec-4669f3bb-78c9-4af1-b44e-daa340d2c538",
1191 );
1192 const persistentMonitorItemId = TurnItemId.make(
1193 "turn-item:provider:codex:native-item:exec-persistent-monitor",
1194 );
1195 const orphanWaitNodeId = NodeId.make("node_recovery_orphan_wait");
1196 const persistentMonitorNodeId = NodeId.make("node_recovery_persistent_monitor");
1197 const codexInstanceId = ProviderInstanceId.make("codex");
1198 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
1199 null;
1200 const projection = {
1201 thread: { id: threadId, providerInstanceId: codexInstanceId },
1202 runtimeRequests: [],
1203 providerSessions: [],
1204 providerThreads: [
1205 {
1206 id: providerThreadId,
1207 driver: ProviderDriverKind.make("codex"),
1208 providerInstanceId: codexInstanceId,
1209 status: "idle",
1210 pendingBackgroundTasks: [],
1211 },
1212 ],
1213 providerTurns: [],
1214 runs: [{ id: settledRunId, status: "completed", providerInstanceId: codexInstanceId }],
1215 attempts: [],
1216 nodes: [
1217 { id: orphanWaitNodeId, runId: settledRunId, status: "running", kind: "tool_call" },
1218 {
1219 id: persistentMonitorNodeId,
1220 runId: settledRunId,
1221 status: "running",
1222 kind: "tool_call",
1223 },
1224 ],
1225 subagents: [],
1226 messages: [],
1227 turnItems: [
1228 {
1229 id: orphanWaitItemId,
1230 runId: settledRunId,
1231 nodeId: orphanWaitNodeId,
1232 providerThreadId,
1233 type: "dynamic_tool",
1234 status: "running",
1235 toolName: "t3-code.t3_thread_wait",
1236 input: {
1237 threadId:
1238 "thread:delegated-task:command%3Amcp%3Aaafffab1-e811-458a-ae83-558e542c61ff%3Adelegate-task%3Areview-mobile-reconnect-opus-20260815",
1239 timeoutMs: 30000,
1240 },
1241 },
1242 {
1243 id: persistentMonitorItemId,
1244 runId: settledRunId,
1245 nodeId: persistentMonitorNodeId,
1246 providerThreadId,
1247 type: "dynamic_tool",
1248 status: "running",
1249 toolName: "grok.monitor",
1250 input: { persistent: true, command: "tail -f" },
1251 },
1252 ],
1253 } as unknown as OrchestrationV2ThreadProjection;
1254 const layer = ProviderRuntimeRecovery.layer.pipe(
1255 Layer.provide(ServerSettings.layerTest()),
1256 Layer.provide(
1257 Layer.mergeAll(
1258 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1259 getRecoveryThreadIds: () => Effect.succeed([threadId]),
1260 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
1261 }),
1262 Layer.mock(EventSink.EventSinkV2)({
1263 commitCommand: (input) => {
1264 committedInput = input;
1265 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
1266 },
1267 }),
1268 IdAllocator.layer,
1269 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
1270 runRecoveryOnce: Effect.succeed(false),
1271 }),
1272 Layer.mock(EffectOutbox.EffectOutboxV2)({
1273 listByCommandId: () => Effect.succeed([]),
1274 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
1275 }),
1276 ),
1277 ),
1278 );
1279
1280 return Effect.gen(function* () {
1281 const summary =
1282 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
1283 assert.equal(summary.terminalizedRuns, 0);
1284 const turnItemCancels = (committedInput?.events ?? []).filter(
1285 (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled",
1286 );
1287 // Process loss means the provider is gone, so even a persistent monitor
1288 // cannot still be alive. The leftover wait and the monitor both close.
1289 assert.equal(turnItemCancels.length, 2);
1290 assert.deepEqual(
1291 turnItemCancels
1292 .map((event) => event.type === "turn-item.updated" && event.payload.id)
1293 .sort(),
1294 [orphanWaitItemId, persistentMonitorItemId].sort(),
1295 );
1296 const nodeCancels = (committedInput?.events ?? []).filter(
1297 (event) => event.type === "node.updated" && event.payload.status === "cancelled",
1298 );
1299 assert.equal(nodeCancels.length, 2);
1300 assert.deepEqual(
1301 nodeCancels.map((event) => event.type === "node.updated" && event.payload.id).sort(),
1302 [orphanWaitNodeId, persistentMonitorNodeId].sort(),
1303 );
1304 }).pipe(Effect.provide(layer));
1305 },
1306);
1307
1308it.effect(
1309 "terminalizes the linked subagent and node for a stale subagent item on a settled run",
1310 () => {
1311 const threadId = ThreadId.make("thread_recovery_subagent");
1312 const settledRunId = RunId.make("run_recovery_subagent_settled");
1313 const providerThreadId = ProviderThreadId.make("provider_thread_recovery_subagent");
1314 const staleSubagentNodeId = NodeId.make("node_recovery_subagent_stale");
1315 const doneSubagentNodeId = NodeId.make("node_recovery_subagent_done");
1316 const staleItemId = TurnItemId.make("turn_item_recovery_subagent_stale");
1317 const doneItemId = TurnItemId.make("turn_item_recovery_subagent_done");
1318 const claudeInstanceId = ProviderInstanceId.make("claude");
1319 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
1320 null;
1321 const projection = {
1322 thread: { id: threadId },
1323 runtimeRequests: [],
1324 providerSessions: [],
1325 providerThreads: [
1326 {
1327 id: providerThreadId,
1328 driver: ProviderDriverKind.make("claude"),
1329 providerInstanceId: claudeInstanceId,
1330 status: "idle",
1331 pendingBackgroundTasks: [],
1332 },
1333 ],
1334 providerTurns: [],
1335 // Settled run: the stale-item loop owns it, not the nonterminal loop.
1336 runs: [{ id: settledRunId, status: "completed", providerInstanceId: claudeInstanceId }],
1337 attempts: [],
1338 nodes: [
1339 { id: staleSubagentNodeId, runId: settledRunId, status: "running" },
1340 { id: doneSubagentNodeId, runId: settledRunId, status: "completed" },
1341 ],
1342 subagents: [
1343 {
1344 id: staleSubagentNodeId,
1345 runId: settledRunId,
1346 origin: "provider_native",
1347 driver: ProviderDriverKind.make("claude"),
1348 providerInstanceId: claudeInstanceId,
1349 status: "running",
1350 },
1351 {
1352 // Already finished with a real result: must never be overwritten.
1353 id: doneSubagentNodeId,
1354 runId: settledRunId,
1355 origin: "provider_native",
1356 driver: ProviderDriverKind.make("claude"),
1357 providerInstanceId: claudeInstanceId,
1358 status: "completed",
1359 result: "done",
1360 },
1361 ],
1362 messages: [],
1363 turnItems: [
1364 {
1365 id: staleItemId,
1366 runId: settledRunId,
1367 nodeId: staleSubagentNodeId,
1368 providerThreadId,
1369 type: "subagent",
1370 status: "running",
1371 subagentId: staleSubagentNodeId,
1372 providerInstanceId: claudeInstanceId,
1373 },
1374 {
1375 id: doneItemId,
1376 runId: settledRunId,
1377 nodeId: doneSubagentNodeId,
1378 providerThreadId,
1379 type: "subagent",
1380 status: "completed",
1381 subagentId: doneSubagentNodeId,
1382 providerInstanceId: claudeInstanceId,
1383 },
1384 ],
1385 } as unknown as OrchestrationV2ThreadProjection;
1386 const layer = ProviderRuntimeRecovery.layer.pipe(
1387 Layer.provide(ServerSettings.layerTest()),
1388 Layer.provide(
1389 Layer.mergeAll(
1390 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1391 getRecoveryThreadIds: () => Effect.succeed([threadId]),
1392 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
1393 }),
1394 Layer.mock(EventSink.EventSinkV2)({
1395 commitCommand: (input) => {
1396 committedInput = input;
1397 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
1398 },
1399 }),
1400 IdAllocator.layer,
1401 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
1402 runRecoveryOnce: Effect.succeed(false),
1403 }),
1404 Layer.mock(EffectOutbox.EffectOutboxV2)({
1405 listByCommandId: () => Effect.succeed([]),
1406 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
1407 }),
1408 ),
1409 ),
1410 );
1411
1412 return Effect.gen(function* () {
1413 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
1414 const events = committedInput?.events ?? [];
1415
1416 // Only the nonterminal subagent item is cancelled.
1417 const turnItemCancels = events.filter(
1418 (event) => event.type === "turn-item.updated" && event.payload.status === "cancelled",
1419 );
1420 assert.equal(turnItemCancels.length, 1);
1421
1422 // The linked subagent entity is terminalized alongside its turn item.
1423 const subagentCancels = events.filter((event) => event.type === "subagent.updated");
1424 assert.equal(subagentCancels.length, 1);
1425 const subagentCancel = subagentCancels[0];
1426 assert.equal(
1427 subagentCancel?.type === "subagent.updated" ? subagentCancel.payload.id : null,
1428 staleSubagentNodeId,
1429 );
1430 assert.equal(
1431 subagentCancel?.type === "subagent.updated" ? subagentCancel.payload.status : null,
1432 "cancelled",
1433 );
1434
1435 // So is its execution node, which no live process can terminalize.
1436 const nodeCancels = events.filter((event) => event.type === "node.updated");
1437 assert.equal(nodeCancels.length, 1);
1438 const nodeCancel = nodeCancels[0];
1439 assert.equal(
1440 nodeCancel?.type === "node.updated" ? nodeCancel.payload.id : null,
1441 staleSubagentNodeId,
1442 );
1443
1444 // The already-completed subagent and node are left untouched.
1445 assert.isFalse(
1446 events.some(
1447 (event) => event.type === "subagent.updated" && event.payload.id === doneSubagentNodeId,
1448 ),
1449 );
1450 assert.isFalse(
1451 events.some(
1452 (event) => event.type === "node.updated" && event.payload.id === doneSubagentNodeId,
1453 ),
1454 );
1455 }).pipe(Effect.provide(layer));
1456 },
1457);
1458
1459it.effect("leaves delegated tasks to their own child threads after process loss", () => {
1460 const threadId = ThreadId.make("thread_recovery_delegation");
1461 const settledRunId = RunId.make("run_recovery_delegation_settled");
1462 const runningRunId = RunId.make("run_recovery_delegation_running");
1463 const providerThreadId = ProviderThreadId.make("provider_thread_recovery_delegation");
1464 const claudeInstanceId = ProviderInstanceId.make("claude");
1465 const driver = ProviderDriverKind.make("claude");
1466 const settledTaskId = NodeId.make("node_recovery_delegation_settled");
1467 const runningTaskId = NodeId.make("node_recovery_delegation_running");
1468 const nativeSubagentId = NodeId.make("node_recovery_native_subagent");
1469 const delegatedTask = (id: NodeId, runId: RunId) => ({
1470 id,
1471 runId,
1472 origin: "app_owned",
1473 childThreadId: ThreadId.make(`thread:delegated-task:${id}`),
1474 driver,
1475 providerInstanceId: claudeInstanceId,
1476 status: "running",
1477 });
1478 const subagentItem = (subagentId: NodeId, runId: RunId, origin: string) => ({
1479 id: TurnItemId.make(`turn_item:${subagentId}`),
1480 runId,
1481 nodeId: subagentId,
1482 providerThreadId,
1483 type: "subagent",
1484 status: "running",
1485 subagentId,
1486 origin,
1487 childThreadId:
1488 origin === "app_owned" ? ThreadId.make(`thread:delegated-task:${subagentId}`) : null,
1489 providerInstanceId: claudeInstanceId,
1490 title: `subagent ${subagentId}`,
1491 });
1492 let committedInput: Parameters<EventSink.EventSinkV2["Service"]["commitCommand"]>[0] | null =
1493 null;
1494 const projection = {
1495 thread: { id: threadId, providerInstanceId: claudeInstanceId },
1496 runtimeRequests: [],
1497 providerSessions: [],
1498 providerThreads: [
1499 {
1500 id: providerThreadId,
1501 driver,
1502 providerInstanceId: claudeInstanceId,
1503 ownerNodeId: null,
1504 status: "idle",
1505 pendingBackgroundTasks: [],
1506 },
1507 ],
1508 providerTurns: [],
1509 runs: [
1510 {
1511 id: settledRunId,
1512 ordinal: 1,
1513 status: "completed",
1514 providerThreadId,
1515 providerInstanceId: claudeInstanceId,
1516 },
1517 {
1518 id: runningRunId,
1519 ordinal: 2,
1520 status: "running",
1521 providerThreadId,
1522 providerInstanceId: claudeInstanceId,
1523 },
1524 ],
1525 attempts: [],
1526 nodes: [
1527 { id: settledTaskId, runId: settledRunId, status: "running", kind: "subagent" },
1528 { id: runningTaskId, runId: runningRunId, status: "running", kind: "subagent" },
1529 { id: nativeSubagentId, runId: settledRunId, status: "running", kind: "subagent" },
1530 ],
1531 subagents: [
1532 delegatedTask(settledTaskId, settledRunId),
1533 delegatedTask(runningTaskId, runningRunId),
1534 {
1535 ...delegatedTask(nativeSubagentId, settledRunId),
1536 origin: "provider_native",
1537 childThreadId: null,
1538 },
1539 ],
1540 messages: [],
1541 turnItems: [
1542 subagentItem(settledTaskId, settledRunId, "app_owned"),
1543 subagentItem(runningTaskId, runningRunId, "app_owned"),
1544 subagentItem(nativeSubagentId, settledRunId, "provider_native"),
1545 ],
1546 } as unknown as OrchestrationV2ThreadProjection;
1547 const layer = ProviderRuntimeRecovery.layer.pipe(
1548 Layer.provide(ServerSettings.layerTest()),
1549 Layer.provide(
1550 Layer.mergeAll(
1551 Layer.mock(ProjectionStore.ProjectionStoreV2)({
1552 getRecoveryThreadIds: () => Effect.succeed([threadId]),
1553 getRuntimeRecoveryProjection: () => Effect.succeed(projection),
1554 }),
1555 Layer.mock(EventSink.EventSinkV2)({
1556 commitCommand: (input) => {
1557 committedInput = input;
1558 return Effect.succeed({ committed: true, cancelledEffectCount: 0 } as never);
1559 },
1560 }),
1561 IdAllocator.layer,
1562 Layer.mock(EffectWorker.OrchestrationEffectWorkerV2)({
1563 runRecoveryOnce: Effect.succeed(false),
1564 }),
1565 Layer.mock(EffectOutbox.EffectOutboxV2)({
1566 listByCommandId: () => Effect.succeed([]),
1567 reconcileAfterProcessLoss: Effect.succeed({ requeued: 0, cancelled: 0 }),
1568 }),
1569 ),
1570 ),
1571 );
1572
1573 return Effect.gen(function* () {
1574 yield* (yield* ProviderRuntimeRecovery.ProviderRuntimeRecoveryService).reconcile("startup");
1575 const events = committedInput?.events ?? [];
1576 const touched = events.flatMap((event) =>
1577 event.type === "turn-item.updated"
1578 ? event.payload.type === "subagent"
1579 ? [event.payload.subagentId]
1580 : []
1581 : event.type === "subagent.updated" || event.type === "node.updated"
1582 ? [event.payload.id]
1583 : [],
1584 );
1585 // The cut run itself is cancelled; only the provider-native subagent dies with it.
1586 assert.isTrue(
1587 events.some(
1588 (event) =>
1589 event.type === "run.updated" &&
1590 event.payload.id === runningRunId &&
1591 event.payload.status === "cancelled",
1592 ),
1593 );
1594 assert.notInclude(touched, settledTaskId);
1595 assert.notInclude(touched, runningTaskId);
1596 assert.include(touched, nativeSubagentId);
1597 // The note lists only provider-native work: delegated tasks report back on their own.
1598 const noted = events.flatMap((event) =>
1599 event.type === "run.background-work-cancelled"
1600 ? event.payload.restartCancelledBackgroundWork.map((work) => work.label)
1601 : [],
1602 );
1603 assert.deepEqual(noted, [`subagent ${nativeSubagentId}`]);
1604 }).pipe(Effect.provide(layer));
1605});