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