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 Layer from "effect/Layer";
9import * as Metric from "effect/Metric";
10import * as Option from "effect/Option";
11import * as Schema from "effect/Schema";
12
13import {
14 increment,
15 metricAttributes,
16 orchestrationEffectClaimsTotal,
17 orchestrationEffectQueueWait,
18} from "../observability/Metrics.ts";
19import * as RunFinalizationService from "./RunFinalizationService.ts";
20import * as ResourceCleanupService from "./ResourceCleanupService.ts";
21import * as EffectOutbox from "./EffectOutbox.ts";
22import * as CheckpointRollbackService from "./CheckpointRollbackService.ts";
23import * as ProviderSessionManager from "./ProviderSessionManager.ts";
24import * as ProviderTurnControlService from "./ProviderTurnControlService.ts";
25import * as ProviderTurnStartService from "./ProviderTurnStartService.ts";
26import * as RuntimeRequestService from "./RuntimeRequestService.ts";
27import * as ThreadTitleRegenerationService from "./ThreadTitleRegenerationService.ts";
28import * as ThreadManagementService from "./ThreadManagementService.ts";
29import * as ServerSettings from "../serverSettings.ts";
30import { continueRestartedRun } from "./RestartContinuation.ts";
31
32export class OrchestrationEffectExecutionError extends Schema.TaggedError<OrchestrationEffectExecutionError>()(
33 "OrchestrationEffectExecutionError",
34 {
35 effectId: Schema.String,
36 effectType: Schema.String,
37 cause: Schema.optional(Schema.Defect()),
38 },
39) {}
40
41/**
42 * Pure interrupt races with hard process teardown or a dead session produce
43 * "not active" protocol errors. Retrying those only delays recovery.
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 */
50export function isNonRetryableProviderTurnControlFailure(
51 effectType: string,
52 errorText: string,
53): boolean {
54 if (effectType !== "provider-turn.interrupt") {
55 return false;
56 }
57 return (
58 /is not active/i.test(errorText) ||
59 /hard teardown is already in progress/i.test(errorText) ||
60 /treating as already interrupted/i.test(errorText) ||
61 /treating as already stopped/i.test(errorText)
62 );
63}
64
65export interface OrchestrationEffectExecutorV2Shape {
66 /**
67 * Runs one claimed effect. `willRetry` is true when the worker will retry a
68 * failure, so a step can fail and try again instead of settling the run.
69 */
70 readonly execute: (
71 effect: EffectOutbox.OrchestrationEffectV2,
72 options?: { readonly willRetry: boolean },
73 ) => Effect.Effect<void, OrchestrationEffectExecutionError>;
74}
75
76export class OrchestrationEffectExecutorV2 extends Context.Service<
77 OrchestrationEffectExecutorV2,
78 OrchestrationEffectExecutorV2Shape
79>()("t3/orchestration-v2/EffectWorker/OrchestrationEffectExecutorV2") {}
80
81export const layerExecutor: Layer.Layer<
82 OrchestrationEffectExecutorV2,
83 never,
84 | ProviderSessionManager.ProviderSessionManagerV2
85 | RunFinalizationService.RunFinalizationService
86 | CheckpointRollbackService.CheckpointRollbackServiceV2
87 | ProviderTurnControlService.ProviderTurnControlServiceV2
88 | ProviderTurnStartService.ProviderTurnStartServiceV2
89 | RuntimeRequestService.RuntimeRequestServiceV2
90 | ThreadTitleRegenerationService.ThreadTitleRegenerationService
91 | ThreadManagementService.ThreadManagementService
92 | ServerSettings.ServerSettingsService
93> = Layer.effect(
94 OrchestrationEffectExecutorV2,
95 Effect.gen(function* () {
96 const runFinalization = yield* RunFinalizationService.RunFinalizationService;
97 const resourceCleanup = yield* ResourceCleanupService.ResourceCleanupService;
98 const checkpointRollback = yield* CheckpointRollbackService.CheckpointRollbackServiceV2;
99 const providerSessions = yield* ProviderSessionManager.ProviderSessionManagerV2;
100 const providerTurnControl = yield* ProviderTurnControlService.ProviderTurnControlServiceV2;
101 const providerTurnStart = yield* ProviderTurnStartService.ProviderTurnStartServiceV2;
102 const runtimeRequests = yield* RuntimeRequestService.RuntimeRequestServiceV2;
103 const threadTitleRegeneration =
104 yield* ThreadTitleRegenerationService.ThreadTitleRegenerationService;
105 const threads = yield* ThreadManagementService.ThreadManagementService;
106 const settings = yield* ServerSettings.ServerSettingsService;
107 return OrchestrationEffectExecutorV2.of({
108 execute: (effect, options) => {
109 const willRetry = options?.willRetry ?? false;
110 switch (effect.request.type) {
111 case "provider-runtime.continue": {
112 const sourceRunId = effect.request.sourceRunId;
113 return continueRestartedRun({ threadId: effect.threadId, sourceRunId }).pipe(
114 Effect.provideService(ThreadManagementService.ThreadManagementService, threads),
115 Effect.provideService(ServerSettings.ServerSettingsService, settings),
116 // A continuation that will never run still owes a delegated parent a result.
117 Effect.tapError(() =>
118 willRetry
119 ? Effect.void
120 : threads.recoverDelegatedTask(effect.threadId, sourceRunId),
121 ),
122 Effect.mapError(
123 (cause) =>
124 new OrchestrationEffectExecutionError({
125 effectId: effect.id,
126 effectType: effect.request.type,
127 cause,
128 }),
129 ),
130 );
131 }
132 case "provider-session.detach":
133 return providerSessions
134 .detach({
135 providerSessionId: effect.request.providerSessionId,
136 threadId: effect.threadId,
137 ...(effect.request.detail === undefined ? {} : { detail: effect.request.detail }),
138 ...(effect.request.revokeMcpCredential === undefined
139 ? {}
140 : { revokeMcpCredential: effect.request.revokeMcpCredential }),
141 })
142 .pipe(
143 Effect.mapError(
144 (cause) =>
145 new OrchestrationEffectExecutionError({
146 effectId: effect.id,
147 effectType: effect.request.type,
148 cause,
149 }),
150 ),
151 );
152 case "provider-turn.start":
153 return providerTurnStart
154 .start({ threadId: effect.threadId, runId: effect.request.runId, willRetry })
155 .pipe(
156 Effect.mapError(
157 (cause) =>
158 new OrchestrationEffectExecutionError({
159 effectId: effect.id,
160 effectType: effect.request.type,
161 cause,
162 }),
163 ),
164 );
165 case "provider-turn.interrupt": {
166 const request = effect.request;
167 return providerTurnControl
168 .interrupt({
169 threadId: effect.threadId,
170 providerSessionId: request.providerSessionId,
171 providerThreadId: request.providerThreadId,
172 providerTurnId: request.providerTurnId,
173 ...(request.subagent === undefined
174 ? {}
175 : {
176 subagent: {
177 ...request.subagent,
178 requestedAt: DateTime.makeUnsafe(effect.createdAt),
179 },
180 }),
181 })
182 .pipe(
183 Effect.tapError((cause) =>
184 request.subagent === undefined
185 ? Effect.void
186 : Effect.logWarning("orchestration-v2.subagent-stop-failed", {
187 effectId: effect.id,
188 cause,
189 }),
190 ),
191 Effect.catch((cause) =>
192 request.subagent === undefined &&
193 isNonRetryableProviderTurnControlFailure(
194 request.type,
195 Cause.pretty(Cause.fail(cause)),
196 )
197 ? Effect.void
198 : Effect.fail(cause),
199 ),
200 // A targeted Stop repairs only an outcome the provider actually observed.
201 // A full-turn Stop retains its existing process-teardown repair.
202 Effect.andThen((outcome) =>
203 request.subagent !== undefined && outcome === undefined
204 ? Effect.void
205 : threads.dispatch({
206 type: "thread.background-work.settle",
207 commandId: CommandId.make(`${effect.id}:background-work-settled`),
208 ...(request.subagent === undefined || outcome === undefined
209 ? {}
210 : {
211 subagent: {
212 id: request.subagent.id,
213 stoppedAt: DateTime.makeUnsafe(effect.createdAt),
214 outcome,
215 },
216 }),
217 threadId: effect.threadId,
218 providerThreadId: request.providerThreadId,
219 providerTurnId: request.providerTurnId,
220 }),
221 ),
222 Effect.mapError(
223 (cause) =>
224 new OrchestrationEffectExecutionError({
225 effectId: effect.id,
226 effectType: request.type,
227 cause,
228 }),
229 ),
230 );
231 }
232 case "provider-turn.steer":
233 return providerTurnControl
234 .steer({
235 threadId: effect.threadId,
236 providerSessionId: effect.request.providerSessionId,
237 providerThreadId: effect.request.providerThreadId,
238 providerTurnId: effect.request.providerTurnId,
239 messageId: effect.request.messageId,
240 })
241 .pipe(
242 Effect.tap(() =>
243 Effect.gen(function* () {
244 if (effect.request.type !== "provider-turn.steer") return;
245 const messageId = effect.request.messageId;
246 const projection = yield* threads.getThreadRecords(
247 effect.threadId,
248 ["messages", "runs"],
249 { messageIds: [effect.request.messageId] },
250 );
251 const message = projection.messages.find((row) => row.id === messageId);
252 if (message?.delegatedCompletion === undefined) return;
253 yield* threads.dispatch({
254 type: "notification.delivery.accept",
255 commandId: CommandId.make(`command:mailbox-accepted:${effect.id}`),
256 threadId: effect.threadId,
257 messageId: message.id,
258 });
259 }),
260 ),
261 Effect.catch((error) =>
262 Effect.gen(function* () {
263 if (
264 !("turnCompleted" in error) ||
265 !error.turnCompleted ||
266 effect.request.type !== "provider-turn.steer"
267 ) {
268 return yield* error;
269 }
270 const projection = yield* threads.getThreadRecords(
271 effect.threadId,
272 ["messages", "runs"],
273 { messageIds: [effect.request.messageId] },
274 );
275 const messageId = effect.request.messageId;
276 const message = projection.messages.find((item) => item.id === messageId);
277 const run = projection.runs.find((item) => item.id === message?.runId);
278 if (message === undefined || run === undefined) return yield* error;
279 // Reuse the message identity and a stable command receipt so an outbox
280 // retry cannot append a duplicate message or start a second follow-up.
281 yield* threads
282 .dispatch({
283 type: "message.dispatch",
284 commandId: CommandId.make(`command:steer-follow-up:${effect.id}`),
285 threadId: effect.threadId,
286 messageId: message.id,
287 text: message.text,
288 ...(message.context ? { context: message.context } : {}),
289 attachments: message.attachments,
290 // A user's follow-up starts on the thread's saved selection,
291 // which already holds the steer's choice. A delegated
292 // completion stays pinned to the run it reports to.
293 ...(message.delegatedCompletion === undefined
294 ? {}
295 : { modelSelection: run.modelSelection }),
296 dispatchMode: {
297 type:
298 message.delegatedCompletion === undefined
299 ? "start_immediately"
300 : "queue_after_active",
301 },
302 createdBy: message.createdBy,
303 creationSource: message.creationSource,
304 ...(message.delegatedCompletion === undefined
305 ? {}
306 : { delegatedCompletion: message.delegatedCompletion }),
307 ...(message.notification === undefined
308 ? {}
309 : { notification: message.notification }),
310 ...(message.scheduledTaskId === undefined
311 ? {}
312 : { scheduledTaskId: message.scheduledTaskId }),
313 ...(message.senderThreadId === undefined
314 ? {}
315 : { senderThreadId: message.senderThreadId }),
316 })
317 .pipe(
318 // The turn that just completed leaves no further retry
319 // target once the instance is off: the completed turn
320 // stands as the thread's result instead of going
321 // through the outbox's bounded retry on a refusal that
322 // retrying cannot change.
323 Effect.catchTags({
324 ThreadManagementProviderInstanceDisabledError: (disabledError) =>
325 Effect.logInfo(
326 "orchestration-v2.effect-worker.steer-follow-up-disabled",
327 {
328 threadId: effect.threadId,
329 messageId: message.id,
330 projectId: disabledError.projectId,
331 instanceId: disabledError.instanceId,
332 },
333 ),
334 }),
335 );
336 }),
337 ),
338 Effect.mapError(
339 (cause) =>
340 new OrchestrationEffectExecutionError({
341 effectId: effect.id,
342 effectType: effect.request.type,
343 cause,
344 }),
345 ),
346 );
347 case "provider-turn.restart":
348 return providerTurnControl
349 .interruptAndAwaitTerminal({
350 threadId: effect.threadId,
351 providerSessionId: effect.request.providerSessionId,
352 providerThreadId: effect.request.providerThreadId,
353 providerTurnId: effect.request.providerTurnId,
354 interruptedAttemptId: effect.request.interruptedAttemptId,
355 ...(effect.request.sessionTransition?.type === "replace"
356 ? {
357 replacementProviderSessionId:
358 effect.request.sessionTransition.replacementProviderSessionId,
359 }
360 : {}),
361 })
362 .pipe(
363 Effect.andThen(
364 effect.request.sessionTransition?.type === "replace"
365 ? providerSessions.detach({
366 providerSessionId: effect.request.providerSessionId,
367 threadId: effect.threadId,
368 detail: "Selection change requires a provider session restart.",
369 })
370 : effect.request.sessionTransition?.type === "detach"
371 ? providerSessions.detach({
372 providerSessionId: effect.request.providerSessionId,
373 threadId: effect.threadId,
374 detail: "Provider thread handoff replaced this session binding.",
375 })
376 : Effect.void,
377 ),
378 Effect.andThen(
379 providerTurnStart.start({
380 threadId: effect.threadId,
381 runId: effect.request.runId,
382 willRetry,
383 }),
384 ),
385 Effect.mapError(
386 (cause) =>
387 new OrchestrationEffectExecutionError({
388 effectId: effect.id,
389 effectType: effect.request.type,
390 cause,
391 }),
392 ),
393 );
394 case "runtime-request.respond":
395 return runtimeRequests
396 .respond({
397 threadId: effect.threadId,
398 providerSessionId: effect.request.providerSessionId,
399 requestId: effect.request.requestId,
400 ...(effect.request.decision === undefined
401 ? {}
402 : { decision: effect.request.decision }),
403 ...(effect.request.answers === undefined
404 ? {}
405 : { answers: effect.request.answers }),
406 })
407 .pipe(
408 Effect.mapError(
409 (cause) =>
410 new OrchestrationEffectExecutionError({
411 effectId: effect.id,
412 effectType: effect.request.type,
413 cause,
414 }),
415 ),
416 );
417 case "provider-thread.rollback":
418 return checkpointRollback
419 .execute({
420 threadId: effect.threadId,
421 providerThreadId: effect.request.providerThreadId,
422 checkpointId: effect.request.checkpointId,
423 scopeId: effect.request.scopeId,
424 ...(effect.request.restoreFiles === undefined
425 ? {}
426 : { restoreFiles: effect.request.restoreFiles }),
427 })
428 .pipe(
429 // The last failed attempt tells waiting clients it failed,
430 // instead of leaving them to time out. Clients get a fixed
431 // message; the worker logs the full cause for each attempt.
432 Effect.tapCause((cause) =>
433 willRetry || Cause.hasInterruptsOnly(cause)
434 ? Effect.void
435 : threads
436 .dispatch({
437 type: "checkpoint.rollback.fail",
438 commandId: CommandId.make(`${effect.commandId}:rollback-failed`),
439 threadId: effect.threadId,
440 requestId: effect.commandId,
441 message: CheckpointRollbackService.ROLLBACK_FAILED_MESSAGE,
442 })
443 .pipe(
444 Effect.catchCause((recordCause) =>
445 Effect.logWarning("Failed to record rollback failure", {
446 effectId: effect.id,
447 cause: recordCause,
448 }),
449 ),
450 ),
451 ),
452 Effect.mapError(
453 (cause) =>
454 new OrchestrationEffectExecutionError({
455 effectId: effect.id,
456 effectType: effect.request.type,
457 cause,
458 }),
459 ),
460 );
461 case "checkpoint.capture":
462 return runFinalization
463 .finalize({
464 threadId: effect.threadId,
465 runId: effect.request.runId,
466 scopeId: effect.request.scopeId,
467 })
468 .pipe(
469 Effect.mapError(
470 (cause) =>
471 new OrchestrationEffectExecutionError({
472 effectId: effect.id,
473 effectType: effect.request.type,
474 cause,
475 }),
476 ),
477 );
478 case "terminal.cleanup":
479 return resourceCleanup.cleanupTerminals(effect.threadId).pipe(
480 Effect.mapError(
481 (cause) =>
482 new OrchestrationEffectExecutionError({
483 effectId: effect.id,
484 effectType: effect.request.type,
485 cause,
486 }),
487 ),
488 );
489 case "preview.cleanup":
490 return resourceCleanup.cleanupPreviews(effect.threadId).pipe(
491 Effect.mapError(
492 (cause) =>
493 new OrchestrationEffectExecutionError({
494 effectId: effect.id,
495 effectType: effect.request.type,
496 cause,
497 }),
498 ),
499 );
500 case "attachment.cleanup":
501 return resourceCleanup.cleanupAttachments(effect.request.attachmentIds).pipe(
502 Effect.mapError(
503 (cause) =>
504 new OrchestrationEffectExecutionError({
505 effectId: effect.id,
506 effectType: effect.request.type,
507 cause,
508 }),
509 ),
510 );
511 case "thread-title.generate":
512 return threadTitleRegeneration
513 .execute({
514 threadId: effect.threadId,
515 requestId: effect.commandId,
516 kind: effect.request.kind,
517 })
518 .pipe(
519 Effect.mapError(
520 (cause) =>
521 new OrchestrationEffectExecutionError({
522 effectId: effect.id,
523 effectType: effect.request.type,
524 cause,
525 }),
526 ),
527 );
528 case "delegated-tasks.stop":
529 return threads
530 .stopDelegatedTasks({
531 threadId: effect.threadId,
532 commandId: effect.commandId,
533 reason: effect.request.reason,
534 })
535 .pipe(
536 Effect.mapError(
537 (cause) =>
538 new OrchestrationEffectExecutionError({
539 effectId: effect.id,
540 effectType: effect.request.type,
541 cause,
542 }),
543 ),
544 );
545 }
546 },
547 });
548 }),
549);
550
551export class OrchestrationEffectWorkerError extends Schema.TaggedError<OrchestrationEffectWorkerError>()(
552 "OrchestrationEffectWorkerError",
553 {
554 operation: Schema.String,
555 effectId: Schema.optional(Schema.String),
556 cause: Schema.optional(Schema.Defect()),
557 },
558) {}
559
560const isOrchestrationEffectWorkerError = Schema.is(OrchestrationEffectWorkerError);
561
562export interface OrchestrationEffectWorkerV2Shape {
563 readonly awaitWork: Effect.Effect<void>;
564 readonly runOnce: Effect.Effect<boolean, OrchestrationEffectWorkerError>;
565 readonly runRecoveryOnce: Effect.Effect<boolean, OrchestrationEffectWorkerError>;
566 readonly nextClaimableAt: Effect.Effect<
567 Option.Option<DateTime.Utc>,
568 OrchestrationEffectWorkerError
569 >;
570 readonly drain: (maxEffects?: number) => Effect.Effect<number, OrchestrationEffectWorkerError>;
571}
572
573export class OrchestrationEffectWorkerV2 extends Context.Service<
574 OrchestrationEffectWorkerV2,
575 OrchestrationEffectWorkerV2Shape
576>()("t3/orchestration-v2/EffectWorker/OrchestrationEffectWorkerV2") {}
577
578export interface OrchestrationEffectWorkerOptions {
579 readonly workerId?: string;
580 readonly leaseDurationMs?: number;
581 readonly maxAttempts?: number;
582}
583
584export const layerWithOptions = (
585 options: OrchestrationEffectWorkerOptions = {},
586): Layer.Layer<
587 OrchestrationEffectWorkerV2,
588 never,
589 EffectOutbox.EffectOutboxV2 | OrchestrationEffectExecutorV2
590> =>
591 Layer.effect(
592 OrchestrationEffectWorkerV2,
593 Effect.gen(function* () {
594 const outbox = yield* EffectOutbox.EffectOutboxV2;
595 const executor = yield* OrchestrationEffectExecutorV2;
596 const workerId = options.workerId ?? `orchestration-v2:${process.pid}`;
597 const leaseDurationMs = Math.max(1, options.leaseDurationMs ?? 30_000);
598 const maxAttempts = Math.max(1, options.maxAttempts ?? 5);
599 const wasCancelled = (effectId: string) =>
600 outbox.get(effectId).pipe(
601 Effect.map(
602 Option.match({
603 onNone: () => false,
604 onSome: (effect) => effect.status === "cancelled",
605 }),
606 ),
607 );
608 const requeueClaim = (
609 effect: EffectOutbox.OrchestrationEffectV2,
610 cause: Cause.Cause<unknown>,
611 ) =>
612 Cause.hasInterruptsOnly(cause)
613 ? Effect.void
614 : outbox
615 .retry({
616 effectId: effect.id,
617 workerId,
618 error: `Worker failed before settling the claimed effect: ${Cause.pretty(cause)}`,
619 delayMs: 0,
620 })
621 .pipe(
622 Effect.flatMap((requeued) =>
623 requeued
624 ? Effect.logWarning("Requeued effect after unexpected worker failure", {
625 effectId: effect.id,
626 effectType: effect.request.type,
627 })
628 : Effect.logWarning("Could not requeue effect after worker lost its lease", {
629 effectId: effect.id,
630 effectType: effect.request.type,
631 }),
632 ),
633 Effect.catchCause((requeueCause) =>
634 Effect.logError("Failed to requeue effect after unexpected worker failure", {
635 effectId: effect.id,
636 effectType: effect.request.type,
637 error: Cause.pretty(requeueCause),
638 }),
639 ),
640 );
641 const terminalizeClaim = (
642 effect: EffectOutbox.OrchestrationEffectV2,
643 cause: Cause.Cause<unknown>,
644 ) => {
645 if (Cause.hasInterruptsOnly(cause)) return Effect.void;
646 return outbox
647 .fail({
648 effectId: effect.id,
649 workerId,
650 error: `Worker failed to settle a process-bound effect after execution started: ${Cause.pretty(cause)}`,
651 })
652 .pipe(
653 Effect.flatMap((failed) =>
654 failed
655 ? Effect.logError("Terminalized process-bound effect after settlement failure", {
656 effectId: effect.id,
657 effectType: effect.request.type,
658 })
659 : Effect.logWarning(
660 "Could not terminalize process-bound effect after worker lost its lease",
661 {
662 effectId: effect.id,
663 effectType: effect.request.type,
664 },
665 ),
666 ),
667 Effect.catchCause((failCause) =>
668 Effect.logError(
669 "Failed to terminalize process-bound effect after settlement failure",
670 {
671 effectId: effect.id,
672 effectType: effect.request.type,
673 error: Cause.pretty(failCause),
674 },
675 ),
676 ),
677 );
678 };
679 const recoverPostSuccessSettlement = (
680 effect: EffectOutbox.OrchestrationEffectV2,
681 cause: Cause.Cause<unknown>,
682 ) =>
683 EffectOutbox.REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS.some(
684 (effectType) => effectType === effect.request.type,
685 )
686 ? requeueClaim(effect, cause)
687 : terminalizeClaim(effect, cause);
688
689 const runOnce = (excludeRestartContinuations = false) =>
690 Effect.gen(function* () {
691 const claimExit = yield* Effect.exit(
692 outbox.claimNext({ workerId, leaseDurationMs, excludeRestartContinuations }),
693 );
694 yield* increment(orchestrationEffectClaimsTotal, {
695 result: Exit.isFailure(claimExit)
696 ? "error"
697 : Option.isNone(claimExit.value)
698 ? "empty"
699 : "claimed",
700 });
701 if (Exit.isFailure(claimExit)) return yield* Effect.failCause(claimExit.cause);
702 const claimed = claimExit.value;
703 if (Option.isNone(claimed)) {
704 return false;
705 }
706 const effect = claimed.value;
707 // Arm the process-local cancellation signal before re-reading durable
708 // state. A cancellation that commits after the row read has begun can
709 // then still win the execution race instead of falling into the gap
710 // between the read and signal registration.
711 const cancellation = outbox
712 .awaitCancellation(effect.id)
713 .pipe(Effect.as("cancelled" as const));
714 const cancelledBeforeExecution = yield* Effect.gen(function* () {
715 const claimedAt = DateTime.toEpochMillis(yield* DateTime.now);
716 const eligibleAt = Math.max(
717 DateTime.toEpochMillis(DateTime.makeUnsafe(effect.createdAt)),
718 DateTime.toEpochMillis(DateTime.makeUnsafe(effect.availableAt)),
719 );
720 yield* Metric.update(
721 Metric.withAttributes(
722 orchestrationEffectQueueWait,
723 metricAttributes({ effect_type: effect.request.type }),
724 ),
725 Duration.millis(Math.max(0, claimedAt - eligibleAt)),
726 );
727 // Cancellation can commit after the durable claim but before the
728 // process-local Deferred is registered. Re-read the authoritative row
729 // once before starting external work; later cancellations use the
730 // Deferred raced below.
731 if (yield* wasCancelled(effect.id)) {
732 yield* outbox.clearCancellation(effect.id);
733 return true;
734 }
735 return false;
736 }).pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
737 if (cancelledBeforeExecution) return true;
738
739 const execution = executor
740 .execute(effect, { willRetry: effect.attemptCount < maxAttempts })
741 .pipe(Effect.as("executed" as const));
742 const exit = yield* Effect.exit(Effect.raceFirst(execution, cancellation)).pipe(
743 Effect.ensuring(outbox.clearCancellation(effect.id)),
744 );
745 if (Exit.isSuccess(exit) && exit.value === "cancelled") {
746 return true;
747 }
748 if (Exit.isSuccess(exit)) {
749 return yield* Effect.gen(function* () {
750 const completed = yield* outbox.succeed({ effectId: effect.id, workerId });
751 if (!completed) {
752 if (yield* wasCancelled(effect.id)) return true;
753 return yield* new OrchestrationEffectWorkerError({
754 operation: "complete",
755 effectId: effect.id,
756 cause: "The worker no longer owns the effect lease.",
757 });
758 }
759 return true;
760 }).pipe(Effect.onError((cause) => recoverPostSuccessSettlement(effect, cause)));
761 }
762
763 const error = Cause.pretty(exit.cause);
764 const nonRetryable = isNonRetryableProviderTurnControlFailure(effect.request.type, error);
765 yield* Effect.logWarning("Orchestration effect execution failed", {
766 effectId: effect.id,
767 effectType: effect.request.type,
768 attemptCount: effect.attemptCount,
769 nonRetryable,
770 error,
771 });
772 // Prefer succeed for terminal interrupt races so the outbox does not
773 // keep a failed interrupt around; fail only when we must not retry.
774 const updated = nonRetryable
775 ? yield* outbox
776 .succeed({ effectId: effect.id, workerId })
777 .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause)))
778 : effect.attemptCount >= maxAttempts
779 ? yield* outbox
780 .fail({ effectId: effect.id, workerId, error })
781 .pipe(Effect.onError((cause) => terminalizeClaim(effect, cause)))
782 : yield* outbox
783 .retry({
784 effectId: effect.id,
785 workerId,
786 error,
787 delayMs: Math.min(30_000, 100 * 2 ** Math.max(0, effect.attemptCount - 1)),
788 })
789 .pipe(Effect.onError((cause) => requeueClaim(effect, cause)));
790 if (!updated) {
791 if (yield* wasCancelled(effect.id)) return true;
792 return yield* new OrchestrationEffectWorkerError({
793 operation: "reschedule",
794 effectId: effect.id,
795 cause: "The worker no longer owns the effect lease.",
796 });
797 }
798 return true;
799 }).pipe(
800 Effect.mapError((cause) =>
801 isOrchestrationEffectWorkerError(cause)
802 ? cause
803 : new OrchestrationEffectWorkerError({ operation: "run", cause }),
804 ),
805 );
806
807 return OrchestrationEffectWorkerV2.of({
808 awaitWork: outbox.awaitAvailable,
809 runOnce: runOnce(),
810 runRecoveryOnce: runOnce(true),
811 nextClaimableAt: outbox.nextClaimableAt.pipe(
812 Effect.mapError(
813 (cause) =>
814 new OrchestrationEffectWorkerError({
815 operation: "next-claimable",
816 cause,
817 }),
818 ),
819 ),
820 drain: (maxEffects = Number.MAX_SAFE_INTEGER) =>
821 Effect.gen(function* () {
822 let completed = 0;
823 while (completed < maxEffects && (yield* runOnce())) {
824 completed += 1;
825 }
826 return completed;
827 }),
828 });
829 }),
830 );
831
832export const layer = layerWithOptions();
833
834export interface OrchestrationEffectDaemonOptions {
835 readonly concurrency?: number;
836 readonly livenessPollIntervalMs?: number;
837}
838
839const DEFAULT_EFFECT_WORKER_CONCURRENCY = 4;
840const DEFAULT_EFFECT_WORKER_LIVENESS_POLL_INTERVAL_MS = 30_000;
841
842export const runDaemonWithOptions = (options: OrchestrationEffectDaemonOptions = {}) =>
843 Effect.scoped(
844 Effect.gen(function* () {
845 const worker = yield* OrchestrationEffectWorkerV2;
846 const requestedConcurrency = options.concurrency ?? DEFAULT_EFFECT_WORKER_CONCURRENCY;
847 const concurrency = Number.isFinite(requestedConcurrency)
848 ? Math.max(1, Math.floor(requestedConcurrency))
849 : DEFAULT_EFFECT_WORKER_CONCURRENCY;
850 const requestedLivenessPollIntervalMs =
851 options.livenessPollIntervalMs ?? DEFAULT_EFFECT_WORKER_LIVENESS_POLL_INTERVAL_MS;
852 const livenessPollIntervalMs = Number.isFinite(requestedLivenessPollIntervalMs)
853 ? Math.max(1, Math.floor(requestedLivenessPollIntervalMs))
854 : DEFAULT_EFFECT_WORKER_LIVENESS_POLL_INTERVAL_MS;
855 // Post-commit notifications are the low-latency path. `availableAt` is the
856 // durable retry schedule, and the long liveness poll only recovers from a
857 // missed in-process notification or work inserted by another process.
858 const runWorker = Effect.gen(function* () {
859 while (true) {
860 const outcome = yield* worker.runOnce.pipe(
861 Effect.map((worked) => (worked ? ("worked" as const) : ("idle" as const))),
862 Effect.catchCause((cause) =>
863 Effect.logWarning("Orchestration effect worker failed", cause).pipe(
864 Effect.as("failed" as const),
865 ),
866 ),
867 );
868 if (outcome === "worked") {
869 yield* Effect.yieldNow;
870 continue;
871 }
872 if (outcome === "failed") {
873 // A due row can remain visible when a claim UPDATE fails. Do not
874 // feed that past deadline back into the scheduler and retry at the
875 // one-millisecond floor; let transient database failures cool off.
876 yield* Effect.sleep(Duration.millis(Math.min(1_000, livenessPollIntervalMs)));
877 continue;
878 }
879
880 const nextClaimableAt = yield* worker.nextClaimableAt.pipe(
881 Effect.catchCause((cause) =>
882 Effect.logWarning(
883 "Failed to read the next orchestration effect deadline",
884 cause,
885 ).pipe(Effect.as(Option.none<DateTime.Utc>())),
886 ),
887 );
888 const now = DateTime.toEpochMillis(yield* DateTime.now);
889 const sleepMs = Option.match(nextClaimableAt, {
890 onNone: () => livenessPollIntervalMs,
891 onSome: (availableAt) => {
892 const untilAvailable = DateTime.toEpochMillis(availableAt) - now;
893 return Math.min(livenessPollIntervalMs, untilAvailable > 0 ? untilAvailable : 25);
894 },
895 });
896 yield* Effect.raceFirst(
897 worker.awaitWork.pipe(Effect.as("notified" as const)),
898 Effect.sleep(Duration.millis(sleepMs)).pipe(Effect.as("scheduled" as const)),
899 );
900 }
901 });
902
903 return yield* Effect.all(
904 Array.from({ length: concurrency }, () => runWorker),
905 {
906 concurrency: "unbounded",
907 discard: true,
908 },
909 );
910 }),
911 );
912
913export const runDaemon = runDaemonWithOptions();
914
915const layerDaemon: Layer.Layer<never, never, OrchestrationEffectWorkerV2> = Layer.effectDiscard(
916 runDaemon.pipe(Effect.forkScoped),
917);