1import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError";
2import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
3import {
4 CommandId,
5 isProviderNativeSubagentThread,
6 type OrchestrationV2DomainEvent,
7 type ProviderThreadId,
8 type OrchestrationV2RestartCancelledBackgroundWork,
9 type OrchestrationV2Subagent,
10 type OrchestrationV2ThreadProjection,
11 isOrchestrationV2WorkActive,
12 ThreadId,
13} from "@t3tools/contracts";
14import * as Cause from "effect/Cause";
15import * as Context from "effect/Context";
16import * as DateTime from "effect/DateTime";
17import * as Effect from "effect/Effect";
18import * as Layer from "effect/Layer";
19import * as Schema from "effect/Schema";
20
21import * as EffectOutbox from "./EffectOutbox.ts";
22import * as EventSink from "./EventSink.ts";
23import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
24import * as ProjectionStore from "./ProjectionStore.ts";
25import * as ServerSettings from "../serverSettings.ts";
26import { restartContinuationRun } from "./RestartContinuation.ts";
27import { isBufferedProviderNotification } from "./NotificationMailbox.ts";
28import { queuedRunsInDeliveryOrder } from "./QueuedRunOrder.ts";
29import {
30 endOrphanedNativeSubagent,
31 endRunlessRootTurns,
32} from "@t3tools/provider-core/server/subagentProjection";
33import {
34 awaitingProviderThreadId,
35 cancelledRosterTaskWork,
36 cancelledTurnItemWork,
37 mergeRestartCancelledBackgroundWork,
38 neverStarted,
39} from "./RestartBackgroundNote.ts";
40
41export class ProviderRuntimeRecoveryError extends Schema.TaggedError<ProviderRuntimeRecoveryError>()(
42 "ProviderRuntimeRecoveryError",
43 {
44 operation: Schema.Literals(["read-projections", "reconcile", "drain-outbox"]),
45 threadId: Schema.optional(ThreadId),
46 cause: Schema.Defect(),
47 },
48) {
49 override get message(): string {
50 return `Provider runtime recovery failed during ${this.operation}.`;
51 }
52}
53
54const isProviderRuntimeRecoveryError = Schema.is(ProviderRuntimeRecoveryError);
55
56export interface ProviderRuntimeRecoverySummary {
57 readonly terminalizedRuns: number;
58 readonly stoppedSessions: number;
59 readonly closedRequests: number;
60 readonly retiredEffects: number;
61 readonly requeuedEffects: number;
62}
63
64export interface ProviderRuntimeReconciliationSummary {
65 readonly terminalizedRuns: number;
66 readonly stoppedSessions: number;
67 readonly closedRequests: number;
68 readonly retiredEffects: number;
69 readonly requeuedEffects: number;
70}
71
72export class ProviderRuntimeRecoveryService extends Context.Service<
73 ProviderRuntimeRecoveryService,
74 {
75 readonly reconcile: (
76 trigger: "startup" | "shutdown",
77 ) => Effect.Effect<ProviderRuntimeReconciliationSummary, ProviderRuntimeRecoveryError>;
78 readonly prepareForShutdown: Effect.Effect<void, ProviderRuntimeRecoveryError>;
79 readonly recover: Effect.Effect<ProviderRuntimeRecoverySummary, ProviderRuntimeRecoveryError>;
80 }
81>()("t3/orchestration-v2/ProviderRuntimeRecoveryService") {}
82
83function nonterminalRuns(projection: ProjectionStore.ProjectionRuntimeRecoveryState) {
84 return projection.runs.filter((run) => {
85 const status: string = run.status;
86 return (
87 status === "preparing" ||
88 run.status === "starting" ||
89 run.status === "running" ||
90 run.status === "waiting"
91 );
92 });
93}
94
95function isBackgroundCapableTurnItemType(type: string): boolean {
96 return type === "command_execution" || type === "dynamic_tool" || type === "subagent";
97}
98
99function isNonterminalTurnItemStatus(status: string): boolean {
100 return status === "pending" || status === "running" || status === "waiting";
101}
102
103/**
104 * A delegate_task child. Its own thread is reconciled and continued on its own
105 * (here, or in the linked environment it runs in) and reports back through
106 * the app, so it is not provider background work.
107 */
108function isAppOwnedDelegation(task: {
109 readonly origin: OrchestrationV2Subagent["origin"];
110 readonly childThreadId: ThreadId | null;
111 readonly remoteChild?: OrchestrationV2Subagent["remoteChild"];
112}): boolean {
113 return (
114 task.origin === "app_owned" && (task.childThreadId !== null || task.remoteChild !== undefined)
115 );
116}
117
118function isAppOwnedDelegationItem(
119 item: OrchestrationV2ThreadProjection["turnItems"][number],
120): boolean {
121 return item.type === "subagent" && isAppOwnedDelegation(item);
122}
123
124function providerThreadHasPendingBackgroundTasks(
125 providerThread: OrchestrationV2ThreadProjection["providerThreads"][number],
126): boolean {
127 return (providerThread.pendingBackgroundTasks?.length ?? 0) > 0;
128}
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
154const cancelledItem = (
155 item: OrchestrationV2ThreadProjection["turnItems"][number],
156 now: DateTime.Utc,
157): OrchestrationV2ThreadProjection["turnItems"][number] =>
158 item.type === "secret_request" && item.secretStatus === "pending"
159 ? { ...item, status: "cancelled", secretStatus: "cancelled", completedAt: now, updatedAt: now }
160 : {
161 ...item,
162 ...("streaming" in item ? { streaming: false } : {}),
163 status: "cancelled",
164 completedAt: now,
165 updatedAt: now,
166 };
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 !neverStarted(run) &&
178 (latest === undefined || runRanAfter(run, latest))
179 ? run
180 : latest,
181 undefined,
182 );
183}
184
185export const make = Effect.gen(function* () {
186 const settings = yield* ServerSettings.ServerSettingsService;
187 const projections = yield* ProjectionStore.ProjectionStoreV2;
188 const eventSink = yield* EventSink.EventSinkV2;
189 const ids = yield* IdAllocator.IdAllocatorV2;
190 const outbox = yield* EffectOutbox.EffectOutboxV2;
191 type RecoverySession = ProjectionStore.ProjectionRuntimeRecoveryState["providerSessions"][number];
192 /** Shared sessions stop on their first recovered thread; later threads still decide wakes from pre-recovery liveness. */
193 const sessionsLiveAtStart = (
194 projection: ProjectionStore.ProjectionRuntimeRecoveryState,
195 stoppedThisPass: ReadonlyMap<string, RecoverySession>,
196 ) => {
197 const sessions = new Map(projection.providerSessions.map((session) => [session.id, session]));
198 for (const providerThread of projection.providerThreads) {
199 const stopped =
200 providerThread.providerSessionId === null
201 ? undefined
202 : stoppedThisPass.get(providerThread.providerSessionId);
203 if (stopped !== undefined) sessions.set(stopped.id, stopped);
204 }
205 return [...sessions.values()];
206 };
207
208 const reconcileProjection = Effect.fn("ProviderRuntimeRecoveryService.reconcileProjection")(
209 function* (
210 projection: ProjectionStore.ProjectionRuntimeRecoveryState,
211 trigger: "startup" | "shutdown",
212 continueAfterRestart: boolean,
213 stoppedThisPass: ReadonlyMap<string, RecoverySession> = new Map(),
214 ) {
215 const now = yield* DateTime.now;
216 let runs = [] as Array<OrchestrationV2ThreadProjection["runs"][number]>;
217 for (const run of nonterminalRuns(projection)) {
218 if (run.status === "waiting") {
219 const checkpointEffects = yield* outbox
220 .listByCommandId(CommandId.make(`command:effect:checkpoint.capture:${run.id}`))
221 .pipe(
222 Effect.mapError(
223 (cause) =>
224 new ProviderRuntimeRecoveryError({
225 operation: "reconcile",
226 threadId: projection.thread.id,
227 cause,
228 }),
229 ),
230 );
231 const hasReplayableCheckpoint = checkpointEffects.some(
232 (effect) =>
233 effect.request.type === "checkpoint.capture" &&
234 effect.request.runId === run.id &&
235 (effect.status === "pending" || effect.status === "running"),
236 );
237 if (hasReplayableCheckpoint) continue;
238 }
239 runs.push(run);
240 }
241 const pendingRequests = projection.runtimeRequests.filter(
242 (request) => request.status === "pending",
243 );
244 // The lost provider callback cannot resume; persisted questions can be
245 // answered by a new user message in threads that accept messages.
246 const questions = pendingRequests.filter(
247 (request) =>
248 request.kind === "user_input" &&
249 request.responseCapability.type === "live" &&
250 projection.turnItems.some(
251 (item) =>
252 item.type === "user_input_request" &&
253 item.requestId === request.id &&
254 item.nodeId === request.nodeId &&
255 item.questions.length > 0 &&
256 item.questions.every((question) => question.question.trim().length > 0),
257 ) &&
258 !isProviderNativeSubagentThread(projection.thread),
259 );
260 const recoveredRequestIds = new Set(questions.map((request) => request.id));
261 const messageRequestNodeIds = new Set(
262 pendingRequests
263 .filter(
264 (request) =>
265 request.responseCapability.type === "message" || recoveredRequestIds.has(request.id),
266 )
267 .map((request) => request.nodeId),
268 );
269 const requests = pendingRequests.filter(
270 (request) =>
271 request.responseCapability.type !== "message" && !recoveredRequestIds.has(request.id),
272 );
273 // Delegated task rows, items and nodes stay open: the child settles them.
274 const delegatedTaskNodeIds = new Set<string>([
275 ...(projection.subagents ?? []).filter(isAppOwnedDelegation).map((subagent) => subagent.id),
276 ...(projection.turnItems ?? []).flatMap((item) =>
277 item.type === "subagent" && isAppOwnedDelegation(item) ? [item.subagentId] : [],
278 ),
279 ]);
280 const detail = `Cancelled because the server ${trigger === "startup" ? "restarted" : "shut down"} before the provider work completed.`;
281 const commandId = CommandId.make(
282 `command:runtime-reconcile:${trigger}:${projection.thread.id}:${DateTime.formatIso(now)}`,
283 );
284 const allocateEventId = () =>
285 ids.allocate.event({ threadId: projection.thread.id, commandId }).pipe(
286 Effect.mapError(
287 (cause) =>
288 new ProviderRuntimeRecoveryError({
289 operation: "reconcile",
290 threadId: projection.thread.id,
291 cause,
292 }),
293 ),
294 );
295 const events: Array<OrchestrationV2DomainEvent> = [];
296 const cancelledBackgroundWork = new Map<
297 ProviderThreadId,
298 Array<OrchestrationV2RestartCancelledBackgroundWork>
299 >();
300 const cancelledBackgroundNativeIds = new Set<string>();
301 const recordCancelledBackgroundWork = (
302 providerThreadId: ProviderThreadId | null | undefined,
303 work: OrchestrationV2RestartCancelledBackgroundWork,
304 ) => {
305 if (providerThreadId == null) return;
306 const existing = cancelledBackgroundWork.get(providerThreadId);
307 if (existing === undefined) cancelledBackgroundWork.set(providerThreadId, [work]);
308 else existing.push(work);
309 };
310 const recordCancelledBackgroundItem = (
311 item: OrchestrationV2ThreadProjection["turnItems"][number],
312 ) => {
313 if (!isBackgroundCapableTurnItemType(item.type)) return;
314 const awaitingId = awaitingProviderThreadId(item, projection.runs);
315 const ownProviderThread =
316 item.providerThreadId === awaitingId
317 ? undefined
318 : (projection.providerThreads ?? []).find(
319 (candidate) => candidate.id === item.providerThreadId,
320 );
321 const work = cancelledTurnItemWork(
322 item,
323 ownProviderThread?.nativeThreadRef?.nativeId ?? undefined,
324 );
325 if (work === undefined) return;
326 recordCancelledBackgroundWork(awaitingId, work);
327 if (item.nativeItemRef?.nativeId != null) {
328 cancelledBackgroundNativeIds.add(item.nativeItemRef.nativeId);
329 }
330 };
331 const unstartedRuns = runs.filter(
332 (run) =>
333 run.status === "starting" &&
334 neverStarted(run) &&
335 !projection.providerTurns.some((turn) => turn.runAttemptId === run.activeAttemptId) &&
336 !projection.runtimeRequests.some(
337 (request) =>
338 request.status === "pending" &&
339 projection.nodes.some((node) => node.id === request.nodeId && node.runId === run.id),
340 ),
341 );
342 const queuedRuns = projection.runs.filter((run) => run.status === "queued");
343 const pendingRuns = [...unstartedRuns, ...queuedRuns];
344 const queuedMessages =
345 pendingRuns.length === 0
346 ? []
347 : [
348 ...(yield* projections
349 .getMessagesById(
350 projection.thread.id,
351 pendingRuns.map((run) => run.userMessageId),
352 )
353 .pipe(
354 Effect.mapError(
355 (cause) =>
356 new ProviderRuntimeRecoveryError({
357 operation: "read-projections",
358 threadId: projection.thread.id,
359 cause,
360 }),
361 ),
362 )).values(),
363 ];
364 const bufferedMessageIds = new Set(
365 queuedMessages.filter(isBufferedProviderNotification).map((message) => message.id),
366 );
367 const retryableMessageIds = new Set(
368 queuedMessages
369 .filter(
370 (message) =>
371 message.createdBy === "user" &&
372 message.scheduledTaskId === undefined &&
373 message.notification === undefined &&
374 message.delegatedCompletion === undefined,
375 )
376 .map((message) => message.id),
377 );
378 // A saved user message the provider never accepted stays available through the held queue.
379 const retryableRuns = unstartedRuns.filter((run) =>
380 retryableMessageIds.has(run.userMessageId),
381 );
382 const retryableRunIds = new Set(retryableRuns.map((run) => run.id));
383 const retryableItems =
384 retryableRuns.length === 0
385 ? []
386 : (yield* projections
387 .getThreadRecords(projection.thread.id, ["turnItems"], {
388 turnItemTypes: ["user_message"],
389 turnItemRunIds: retryableRuns.map((run) => run.id),
390 })
391 .pipe(
392 Effect.mapError(
393 (cause) =>
394 new ProviderRuntimeRecoveryError({
395 operation: "read-projections",
396 threadId: projection.thread.id,
397 cause,
398 }),
399 ),
400 )).turnItems;
401 for (const item of retryableItems) {
402 if (item.type !== "user_message" || !retryableMessageIds.has(item.messageId)) continue;
403 events.push({
404 id: yield* allocateEventId(),
405 type: "turn-item.updated",
406 threadId: projection.thread.id,
407 ...(item.runId === null ? {} : { runId: item.runId }),
408 occurredAt: now,
409 payload: {
410 ...item,
411 inputIntent: "queued_turn",
412 status: "pending",
413 startedAt: null,
414 completedAt: null,
415 updatedAt: now,
416 },
417 });
418 }
419 runs = runs.filter((run) => !retryableRunIds.has(run.id));
420 const heldRuns = [
421 ...retryableRuns,
422 ...queuedRunsInDeliveryOrder({ runs: queuedRuns, messages: queuedMessages }),
423 ];
424 for (const [index, run] of heldRuns.entries()) {
425 if (bufferedMessageIds.has(run.userMessageId)) {
426 runs.push(run);
427 continue;
428 }
429 if (run.queueHeld === true && retryableRuns.length === 0) continue;
430 events.push({
431 id: yield* allocateEventId(),
432 type: "run.updated",
433 threadId: projection.thread.id,
434 runId: run.id,
435 providerInstanceId: run.providerInstanceId,
436 occurredAt: now,
437 payload: {
438 ...run,
439 status: "queued",
440 queueHeld: true,
441 ...(retryableRuns.length === 0 ? {} : { queuePosition: index + 1 }),
442 },
443 });
444 }
445 const unsentResumeRunIds = runs.flatMap((run) =>
446 run.startedAt === null && run.usageLimitContinuationOfRunId !== undefined ? [run.id] : [],
447 );
448 const limitRecovery = projection.thread.limitRecovery;
449 // The unsent resume spent the worker's request id, so a fresh one lets it send again. An
450 // untagged resume from an older server stays cancelled: its copy couldn't be hidden.
451 if (limitRecovery?.autoResume === true && unsentResumeRunIds.length > 0) {
452 events.push({
453 id: yield* allocateEventId(),
454 type: "thread.metadata-updated",
455 threadId: projection.thread.id,
456 occurredAt: now,
457 payload: {
458 ...projection.thread,
459 limitRecovery: { ...limitRecovery, requestId: commandId },
460 updatedAt: now,
461 },
462 });
463 }
464 // The worker sends the auto-resume again, so the unsent copy leaves the transcript the way
465 // cancelled queued input does.
466 const unsentResumePrompts =
467 unsentResumeRunIds.length === 0
468 ? []
469 : (yield* projections
470 .getThreadRecords(projection.thread.id, ["turnItems"], {
471 turnItemTypes: ["user_message"],
472 turnItemRunIds: unsentResumeRunIds,
473 })
474 .pipe(
475 Effect.mapError(
476 (cause) =>
477 new ProviderRuntimeRecoveryError({
478 operation: "read-projections",
479 threadId: projection.thread.id,
480 cause,
481 }),
482 ),
483 )).turnItems;
484 const unsentResumePromptIds = new Set(unsentResumePrompts.map((item) => item.id));
485 for (const item of unsentResumePrompts) {
486 if (item.type !== "user_message") continue;
487 const active =
488 item.status === "pending" || item.status === "running" || item.status === "waiting";
489 events.push({
490 id: yield* allocateEventId(),
491 type: "turn-item.updated",
492 threadId: projection.thread.id,
493 ...(item.runId === null ? {} : { runId: item.runId }),
494 occurredAt: now,
495 payload: {
496 ...item,
497 ...(active ? { status: "cancelled", completedAt: now } : {}),
498 inputIntent: "queued_turn",
499 updatedAt: now,
500 },
501 });
502 }
503 for (const question of questions) {
504 events.push({
505 id: yield* allocateEventId(),
506 type: "runtime-request.updated",
507 threadId: projection.thread.id,
508 nodeId: question.nodeId,
509 occurredAt: now,
510 payload: { ...question, responseCapability: { type: "message" } },
511 });
512 }
513 for (const request of requests) {
514 events.push({
515 id: yield* allocateEventId(),
516 type: "runtime-request.updated",
517 threadId: projection.thread.id,
518 nodeId: request.nodeId,
519 occurredAt: now,
520 payload: {
521 ...request,
522 status: trigger === "startup" ? "expired" : "cancelled",
523 responseCapability: {
524 type: "not_resumable",
525 reason: `The server ${trigger === "startup" ? "restarted" : "shut down"} before this runtime request was resolved.`,
526 },
527 resolvedAt: now,
528 },
529 });
530 }
531 for (const run of runs) {
532 events.push({
533 id: yield* allocateEventId(),
534 type: "run.updated",
535 threadId: projection.thread.id,
536 runId: run.id,
537 providerInstanceId: run.providerInstanceId,
538 occurredAt: now,
539 payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now },
540 });
541 for (const attempt of projection.attempts.filter(
542 (candidate) =>
543 candidate.runId === run.id &&
544 (candidate.status === "pending" || candidate.status === "running"),
545 )) {
546 events.push({
547 id: yield* allocateEventId(),
548 type: "run-attempt.updated",
549 threadId: projection.thread.id,
550 runId: run.id,
551 nodeId: attempt.rootNodeId,
552 providerInstanceId: run.providerInstanceId,
553 occurredAt: now,
554 payload: { ...attempt, status: "cancelled", completedAt: now },
555 });
556 }
557 for (const node of projection.nodes.filter(
558 (candidate) =>
559 candidate.runId === run.id &&
560 !messageRequestNodeIds.has(candidate.id) &&
561 !delegatedTaskNodeIds.has(candidate.id) &&
562 (candidate.status === "pending" ||
563 candidate.status === "running" ||
564 candidate.status === "waiting"),
565 )) {
566 events.push({
567 id: yield* allocateEventId(),
568 type: "node.updated",
569 threadId: projection.thread.id,
570 runId: run.id,
571 nodeId: node.id,
572 providerInstanceId: run.providerInstanceId,
573 occurredAt: now,
574 payload: { ...node, status: "cancelled", completedAt: now },
575 });
576 }
577 for (const providerTurn of projection.providerTurns.filter(
578 (candidate) =>
579 candidate.runAttemptId !== null &&
580 projection.attempts.some(
581 (attempt) => attempt.id === candidate.runAttemptId && attempt.runId === run.id,
582 ) &&
583 (candidate.status === "pending" || candidate.status === "running"),
584 )) {
585 events.push({
586 id: yield* allocateEventId(),
587 type: "provider-turn.updated",
588 threadId: projection.thread.id,
589 runId: run.id,
590 nodeId: providerTurn.nodeId,
591 providerInstanceId: run.providerInstanceId,
592 occurredAt: now,
593 payload: { ...providerTurn, status: "cancelled", completedAt: now },
594 });
595 }
596 for (const message of projection.messages.filter(
597 (candidate) => candidate.runId === run.id && candidate.streaming,
598 )) {
599 events.push({
600 id: yield* allocateEventId(),
601 type: "message.updated",
602 threadId: projection.thread.id,
603 runId: run.id,
604 ...(message.nodeId === null ? {} : { nodeId: message.nodeId }),
605 providerInstanceId: run.providerInstanceId,
606 occurredAt: now,
607 payload: { ...message, streaming: false, updatedAt: now },
608 });
609 }
610 for (const item of projection.turnItems.filter(
611 (candidate) =>
612 candidate.runId === run.id &&
613 !unsentResumePromptIds.has(candidate.id) &&
614 (candidate.nodeId === null || !messageRequestNodeIds.has(candidate.nodeId)) &&
615 !isAppOwnedDelegationItem(candidate) &&
616 // A link card waits on the user, not the run, and answers by message.
617 candidate.type !== "link_request" &&
618 (candidate.status === "pending" ||
619 candidate.status === "running" ||
620 candidate.status === "waiting"),
621 )) {
622 // Cut native subagents do not resume; record them for the continuation to restart.
623 if (run.status === "waiting") recordCancelledBackgroundItem(item);
624 else if (item.type === "subagent" && item.origin === "provider_native")
625 recordCancelledBackgroundItem(item);
626 events.push({
627 id: yield* allocateEventId(),
628 type: "turn-item.updated",
629 threadId: projection.thread.id,
630 runId: run.id,
631 ...(item.nodeId === null ? {} : { nodeId: item.nodeId }),
632 providerInstanceId: run.providerInstanceId,
633 occurredAt: now,
634 payload: cancelledItem(item, now),
635 });
636 }
637 // Queued runs and unsent resumes leave no visible turn behind, so there is nothing to explain.
638 if (run.status === "queued" || unsentResumeRunIds.includes(run.id)) continue;
639 events.push({
640 id: yield* allocateEventId(),
641 type: "turn-item.updated",
642 threadId: projection.thread.id,
643 runId: run.id,
644 ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }),
645 providerInstanceId: run.providerInstanceId,
646 occurredAt: now,
647 payload: {
648 id: ids.derive.runSignalTurnItem({ runId: run.id, signal: "server-recovery" }),
649 threadId: projection.thread.id,
650 runId: run.id,
651 nodeId: run.rootNodeId,
652 providerThreadId: run.providerThreadId,
653 providerTurnId: null,
654 nativeItemRef: null,
655 parentItemId: null,
656 // EventSink allocates the real position, after the run's existing items.
657 ordinal: 0,
658 status: "completed",
659 title: trigger === "startup" ? "Server restarted" : "Server shut down",
660 startedAt: now,
661 completedAt: now,
662 updatedAt: now,
663 type: "system_notice",
664 message: detail,
665 },
666 });
667 }
668 // Blocking waits die with their turn; let child results wake the parent continuation as async tasks.
669 const cutRunIds = new Set(runs.map((run) => run.id));
670 for (const task of projection.subagents ?? []) {
671 if (
672 task.origin !== "app_owned" ||
673 task.runId === null ||
674 !cutRunIds.has(task.runId) ||
675 task.completionWake !== "settled_only" ||
676 !isOrchestrationV2WorkActive(task.status)
677 )
678 continue;
679 events.push({
680 id: yield* allocateEventId(),
681 type: "subagent.updated",
682 threadId: projection.thread.id,
683 runId: task.runId,
684 nodeId: task.id,
685 driver: task.driver,
686 providerInstanceId: task.providerInstanceId,
687 occurredAt: now,
688 payload: { ...task, completionWake: "always", updatedAt: now },
689 });
690 }
691 const recoveredNonterminalRunIds = new Set(runs.map((run) => run.id));
692 const cancelledStaleNodeIds = new Set<string>();
693 for (const item of projection.turnItems ?? []) {
694 if (item.runId !== null && recoveredNonterminalRunIds.has(item.runId)) {
695 continue;
696 }
697 if (!isBackgroundCapableTurnItemType(item.type)) {
698 continue;
699 }
700 if (!isNonterminalTurnItemStatus(item.status) || isAppOwnedDelegationItem(item)) {
701 continue;
702 }
703 const providerInstanceId = resolveStaleBackgroundItemProviderInstanceId(item, projection);
704 recordCancelledBackgroundItem(item);
705 events.push({
706 id: yield* allocateEventId(),
707 type: "turn-item.updated",
708 threadId: projection.thread.id,
709 ...(item.runId === null ? {} : { runId: item.runId }),
710 ...(item.nodeId === null || item.nodeId === undefined ? {} : { nodeId: item.nodeId }),
711 providerInstanceId,
712 occurredAt: now,
713 payload: cancelledItem(item, now),
714 });
715 if (item.nodeId !== null && item.nodeId !== undefined) {
716 const staleItemNode = projection.nodes.find(
717 (candidate) =>
718 candidate.id === item.nodeId && isNonterminalTurnItemStatus(candidate.status),
719 );
720 if (staleItemNode !== undefined && !cancelledStaleNodeIds.has(staleItemNode.id)) {
721 cancelledStaleNodeIds.add(staleItemNode.id);
722 events.push({
723 id: yield* allocateEventId(),
724 type: "node.updated",
725 threadId: projection.thread.id,
726 ...(item.runId === null ? {} : { runId: item.runId }),
727 nodeId: staleItemNode.id,
728 providerInstanceId,
729 occurredAt: now,
730 payload: { ...staleItemNode, status: "cancelled", completedAt: now },
731 });
732 }
733 }
734 }
735 // No provider process survives the server; open native subagents can never report completion.
736 for (const subagent of projection.subagents ?? []) {
737 events.push(
738 ...(yield* endOrphanedNativeSubagent({
739 projection,
740 subagentId: subagent.id,
741 status: "cancelled",
742 now,
743 emitted: events,
744 allocateEventId,
745 })),
746 );
747 }
748 // Native subagent threads have no runs to settle their runless root turns.
749 events.push(
750 ...(yield* endRunlessRootTurns({
751 threadId: projection.thread.id,
752 providerInstanceId: projection.thread.providerInstanceId,
753 projection,
754 status: "cancelled",
755 now,
756 emitted: events,
757 allocateEventId,
758 })),
759 );
760 for (const providerThread of projection.providerThreads ?? []) {
761 const needsIdle = providerThread.status === "active";
762 const needsRosterClear = providerThreadHasPendingBackgroundTasks(providerThread);
763 if (!needsIdle && !needsRosterClear) {
764 continue;
765 }
766 if (providerThread.ownerNodeId === null) {
767 for (const task of providerThread.pendingBackgroundTasks ?? []) {
768 // A subagent already recorded covers the shells it started.
769 if (
770 cancelledBackgroundNativeIds.has(task.taskId) ||
771 (task.parentTaskId !== undefined &&
772 cancelledBackgroundNativeIds.has(task.parentTaskId))
773 )
774 continue;
775 cancelledBackgroundNativeIds.add(task.taskId);
776 recordCancelledBackgroundWork(providerThread.id, cancelledRosterTaskWork(task));
777 }
778 }
779 events.push({
780 id: yield* allocateEventId(),
781 type: "provider-thread.updated",
782 threadId: projection.thread.id,
783 driver: providerThread.driver,
784 providerInstanceId: providerThread.providerInstanceId,
785 occurredAt: now,
786 payload: {
787 ...providerThread,
788 status: needsIdle ? "idle" : providerThread.status,
789 pendingBackgroundTasks: [],
790 updatedAt: now,
791 },
792 });
793 }
794 for (const session of projection.providerSessions.filter(
795 (candidate) =>
796 (candidate.status !== "stopped" && candidate.status !== "error") ||
797 (candidate.status === "error" && (candidate.extensionStatuses?.length ?? 0) > 0),
798 )) {
799 events.push({
800 id: yield* allocateEventId(),
801 type: "provider-session.updated",
802 threadId: projection.thread.id,
803 driver: session.driver,
804 providerInstanceId: session.providerInstanceId,
805 occurredAt: now,
806 payload: {
807 ...session,
808 status: session.status === "error" ? "error" : "stopped",
809 ...(session.extensionStatuses === undefined ? {} : { extensionStatuses: [] }),
810 updatedAt: now,
811 lastError: session.status === "error" ? session.lastError : null,
812 },
813 });
814 }
815 for (const [providerThreadId, work] of cancelledBackgroundWork) {
816 const noteRun = latestStartedRun(projection, providerThreadId);
817 if (noteRun === undefined) continue;
818 // Use a separate event; stale run snapshots can overwrite intervening lifecycle changes.
819 events.push({
820 id: yield* allocateEventId(),
821 type: "run.background-work-cancelled",
822 threadId: projection.thread.id,
823 runId: noteRun.id,
824 providerInstanceId: noteRun.providerInstanceId,
825 occurredAt: now,
826 payload: {
827 runId: noteRun.id,
828 restartCancelledBackgroundWork: mergeRestartCancelledBackgroundWork(
829 noteRun.restartCancelledBackgroundWork ?? [],
830 work,
831 ),
832 },
833 });
834 }
835 const continuationRun =
836 continueAfterRestart && trigger === "startup"
837 ? restartContinuationRun(
838 { ...projection, providerSessions: sessionsLiveAtStart(projection, stoppedThisPass) },
839 now,
840 )
841 : undefined;
842 const effects: Array<EffectOutbox.PendingOrchestrationEffectV2> = continuationRun
843 ? [
844 {
845 id: `effect:restart-continuation:${continuationRun.id}`,
846 commandId,
847 threadId: projection.thread.id,
848 request: { type: "provider-runtime.continue", sourceRunId: continuationRun.id },
849 },
850 ]
851 : [];
852 const stoppedSessions = projection.providerSessions.filter(
853 (candidate) => candidate.status !== "stopped" && candidate.status !== "error",
854 ).length;
855 let retiredEffects: number;
856 let cancelledEffectIds: ReadonlyArray<string> = [];
857 if (events.length === 0) {
858 const retiredEffectIds = yield* outbox
859 .cancelUnsettled({
860 threadId: projection.thread.id,
861 effectTypes: EffectOutbox.PROCESS_BOUND_EFFECT_TYPES,
862 reason: detail,
863 })
864 .pipe(
865 Effect.mapError(
866 (cause) =>
867 new ProviderRuntimeRecoveryError({
868 operation: "reconcile",
869 threadId: projection.thread.id,
870 cause: { detail, cause },
871 }),
872 ),
873 );
874 cancelledEffectIds = retiredEffectIds;
875 retiredEffects = retiredEffectIds.length;
876 } else {
877 const result = yield* eventSink
878 .commitCommand({
879 commandId,
880 threadId: projection.thread.id,
881 commandType: "provider-runtime.reconcile",
882 acceptedAt: now,
883 events,
884 effects,
885 cancelUnsettledEffects: {
886 effectTypes: EffectOutbox.PROCESS_BOUND_EFFECT_TYPES,
887 reason: detail,
888 },
889 })
890 .pipe(
891 Effect.mapError(
892 (cause) =>
893 new ProviderRuntimeRecoveryError({
894 operation: "reconcile",
895 threadId: projection.thread.id,
896 cause,
897 }),
898 ),
899 );
900 retiredEffects = result.cancelledEffectCount;
901 }
902 return {
903 terminalizedRuns: runs.length,
904 stoppedSessions,
905 closedRequests: requests.length,
906 retiredEffects,
907 cancelledEffectIds,
908 };
909 },
910 );
911
912 const reconcile = Effect.fn("ProviderRuntimeRecoveryService.reconcile")(function* (
913 trigger: "startup" | "shutdown",
914 ) {
915 const continueAfterRestart = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null));
916 const threadIds = yield* projections
917 .getRecoveryThreadIds("runtime")
918 .pipe(
919 Effect.mapError(
920 (cause) => new ProviderRuntimeRecoveryError({ operation: "read-projections", cause }),
921 ),
922 );
923 let terminalizedRuns = 0;
924 let stoppedSessions = 0;
925 let closedRequests = 0;
926 let retiredEffects = 0;
927 const stoppedThisPass = new Map<string, RecoverySession>();
928 const grouped = trigger === "shutdown" && threadIds.length > 1;
929 const groupSize = grouped ? 32 : 1;
930 for (let offset = 0; offset < threadIds.length; offset += groupSize) {
931 const cancelledEffectIds: string[] = [];
932 // Read and decide on the writer so shared sessions see earlier updates in this group.
933 const reconcileGroup = Effect.gen(function* () {
934 for (const threadId of threadIds.slice(offset, offset + groupSize)) {
935 const projection = yield* projections.getRuntimeRecoveryProjection(threadId).pipe(
936 Effect.mapError(
937 (cause) =>
938 new ProviderRuntimeRecoveryError({
939 operation: "read-projections",
940 threadId,
941 cause,
942 }),
943 ),
944 );
945 const enabled =
946 continueAfterRestart !== null &&
947 resolveProjectSettings(continueAfterRestart, projection.thread.projectId).settings
948 .continueThreadsAfterServerUpdate;
949 const result = yield* reconcileProjection(projection, trigger, enabled, stoppedThisPass);
950 for (const session of projection.providerSessions) {
951 if (session.status !== "stopped" && session.status !== "error") {
952 stoppedThisPass.set(session.id, session);
953 }
954 }
955 cancelledEffectIds.push(...result.cancelledEffectIds);
956 terminalizedRuns += result.terminalizedRuns;
957 stoppedSessions += result.stoppedSessions;
958 closedRequests += result.closedRequests;
959 retiredEffects += result.retiredEffects;
960 }
961 });
962 yield* (grouped ? eventSink.withProviderBatch(reconcileGroup) : reconcileGroup).pipe(
963 Effect.mapError((cause) =>
964 isProviderRuntimeRecoveryError(cause)
965 ? cause
966 : new ProviderRuntimeRecoveryError({ operation: "reconcile", cause }),
967 ),
968 );
969 // Eventless outbox cancellations also wait for the group's durable commit.
970 if (cancelledEffectIds.length > 0) yield* outbox.signalCancellations(cancelledEffectIds);
971 }
972 const outboxReconciliation = yield* outbox.reconcileAfterProcessLoss.pipe(
973 Effect.mapError(
974 (cause) => new ProviderRuntimeRecoveryError({ operation: "drain-outbox", cause }),
975 ),
976 );
977 return {
978 terminalizedRuns,
979 stoppedSessions,
980 closedRequests,
981 retiredEffects: retiredEffects + outboxReconciliation.cancelled,
982 requeuedEffects: outboxReconciliation.requeued,
983 } satisfies ProviderRuntimeReconciliationSummary;
984 });
985
986 const prepareForShutdown = Effect.gen(function* () {
987 const enabled = yield* settings.getSettings.pipe(Effect.orElseSucceed(() => null));
988 if (!enabled) return;
989 const enabledProjects = Object.entries(enabled.projectSettingsOverrides)
990 .filter(([, overrides]) => overrides.continueThreadsAfterServerUpdate === true)
991 .map(([id]) => id);
992 if (!enabled.continueThreadsAfterServerUpdate && enabledProjects.length === 0) return;
993 const disabledProjects = Object.entries(enabled.projectSettingsOverrides)
994 .filter(([, overrides]) => overrides.continueThreadsAfterServerUpdate === false)
995 .map(([id]) => id);
996 const threadIds = yield* projections.getRecoveryThreadIds(
997 "restart-continuations",
998 enabled.continueThreadsAfterServerUpdate
999 ? { exclude: disabledProjects }
1000 : { include: enabledProjects },
1001 );
1002 yield* Effect.forEach(
1003 threadIds,
1004 (threadId) =>
1005 Effect.gen(function* () {
1006 const projection = yield* projections.getRuntimeRecoveryProjection(threadId);
1007 if (
1008 !resolveProjectSettings(enabled, projection.thread.projectId).settings
1009 .continueThreadsAfterServerUpdate
1010 )
1011 return;
1012 const run = restartContinuationRun(projection, yield* DateTime.now);
1013 if (!run) return;
1014 const commandId = CommandId.make(`command:restart-prepare:${run.id}`);
1015 yield* eventSink.writeWithEffects({
1016 commandId,
1017 events: [],
1018 effects: [
1019 {
1020 id: `effect:restart-continuation:${run.id}`,
1021 commandId,
1022 threadId,
1023 request: { type: "provider-runtime.continue", sourceRunId: run.id },
1024 },
1025 ],
1026 });
1027 }).pipe(
1028 Effect.catchCauseIf(
1029 (cause) => !Cause.hasInterruptsOnly(cause),
1030 (cause) =>
1031 Effect.logWarning("Failed to prepare a restart continuation", { threadId, cause }),
1032 ),
1033 ),
1034 // Leave room in the four-reader pool for other shutdown reads.
1035 { concurrency: 2, discard: true },
1036 );
1037 }).pipe(
1038 Effect.withSpan("ProviderRuntimeRecoveryService.prepareForShutdown"),
1039 Effect.mapError((cause) => new ProviderRuntimeRecoveryError({ operation: "reconcile", cause })),
1040 );
1041
1042 const recover = Effect.gen(function* () {
1043 return (yield* reconcile("startup")) satisfies ProviderRuntimeRecoverySummary;
1044 });
1045
1046 return ProviderRuntimeRecoveryService.of({ reconcile, prepareForShutdown, recover });
1047});
1048
1049export const layer = Layer.effect(ProviderRuntimeRecoveryService, make);