1import { CommandId } from "@t3tools/contracts";
2import * as Cause from "effect/Cause";
3import * as Context from "effect/Context";
4import * as DateTime from "effect/DateTime";
5import * as Duration from "effect/Duration";
6import * as Effect from "effect/Effect";
7import * as Exit from "effect/Exit";
8import * as Fiber from "effect/Fiber";
9import * as Layer from "effect/Layer";
10import * as Metric from "effect/Metric";
11import * as Option from "effect/Option";
12import * as Schema from "effect/Schema";
13
14import {
15 increment,
16 metricAttributes,
17 orchestrationEffectClaimsTotal,
18 orchestrationEffectQueueWait,
19} from "../observability/Metrics.ts";
20import * as RunFinalizationService from "./RunFinalizationService.ts";
21import * as ResourceCleanupService from "./ResourceCleanupService.ts";
22import * as EffectOutbox from "./EffectOutbox.ts";
23import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
24import * as ProviderSessionManager from "./ProviderSessionManager.ts";
25import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
26import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
27import * as RuntimeRequestService from "./RuntimeRequestService.ts";
28import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
29import * as ThreadManagementService from "./ThreadManagementService.ts";
30import * as ServerSettings from "../serverSettings.ts";
31import { continueRestartedRun } from "./RestartContinuation.ts";
32
33const isProviderTurnControlError = Schema.is(ProviderTurnControlService.ProviderTurnControlError);
34
35export class OrchestrationEffectExecutionError extends Schema.TaggedError<OrchestrationEffectExecutionError>()(
36 "OrchestrationEffectExecutionError",
37 {
38 effectId: Schema.String,
39 effectType: Schema.String,
40 cause: Schema.optional(Schema.Defect()),
41 },
42) {}
43
44/**
45 * Do not apply this to `provider-turn.restart`: that compound effect also runs
46 * detach and start. Swallowing a start failure that happens to mention
47 * "is not active" would drop the outbox item without ever starting the
48 * replacement turn.
49 */
50function isNonRetryableProviderTurnControlFailure(effectType: string, errorText: string): boolean {
51 if (effectType !== "provider-turn.interrupt") {
52 return false;
53 }
54 return (
55 /is not active/i.test(errorText) ||
56 /hard teardown is already in progress/i.test(errorText) ||
57 /treating as already interrupted/i.test(errorText) ||
58 /treating as already stopped/i.test(errorText)
59 );
60}
61
62export interface OrchestrationEffectExecutorV2Shape {
63 readonly execute: (
64 effect: EffectOutbox.OrchestrationEffectV2,
65 options?: { readonly willRetry: boolean },
66 ) => Effect.Effect<void, OrchestrationEffectExecutionError>;
67}
68
69export class OrchestrationEffectExecutorV2 extends Context.Service<
70 OrchestrationEffectExecutorV2,
71 OrchestrationEffectExecutorV2Shape
72>()("t3/orchestration-v2/EffectWorker/OrchestrationEffectExecutorV2") {}
73
74export const layerExecutor: Layer.Layer<
75 OrchestrationEffectExecutorV2,
76 never,
77 | ProviderSessionManager.ProviderSessionManagerV2
78 | RunFinalizationService.RunFinalizationService
79 | CheckpointRollbackService.CheckpointRollbackServiceV2
80 | ProviderTurnControlService.ProviderTurnControlServiceV2
81 | ProviderTurnStartService.ProviderTurnStartServiceV2
82 | RuntimeRequestService.RuntimeRequestServiceV2
83 | ThreadTitleRegenerationService.ThreadTitleRegenerationService
84 | ThreadManagementService.ThreadManagementService
85 | ServerSettings.ServerSettingsService
86> = Layer.effect(
87 OrchestrationEffectExecutorV2,
88 Effect.gen(function* () {
89 const runFinalization = yield* RunFinalizationService.RunFinalizationService;
90 const resourceCleanup = yield* ResourceCleanupService.ResourceCleanupService;
91 const checkpointRollback = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
92 const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
93 const providerTurnControl = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
94 const providerTurnStart = yield* ProviderTurnStartService.ProviderTurnStartServiceV2;
95 const runtimeRequests = yield* RuntimeRequestService.RuntimeRequestServiceV2;
96 const threadTitleRegeneration =
97 yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService;
98 const threads = yield* ThreadManagementService.ThreadManagementService;
99 const settings = yield* ServerSettings.ServerSettingsService;
100 return OrchestrationEffectExecutorV2.of({
101 execute: Effect.fn("OrchestrationEffectExecutorV2.execute")(function (effect, options) {
102 const willRetry = options?.willRetry ?? false;
103 switch (effect.request.type) {
104 case "provider-runtime.continue": {
105 const sourceRunId = effect.request.sourceRunId;
106 return continueRestartedRun({ threadId: effect.threadId, sourceRunId }).pipe(
107 Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
108 Effect.provideService(ServerSettings.ServerSettingsService, settings),
109 Effect.tapError(() =>
110 willRetry
111 ? Effect.void
112 : threads.recoverDelegatedTask(effect.threadId, { sourceRunId }),
113 ),
114 Effect.mapError(
115 (cause) =>
116 new OrchestrationEffectExecutionError({
117 effectId: effect.id,
118 effectType: effect.request.type,
119 cause,
120 }),
121 ),
122 );
123 }
124 case "provider-session.detach":
125 return providerSessions
126 .detach({
127 providerSessionId: effect.request.providerSessionId,
128 threadId: effect.threadId,
129 ...(effect.request.detail === undefined ? {} : { detail: effect.request.detail }),
130 ...(effect.request.revokeMcpCredential === undefined
131 ? {}
132 : { revokeMcpCredential: effect.request.revokeMcpCredential }),
133 })
134 .pipe(
135 Effect.mapError(
136 (cause) =>
137 new OrchestrationEffectExecutionError({
138 effectId: effect.id,
139 effectType: effect.request.type,
140 cause,
141 }),
142 ),
143 );
144 case "provider-session.apply-model-selection":
145 return providerTurnControl
146 .applyModelSelection({
147 threadId: effect.threadId,
148 providerSessionId: effect.request.providerSessionId,
149 providerThreadId: effect.request.providerThreadId,
150 modelSelection: effect.request.modelSelection,
151 })
152 .pipe(
153 Effect.mapError(
154 (cause) =>
155 new OrchestrationEffectExecutionError({
156 effectId: effect.id,
157 effectType: effect.request.type,
158 cause,
159 }),
160 ),
161 );
162 case "provider-turn.start":
163 return providerTurnStart
164 .start({ threadId: effect.threadId, runId: effect.request.runId, willRetry })
165 .pipe(
166 Effect.mapError(
167 (cause) =>
168 new OrchestrationEffectExecutionError({
169 effectId: effect.id,
170 effectType: effect.request.type,
171 cause,
172 }),
173 ),
174 );
175 case "provider-turn.interrupt": {
176 const request = effect.request;
177 return providerTurnControl
178 .interrupt({
179 threadId: effect.threadId,
180 providerSessionId: request.providerSessionId,
181 providerThreadId: request.providerThreadId,
182 providerTurnId: request.providerTurnId,
183 ...(request.subagent === undefined
184 ? {}
185 : {
186 subagent: {
187 ...request.subagent,
188 requestedAt: DateTime.makeUnsafe(effect.createdAt),
189 },
190 }),
191 })
192 .pipe(
193 Effect.tapError((cause) =>
194 request.subagent === undefined
195 ? Effect.void
196 : Effect.logWarning("orchestration-v2.subagent-stop-failed", {
197 effectId: effect.id,
198 cause,
199 }),
200 ),
201 Effect.map((outcome) => ({ unresponsive: false, outcome }) as const),
202 // Settling here would end the run while the agent may still be running; the
203 // user decides with Force stop instead.
204 Effect.catchIf(
205 (cause) => request.subagent === undefined && cause.unresponsive === true,
206 () => Effect.succeed({ unresponsive: true, outcome: undefined } as const),
207 ),
208 Effect.catchIf(
209 (cause) =>
210 request.subagent === undefined &&
211 isNonRetryableProviderTurnControlFailure(
212 request.type,
213 Cause.pretty(Cause.fail(cause)),
214 ),
215 () => Effect.succeed({ unresponsive: false, outcome: undefined } as const),
216 ),
217 Effect.andThen(({ unresponsive, outcome }) =>
218 unresponsive
219 ? threads.dispatch({
220 type: "thread.interrupt.unresponsive",
221 commandId: CommandId.make(`${effect.id}:unresponsive`),
222 threadId: effect.threadId,
223 providerThreadId: request.providerThreadId,
224 providerTurnId: request.providerTurnId,
225 })
226 : request.subagent !== undefined && outcome === undefined
227 ? Effect.void
228 : threads.dispatch({
229 type: "thread.background-work.settle",
230 commandId: CommandId.make(`${effect.id}:background-work-settled`),
231 ...(request.subagent === undefined || outcome === undefined
232 ? {}
233 : {
234 subagent: {
235 id: request.subagent.id,
236 stoppedAt: DateTime.makeUnsafe(effect.createdAt),
237 outcome,
238 },
239 }),
240 threadId: effect.threadId,
241 providerThreadId: request.providerThreadId,
242 providerTurnId: request.providerTurnId,
243 }),
244 ),
245 Effect.mapError(
246 (cause) =>
247 new OrchestrationEffectExecutionError({
248 effectId: effect.id,
249 effectType: request.type,
250 cause,
251 }),
252 ),
253 );
254 }
255 case "provider-turn.steer":
256 return providerTurnControl
257 .steer({
258 threadId: effect.threadId,
259 providerSessionId: effect.request.providerSessionId,
260 providerThreadId: effect.request.providerThreadId,
261 providerTurnId: effect.request.providerTurnId,
262 messageId: effect.request.messageId,
263 ...(effect.request.modelSelection === undefined
264 ? {}
265 : { modelSelection: effect.request.modelSelection }),
266 })
267 .pipe(
268 Effect.tap(() =>
269 Effect.gen(function* () {
270 if (effect.request.type !== "provider-turn.steer") return;
271 const messageId = effect.request.messageId;
272 const projection = yield* threads.getThreadRecords(
273 effect.threadId,
274 ["messages", "runs"],
275 { messageIds: [effect.request.messageId] },
276 );
277 const message = projection.messages.find((row) => row.id === messageId);
278 if (message?.delegatedCompletion === undefined) return;
279 yield* threads.dispatch({
280 type: "notification.delivery.accept",
281 commandId: CommandId.make(`command:mailbox-accepted:${effect.id}`),
282 threadId: effect.threadId,
283 messageId: message.id,
284 });
285 }),
286 ),
287 Effect.catch((error) =>
288 Effect.gen(function* () {
289 if (
290 !isProviderTurnControlError(error) ||
291 (!error.turnCompleted && !error.steerDeclined) ||
292 effect.request.type !== "provider-turn.steer"
293 ) {
294 return yield* error;
295 }
296 const projection = yield* threads.getThreadRecords(
297 effect.threadId,
298 ["messages", "runs"],
299 { messageIds: [effect.request.messageId] },
300 );
301 const messageId = effect.request.messageId;
302 const message = projection.messages.find((item) => item.id === messageId);
303 const run = projection.runs.find((item) => item.id === message?.runId);
304 if (message === undefined || run === undefined) return yield* error;
305 // Notifications must be queued messages.
306 const queued =
307 message.delegatedCompletion !== undefined ||
308 message.notification !== undefined ||
309 error.steerDeclined === true;
310 // Reuse message ids and receipts so retries cannot duplicate messages or follow-ups.
311 yield* threads
312 .dispatch({
313 type: "message.dispatch",
314 commandId: CommandId.make(`command:steer-follow-up:${effect.id}`),
315 threadId: effect.threadId,
316 messageId: message.id,
317 ...(error.steerDeclined ? { steerFollowUpOfRunId: run.id } : {}),
318 ...(error.steerDeclined &&
319 effect.request.steerFollowUpAfterRunOrdinal !== undefined
320 ? {
321 steerFollowUpAfterRunOrdinal:
322 effect.request.steerFollowUpAfterRunOrdinal,
323 }
324 : {}),
325 text: message.text,
326 ...(message.context ? { context: message.context } : {}),
327 attachments: message.attachments,
328 ...(error.steerDeclined && effect.request.modelSelection !== undefined
329 ? { modelSelection: effect.request.modelSelection }
330 : message.delegatedCompletion === undefined
331 ? {}
332 : { modelSelection: run.modelSelection }),
333 dispatchMode: {
334 type: queued ? "queue_after_active" : "start_immediately",
335 },
336 createdBy: message.createdBy,
337 creationSource: message.creationSource,
338 ...(message.delegatedCompletion === undefined
339 ? {}
340 : { delegatedCompletion: message.delegatedCompletion }),
341 ...(message.notification === undefined
342 ? {}
343 : { notification: message.notification }),
344 ...(message.scheduledTaskId === undefined
345 ? {}
346 : { scheduledTaskId: message.scheduledTaskId }),
347 ...(message.senderThreadId === undefined
348 ? {}
349 : { senderThreadId: message.senderThreadId }),
350 ...(message.linkedSender === undefined
351 ? {}
352 : { linkedSender: message.linkedSender }),
353 })
354 .pipe(
355 // A disabled instance cannot retry this completed turn; its result stands.
356 Effect.catchTags({
357 ThreadManagementProviderInstanceDisabledError: (disabledError) =>
358 Effect.logInfo(
359 "orchestration-v2.effect-worker.steer-follow-up-disabled",
360 {
361 threadId: effect.threadId,
362 messageId: message.id,
363 projectId: disabledError.projectId,
364 instanceId: disabledError.instanceId,
365 },
366 ),
367 }),
368 );
369 }),
370 ),
371 Effect.mapError(
372 (cause) =>
373 new OrchestrationEffectExecutionError({
374 effectId: effect.id,
375 effectType: effect.request.type,
376 cause,
377 }),
378 ),
379 );
380 case "provider-turn.restart":
381 return providerTurnControl
382 .interruptAndAwaitTerminal({
383 threadId: effect.threadId,
384 providerSessionId: effect.request.providerSessionId,
385 providerThreadId: effect.request.providerThreadId,
386 providerTurnId: effect.request.providerTurnId,
387 interruptedAttemptId: effect.request.interruptedAttemptId,
388 ...(effect.request.sessionTransition?.type === "replace"
389 ? {
390 replacementProviderSessionId:
391 effect.request.sessionTransition.replacementProviderSessionId,
392 }
393 : {}),
394 })
395 .pipe(
396 Effect.andThen(
397 effect.request.sessionTransition?.type === "replace"
398 ? providerSessions.detach({
399 providerSessionId: effect.request.providerSessionId,
400 threadId: effect.threadId,
401 detail: "Selection change requires a provider session restart.",
402 })
403 : effect.request.sessionTransition?.type === "detach"
404 ? providerSessions.detach({
405 providerSessionId: effect.request.providerSessionId,
406 threadId: effect.threadId,
407 detail: "Provider thread handoff replaced this session binding.",
408 })
409 : Effect.void,
410 ),
411 Effect.andThen(
412 providerTurnStart.start({
413 threadId: effect.threadId,
414 runId: effect.request.runId,
415 willRetry,
416 }),
417 ),
418 Effect.mapError(
419 (cause) =>
420 new OrchestrationEffectExecutionError({
421 effectId: effect.id,
422 effectType: effect.request.type,
423 cause,
424 }),
425 ),
426 );
427 case "runtime-request.respond": {
428 const requestId = effect.request.requestId;
429 return runtimeRequests
430 .respond({
431 threadId: effect.threadId,
432 providerSessionId: effect.request.providerSessionId,
433 requestId: effect.request.requestId,
434 ...(effect.request.decision === undefined
435 ? {}
436 : { decision: effect.request.decision }),
437 ...(effect.request.feedback === undefined
438 ? {}
439 : { feedback: effect.request.feedback }),
440 ...(effect.request.answers === undefined
441 ? {}
442 : { answers: effect.request.answers }),
443 })
444 .pipe(
445 Effect.tapCause((cause) =>
446 willRetry || Cause.hasInterruptsOnly(cause)
447 ? Effect.void
448 : threads
449 .dispatch({
450 type: "runtime-request.delivery.fail",
451 commandId: CommandId.make(`${effect.commandId}:delivery-failed`),
452 threadId: effect.threadId,
453 requestId,
454 })
455 .pipe(
456 Effect.catchCause((recordCause) =>
457 Effect.logWarning("Failed to record answer delivery failure", {
458 effectId: effect.id,
459 cause: recordCause,
460 }),
461 ),
462 ),
463 ),
464 Effect.mapError(
465 (cause) =>
466 new OrchestrationEffectExecutionError({
467 effectId: effect.id,
468 effectType: effect.request.type,
469 cause,
470 }),
471 ),
472 );
473 }
474 case "provider-thread.rollback":
475 return checkpointRollback
476 .execute({
477 threadId: effect.threadId,
478 providerThreadId: effect.request.providerThreadId,
479 checkpointId: effect.request.checkpointId,
480 scopeId: effect.request.scopeId,
481 ...(effect.request.restoreFiles === undefined
482 ? {}
483 : { restoreFiles: effect.request.restoreFiles }),
484 })
485 .pipe(
486 Effect.tapCause((cause) =>
487 willRetry || Cause.hasInterruptsOnly(cause)
488 ? Effect.void
489 : threads
490 .dispatch({
491 type: "checkpoint.rollback.fail",
492 commandId: CommandId.make(`${effect.commandId}:rollback-failed`),
493 threadId: effect.threadId,
494 requestId: effect.commandId,
495 message: Option.match(Cause.findErrorOption(cause), {
496 onNone: () => CheckpointRollbackService.ROLLBACK_FAILED_MESSAGE,
497 onSome: CheckpointRollbackService.rollbackFailureClientMessage,
498 }),
499 })
500 .pipe(
501 Effect.catchCause((recordCause) =>
502 Effect.logWarning("Failed to record rollback failure", {
503 effectId: effect.id,
504 cause: recordCause,
505 }),
506 ),
507 ),
508 ),
509 Effect.mapError(
510 (cause) =>
511 new OrchestrationEffectExecutionError({
512 effectId: effect.id,
513 effectType: effect.request.type,
514 cause,
515 }),
516 ),
517 );
518 case "checkpoint.capture":
519 return runFinalization
520 .finalize({
521 threadId: effect.threadId,
522 runId: effect.request.runId,
523 scopeId: effect.request.scopeId,
524 })
525 .pipe(
526 // Completion bookkeeping must survive a missed live run update, including capture retries.
527 Effect.andThen(
528 threads.recoverDelegatedTask(effect.threadId, {
529 completedRunId: effect.request.runId,
530 }),
531 ),
532 Effect.mapError(
533 (cause) =>
534 new OrchestrationEffectExecutionError({
535 effectId: effect.id,
536 effectType: effect.request.type,
537 cause,
538 }),
539 ),
540 );
541 case "scheduled-tasks.cleanup":
542 return resourceCleanup.cleanupScheduledTasks(effect.threadId).pipe(
543 Effect.mapError(
544 (cause) =>
545 new OrchestrationEffectExecutionError({
546 effectId: effect.id,
547 effectType: effect.request.type,
548 cause,
549 }),
550 ),
551 );
552 // Archive and delete queue this one per-thread teardown, so it also releases the
553 // thread's device claims; the devices stay booted for other threads.
554 case "terminal.cleanup":
555 return Effect.partition(
556 [
557 resourceCleanup.cleanupTerminals(effect.threadId),
558 resourceCleanup.cleanupDevices(effect.threadId),
559 ],
560 (cleanup) => cleanup,
561 ).pipe(
562 Effect.flatMap(([, [failure]]) =>
563 failure === undefined ? Effect.void : Effect.fail(failure),
564 ),
565 Effect.mapError(
566 (cause) =>
567 new OrchestrationEffectExecutionError({
568 effectId: effect.id,
569 effectType: effect.request.type,
570 cause,
571 }),
572 ),
573 );
574 case "preview.cleanup":
575 return resourceCleanup.cleanupPreviews(effect.threadId).pipe(
576 Effect.mapError(
577 (cause) =>
578 new OrchestrationEffectExecutionError({
579 effectId: effect.id,
580 effectType: effect.request.type,
581 cause,
582 }),
583 ),
584 );
585 case "attachment.cleanup":
586 return resourceCleanup.cleanupAttachments(effect.request.attachmentIds).pipe(
587 Effect.mapError(
588 (cause) =>
589 new OrchestrationEffectExecutionError({
590 effectId: effect.id,
591 effectType: effect.request.type,
592 cause,
593 }),
594 ),
595 );
596 case "thread-title.generate":
597 return threadTitleRegeneration
598 .execute({
599 threadId: effect.threadId,
600 requestId: effect.commandId,
601 kind: effect.request.kind,
602 })
603 .pipe(
604 Effect.mapError(
605 (cause) =>
606 new OrchestrationEffectExecutionError({
607 effectId: effect.id,
608 effectType: effect.request.type,
609 cause,
610 }),
611 ),
612 );
613 case "delegated-tasks.stop":
614 return threads
615 .stopDelegatedTasks({
616 threadId: effect.threadId,
617 commandId: effect.commandId,
618 reason: effect.request.reason,
619 })
620 .pipe(
621 Effect.mapError(
622 (cause) =>
623 new OrchestrationEffectExecutionError({
624 effectId: effect.id,
625 effectType: effect.request.type,
626 cause,
627 }),
628 ),
629 );
630 }
631 }),
632 });
633 }),
634);
635
636// A running effect blocks later work on its thread and its lease is never reclaimed in-process,
637// so a stop that hangs must give up and fall through to the normal retry/fail path.
638const DELEGATED_TASKS_STOP_TIMEOUT = Duration.seconds(30);
639
640const boundExecution =
641 (effect: EffectOutbox.OrchestrationEffectV2) =>
642 (execution: Effect.Effect<void, OrchestrationEffectExecutionError>) =>
643 effect.request.type === "delegated-tasks.stop"
644 ? execution.pipe(
645 Effect.timeoutOrElse({
646 duration: DELEGATED_TASKS_STOP_TIMEOUT,
647 orElse: () =>
648 Effect.fail(
649 new OrchestrationEffectExecutionError({
650 effectId: effect.id,
651 effectType: effect.request.type,
652 cause: `Timed out after ${Duration.format(DELEGATED_TASKS_STOP_TIMEOUT)}.`,
653 }),
654 ),
655 }),
656 )
657 : execution;
658
659export class OrchestrationEffectWorkerError extends Schema.TaggedError<OrchestrationEffectWorkerError>()(
660 "OrchestrationEffectWorkerError",
661 {
662 operation: Schema.String,
663 effectId: Schema.optional(Schema.String),
664 cause: Schema.optional(Schema.Defect()),
665 },
666) {}
667
668const isOrchestrationEffectWorkerError = Schema.is(OrchestrationEffectWorkerError);
669
670type ClaimFilter = Omit<Parameters<EffectOutbox.EffectOutboxV2Shape["claimNext"]>[0], "workerId">;
671
672export interface OrchestrationEffectWorkerV2Shape {
673 readonly awaitWork: Effect.Effect<void>;
674 readonly runOnce: (input?: {
675 readonly maxRunningTitleGenerations?: number;
676 }) => Effect.Effect<boolean, OrchestrationEffectWorkerError>;
677 readonly runRecoveryOnce: Effect.Effect<boolean, OrchestrationEffectWorkerError>;
678 readonly nextClaimableAt: (input?: {
679 readonly maxRunningTitleGenerations?: number;
680 }) => Effect.Effect<Option.Option<DateTime.Utc>, OrchestrationEffectWorkerError>;
681 readonly drain: (maxEffects?: number) => Effect.Effect<number, OrchestrationEffectWorkerError>;
682}
683
684export class OrchestrationEffectWorkerV2 extends Context.Service<
685 OrchestrationEffectWorkerV2,
686 OrchestrationEffectWorkerV2Shape
687>()("t3/orchestration-v2/EffectWorker/OrchestrationEffectWorkerV2") {}
688
689export interface OrchestrationEffectWorkerOptions {
690 readonly workerId?: string;
691 readonly maxAttempts?: number;
692}
693
694export const layerWithOptions = (
695 options: OrchestrationEffectWorkerOptions = {},
696): Layer.Layer<
697 OrchestrationEffectWorkerV2,
698 never,
699 EffectOutbox.EffectOutboxV2 | OrchestrationEffectExecutorV2
700> =>
701 Layer.effect(
702 OrchestrationEffectWorkerV2,
703 Effect.gen(function* () {
704 const outbox = yield* EffectOutbox.EffectOutboxV2;
705 const executor = yield* OrchestrationEffectExecutorV2;
706 const workerId = options.workerId ?? `orchestration-v2:${process.pid}`;
707 const maxAttempts = Math.max(1, options.maxAttempts ?? 5);
708 const wasCancelled = (effectId: string) =>
709 outbox.get(effectId).pipe(
710 Effect.map(
711 Option.match({
712 onNone: () => false,
713 onSome: (effect) => effect.status === "cancelled",
714 }),
715 ),
716 );
717 const requeueClaim = (
718 effect: EffectOutbox.OrchestrationEffectV2,
719 cause: Cause.Cause<unknown>,
720 ) =>
721 Cause.hasInterruptsOnly(cause)
722 ? Effect.void
723 : outbox
724 .retry({
725 effectId: effect.id,
726 workerId,
727 error: `Worker failed before settling the claimed effect: ${Cause.pretty(cause)}`,
728 delayMs: 0,
729 })
730 .pipe(
731 Effect.flatMap((requeued) =>
732 requeued
733 ? Effect.logWarning("Requeued effect after unexpected worker failure", {
734 effectId: effect.id,
735 effectType: effect.request.type,
736 })
737 : Effect.logWarning("Could not requeue effect after worker lost its lease", {
738 effectId: effect.id,
739 effectType: effect.request.type,
740 }),
741 ),
742 Effect.catchCause((requeueCause) =>
743 Effect.logError("Failed to requeue effect after unexpected worker failure", {
744 effectId: effect.id,
745 effectType: effect.request.type,
746 error: Cause.pretty(requeueCause),
747 }),
748 ),
749 );
750 const terminalizeClaim = (
751 effect: EffectOutbox.OrchestrationEffectV2,
752 cause: Cause.Cause<unknown>,
753 ) => {
754 if (Cause.hasInterruptsOnly(cause)) return Effect.void;
755 return outbox
756 .fail({
757 effectId: effect.id,
758 workerId,
759 error: `Worker failed to settle a process-bound effect after execution started: ${Cause.pretty(cause)}`,
760 })
761 .pipe(
762 Effect.flatMap((failed) =>
763 failed
764 ? Effect.logError("Terminalized process-bound effect after settlement failure", {
765 effectId: effect.id,
766 effectType: effect.request.type,
767 })
768 : Effect.logWarning(
769 "Could not terminalize process-bound effect after worker lost its lease",
770 {
771 effectId: effect.id,
772 effectType: effect.request.type,
773 },
774 ),
775 ),
776 Effect.catchCause((failCause) =>
777 Effect.logError(
778 "Failed to terminalize process-bound effect after settlement failure",
779 {
780 effectId: effect.id,
781 effectType: effect.request.type,
782 error: Cause.pretty(failCause),
783 },
784 ),
785 ),
786 );
787 };
788 const recoverPostSuccessSettlement = (
789 effect: EffectOutbox.OrchestrationEffectV2,
790 cause: Cause.Cause<unknown>,
791 ) =>
792 EffectOutbox.REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS.some(
793 (effectType) => effectType === effect.request.type,
794 )
795 ? requeueClaim(effect, cause)
796 : terminalizeClaim(effect, cause);
797
798 // The worker probes for work repeatedly, including when the outbox is empty.
799 const runOnce = Effect.fnUntraced(
800 function* (filter: ClaimFilter = {}) {
801 const claimExit = yield* Effect.exit(outbox.claimNext({ workerId, ...filter }));
802 yield* increment(orchestrationEffectClaimsTotal, {
803 result: Exit.isFailure(claimExit)
804 ? "error"
805 : Option.isNone(claimExit.value)
806 ? "empty"
807 : "claimed",
808 });
809 if (Exit.isFailure(claimExit)) return yield* Effect.failCause(claimExit.cause);
810 const claimed = claimExit.value;
811 if (Option.isNone(claimed)) {
812 return false;
813 }
814 const effect = claimed.value;
815 // Arm cancellation before rereading durable state to close the read-to-registration race.
816 const cancellation = outbox
817 .awaitCancellation(effect.id)
818 .pipe(Effect.as("cancelled" as const));
819 const cancelledBeforeExecution = yield* Effect.gen(function* () {
820 const claimedAt = DateTime.toEpochMillis(yield* DateTime.now);
821 const eligibleAt = Math.max(
822 DateTime.toEpochMillis(DateTime.makeUnsafe(effect.createdAt)),
823 DateTime.toEpochMillis(DateTime.makeUnsafe(effect.availableAt)),
824 );
825 yield* Metric.update(
826 Metric.withAttributes(
827 orchestrationEffectQueueWait,
828 metricAttributes({ effect_type: effect.request.type }),
829 ),
830 Duration.millis(Math.max(0, claimedAt - eligibleAt)),
831 );
832 // Recheck durable cancellation after registering the Deferred; cancellation may have committed after claim.
833 if (yield* wasCancelled(effect.id)) {
834 yield* outbox.clearCancellation(effect.id);
835 return true;
836 }
837 return false;
838 }).pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
839 if (cancelledBeforeExecution) return true;
840
841 const execution = executor
842 .execute(effect, { willRetry: effect.attemptCount < maxAttempts })
843 .pipe(boundExecution(effect), Effect.as("executed" as const));
844 const exit = yield* Effect.exit(Effect.raceFirst(execution, cancellation)).pipe(
845 Effect.ensuring(outbox.clearCancellation(effect.id)),
846 );
847 if (Exit.isSuccess(exit) && exit.value === "cancelled") {
848 return true;
849 }
850 if (Exit.isSuccess(exit)) {
851 return yield* Effect.gen(function* () {
852 const completed = yield* outbox.succeed({ effectId: effect.id, workerId });
853 if (!completed) {
854 if (yield* wasCancelled(effect.id)) return true;
855 return yield* new OrchestrationEffectWorkerError({
856 operation: "complete",
857 effectId: effect.id,
858 cause: "The worker no longer owns the effect lease.",
859 });
860 }
861 return true;
862 }).pipe(Effect.onError((cause) => recoverPostSuccessSettlement(effect, cause)));
863 }
864
865 const error = Cause.pretty(exit.cause);
866 const nonRetryable = isNonRetryableProviderTurnControlFailure(effect.request.type, error);
867 yield* Effect.logWarning("Orchestration effect execution failed", {
868 effectId: effect.id,
869 effectType: effect.request.type,
870 attemptCount: effect.attemptCount,
871 nonRetryable,
872 error,
873 });
874 const updated = nonRetryable
875 ? yield* outbox
876 .succeed({ effectId: effect.id, workerId })
877 .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause)))
878 : effect.attemptCount >= maxAttempts
879 ? yield* outbox
880 .fail({ effectId: effect.id, workerId, error })
881 .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause)))
882 : yield* outbox
883 .retry({
884 effectId: effect.id,
885 workerId,
886 error,
887 delayMs: Math.min(30_000, 100 * 2 ** Math.max(0, effect.attemptCount - 1)),
888 })
889 .pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
890 if (!updated) {
891 if (yield* wasCancelled(effect.id)) return true;
892 return yield* new OrchestrationEffectWorkerError({
893 operation: "reschedule",
894 effectId: effect.id,
895 cause: "The worker no longer owns the effect lease.",
896 });
897 }
898 return true;
899 },
900 (effect, _filter?: ClaimFilter) =>
901 effect.pipe(
902 Effect.mapError((cause) =>
903 isOrchestrationEffectWorkerError(cause)
904 ? cause
905 : new OrchestrationEffectWorkerError({ operation: "run", cause }),
906 ),
907 ),
908 );
909
910 return OrchestrationEffectWorkerV2.of({
911 awaitWork: outbox.awaitAvailable,
912 runOnce,
913 runRecoveryOnce: runOnce({ excludeRestartContinuations: true }),
914 // Idle workers poll this; a span per poll would bury useful traces.
915 nextClaimableAt: Effect.fnUntraced(
916 function* (input) {
917 return yield* outbox.nextClaimableAt(input);
918 },
919 Effect.mapError(
920 (cause) =>
921 new OrchestrationEffectWorkerError({
922 operation: "next-claimable",
923 cause,
924 }),
925 ),
926 ),
927 drain: Effect.fn("EffectWorker.drain")(function* (maxEffects = Number.MAX_SAFE_INTEGER) {
928 let completed = 0;
929 while (completed < maxEffects && (yield* runOnce())) {
930 completed += 1;
931 }
932 return completed;
933 }),
934 });
935 }),
936 );
937
938export const layer = layerWithOptions();
939
940export interface OrchestrationEffectDaemonOptions {
941 readonly concurrency?: number;
942 readonly livenessPollIntervalMs?: number;
943}
944
945const DEFAULT_EFFECT_WORKER_CONCURRENCY = 4;
946const DEFAULT_EFFECT_WORKER_LIVENESS_POLL_INTERVAL_MS = 30_000;
947// A provider request can go unanswered indefinitely. After this wait the slot claims other work
948// while the execution keeps its running row (which blocks its own thread) and settles it later.
949const EFFECT_WORKER_SLOT_WAIT = Duration.seconds(30);
950
951export const runDaemonWithOptions = (options: OrchestrationEffectDaemonOptions = {}) =>
952 Effect.scoped(
953 Effect.gen(function* () {
954 const worker = yield* OrchestrationEffectWorkerV2;
955 const scope = yield* Effect.scope;
956 const requestedConcurrency = options.concurrency ?? DEFAULT_EFFECT_WORKER_CONCURRENCY;
957 const concurrency = Number.isFinite(requestedConcurrency)
958 ? Math.max(1, Math.floor(requestedConcurrency))
959 : DEFAULT_EFFECT_WORKER_CONCURRENCY;
960 const requestedLivenessPollIntervalMs =
961 options.livenessPollIntervalMs ?? DEFAULT_EFFECT_WORKER_LIVENESS_POLL_INTERVAL_MS;
962 const livenessPollIntervalMs = Number.isFinite(requestedLivenessPollIntervalMs)
963 ? Math.max(1, Math.floor(requestedLivenessPollIntervalMs))
964 : DEFAULT_EFFECT_WORKER_LIVENESS_POLL_INTERVAL_MS;
965 // Title generation can run for minutes; leave at least one worker free for turn work.
966 const claimLimits = concurrency > 1 ? { maxRunningTitleGenerations: concurrency - 1 } : {};
967 // Notifications wake promptly; availableAt persists retry timing, and liveness polling recovers missed wakeups.
968 const runWorker = Effect.gen(function* () {
969 while (true) {
970 const attempt = yield* worker.runOnce(claimLimits).pipe(
971 Effect.map((worked) => (worked ? ("worked" as const) : ("idle" as const))),
972 Effect.catchCause((cause) =>
973 Effect.logWarning("Orchestration effect worker failed", cause).pipe(
974 Effect.as("failed" as const),
975 ),
976 ),
977 Effect.forkIn(scope),
978 );
979 const outcome = yield* Fiber.join(attempt).pipe(
980 Effect.timeoutOption(EFFECT_WORKER_SLOT_WAIT),
981 Effect.map(Option.getOrElse(() => "worked" as const)),
982 );
983 if (outcome === "worked") {
984 yield* Effect.yieldNow;
985 continue;
986 }
987 if (outcome === "failed") {
988 // Failed claims can leave overdue rows; cool off instead of rescheduling at the one-millisecond floor.
989 yield* Effect.sleep(Duration.millis(Math.min(1_000, livenessPollIntervalMs)));
990 continue;
991 }
992
993 const nextClaimableAt = yield* worker
994 .nextClaimableAt(claimLimits)
995 .pipe(
996 Effect.catchCause((cause) =>
997 Effect.logWarning(
998 "Failed to read the next orchestration effect deadline",
999 cause,
1000 ).pipe(Effect.as(Option.none<DateTime.Utc>())),
1001 ),
1002 );
1003 const now = DateTime.toEpochMillis(yield* DateTime.now);
1004 const sleepMs = Option.match(nextClaimableAt, {
1005 onNone: () => livenessPollIntervalMs,
1006 onSome: (availableAt) => {
1007 const untilAvailable = DateTime.toEpochMillis(availableAt) - now;
1008 return Math.min(livenessPollIntervalMs, untilAvailable > 0 ? untilAvailable : 25);
1009 },
1010 });
1011 yield* Effect.raceFirst(
1012 worker.awaitWork.pipe(Effect.as("notified" as const)),
1013 Effect.sleep(Duration.millis(sleepMs)).pipe(Effect.as("scheduled" as const)),
1014 );
1015 }
1016 });
1017
1018 return yield* Effect.all(
1019 Array.from({ length: concurrency }, () => runWorker),
1020 {
1021 concurrency: "unbounded",
1022 discard: true,
1023 },
1024 );
1025 }),
1026 );
1027
1028export const runDaemon = runDaemonWithOptions();