1import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
2import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
3import {
4 CommandId,
5 type OrchestrationV2DomainEvent,
6 type ProviderThreadId,
7 type OrchestrationV2RestartCancelledBackgroundWork,
8 type OrchestrationV2Subagent,
9 type OrchestrationV2ThreadProjection,
10 isOrchestrationV2WorkActive,
11 ThreadId,
12} from "@t3tools/contracts";
13import * as Cause from "effect/Cause";
14import * as Context from "effect/Context";
15import * as DateTime from "effect/DateTime";
16import * as Effect from "effect/Effect";
17import * as Layer from "effect/Layer";
18import * as Schema from "effect/Schema";
19
20import * as EffectOutbox from "./EffectOutbox.ts";
21import * as EventSink from "./EventSink.ts";
22import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
23import * as ProjectionStore from "./ProjectionStore.ts";
24import * as ServerSettings from "../serverSettings.ts";
25import { restartContinuationRun } from "./RestartContinuation.ts";
26import { isBufferedProviderNotification } from "./NotificationMailbox.ts";
27import {
28 endOrphanedNativeSubagent,
29 endRunlessRootTurns,
30} from "@t3tools/provider-core/server/subagentProjection";
31import {
32 cancelledRosterTaskWork,
33 cancelledTurnItemWork,
34 mergeRestartCancelledBackgroundWork,
35} from "./RestartBackgroundNote.ts";
36
37export class ProviderRuntimeRecoveryError extends Schema.TaggedError<ProviderRuntimeRecoveryError>()(
38 "ProviderRuntimeRecoveryError",
39 {
40 operation: Schema.Literals(["read-projections", "reconcile", "drain-outbox"]),
41 threadId: Schema.optional(ThreadId),
42 cause: Schema.Defect(),
43 },
44) {
45 override get message(): string {
46 return `Provider runtime recovery failed during ${this.operation}.`;
47 }
48}
49
50export interface ProviderRuntimeRecoverySummary {
51 readonly terminalizedRuns: number;
52 readonly stoppedSessions: number;
53 readonly closedRequests: number;
54 readonly retiredEffects: number;
55 readonly requeuedEffects: number;
56}
57
58export interface ProviderRuntimeReconciliationSummary {
59 readonly terminalizedRuns: number;
60 readonly stoppedSessions: number;
61 readonly closedRequests: number;
62 readonly retiredEffects: number;
63 readonly requeuedEffects: number;
64}
65
66export class ProviderRuntimeRecoveryService extends Context.Service<
67 ProviderRuntimeRecoveryService,
68 {
69 readonly reconcile: (
70 trigger: "startup" | "shutdown",
71 ) => Effect.Effect<ProviderRuntimeReconciliationSummary, ProviderRuntimeRecoveryError>;
72 readonly prepareForShutdown: Effect.Effect<void, ProviderRuntimeRecoveryError>;
73 readonly recover: Effect.Effect<ProviderRuntimeRecoverySummary, ProviderRuntimeRecoveryError>;
74 }
75>()("t3/orchestration-v2/ProviderRuntimeRecoveryService") {}
76
77function nonterminalRuns(projection: ProjectionStore.ProjectionRuntimeRecoveryState) {
78 return projection.runs.filter((run) => {
79 const status: string = run.status;
80 return (
81 status === "preparing" ||
82 run.status === "starting" ||
83 run.status === "running" ||
84 run.status === "waiting"
85 );
86 });
87}
88
89function isBackgroundCapableTurnItemType(type: string): boolean {
90 return type === "command_execution" || type === "dynamic_tool" || type === "subagent";
91}
92
93function isNonterminalTurnItemStatus(status: string): boolean {
94 return status === "pending" || status === "running" || status === "waiting";
95}
96
97function isNonterminalNodeStatus(status: string): boolean {
98 return status === "pending" || status === "running" || status === "waiting";
99}
100
101/**
102 * A delegate_task child. Its own thread is reconciled and continued on its own
103 * and reports back through the app, so it is not provider background work.
104 */
105function isAppOwnedDelegation(task: {
106 readonly origin: OrchestrationV2Subagent["origin"];
107 readonly childThreadId: ThreadId | null;
108}): boolean {
109 return task.origin === "app_owned" && task.childThreadId !== null;
110}
111
112function isAppOwnedDelegationItem(
113 item: OrchestrationV2ThreadProjection["turnItems"][number],
114): boolean {
115 return item.type === "subagent" && isAppOwnedDelegation(item);
116}
117
118function providerThreadHasPendingBackgroundTasks(
119 providerThread: OrchestrationV2ThreadProjection["providerThreads"][number],
120): boolean {
121 return (providerThread.pendingBackgroundTasks?.length ?? 0) > 0;
122}
123
124/**
125 * Resolve providerInstanceId for a stale background-capable turn item whose
126 * run is missing/null (or not found). Prefer an existing run, then a subagent
127 * item's own instance id, then the item's provider thread, then a last-resort
128 * first provider thread, then the thread's selected provider.
129 */
130function resolveStaleBackgroundItemProviderInstanceId(
131 item: OrchestrationV2ThreadProjection["turnItems"][number],
132 projection: ProjectionStore.ProjectionRuntimeRecoveryState,
133): OrchestrationV2ThreadProjection["thread"]["providerInstanceId"] {
134 if (item.runId !== null) {
135 const run = projection.runs.find((candidate) => candidate.id === item.runId);
136 if (run !== undefined) {
137 return run.providerInstanceId;
138 }
139 }
140 if (item.type === "subagent") {
141 return item.providerInstanceId;
142 }
143 if (item.providerThreadId !== null && item.providerThreadId !== undefined) {
144 const providerThread = projection.providerThreads.find(
145 (candidate) => candidate.id === item.providerThreadId,
146 );
147 if (providerThread !== undefined) {
148 return providerThread.providerInstanceId;
149 }
150 }
151 return projection.providerThreads[0]?.providerInstanceId ?? projection.thread.providerInstanceId;
152}
153
154/** An open item closed by recovery. A secret card also closes its form, so it takes no answer. */
155const cancelledItem = (
156 item: OrchestrationV2ThreadProjection["turnItems"][number],
157 now: DateTime.Utc,
158): OrchestrationV2ThreadProjection["turnItems"][number] =>
159 item.type === "secret_request" && item.secretStatus === "pending"
160 ? { ...item, status: "cancelled", secretStatus: "cancelled", completedAt: now, updatedAt: now }
161 : { ...item, status: "cancelled", completedAt: now, updatedAt: now };
162
163/**
164 * A provider thread's latest started run: the last turn that provider saw.
165 * Restart recovery records the thread's cancelled background work on it, and
166 * the next run on the same provider thread delivers it with its input.
167 */
168function latestStartedRun(
169 projection: ProjectionStore.ProjectionRuntimeRecoveryState,
170 providerThreadId: ProviderThreadId,
171) {
172 return projection.runs.reduce<OrchestrationV2ThreadProjection["runs"][number] | undefined>(
173 (latest, run) =>
174 run.providerThreadId === providerThreadId &&
175 run.status !== "queued" &&
176 run.status !== "rolled_back" &&
177 (latest === undefined || runRanAfter(run, latest))
178 ? run
179 : latest,
180 undefined,
181 );
182}
183
184export const make = Effect.gen(function* () {
185 const settings = yield* ServerSettings.ServerSettingsService;
186 const projections = yield* ProjectionStore.ProjectionStoreV2;
187 const eventSink = yield* EventSink.EventSinkV2;
188 const ids = yield* IdAllocator.IdAllocatorV2;
189 const outbox = yield* EffectOutbox.EffectOutboxV2;
190 const reconcileProjection = Effect.fn("ProviderRuntimeRecoveryService.reconcileProjection")(
191 function* (
192 projection: ProjectionStore.ProjectionRuntimeRecoveryState,
193 trigger: "startup" | "shutdown",
194 continueAfterRestart: boolean,
195 ) {
196 const now = yield* DateTime.now;
197 const runs = [] as Array<OrchestrationV2ThreadProjection["runs"][number]>;
198 for (const run of nonterminalRuns(projection)) {
199 if (run.status === "waiting") {
200 const checkpointEffects = yield* outbox
201 .listByCommandId(CommandId.make(`command:effect:checkpoint.capture:${run.id}`))
202 .pipe(
203 Effect.mapError(
204 (cause) =>
205 new ProviderRuntimeRecoveryError({
206 operation: "reconcile",
207 threadId: projection.thread.id,
208 cause,
209 }),
210 ),
211 );
212 const hasReplayableCheckpoint = checkpointEffects.some(
213 (effect) =>
214 effect.request.type === "checkpoint.capture" &&
215 effect.request.runId === run.id &&
216 (effect.status === "pending" || effect.status === "running"),
217 );
218 if (hasReplayableCheckpoint) continue;
219 }
220 runs.push(run);
221 }
222 const messageRequestNodeIds = new Set(
223 projection.runtimeRequests
224 .filter(
225 (request) =>
226 request.status === "pending" && request.responseCapability.type === "message",
227 )
228 .map((request) => request.nodeId),
229 );
230 const requests = projection.runtimeRequests.filter(
231 (request) => request.status === "pending" && request.responseCapability.type !== "message",
232 );
233 // Delegated task rows, items and nodes stay open: the child settles them.
234 const delegatedTaskNodeIds = new Set<string>([
235 ...(projection.subagents ?? []).filter(isAppOwnedDelegation).map((subagent) => subagent.id),
236 ...(projection.turnItems ?? []).flatMap((item) =>
237 item.type === "subagent" && isAppOwnedDelegation(item) ? [item.subagentId] : [],
238 ),
239 ]);
240 const detail = `Cancelled because the server ${trigger === "startup" ? "restarted" : "shut down"} before the provider work completed.`;
241 const commandId = CommandId.make(
242 `command:runtime-reconcile:${trigger}:${projection.thread.id}:${DateTime.formatIso(now)}`,
243 );
244 const allocateEventId = () =>
245 ids.allocate.event({ threadId: projection.thread.id, commandId }).pipe(
246 Effect.mapError(
247 (cause) =>
248 new ProviderRuntimeRecoveryError({
249 operation: "reconcile",
250 threadId: projection.thread.id,
251 cause,
252 }),
253 ),
254 );
255 const events: Array<OrchestrationV2DomainEvent> = [];
256 // Background work that outlived its settled turn. The provider transcript
257 // cannot record its death, so the next provider turn is told instead.
258 // Shutdown records it too: a graceful restart cancels it there first.
259 // Keyed by the provider thread that lost the work: only its turns are told.
260 const cancelledBackgroundWork = new Map<
261 ProviderThreadId,
262 Array<OrchestrationV2RestartCancelledBackgroundWork>
263 >();
264 const cancelledBackgroundNativeIds = new Set<string>();
265 const recordCancelledBackgroundWork = (
266 providerThreadId: ProviderThreadId | null | undefined,
267 work: OrchestrationV2RestartCancelledBackgroundWork,
268 ) => {
269 if (providerThreadId == null) return;
270 const existing = cancelledBackgroundWork.get(providerThreadId);
271 if (existing === undefined) cancelledBackgroundWork.set(providerThreadId, [work]);
272 else existing.push(work);
273 };
274 const recordCancelledBackgroundItem = (
275 item: OrchestrationV2ThreadProjection["turnItems"][number],
276 ) => {
277 if (!isBackgroundCapableTurnItemType(item.type)) return;
278 const work = cancelledTurnItemWork(item);
279 if (work === undefined) return;
280 recordCancelledBackgroundWork(
281 item.providerThreadId ??
282 projection.runs.find((run) => run.id === item.runId)?.providerThreadId,
283 work,
284 );
285 if (item.nativeItemRef?.nativeId != null) {
286 cancelledBackgroundNativeIds.add(item.nativeItemRef.nativeId);
287 }
288 };
289 const queuedRuns = projection.runs.filter((run) => run.status === "queued");
290 const queuedMessages =
291 queuedRuns.length === 0
292 ? []
293 : (yield* projections
294 .getThreadRecords(projection.thread.id, ["messages"], {
295 messageIds: queuedRuns.map((run) => run.userMessageId),
296 })
297 .pipe(
298 Effect.mapError(
299 (cause) =>
300 new ProviderRuntimeRecoveryError({
301 operation: "read-projections",
302 threadId: projection.thread.id,
303 cause,
304 }),
305 ),
306 )).messages;
307 const bufferedMessageIds = new Set(
308 queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id),
309 );
310 // Buffered wakes lost their output with the process. Durable queued messages
311 // keep their execution identities and order until the user resumes them.
312 for (const run of queuedRuns) {
313 if (bufferedMessageIds.has(run.userMessageId)) {
314 runs.push(run);
315 continue;
316 }
317 if (run.queueHeld === true) continue;
318 events.push({
319 id: yield* allocateEventId(),
320 type: "run.updated",
321 threadId: projection.thread.id,
322 runId: run.id,
323 providerInstanceId: run.providerInstanceId,
324 occurredAt: now,
325 payload: { ...run, queueHeld: true },
326 });
327 }
328 for (const request of requests) {
329 events.push({
330 id: yield* allocateEventId(),
331 type: "runtime-request.updated",
332 threadId: projection.thread.id,
333 nodeId: request.nodeId,
334 occurredAt: now,
335 payload: {
336 ...request,
337 status: trigger === "startup" ? "expired" : "cancelled",
338 responseCapability: {
339 type: "not_resumable",
340 reason: `The server ${trigger === "startup" ? "restarted" : "shut down"} before this runtime request was resolved.`,
341 },
342 resolvedAt: now,
343 },
344 });
345 }
346 for (const run of runs) {
347 events.push({
348 id: yield* allocateEventId(),
349 type: "run.updated",
350 threadId: projection.thread.id,
351 runId: run.id,
352 providerInstanceId: run.providerInstanceId,
353 occurredAt: now,
354 payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now },
355 });
356 for (const attempt of projection.attempts.filter(
357 (candidate) =>
358 candidate.runId === run.id &&
359 (candidate.status === "pending" || candidate.status === "running"),
360 )) {
361 events.push({
362 id: yield* allocateEventId(),
363 type: "run-attempt.updated",
364 threadId: projection.thread.id,
365 runId: run.id,
366 nodeId: attempt.rootNodeId,
367 providerInstanceId: run.providerInstanceId,
368 occurredAt: now,
369 payload: { ...attempt, status: "cancelled", completedAt: now },
370 });
371 }
372 for (const node of projection.nodes.filter(
373 (candidate) =>
374 candidate.runId === run.id &&
375 !messageRequestNodeIds.has(candidate.id) &&
376 !delegatedTaskNodeIds.has(candidate.id) &&
377 (candidate.status === "pending" ||
378 candidate.status === "running" ||
379 candidate.status === "waiting"),
380 )) {
381 events.push({
382 id: yield* allocateEventId(),
383 type: "node.updated",
384 threadId: projection.thread.id,
385 runId: run.id,
386 nodeId: node.id,
387 providerInstanceId: run.providerInstanceId,
388 occurredAt: now,
389 payload: { ...node, status: "cancelled", completedAt: now },
390 });
391 }
392 for (const providerTurn of projection.providerTurns.filter(
393 (candidate) =>
394 candidate.runAttemptId !== null &&
395 projection.attempts.some(
396 (attempt) => attempt.id === candidate.runAttemptId && attempt.runId === run.id,
397 ) &&
398 (candidate.status === "pending" || candidate.status === "running"),
399 )) {
400 events.push({
401 id: yield* allocateEventId(),
402 type: "provider-turn.updated",
403 threadId: projection.thread.id,
404 runId: run.id,
405 nodeId: providerTurn.nodeId,
406 providerInstanceId: run.providerInstanceId,
407 occurredAt: now,
408 payload: { ...providerTurn, status: "cancelled", completedAt: now },
409 });
410 }
411 for (const message of projection.messages.filter(
412 (candidate) => candidate.runId === run.id && candidate.streaming,
413 )) {
414 events.push({
415 id: yield* allocateEventId(),
416 type: "message.updated",
417 threadId: projection.thread.id,
418 runId: run.id,
419 ...(message.nodeId === null ? {} : { nodeId: message.nodeId }),
420 providerInstanceId: run.providerInstanceId,
421 occurredAt: now,
422 payload: { ...message, streaming: false, updatedAt: now },
423 });
424 }
425 for (const item of projection.turnItems.filter(
426 (candidate) =>
427 candidate.runId === run.id &&
428 (candidate.nodeId === null || !messageRequestNodeIds.has(candidate.nodeId)) &&
429 !isAppOwnedDelegationItem(candidate) &&
430 (candidate.status === "pending" ||
431 candidate.status === "running" ||
432 candidate.status === "waiting"),
433 )) {
434 // A waiting run's provider turn already settled; its open items are
435 // background work. A running run's items die with its turn.
436 if (run.status === "waiting") recordCancelledBackgroundItem(item);
437 events.push({
438 id: yield* allocateEventId(),
439 type: "turn-item.updated",
440 threadId: projection.thread.id,
441 runId: run.id,
442 ...(item.nodeId === null ? {} : { nodeId: item.nodeId }),
443 providerInstanceId: run.providerInstanceId,
444 occurredAt: now,
445 payload: cancelledItem(item, now),
446 });
447 }
448 }
449 // A blocking delegate_task wait dies with the turn that made it, and its
450 // continuation never got the tool result. Let the child's result wake
451 // the parent like an async task, so it reaches the continued turn.
452 const cutRunIds = new Set(runs.map((run) => run.id));
453 for (const task of projection.subagents ?? []) {
454 if (
455 task.origin !== "app_owned" ||
456 task.runId === null ||
457 !cutRunIds.has(task.runId) ||
458 task.completionWake !== "settled_only" ||
459 !isOrchestrationV2WorkActive(task.status)
460 )
461 continue;
462 events.push({
463 id: yield* allocateEventId(),
464 type: "subagent.updated",
465 threadId: projection.thread.id,
466 runId: task.runId,
467 nodeId: task.id,
468 driver: task.driver,
469 providerInstanceId: task.providerInstanceId,
470 occurredAt: now,
471 payload: { ...task, completionWake: "always", updatedAt: now },
472 });
473 }
474 // Process loss also orphans background-capable turn items on already-
475 // settled runs (e.g. post-settle Waiting work). Skip items already
476 // cancelled above for recovered nonterminal runs to avoid duplicate
477 // cancellation events.
478 const recoveredNonterminalRunIds = new Set(runs.map((run) => run.id));
479 const cancelledStaleNodeIds = new Set<string>();
480 for (const item of projection.turnItems ?? []) {
481 if (item.runId !== null && recoveredNonterminalRunIds.has(item.runId)) {
482 continue;
483 }
484 if (!isBackgroundCapableTurnItemType(item.type)) {
485 continue;
486 }
487 if (!isNonterminalTurnItemStatus(item.status) || isAppOwnedDelegationItem(item)) {
488 continue;
489 }
490 const providerInstanceId = resolveStaleBackgroundItemProviderInstanceId(item, projection);
491 recordCancelledBackgroundItem(item);
492 events.push({
493 id: yield* allocateEventId(),
494 type: "turn-item.updated",
495 threadId: projection.thread.id,
496 ...(item.runId === null ? {} : { runId: item.runId }),
497 ...(item.nodeId === null || item.nodeId === undefined ? {} : { nodeId: item.nodeId }),
498 providerInstanceId,
499 occurredAt: now,
500 payload: cancelledItem(item, now),
501 });
502 if (item.nodeId !== null && item.nodeId !== undefined) {
503 const staleItemNode = projection.nodes.find(
504 (candidate) =>
505 candidate.id === item.nodeId && isNonterminalNodeStatus(candidate.status),
506 );
507 if (staleItemNode !== undefined && !cancelledStaleNodeIds.has(staleItemNode.id)) {
508 cancelledStaleNodeIds.add(staleItemNode.id);
509 events.push({
510 id: yield* allocateEventId(),
511 type: "node.updated",
512 threadId: projection.thread.id,
513 ...(item.runId === null ? {} : { runId: item.runId }),
514 nodeId: staleItemNode.id,
515 providerInstanceId,
516 occurredAt: now,
517 payload: { ...staleItemNode, status: "cancelled", completedAt: now },
518 });
519 }
520 }
521 }
522 // No provider process outlives the server, so a native subagent still open
523 // here never reports its end, whatever state its run and item are in.
524 for (const subagent of projection.subagents ?? []) {
525 events.push(
526 ...(yield* endOrphanedNativeSubagent({
527 projection,
528 subagentId: subagent.id,
529 status: "cancelled",
530 now,
531 emitted: events,
532 allocateEventId,
533 })),
534 );
535 }
536 // A provider-native subagent thread has no runs; left running, the child
537 // would show as working forever.
538 events.push(
539 ...(yield* endRunlessRootTurns({
540 threadId: projection.thread.id,
541 providerInstanceId: projection.thread.providerInstanceId,
542 projection,
543 status: "cancelled",
544 now,
545 emitted: events,
546 allocateEventId,
547 })),
548 );
549 // All provider processes are gone on startup/shutdown: clear any
550 // persisted Waiting roster (including idle threads from settled roots)
551 // and idle active threads without resurrecting active status.
552 for (const providerThread of projection.providerThreads ?? []) {
553 const needsIdle = providerThread.status === "active";
554 const needsRosterClear = providerThreadHasPendingBackgroundTasks(providerThread);
555 if (!needsIdle && !needsRosterClear) {
556 continue;
557 }
558 if (providerThread.ownerNodeId === null) {
559 for (const task of providerThread.pendingBackgroundTasks ?? []) {
560 if (cancelledBackgroundNativeIds.has(task.taskId)) continue;
561 cancelledBackgroundNativeIds.add(task.taskId);
562 recordCancelledBackgroundWork(providerThread.id, cancelledRosterTaskWork(task));
563 }
564 }
565 events.push({
566 id: yield* allocateEventId(),
567 type: "provider-thread.updated",
568 threadId: projection.thread.id,
569 driver: providerThread.driver,
570 providerInstanceId: providerThread.providerInstanceId,
571 occurredAt: now,
572 payload: {
573 ...providerThread,
574 status: needsIdle ? "idle" : providerThread.status,
575 pendingBackgroundTasks: [],
576 updatedAt: now,
577 },
578 });
579 }
580 for (const session of projection.providerSessions.filter(
581 (candidate) => candidate.status !== "stopped" && candidate.status !== "error",
582 )) {
583 events.push({
584 id: yield* allocateEventId(),
585 type: "provider-session.updated",
586 threadId: projection.thread.id,
587 driver: session.driver,
588 providerInstanceId: session.providerInstanceId,
589 occurredAt: now,
590 payload: { ...session, status: "stopped", updatedAt: now, lastError: null },
591 });
592 }
593 for (const [providerThreadId, work] of cancelledBackgroundWork) {
594 const noteRun = latestStartedRun(projection, providerThreadId);
595 if (noteRun === undefined) continue;
596 // Its own event: a run snapshot read before this commit could regress
597 // a lifecycle change (e.g. a checkpoint completing the run) made since.
598 events.push({
599 id: yield* allocateEventId(),
600 type: "run.background-work-cancelled",
601 threadId: projection.thread.id,
602 runId: noteRun.id,
603 providerInstanceId: noteRun.providerInstanceId,
604 occurredAt: now,
605 payload: {
606 runId: noteRun.id,
607 restartCancelledBackgroundWork: mergeRestartCancelledBackgroundWork(
608 noteRun.restartCancelledBackgroundWork ?? [],
609 work,
610 ),
611 },
612 });
613 }
614 const continuationRun =
615 continueAfterRestart && trigger === "startup"
616 ? restartContinuationRun(projection, now)
617 : undefined;
618 const effects: Array<EffectOutbox.PendingOrchestrationEffectV2> = continuationRun
619 ? [
620 {
621 id: `effect:restart-continuation:${continuationRun.id}`,
622 commandId,
623 threadId: projection.thread.id,
624 request: { type: "provider-runtime.continue", sourceRunId: continuationRun.id },
625 },
626 ]
627 : [];
628 const stoppedSessions = projection.providerSessions.filter(
629 (candidate) => candidate.status !== "stopped" && candidate.status !== "error",
630 ).length;
631 let retiredEffects: number;
632 if (events.length === 0) {
633 const retiredEffectIds = yield* outbox
634 .cancelUnsettled({
635 threadId: projection.thread.id,
636 effectTypes: EffectOutbox.PROCESS_BOUND_EFFECT_TYPES,
637 reason: detail,
638 })
639 .pipe(
640 Effect.mapError(
641 (cause) =>
642 new ProviderRuntimeRecoveryError({
643 operation: "reconcile",
644 threadId: projection.thread.id,
645 cause: { detail, cause },
646 }),
647 ),
648 );
649 yield* outbox.signalCancellations(retiredEffectIds);
650 retiredEffects = retiredEffectIds.length;
651 } else {
652 const result = yield* eventSink
653 .commitCommand({
654 commandId,
655 threadId: projection.thread.id,
656 commandType: "provider-runtime.reconcile",
657 acceptedAt: now,
658 events,
659 effects,
660 cancelUnsettledEffects: {
661 effectTypes: EffectOutbox.PROCESS_BOUND_EFFECT_TYPES,
662 reason: detail,
663 },
664 })
665 .pipe(
666 Effect.mapError(
667 (cause) =>
668 new ProviderRuntimeRecoveryError({
669 operation: "reconcile",
670 threadId: projection.thread.id,
671 cause,
672 }),
673 ),
674 );
675 retiredEffects = result.cancelledEffectCount;
676 }
677 return {
678 terminalizedRuns: runs.length,
679 stoppedSessions,
680 closedRequests: requests.length,
681 retiredEffects,
682 };
683 },
684 );
685
686 const reconcile = (trigger: "startup" | "shutdown") =>
687 Effect.gen(function* () {
688 const continueAfterRestart = yield* settings.getSettings.pipe(
689 Effect.orElseSucceed(() => null),
690 );
691 const threadIds = yield* projections
692 .getRecoveryThreadIds("runtime")
693 .pipe(
694 Effect.mapError(
695 (cause) => new ProviderRuntimeRecoveryError({ operation: "read-projections", cause }),
696 ),
697 );
698 let terminalizedRuns = 0;
699 let stoppedSessions = 0;
700 let closedRequests = 0;
701 let retiredEffects = 0;
702 for (const threadId of threadIds) {
703 const projection = yield* projections.getRuntimeRecoveryProjection(threadId).pipe(
704 Effect.mapError(
705 (cause) =>
706 new ProviderRuntimeRecoveryError({
707 operation: "read-projections",
708 threadId,
709 cause,
710 }),
711 ),
712 );
713 const enabled =
714 continueAfterRestart !== null &&
715 resolveProjectSettings(continueAfterRestart, projection.thread.projectId).settings
716 .continueThreadsAfterServerUpdate;
717 const result = yield* reconcileProjection(projection, trigger, enabled);
718 terminalizedRuns += result.terminalizedRuns;
719 stoppedSessions += result.stoppedSessions;
720 closedRequests += result.closedRequests;
721 retiredEffects += result.retiredEffects;
722 }
723 const outboxReconciliation = yield* outbox.reconcileAfterProcessLoss.pipe(
724 Effect.mapError(
725 (cause) => new ProviderRuntimeRecoveryError({ operation: "drain-outbox", cause }),
726 ),
727 );
728 return {
729 terminalizedRuns,
730 stoppedSessions,
731 closedRequests,
732 retiredEffects: retiredEffects + outboxReconciliation.cancelled,
733 requeuedEffects: outboxReconciliation.requeued,
734 } satisfies ProviderRuntimeReconciliationSummary;
735 });
736
737 // Snapshot intent only while providers are live. A provider may finish while
738 // this commits; reconciliation reads fresh state after shutdown, and delivery
739 // rejects any source run that actually completed.
740 const prepareForShutdown = Effect.gen(function* () {
741 const enabled = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null));
742 if (!enabled) return;
743 const threadIds = yield* projections.getRecoveryThreadIds("runtime");
744 for (const threadId of threadIds) {
745 yield* Effect.gen(function* () {
746 const projection = yield* projections.getRuntimeRecoveryProjection(threadId);
747 if (
748 !resolveProjectSettings(enabled, projection.thread.projectId).settings
749 .continueThreadsAfterServerUpdate
750 )
751 return;
752 const run = restartContinuationRun(projection, yield* DateTime.now);
753 if (!run) return;
754 const commandId = CommandId.make(`command:restart-prepare:${run.id}`);
755 yield* eventSink.writeWithEffects({
756 commandId,
757 events: [],
758 effects: [
759 {
760 id: `effect:restart-continuation:${run.id}`,
761 commandId,
762 threadId,
763 request: { type: "provider-runtime.continue", sourceRunId: run.id },
764 },
765 ],
766 });
767 }).pipe(
768 // One failing thread must not cost the threads after it their continuation.
769 Effect.catchCauseIf(
770 (cause) => !Cause.hasInterruptsOnly(cause),
771 (cause) =>
772 Effect.logWarning("Failed to prepare a restart continuation", { threadId, cause }),
773 ),
774 );
775 }
776 }).pipe(
777 Effect.mapError((cause) => new ProviderRuntimeRecoveryError({ operation: "reconcile", cause })),
778 );
779
780 const recover = Effect.gen(function* () {
781 return (yield* reconcile("startup")) satisfies ProviderRuntimeRecoverySummary;
782 });
783
784 return ProviderRuntimeRecoveryService.of({ reconcile, prepareForShutdown, recover });
785});
786
787export const layer = Layer.effect(ProviderRuntimeRecoveryService, make);