1import {
2 CheckpointId,
3 CheckpointScopeId,
4 CommandId,
5 MessageId,
6 ModelSelection,
7 NodeId,
8 NonNegativeInt,
9 ProviderSessionId,
10 RunAttemptId,
11 ProviderApprovalDecision,
12 ProviderUserInputAnswers,
13 ProviderThreadId,
14 ProviderTurnId,
15 RunId,
16 RuntimeRequestId,
17 ThreadId,
18} from "@t3tools/contracts";
19import * as Context from "effect/Context";
20import * as NodeSqliteClient from "@t3tools/shared/nodeSqliteClient";
21import * as DateTime from "effect/DateTime";
22import * as Deferred from "effect/Deferred";
23import * as Duration from "effect/Duration";
24import * as Effect from "effect/Effect";
25import * as Layer from "effect/Layer";
26import * as Option from "effect/Option";
27import * as Queue from "effect/Queue";
28import * as Schedule from "effect/Schedule";
29import * as Schema from "effect/Schema";
30import * as SqlClient from "effect/sql/SqlClient";
31
32import { forkParked } from "../serverActivation.ts";
33
34export const OrchestrationEffectRequestV2 = Schema.Union([
35 Schema.Struct({
36 type: Schema.Literal("provider-runtime.continue"),
37 sourceRunId: RunId,
38 }),
39 Schema.Struct({
40 type: Schema.Literal("provider-session.detach"),
41 providerSessionId: ProviderSessionId,
42 detail: Schema.optional(Schema.String),
43 revokeMcpCredential: Schema.optional(Schema.Boolean),
44 }),
45 Schema.Struct({
46 type: Schema.Literal("provider-session.apply-model-selection"),
47 providerSessionId: ProviderSessionId,
48 providerThreadId: ProviderThreadId,
49 modelSelection: ModelSelection,
50 }),
51 Schema.Struct({
52 type: Schema.Literal("provider-turn.start"),
53 runId: RunId,
54 }),
55 Schema.Struct({
56 type: Schema.Literal("provider-turn.interrupt"),
57 providerSessionId: ProviderSessionId,
58 providerThreadId: ProviderThreadId,
59 providerTurnId: ProviderTurnId,
60 subagent: Schema.optional(Schema.Struct({ id: NodeId, nativeTaskId: Schema.String })),
61 }),
62 Schema.Struct({
63 type: Schema.Literal("provider-turn.steer"),
64 providerSessionId: ProviderSessionId,
65 providerThreadId: ProviderThreadId,
66 providerTurnId: ProviderTurnId,
67 messageId: MessageId,
68 modelSelection: Schema.optional(ModelSelection),
69 steerFollowUpAfterRunOrdinal: Schema.optional(NonNegativeInt),
70 }),
71 Schema.Struct({
72 type: Schema.Literal("provider-turn.restart"),
73 providerSessionId: ProviderSessionId,
74 providerThreadId: ProviderThreadId,
75 providerTurnId: ProviderTurnId,
76 interruptedAttemptId: RunAttemptId,
77 runId: RunId,
78 sessionTransition: Schema.optional(
79 Schema.Union([
80 Schema.Struct({
81 type: Schema.Literal("replace"),
82 replacementProviderSessionId: ProviderSessionId,
83 }),
84 Schema.Struct({ type: Schema.Literal("detach") }),
85 ]),
86 ),
87 }),
88 Schema.Struct({
89 type: Schema.Literal("runtime-request.respond"),
90 providerSessionId: ProviderSessionId,
91 requestId: RuntimeRequestId,
92 decision: Schema.optional(ProviderApprovalDecision),
93 feedback: Schema.optional(Schema.String),
94 answers: Schema.optional(ProviderUserInputAnswers),
95 }),
96 Schema.Struct({
97 type: Schema.Literal("provider-thread.rollback"),
98 restoreFiles: Schema.optional(Schema.Boolean),
99 providerThreadId: ProviderThreadId,
100 checkpointId: CheckpointId,
101 scopeId: CheckpointScopeId,
102 }),
103 Schema.Struct({
104 type: Schema.Literal("checkpoint.capture"),
105 runId: RunId,
106 scopeId: CheckpointScopeId,
107 }),
108 Schema.Struct({
109 type: Schema.Literal("scheduled-tasks.cleanup"),
110 }),
111 Schema.Struct({
112 type: Schema.Literal("terminal.cleanup"),
113 }),
114 Schema.Struct({
115 type: Schema.Literal("preview.cleanup"),
116 }),
117 Schema.Struct({
118 type: Schema.Literal("attachment.cleanup"),
119 attachmentIds: Schema.Array(Schema.String),
120 }),
121 Schema.Struct({
122 type: Schema.Literal("thread-title.generate"),
123 kind: Schema.Union([
124 Schema.Struct({ type: Schema.Literal("initial"), messageId: MessageId }),
125 Schema.Struct({ type: Schema.Literal("regenerate") }),
126 ]),
127 }),
128 Schema.Struct({
129 type: Schema.Literal("delegated-tasks.stop"),
130 reason: Schema.optional(Schema.String),
131 }),
132]);
133export type OrchestrationEffectRequestV2 = typeof OrchestrationEffectRequestV2.Type;
134
135export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [
136 "provider-runtime.continue",
137 "provider-session.detach",
138 "provider-thread.rollback",
139 "checkpoint.capture",
140 "scheduled-tasks.cleanup",
141 "terminal.cleanup",
142 "preview.cleanup",
143 "attachment.cleanup",
144 "thread-title.generate",
145 "delegated-tasks.stop",
146] as const satisfies ReadonlyArray<OrchestrationEffectRequestV2["type"]>;
147
148export const PROCESS_BOUND_EFFECT_TYPES = [
149 "provider-session.apply-model-selection",
150 "provider-turn.start",
151 "provider-turn.interrupt",
152 "provider-turn.steer",
153 "provider-turn.restart",
154 "runtime-request.respond",
155] as const satisfies ReadonlyArray<OrchestrationEffectRequestV2["type"]>;
156
157export const OrchestrationEffectStatusV2 = Schema.Literals([
158 "pending",
159 "running",
160 "succeeded",
161 "failed",
162 "cancelled",
163]);
164export type OrchestrationEffectStatusV2 = typeof OrchestrationEffectStatusV2.Type;
165
166/** Receipts deduplicate retries, and restart re-enqueues need far less than a week. Retain failed rows to keep deleted-thread worktrees. */
167export const SETTLED_EFFECT_RETENTION = Duration.days(7);
168const PRUNE_BATCH_SIZE = 500;
169
170export interface OrchestrationEffectV2 {
171 readonly id: string;
172 readonly commandId: CommandId;
173 readonly threadId: ThreadId;
174 readonly request: OrchestrationEffectRequestV2;
175 readonly status: OrchestrationEffectStatusV2;
176 readonly attemptCount: number;
177 readonly availableAt: string;
178 readonly leaseOwner: string | null;
179 readonly createdAt: string;
180 readonly updatedAt: string;
181 readonly completedAt: string | null;
182 readonly lastError: string | null;
183}
184
185export interface PendingOrchestrationEffectV2 {
186 readonly id: string;
187 readonly commandId: CommandId;
188 readonly threadId: ThreadId;
189 readonly request: OrchestrationEffectRequestV2;
190 readonly availableAt?: DateTime.Utc;
191}
192
193export class EffectOutboxError extends Schema.TaggedError<EffectOutboxError>()(
194 "EffectOutboxError",
195 {
196 operation: Schema.String,
197 effectId: Schema.optional(Schema.String),
198 cause: Schema.optional(Schema.Defect()),
199 },
200) {
201 override get message(): string {
202 return `Orchestration effect outbox ${this.operation} failed${this.effectId === undefined ? "" : ` for ${this.effectId}`}.`;
203 }
204}
205
206const isEffectOutboxError = Schema.is(EffectOutboxError);
207
208export interface EffectOutboxV2Shape {
209 readonly awaitAvailable: Effect.Effect<void>;
210 readonly notifyAvailable: (count?: number) => Effect.Effect<void>;
211 /** Persist rows only. Notify workers after the surrounding transaction commits. */
212 readonly enqueue: (
213 effects: ReadonlyArray<PendingOrchestrationEffectV2>,
214 ) => Effect.Effect<void, EffectOutboxError>;
215 readonly get: (
216 effectId: string,
217 ) => Effect.Effect<Option.Option<OrchestrationEffectV2>, EffectOutboxError>;
218 readonly listByCommandId: (
219 commandId: CommandId,
220 ) => Effect.Effect<ReadonlyArray<OrchestrationEffectV2>, EffectOutboxError>;
221 readonly cancelUnsettled: (input: {
222 readonly threadId: ThreadId;
223 readonly effectTypes: ReadonlyArray<OrchestrationEffectRequestV2["type"]>;
224 readonly reason: string;
225 }) => Effect.Effect<ReadonlyArray<string>, EffectOutboxError>;
226 readonly signalCancellations: (effectIds: ReadonlyArray<string>) => Effect.Effect<void>;
227 readonly awaitCancellation: (effectId: string) => Effect.Effect<void>;
228 readonly clearCancellation: (effectId: string) => Effect.Effect<void>;
229 /** Resolves once the effect is neither pending nor running, or right away when it does not exist. */
230 readonly awaitSettled: (effectId: string) => Effect.Effect<void, EffectOutboxError>;
231 readonly reconcileAfterProcessLoss: Effect.Effect<
232 { readonly requeued: number; readonly cancelled: number },
233 EffectOutboxError
234 >;
235 readonly claimNext: (input: {
236 readonly workerId: string;
237 readonly excludeRestartContinuations?: boolean;
238 readonly maxRunningTitleGenerations?: number;
239 }) => Effect.Effect<Option.Option<OrchestrationEffectV2>, EffectOutboxError>;
240 readonly nextClaimableAt: (input?: {
241 readonly maxRunningTitleGenerations?: number;
242 }) => Effect.Effect<Option.Option<DateTime.Utc>, EffectOutboxError>;
243 readonly succeed: (input: {
244 readonly effectId: string;
245 readonly workerId: string;
246 }) => Effect.Effect<boolean, EffectOutboxError>;
247 readonly retry: (input: {
248 readonly effectId: string;
249 readonly workerId: string;
250 readonly error: string;
251 readonly delayMs: number;
252 }) => Effect.Effect<boolean, EffectOutboxError>;
253 readonly fail: (input: {
254 readonly effectId: string;
255 readonly workerId: string;
256 readonly error: string;
257 }) => Effect.Effect<boolean, EffectOutboxError>;
258 readonly pruneSettled: Effect.Effect<number, EffectOutboxError>;
259}
260
261export class EffectOutboxV2 extends Context.Service<EffectOutboxV2, EffectOutboxV2Shape>()(
262 "t3/orchestration-v2/EffectOutbox/EffectOutboxV2",
263) {}
264
265type EffectRow = {
266 readonly effect_id: string;
267 readonly command_id: string;
268 readonly thread_id: string;
269 readonly effect_type: string;
270 readonly payload_json: string;
271 readonly status: string;
272 readonly attempt_count: number;
273 readonly available_at: string;
274 readonly lease_owner: string | null;
275 readonly created_at: string;
276 readonly updated_at: string;
277 readonly completed_at: string | null;
278 readonly last_error: string | null;
279};
280
281const encodeRequest = Schema.encodeSync(Schema.fromJsonString(OrchestrationEffectRequestV2));
282const decodeRequest = Schema.decodeUnknownEffect(
283 Schema.fromJsonString(OrchestrationEffectRequestV2),
284);
285
286const rowToEffect = (row: EffectRow) =>
287 decodeRequest(row.payload_json).pipe(
288 Effect.map((request): OrchestrationEffectV2 => ({
289 id: row.effect_id,
290 commandId: CommandId.make(row.command_id),
291 threadId: ThreadId.make(row.thread_id),
292 request,
293 status: row.status as OrchestrationEffectStatusV2,
294 attemptCount: row.attempt_count,
295 availableAt: row.available_at,
296 leaseOwner: row.lease_owner,
297 createdAt: row.created_at,
298 updatedAt: row.updated_at,
299 completedAt: row.completed_at,
300 lastError: row.last_error,
301 })),
302 );
303
304export const layer: Layer.Layer<EffectOutboxV2, never, SqlClient.SqlClient> = Layer.effect(
305 EffectOutboxV2,
306 Effect.gen(function* () {
307 const sql = yield* SqlClient.SqlClient;
308 const available = yield* Queue.dropping<void>(64);
309 const cancellationSignals = new Map<string, Deferred.Deferred<void>>();
310 const settledSignals = new Map<string, Deferred.Deferred<void>>();
311 const signalSettled = (effectIds: ReadonlyArray<string>) =>
312 Effect.forEach(
313 effectIds,
314 (effectId) => {
315 const signal = settledSignals.get(effectId);
316 if (signal === undefined) return Effect.void;
317 settledSignals.delete(effectId);
318 return Deferred.succeed(signal, undefined);
319 },
320 { discard: true },
321 );
322 const notifyAvailable = (count = 1) =>
323 Queue.offerAll(
324 available,
325 Array.from({ length: Math.min(64, Math.max(0, Math.floor(count))) }, () => undefined),
326 ).pipe(Effect.asVoid);
327 // Retry backoff must block later effects so turns cannot race a pending rollback.
328 const claimableCandidatePredicate = (
329 availableBefore?: string,
330 excludeRestartContinuations = false,
331 maxRunningTitleGenerations?: number,
332 ) =>
333 sql`
334 ${
335 availableBefore === undefined
336 ? sql`1 = 1`
337 : sql`candidate.available_at <= ${availableBefore}`
338 }
339 AND candidate.status = 'pending'
340 AND NOT EXISTS (
341 SELECT 1
342 FROM orchestration_v2_effect_outbox AS active
343 WHERE active.thread_id = candidate.thread_id
344 AND (
345 active.status = 'running'
346 OR (
347 active.status = 'pending'
348 AND active.rowid < candidate.rowid
349 AND ${
350 excludeRestartContinuations
351 ? sql`active.effect_type != 'provider-runtime.continue'`
352 : sql`1 = 1`
353 }
354 )
355 )
356 AND (
357 (
358 candidate.effect_type = 'thread-title.generate'
359 AND active.effect_type = 'thread-title.generate'
360 )
361 OR
362 (
363 candidate.effect_type != 'thread-title.generate'
364 AND active.effect_type != 'thread-title.generate'
365 )
366 )
367 )
368 AND ${
369 maxRunningTitleGenerations === undefined
370 ? sql`1 = 1`
371 : sql`(
372 candidate.effect_type != 'thread-title.generate'
373 OR (
374 SELECT COUNT(*)
375 FROM orchestration_v2_effect_outbox AS running_title
376 WHERE running_title.status = 'running'
377 AND running_title.effect_type = 'thread-title.generate'
378 ) < ${maxRunningTitleGenerations}
379 )`
380 }
381 `;
382
383 const cancellationSignal = (effectId: string) => {
384 const existing = cancellationSignals.get(effectId);
385 if (existing !== undefined) return existing;
386 const created = Deferred.makeUnsafe<void>();
387 cancellationSignals.set(effectId, created);
388 return created;
389 };
390
391 const decodeRows = (operation: string, rows: ReadonlyArray<EffectRow>) =>
392 Effect.forEach(rows, rowToEffect).pipe(
393 Effect.mapError((cause) => new EffectOutboxError({ operation, cause })),
394 );
395
396 const service: EffectOutboxV2Shape = {
397 enqueue: Effect.fn("EffectOutbox.enqueue")(
398 function* (effects) {
399 const now = yield* DateTime.now;
400 const nowIso = DateTime.formatIso(now);
401 yield* Effect.forEach(
402 effects,
403 (effect) => sql`
404 INSERT INTO orchestration_v2_effect_outbox (
405 effect_id,
406 command_id,
407 thread_id,
408 effect_type,
409 payload_json,
410 status,
411 attempt_count,
412 available_at,
413 created_at,
414 updated_at
415 )
416 VALUES (
417 ${effect.id},
418 ${effect.commandId},
419 ${effect.threadId},
420 ${effect.request.type},
421 ${encodeRequest(effect.request)},
422 'pending',
423 0,
424 ${DateTime.formatIso(effect.availableAt ?? now)},
425 ${nowIso},
426 ${nowIso}
427 )
428 ON CONFLICT(effect_id) DO NOTHING
429 `,
430 { concurrency: 1, discard: true },
431 );
432 // EventSink owns the corresponding post-commit notification.
433 },
434 Effect.mapError((cause) => new EffectOutboxError({ operation: "enqueue", cause })),
435 ),
436 get: (effectId) =>
437 sql<EffectRow>`
438 SELECT *
439 FROM orchestration_v2_effect_outbox
440 WHERE effect_id = ${effectId}
441 LIMIT 1
442 `.pipe(
443 NodeSqliteClient.readOnly,
444 Effect.flatMap((rows) => {
445 const row = rows[0];
446 return row === undefined ? Effect.succeedNone : rowToEffect(row).pipe(Effect.asSome);
447 }),
448 Effect.mapError((cause) => new EffectOutboxError({ operation: "get", effectId, cause })),
449 ),
450 awaitAvailable: Queue.take(available),
451 notifyAvailable,
452 listByCommandId: (commandId) =>
453 sql<EffectRow>`
454 SELECT *
455 FROM orchestration_v2_effect_outbox
456 WHERE command_id = ${commandId}
457 ORDER BY created_at ASC, effect_id ASC
458 `.pipe(
459 NodeSqliteClient.readOnly,
460 Effect.flatMap((rows) => decodeRows("list", rows)),
461 Effect.mapError((cause) =>
462 isEffectOutboxError(cause)
463 ? cause
464 : new EffectOutboxError({ operation: "list", cause }),
465 ),
466 ),
467 cancelUnsettled: Effect.fn("EffectOutbox.cancelUnsettled")(
468 function* ({ threadId, effectTypes, reason }) {
469 if (effectTypes.length === 0) return [];
470 const now = DateTime.formatIso(yield* DateTime.now);
471 const rows = yield* sql<{ readonly effect_id: string }>`
472 UPDATE orchestration_v2_effect_outbox
473 SET
474 status = 'cancelled',
475 lease_owner = NULL,
476 completed_at = ${now},
477 updated_at = ${now},
478 last_error = ${reason}
479 WHERE thread_id = ${threadId}
480 AND status IN ('pending', 'running')
481 AND effect_type IN ${sql.in(effectTypes)}
482 RETURNING effect_id
483 `;
484 return rows.map(({ effect_id }) => effect_id);
485 },
486 Effect.mapError((cause) => new EffectOutboxError({ operation: "cancel-unsettled", cause })),
487 ),
488 signalCancellations: Effect.fn("EffectOutbox.signalCancellations")(function* (effectIds) {
489 yield* Effect.forEach(
490 effectIds,
491 (effectId) => {
492 const signal = cancellationSignals.get(effectId);
493 return signal === undefined ? Effect.void : Deferred.succeed(signal, undefined);
494 },
495 { discard: true },
496 );
497 yield* signalSettled(effectIds);
498 if (effectIds.length > 0) yield* notifyAvailable(effectIds.length);
499 }),
500 awaitCancellation: (effectId) => Deferred.await(cancellationSignal(effectId)),
501 clearCancellation: (effectId) =>
502 Effect.sync(() => {
503 cancellationSignals.delete(effectId);
504 }),
505 awaitSettled: Effect.fn("EffectOutbox.awaitSettled")(function* (effectId) {
506 // Registered before the read, so a settle right after it still resolves.
507 const signal = settledSignals.get(effectId) ?? Deferred.makeUnsafe<void>();
508 settledSignals.set(effectId, signal);
509 const effect = yield* service.get(effectId);
510 if (Option.exists(effect, ({ status }) => status === "pending" || status === "running")) {
511 return yield* Deferred.await(signal);
512 }
513 yield* signalSettled([effectId]);
514 }),
515 reconcileAfterProcessLoss: Effect.gen(function* () {
516 const now = DateTime.formatIso(yield* DateTime.now);
517 const cancelledRows = yield* sql<{ readonly effect_id: string }>`
518 UPDATE orchestration_v2_effect_outbox
519 SET
520 status = 'cancelled',
521 lease_owner = NULL,
522 completed_at = ${now},
523 updated_at = ${now},
524 last_error = 'Cancelled because the server process ended before the effect completed.'
525 WHERE status IN ('pending', 'running')
526 AND effect_type IN ${sql.in(PROCESS_BOUND_EFFECT_TYPES)}
527 RETURNING effect_id
528 `;
529 const requeuedRows = yield* sql<{ readonly effect_id: string }>`
530 UPDATE orchestration_v2_effect_outbox
531 SET
532 status = 'pending',
533 lease_owner = NULL,
534 available_at = ${now},
535 updated_at = ${now},
536 last_error = 'Requeued after the previous server process ended.'
537 WHERE status = 'running'
538 AND effect_type IN ${sql.in(REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS)}
539 RETURNING effect_id
540 `;
541 if (requeuedRows.length > 0) yield* notifyAvailable(requeuedRows.length);
542 return { requeued: requeuedRows.length, cancelled: cancelledRows.length };
543 }).pipe(
544 Effect.mapError(
545 (cause) => new EffectOutboxError({ operation: "reconcile-process-loss", cause }),
546 ),
547 ),
548 // The worker polls for the next effect even when there is no work to claim.
549 claimNext: Effect.fnUntraced(
550 function* ({ workerId, excludeRestartContinuations = false, maxRunningTitleGenerations }) {
551 const now = yield* DateTime.now;
552 const nowIso = DateTime.formatIso(now);
553 // Idle claims return empty; tracing their full SQL on every poll overwhelms useful traces.
554 const claimStatement = sql<EffectRow>`
555 UPDATE orchestration_v2_effect_outbox
556 SET
557 status = 'running',
558 attempt_count = attempt_count + 1,
559 lease_owner = ${workerId},
560 updated_at = ${nowIso},
561 last_error = NULL
562 WHERE effect_id = (
563 SELECT candidate.effect_id
564 FROM orchestration_v2_effect_outbox AS candidate
565 WHERE ${claimableCandidatePredicate(
566 nowIso,
567 excludeRestartContinuations,
568 maxRunningTitleGenerations,
569 )}
570 AND ${excludeRestartContinuations ? sql`candidate.effect_type != 'provider-runtime.continue'` : sql`1 = 1`}
571 ORDER BY candidate.available_at ASC, candidate.created_at ASC, candidate.effect_id ASC
572 LIMIT 1
573 )
574 RETURNING *
575 `;
576 const rows = yield* claimStatement.pipe(Effect.withTracerEnabled(false));
577 const row = rows[0];
578 if (row === undefined) return Option.none();
579 cancellationSignals.set(row.effect_id, Deferred.makeUnsafe<void>());
580 return Option.some(yield* rowToEffect(row));
581 },
582 Effect.mapError((cause) => new EffectOutboxError({ operation: "claim", cause })),
583 ),
584 nextClaimableAt: Effect.fnUntraced(
585 function* ({
586 maxRunningTitleGenerations,
587 }: { readonly maxRunningTitleGenerations?: number } = {}) {
588 const rows = yield* sql<{ readonly available_at: string | null }>`
589 SELECT MIN(candidate.available_at) AS available_at
590 FROM orchestration_v2_effect_outbox AS candidate
591 WHERE ${claimableCandidatePredicate(undefined, false, maxRunningTitleGenerations)}
592 `.pipe(NodeSqliteClient.readOnly, Effect.withTracerEnabled(false));
593 const availableAt = rows[0]?.available_at;
594 if (availableAt === undefined || availableAt === null) return Option.none();
595 const parsed = DateTime.make(availableAt);
596 if (Option.isNone(parsed)) {
597 return yield* new EffectOutboxError({
598 operation: "next-claimable",
599 cause: `Invalid available_at timestamp: ${availableAt}`,
600 });
601 }
602 return parsed;
603 },
604 Effect.mapError((cause) =>
605 isEffectOutboxError(cause)
606 ? cause
607 : new EffectOutboxError({ operation: "next-claimable", cause }),
608 ),
609 ),
610 succeed: Effect.fn("EffectOutbox.succeed")(
611 function* ({ effectId, workerId }) {
612 const now = DateTime.formatIso(yield* DateTime.now);
613 const rows = yield* sql<{ readonly effect_id: string }>`
614 UPDATE orchestration_v2_effect_outbox
615 SET
616 status = 'succeeded',
617 lease_owner = NULL,
618 completed_at = ${now},
619 updated_at = ${now},
620 last_error = NULL
621 WHERE effect_id = ${effectId}
622 AND status = 'running'
623 AND lease_owner = ${workerId}
624 RETURNING effect_id
625 `;
626 if (rows.length === 1) {
627 cancellationSignals.delete(effectId);
628 yield* signalSettled([effectId]);
629 yield* notifyAvailable();
630 }
631 return rows.length === 1;
632 },
633 (effect, { effectId }) =>
634 effect.pipe(
635 Effect.mapError(
636 (cause) => new EffectOutboxError({ operation: "succeed", effectId, cause }),
637 ),
638 ),
639 ),
640 retry: Effect.fn("EffectOutbox.retry")(
641 function* ({ effectId, workerId, error, delayMs }) {
642 const now = yield* DateTime.now;
643 const nowIso = DateTime.formatIso(now);
644 const availableAt = DateTime.formatIso(
645 DateTime.add(now, { milliseconds: Math.max(0, delayMs) }),
646 );
647 const rows = yield* sql<{ readonly effect_id: string }>`
648 UPDATE orchestration_v2_effect_outbox
649 SET
650 status = 'pending',
651 available_at = ${availableAt},
652 lease_owner = NULL,
653 updated_at = ${nowIso},
654 last_error = ${error}
655 WHERE effect_id = ${effectId}
656 AND status = 'running'
657 AND lease_owner = ${workerId}
658 RETURNING effect_id
659 `;
660 if (rows.length === 1) {
661 cancellationSignals.delete(effectId);
662 yield* notifyAvailable();
663 }
664 return rows.length === 1;
665 },
666 (effect, { effectId }) =>
667 effect.pipe(
668 Effect.mapError(
669 (cause) => new EffectOutboxError({ operation: "retry", effectId, cause }),
670 ),
671 ),
672 ),
673 fail: Effect.fn("EffectOutbox.fail")(
674 function* ({ effectId, workerId, error }) {
675 const now = DateTime.formatIso(yield* DateTime.now);
676 const rows = yield* sql<{ readonly effect_id: string }>`
677 UPDATE orchestration_v2_effect_outbox
678 SET
679 status = 'failed',
680 lease_owner = NULL,
681 completed_at = ${now},
682 updated_at = ${now},
683 last_error = ${error}
684 WHERE effect_id = ${effectId}
685 AND status = 'running'
686 AND lease_owner = ${workerId}
687 RETURNING effect_id
688 `;
689 if (rows.length === 1) {
690 cancellationSignals.delete(effectId);
691 yield* signalSettled([effectId]);
692 yield* notifyAvailable();
693 }
694 return rows.length === 1;
695 },
696 (effect, { effectId }) =>
697 effect.pipe(
698 Effect.mapError(
699 (cause) => new EffectOutboxError({ operation: "fail", effectId, cause }),
700 ),
701 ),
702 ),
703 pruneSettled: Effect.gen(function* () {
704 const cutoff = DateTime.formatIso(
705 DateTime.subtractDuration(yield* DateTime.now, SETTLED_EFFECT_RETENTION),
706 );
707 // Separate batch statements release the write lock between batches.
708 const deleteBatch = sql<{ readonly effect_id: string }>`
709 DELETE FROM orchestration_v2_effect_outbox
710 WHERE rowid IN (
711 SELECT rowid
712 FROM orchestration_v2_effect_outbox
713 WHERE status IN ('succeeded', 'cancelled')
714 AND completed_at < ${cutoff}
715 LIMIT ${PRUNE_BATCH_SIZE}
716 )
717 RETURNING effect_id
718 `;
719 let pruned = 0;
720 while (true) {
721 const deleted = (yield* deleteBatch).length;
722 pruned += deleted;
723 if (deleted < PRUNE_BATCH_SIZE) return pruned;
724 yield* Effect.yieldNow;
725 }
726 }).pipe(
727 Effect.mapError((cause) => new EffectOutboxError({ operation: "prune-settled", cause })),
728 ),
729 };
730
731 return service;
732 }),
733);
734
735export const layerPruneWorker = Layer.effectDiscard(
736 Effect.gen(function* () {
737 const outbox = yield* EffectOutboxV2;
738 yield* forkParked(
739 outbox.pruneSettled.pipe(
740 Effect.catch((cause) => Effect.logWarning("Failed to prune settled effects", { cause })),
741 Effect.repeat(Schedule.spaced(Duration.hours(1))),
742 ),
743 );
744 }),
745);