1import {
2 CommandId,
3 type OrchestrationV2Run,
4 OrchestrationV2DomainEvent,
5 type OrchestrationV2ProviderThread,
6 type OrchestrationV2TurnItem,
7 OrchestrationV2StoredEvent,
8 ProviderThreadId,
9 RunAttemptId,
10 RunId,
11 RuntimeRequestId,
12 NodeId,
13 type ProjectId,
14 ThreadId,
15} from "@t3tools/contracts";
16import * as ProviderCommitBatch from "./ProviderCommitBatch.ts";
17import * as Context from "effect/Context";
18import * as DateTime from "effect/DateTime";
19import * as Effect from "effect/Effect";
20import * as Layer from "effect/Layer";
21import * as Option from "effect/Option";
22import * as PubSub from "effect/PubSub";
23import * as Semaphore from "effect/Semaphore";
24import * as Schema from "effect/Schema";
25import * as Stream from "effect/Stream";
26import * as SqlClient from "effect/sql/SqlClient";
27import * as SqlError from "effect/sql/SqlError";
28
29import { replayAndBufferProjectedLiveEvents } from "./LiveStreamBudget.ts";
30import type { UnsequencedProjectEvent } from "../persistence/OrchestrationEventStore.ts";
31import { projectStoredEventForWire } from "./WireProjection.ts";
32import { makeRunOutcomeRecorder, type RunOutcome } from "./RunMetrics.ts";
33
34import * as CommandReceiptStore from "./CommandReceiptStore.ts";
35import * as EffectOutbox from "./EffectOutbox.ts";
36import * as EventStore from "./EventStore.ts";
37import * as ProjectionStore from "./ProjectionStore.ts";
38import * as ProjectStore from "./ProjectStore.ts";
39import * as TurnItemPositionStore from "./TurnItemPositionStore.ts";
40
41export class EventSinkWriteError extends Schema.TaggedError<EventSinkWriteError>()(
42 "EventSinkWriteError",
43 {
44 eventCount: Schema.Number,
45 commandId: Schema.optional(CommandId),
46 cause: Schema.optional(Schema.Defect()),
47 },
48) {
49 override get message(): string {
50 return `Failed to write ${this.eventCount} orchestration V2 event(s).`;
51 }
52}
53
54export class EventSinkStreamError extends Schema.TaggedError<EventSinkStreamError>()(
55 "EventSinkStreamError",
56 {
57 threadId: Schema.optional(ThreadId),
58 afterSequence: Schema.optional(Schema.Number),
59 cause: Schema.optional(Schema.Defect()),
60 },
61) {
62 override get message(): string {
63 return this.threadId === undefined
64 ? "Failed to stream orchestration V2 events."
65 : `Failed to stream orchestration V2 events for thread ${this.threadId}.`;
66 }
67}
68
69export const EventSinkV2Error = Schema.Union([EventSinkWriteError, EventSinkStreamError]);
70export type EventSinkV2Error = typeof EventSinkV2Error.Type;
71
72interface ProviderBatchPublication {
73 readonly publish: Array<Effect.Effect<void>>;
74 eventCount: number;
75}
76const CurrentProviderBatch = Context.Reference<ProviderBatchPublication | undefined>(
77 "t3/orchestration-v2/EventSink/CurrentProviderBatch",
78 { defaultValue: () => undefined },
79);
80
81export interface EventSinkV2Shape {
82 /** Items that are not provider snapshots, such as a consumer's own control items, never batch. */
83 readonly consumeProviderEvents: <A extends { readonly type: string }, E, R, E2, R2>(
84 source: Stream.Stream<A, E, R>,
85 process: (event: A) => Effect.Effect<void, E2, R2>,
86 after: Effect.Effect<boolean, E2, R2>,
87 canBatch: Effect.Effect<boolean, never, R2>,
88 ) => Effect.Effect<
89 void,
90 E | E2 | ProviderCommitBatch.CommitBatchError | Schema.SchemaError,
91 R | R2
92 >;
93
94 /** Normalize and persist an ordered group against one transaction; publish after its commit. */
95 readonly withProviderBatch: <A, E, R>(
96 effect: Effect.Effect<A, E, R>,
97 ) => Effect.Effect<A, E | EventSinkV2Error, R>;
98
99 readonly write: (input: {
100 /** Return no events if another thread owns this native session at commit time. */
101 readonly unlessNativeThreadOwned?: Pick<
102 OrchestrationV2ProviderThread,
103 "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
104 >;
105 readonly guardPendingUserInputCancellations?: boolean;
106 readonly commandId?: CommandId;
107 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
108 }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
109 readonly writeWithEffects: (input: {
110 /**
111 * Rebuilds each run.updated from the stored row with only its delegatedCompletion replaced.
112 * The batch must not also change that run's lifecycle.
113 */
114 readonly updateDelegatedCompletionOnly?: boolean;
115 readonly unlessNativeThreadOwned?: Pick<
116 OrchestrationV2ProviderThread,
117 "driver" | "providerInstanceId" | "appThreadId" | "nativeThreadRef"
118 >;
119 readonly guardPendingUserInputCancellations?: boolean;
120 readonly commandId?: CommandId;
121 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
122 readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
123 }) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, EventSinkV2Error>;
124 readonly writeIfRunCurrent: (input: {
125 readonly guardPendingUserInputCancellations?: boolean;
126 readonly commandId?: CommandId;
127 readonly threadId: ThreadId;
128 readonly runId: RunId;
129 readonly activeAttemptId: RunAttemptId;
130 readonly expectedStatus: OrchestrationV2Run["status"];
131 /** Also refuse while this run interrupt request is stored without its result. */
132 readonly unlessInterruptRequested?: {
133 readonly requestId: OrchestrationV2TurnItem["id"];
134 readonly resultId: OrchestrationV2TurnItem["id"];
135 };
136 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
137 readonly effects?: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
138 }) => Effect.Effect<
139 {
140 readonly committed: boolean;
141 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
142 },
143 EventSinkV2Error
144 >;
145 /** Commits only if the expected attempt and ordinal still own the provider thread, protecting newer attempts. */
146 readonly writeIfProviderThreadOwner: (input: {
147 readonly guardPendingUserInputCancellations?: boolean;
148 readonly commandId?: CommandId;
149 readonly providerThreadId: ProviderThreadId;
150 readonly runId: RunId;
151 readonly activeAttemptId: RunAttemptId;
152 readonly expectedLastRunOrdinal: number;
153 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
154 }) => Effect.Effect<
155 {
156 readonly committed: boolean;
157 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
158 },
159 EventSinkV2Error
160 >;
161 /** Commits only if the full provider row equals expected; some writers do not advance updatedAt. */
162 readonly writeIfProviderThreadUnchanged: (input: {
163 readonly expected: OrchestrationV2ProviderThread;
164 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
165 }) => Effect.Effect<
166 {
167 readonly committed: boolean;
168 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
169 },
170 EventSinkV2Error
171 >;
172 /** Commits only if the stored turn item still equals expected. */
173 readonly writeIfTurnItemUnchanged: (input: {
174 readonly expected: OrchestrationV2TurnItem;
175 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
176 }) => Effect.Effect<
177 {
178 readonly committed: boolean;
179 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
180 },
181 EventSinkV2Error
182 >;
183 readonly commitCommand: (input: {
184 readonly commandId: CommandId;
185 readonly threadId: ThreadId;
186 readonly commandType: string;
187 readonly acceptedAt: DateTime.Utc;
188 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
189 readonly effects: ReadonlyArray<EffectOutbox.PendingOrchestrationEffectV2>;
190 readonly cancelUnsettledEffects?: {
191 readonly effectTypes: ReadonlyArray<EffectOutbox.OrchestrationEffectRequestV2["type"]>;
192 readonly reason: string;
193 };
194 /** Like writeWithEffects' updateDelegatedCompletionOnly, for this run's run.updated only. */
195 readonly updateDelegatedCompletionOnlyRunId?: RunId;
196 }) => Effect.Effect<
197 {
198 readonly receipt: CommandReceiptStore.CommandReceiptV2;
199 readonly storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>;
200 readonly committed: boolean;
201 readonly cancelledEffectCount: number;
202 },
203 EventSinkV2Error
204 >;
205 readonly commitRejectedCommand: (input: {
206 readonly commandId: CommandId;
207 readonly threadId: ThreadId;
208 readonly commandType: string;
209 readonly rejectedAt: DateTime.Utc;
210 readonly error: string;
211 }) => Effect.Effect<CommandReceiptStore.CommandReceiptV2, EventSinkV2Error>;
212 readonly commitProjectCommand: (input: {
213 readonly commandId: CommandId;
214 readonly projectId: ProjectId;
215 readonly commandType: string;
216 readonly acceptedAt: DateTime.Utc;
217 readonly event: UnsequencedProjectEvent;
218 }) => Effect.Effect<
219 { readonly receipt: CommandReceiptStore.ProjectCommandReceiptV2; readonly committed: boolean },
220 EventSinkV2Error
221 >;
222 readonly commitRejectedProjectCommand: (input: {
223 readonly commandId: CommandId;
224 readonly projectId: ProjectId;
225 readonly commandType: string;
226 readonly rejectedAt: DateTime.Utc;
227 readonly error: string;
228 }) => Effect.Effect<CommandReceiptStore.ProjectCommandReceiptV2, EventSinkV2Error>;
229 readonly stream: (input?: {
230 readonly threadId?: ThreadId;
231 readonly afterSequence?: number;
232 readonly eventType?: OrchestrationV2DomainEvent["type"];
233 /** Bound RPC subscribers; internal workers must not drop their subscription under load. */
234 readonly bounded?: boolean;
235 }) => Stream.Stream<OrchestrationV2StoredEvent, EventSinkV2Error>;
236 readonly latestSequence: (input?: {
237 readonly threadId?: ThreadId;
238 }) => Effect.Effect<number, EventSinkV2Error>;
239 readonly readByCommandId: (input: {
240 readonly commandId: CommandId;
241 }) => Stream.Stream<OrchestrationV2StoredEvent, EventSinkV2Error>;
242}
243
244export class EventSinkV2 extends Context.Service<EventSinkV2, EventSinkV2Shape>()(
245 "t3/orchestration-v2/EventSink/EventSinkV2",
246) {}
247
248const layerBase: Layer.Layer<
249 EventSinkV2,
250 never,
251 | CommandReceiptStore.CommandReceiptStoreV2
252 | EffectOutbox.EffectOutboxV2
253 | EventStore.EventStoreV2
254 | ProjectionStore.ProjectionStoreV2
255 | ProjectStore.ProjectStoreV2
256 | SqlClient.SqlClient
257 | TurnItemPositionStore.TurnItemPositionStoreV2
258> = Layer.effect(
259 EventSinkV2,
260 Effect.gen(function* () {
261 const sql = yield* SqlClient.SqlClient;
262 const commandReceipts = yield* CommandReceiptStore.CommandReceiptStoreV2;
263 const effectOutbox = yield* EffectOutbox.EffectOutboxV2;
264 const eventStore = yield* EventStore.EventStoreV2;
265 const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
266 const projectStore = yield* ProjectStore.ProjectStoreV2;
267 const turnItemPositions = yield* TurnItemPositionStore.TurnItemPositionStoreV2;
268 const liveEvents = yield* PubSub.unbounded<OrchestrationV2StoredEvent>();
269 const liveEventsByType = new Map<
270 OrchestrationV2DomainEvent["type"],
271 PubSub.PubSub<OrchestrationV2StoredEvent>
272 >();
273 const publishLiveEvents =
274 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
275 Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2StoredEvent>) {
276 yield* PubSub.publishAll(liveEvents, events);
277 for (const [type, pubsub] of liveEventsByType) {
278 yield* PubSub.publishAll(
279 pubsub,
280 events.filter((stored) => stored.event.type === type),
281 );
282 }
283 });
284 const runOutcomes = makeRunOutcomeRecorder((threadId) =>
285 // One payload row; unlike the shell, it still resolves a deleted child thread.
286 projectionStore
287 .getThread(threadId)
288 .pipe(Effect.map((thread) => thread.lineage.relationshipToParent === "subagent")),
289 );
290 // Recorded before publication, so an interrupted publish cannot lose a committed outcome.
291 const publishStoredEvents = (
292 events: ReadonlyArray<OrchestrationV2StoredEvent>,
293 outcomes: ReadonlyArray<RunOutcome>,
294 ) =>
295 runOutcomes
296 .record(outcomes)
297 .pipe(
298 Effect.andThen(eventStore.publishCommitted(events)),
299 Effect.andThen(publishLiveEvents(events)),
300 );
301
302 // Serialize commit and publish; reordered publication makes clients drop earlier sequences.
303 const publishLane = yield* Semaphore.make(1);
304 const commitThenPublishNow = <A, E, R>(
305 transaction: Effect.Effect<A, E, R>,
306 publish: (committed: A) => Effect.Effect<void>,
307 ) =>
308 Effect.uninterruptibleMask((restore) => {
309 let holdsLane = false;
310 const takeLane = publishLane.take(1).pipe(
311 Effect.andThen(
312 Effect.sync(() => {
313 holdsLane = true;
314 }),
315 ),
316 Effect.uninterruptible,
317 );
318 // Cancellation may roll back the transaction's work. Once it commits,
319 // every subscriber must hear about it before this writer can exit.
320 // Otherwise a failed child run can leave its parent's task running.
321 return sql
322 .withTransaction(Effect.tap(restore(transaction), () => takeLane))
323 .pipe(
324 Effect.tap(publish),
325 Effect.ensuring(
326 Effect.suspend(() => (holdsLane ? publishLane.release(1) : Effect.void)),
327 ),
328 );
329 });
330
331 // A user can answer after terminal normalization reads the pending request.
332 // Recheck inside the write transaction so stale cleanup cannot erase answers.
333 const guardUserInputCancellations =
334 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
335 Effect.fnUntraced(function* (events: ReadonlyArray<OrchestrationV2DomainEvent>) {
336 const staleRequests = new Set<RuntimeRequestId>();
337 const staleNodes = new Set<NodeId>();
338 for (const event of events) {
339 if (
340 event.type !== "runtime-request.updated" ||
341 event.payload.kind !== "user_input" ||
342 event.payload.status !== "cancelled"
343 )
344 continue;
345 const current = yield* projectionStore.getRuntimeRequest(
346 event.threadId,
347 event.payload.id,
348 );
349 if (
350 current?.status !== "pending" ||
351 current.kind !== "user_input" ||
352 current.providerTurnId !== event.payload.providerTurnId ||
353 current.responseCapability.type === "message"
354 ) {
355 staleRequests.add(event.payload.id);
356 staleNodes.add(event.payload.nodeId);
357 }
358 }
359 return events.filter((event) => {
360 switch (event.type) {
361 case "runtime-request.updated":
362 return event.payload.status !== "cancelled" || !staleRequests.has(event.payload.id);
363 case "node.updated":
364 return event.payload.status !== "cancelled" || !staleNodes.has(event.payload.id);
365 case "turn-item.updated":
366 return (
367 event.payload.type !== "user_input_request" ||
368 event.payload.status !== "cancelled" ||
369 !staleRequests.has(event.payload.requestId)
370 );
371 default:
372 return true;
373 }
374 });
375 });
376
377 const normalizeEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) => {
378 const runOrdinals = new Map(
379 events.flatMap((event) =>
380 event.type === "run.created" || event.type === "run.updated"
381 ? [[event.payload.id, event.payload.ordinal] as const]
382 : [],
383 ),
384 );
385 return Effect.forEach(
386 events,
387 (event): Effect.Effect<OrchestrationV2DomainEvent, unknown> =>
388 event.type === "turn-item.updated"
389 ? turnItemPositions
390 .normalize(
391 event.payload,
392 event.payload.runId === null ? undefined : runOrdinals.get(event.payload.runId),
393 )
394 .pipe(Effect.map((payload) => ({ ...event, payload })))
395 : Effect.succeed(event),
396 { concurrency: 1 },
397 );
398 };
399
400 const applyStoredEvents =
401 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
402 Effect.fnUntraced(function* (storedEvents: ReadonlyArray<OrchestrationV2StoredEvent>) {
403 yield* Effect.forEach(storedEvents, (stored) => projectionStore.apply(stored.event), {
404 concurrency: 1,
405 });
406 const sequence = storedEvents.at(-1)?.sequence;
407 if (sequence !== undefined) {
408 const now = DateTime.formatIso(yield* DateTime.now);
409 yield* sql`
410 INSERT INTO orchestration_v2_projection_metadata (
411 projection_name,
412 schema_version,
413 last_sequence,
414 updated_at
415 )
416 VALUES (
417 'thread-projections',
418 ${ProjectionStore.ORCHESTRATION_V2_PROJECTION_SCHEMA_VERSION},
419 ${sequence},
420 ${now}
421 )
422 ON CONFLICT(projection_name)
423 DO UPDATE SET
424 schema_version = excluded.schema_version,
425 last_sequence = excluded.last_sequence,
426 updated_at = excluded.updated_at
427 `;
428 }
429 // Read lineage here, on the connection this transaction holds; see makeRunOutcomeRecorder.
430 return yield* runOutcomes.prepare(storedEvents);
431 });
432
433 // The enclosing write or commit span already traces this per-event projection merge.
434 const updateDelegatedCompletionOnly = Effect.fnUntraced(function* (
435 events: ReadonlyArray<OrchestrationV2DomainEvent>,
436 includesRun: (runId: RunId) => boolean,
437 ) {
438 return yield* Effect.forEach(events, (event) =>
439 Effect.gen(function* () {
440 if (event.type !== "run.updated" || !includesRun(event.payload.id)) return event;
441 const { runs } = yield* projectionStore.getThreadRecords(event.threadId, ["runs"], {
442 runIds: [event.payload.id],
443 });
444 const current = runs[0];
445 return current === undefined
446 ? event
447 : {
448 ...event,
449 payload: { ...current, delegatedCompletion: event.payload.delegatedCompletion },
450 };
451 }),
452 );
453 });
454
455 const writeEffect = Effect.fn("orchestrationV2.EventSink.write")(function* (
456 input: Parameters<EventSinkV2Shape["writeWithEffects"]>[0],
457 ) {
458 yield* Effect.annotateCurrentSpan({
459 "orchestration_v2.command_id": input.commandId ?? null,
460 "orchestration_v2.event_count": input.events.length,
461 "orchestration_v2.thread_id": input.events[0]?.threadId ?? null,
462 });
463
464 const batch = yield* CurrentProviderBatch;
465 if (batch !== undefined) batch.eventCount += input.events.length;
466 const { committed } = yield* commitThenPublish(
467 Effect.gen(function* () {
468 // Check ownership in the same transaction that claims the native session.
469 const imported = input.unlessNativeThreadOwned;
470 if (imported?.nativeThreadRef != null) {
471 const owners = yield* sql`
472 SELECT provider_thread_id
473 FROM orchestration_v2_projection_provider_threads
474 WHERE driver = ${imported.driver}
475 AND provider_instance_id = ${imported.providerInstanceId}
476 AND thread_id IS NOT ${imported.appThreadId}
477 AND json_extract(payload_json, '$.nativeThreadRef.nativeId') = ${imported.nativeThreadRef.nativeId}
478 LIMIT 1
479 `;
480 if (owners.length > 0) return { committed: [], outcomes: [] };
481 }
482 // Checkpoint capture can finish after a cohort writer reads its parent run.
483 // Merge in this transaction so the event log and projection keep that completion.
484 const events = input.updateDelegatedCompletionOnly
485 ? yield* updateDelegatedCompletionOnly(input.events, () => true)
486 : input.events;
487 const normalized = yield* normalizeEvents(
488 input.guardPendingUserInputCancellations === true
489 ? yield* guardUserInputCancellations(events)
490 : events,
491 );
492 const committed = yield* eventStore.append({
493 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
494 events: normalized,
495 });
496 const outcomes = yield* applyStoredEvents(committed);
497 yield* effectOutbox.enqueue(input.effects);
498 return { committed, outcomes };
499 }).pipe(Effect.withTracerEnabled(false)),
500 ({ committed, outcomes }) =>
501 Effect.gen(function* () {
502 if (input.effects.length > 0) {
503 yield* effectOutbox.notifyAvailable(input.effects.length);
504 }
505 yield* publishStoredEvents(committed, outcomes);
506 }),
507 );
508 return committed;
509 });
510
511 const writeIfRunCurrentEffect = Effect.fn("orchestrationV2.EventSink.writeIfRunCurrent")(
512 function* (input: Parameters<EventSinkV2Shape["writeIfRunCurrent"]>[0]) {
513 yield* Effect.annotateCurrentSpan({
514 "orchestration_v2.command_id": input.commandId ?? null,
515 "orchestration_v2.event_count": input.events.length,
516 "orchestration_v2.run_id": input.runId,
517 "orchestration_v2.thread_id": input.threadId,
518 });
519
520 return yield* commitThenPublish(
521 Effect.gen(function* () {
522 const rows = yield* sql<{
523 readonly status: string;
524 readonly active_attempt_id: string | null;
525 }>`
526 SELECT
527 status,
528 json_extract(payload_json, '$.activeAttemptId') AS active_attempt_id
529 FROM orchestration_v2_projection_runs
530 WHERE run_id = ${input.runId}
531 AND thread_id = ${input.threadId}
532 LIMIT 1
533 `;
534 const current = rows[0];
535 if (
536 current === undefined ||
537 current.status !== input.expectedStatus ||
538 current.active_attempt_id !== input.activeAttemptId
539 ) {
540 return {
541 committed: false as const,
542 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
543 };
544 }
545 if (input.unlessInterruptRequested !== undefined) {
546 const { requestId, resultId } = input.unlessInterruptRequested;
547 const pending = yield* sql<{ readonly pending: number }>`
548 SELECT EXISTS (
549 SELECT 1 FROM orchestration_v2_projection_turn_items
550 WHERE thread_id = ${input.threadId} AND turn_item_id = ${requestId}
551 ) AND NOT EXISTS (
552 SELECT 1 FROM orchestration_v2_projection_turn_items
553 WHERE thread_id = ${input.threadId} AND turn_item_id = ${resultId}
554 ) AS pending
555 `;
556 if (pending[0]?.pending === 1) {
557 return {
558 committed: false as const,
559 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
560 };
561 }
562 }
563
564 const normalized = yield* normalizeEvents(
565 input.guardPendingUserInputCancellations === true
566 ? yield* guardUserInputCancellations(input.events)
567 : input.events,
568 );
569 const storedEvents = yield* eventStore.append({
570 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
571 events: normalized,
572 });
573 const outcomes = yield* applyStoredEvents(storedEvents);
574 yield* effectOutbox.enqueue(input.effects ?? []);
575 return { committed: true as const, storedEvents, outcomes };
576 }),
577 (result) =>
578 Effect.gen(function* () {
579 if (!result.committed) return;
580 if (input.effects !== undefined && input.effects.length > 0) {
581 yield* effectOutbox.notifyAvailable(input.effects.length);
582 }
583 yield* publishStoredEvents(result.storedEvents, result.outcomes);
584 }),
585 );
586 },
587 );
588
589 const writeIfProviderThreadOwnerEffect = Effect.fn(
590 "orchestrationV2.EventSink.writeIfProviderThreadOwner",
591 )(function* (input: Parameters<EventSinkV2Shape["writeIfProviderThreadOwner"]>[0]) {
592 yield* Effect.annotateCurrentSpan({
593 "orchestration_v2.command_id": input.commandId ?? null,
594 "orchestration_v2.event_count": input.events.length,
595 "orchestration_v2.provider_thread_id": input.providerThreadId,
596 "orchestration_v2.run_id": input.runId,
597 "orchestration_v2.active_attempt_id": input.activeAttemptId,
598 "orchestration_v2.expected_last_run_ordinal": input.expectedLastRunOrdinal,
599 });
600
601 return yield* commitThenPublish(
602 Effect.gen(function* () {
603 const rows = yield* sql<{
604 readonly active_attempt_id: string | null;
605 readonly last_run_ordinal: number | null;
606 }>`
607 SELECT
608 json_extract(r.payload_json, '$.activeAttemptId') AS active_attempt_id,
609 p.last_run_ordinal
610 FROM orchestration_v2_projection_provider_threads p
611 JOIN orchestration_v2_projection_runs r
612 ON r.run_id = ${input.runId}
613 AND r.thread_id = p.thread_id
614 WHERE p.provider_thread_id = ${input.providerThreadId}
615 LIMIT 1
616 `;
617 const current = rows[0];
618 if (
619 current === undefined ||
620 current.active_attempt_id !== input.activeAttemptId ||
621 current.last_run_ordinal !== input.expectedLastRunOrdinal
622 ) {
623 return {
624 committed: false as const,
625 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
626 };
627 }
628
629 const normalized = yield* normalizeEvents(
630 input.guardPendingUserInputCancellations === true
631 ? yield* guardUserInputCancellations(input.events)
632 : input.events,
633 );
634 const storedEvents = yield* eventStore.append({
635 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
636 events: normalized,
637 });
638 const outcomes = yield* applyStoredEvents(storedEvents);
639 return { committed: true as const, storedEvents, outcomes };
640 }),
641 (result) =>
642 result.committed
643 ? publishStoredEvents(result.storedEvents, result.outcomes)
644 : Effect.void,
645 );
646 });
647
648 // Appends the events only if the stored payload still matches, in one transaction.
649 const writeIfStoredPayloadMatches = <E>(
650 readStoredPayload: Effect.Effect<ReadonlyArray<{ readonly payload_json: string }>, E>,
651 expectedJson: string,
652 events: ReadonlyArray<OrchestrationV2DomainEvent>,
653 ) =>
654 commitThenPublish(
655 Effect.gen(function* () {
656 const rows = yield* readStoredPayload;
657 if (rows[0]?.payload_json !== expectedJson) {
658 return {
659 committed: false as const,
660 storedEvents: [] as ReadonlyArray<OrchestrationV2StoredEvent>,
661 };
662 }
663
664 const normalized = yield* normalizeEvents(events);
665 const storedEvents = yield* eventStore.append({ events: normalized });
666 const outcomes = yield* applyStoredEvents(storedEvents);
667 return { committed: true as const, storedEvents, outcomes };
668 }),
669 (result) =>
670 result.committed
671 ? publishStoredEvents(result.storedEvents, result.outcomes)
672 : Effect.void,
673 );
674
675 const writeIfProviderThreadUnchangedEffect = Effect.fn(
676 "orchestrationV2.EventSink.writeIfProviderThreadUnchanged",
677 )(function* (input: Parameters<EventSinkV2Shape["writeIfProviderThreadUnchanged"]>[0]) {
678 yield* Effect.annotateCurrentSpan({
679 "orchestration_v2.event_count": input.events.length,
680 "orchestration_v2.provider_thread_id": input.expected.id,
681 });
682 const expectedJson = yield* ProjectionStore.encodeProviderThreadPayload(input.expected);
683 return yield* writeIfStoredPayloadMatches(
684 sql<{ readonly payload_json: string }>`
685 SELECT payload_json
686 FROM orchestration_v2_projection_provider_threads
687 WHERE provider_thread_id = ${input.expected.id}
688 LIMIT 1
689 `,
690 expectedJson,
691 input.events,
692 );
693 });
694
695 const writeIfTurnItemUnchangedEffect = Effect.fn(
696 "orchestrationV2.EventSink.writeIfTurnItemUnchanged",
697 )(function* (input: Parameters<EventSinkV2Shape["writeIfTurnItemUnchanged"]>[0]) {
698 yield* Effect.annotateCurrentSpan({
699 "orchestration_v2.event_count": input.events.length,
700 "orchestration_v2.turn_item_id": input.expected.id,
701 });
702 const expectedJson = yield* ProjectionStore.encodeTurnItemPayload(input.expected);
703 return yield* writeIfStoredPayloadMatches(
704 sql<{ readonly payload_json: string }>`
705 SELECT payload_json
706 FROM orchestration_v2_projection_turn_items
707 WHERE turn_item_id = ${input.expected.id}
708 LIMIT 1
709 `,
710 expectedJson,
711 input.events,
712 );
713 });
714
715 const existingCommandResult =
716 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
717 Effect.fnUntraced(function* (commandId: CommandId) {
718 const existing = yield* commandReceipts.getByCommandId(commandId);
719 if (Option.isNone(existing)) {
720 return yield* Effect.die(
721 new Error(`Command receipt ${commandId} disappeared during its transaction.`),
722 );
723 }
724 const storedEvents = yield* eventStore.readByCommandId({ commandId }).pipe(
725 Stream.runCollect,
726 Effect.map((events): ReadonlyArray<OrchestrationV2StoredEvent> => Array.from(events)),
727 );
728 return { receipt: existing.value, storedEvents };
729 });
730
731 const commitCommandEffect = Effect.fn("orchestrationV2.EventSink.commitCommand")(function* (
732 input: Parameters<EventSinkV2Shape["commitCommand"]>[0],
733 ) {
734 const result = yield* commitThenPublish(
735 Effect.gen(function* () {
736 const reserved = yield* commandReceipts.insertIfAbsent({
737 commandId: input.commandId,
738 threadId: input.threadId,
739 commandType: input.commandType,
740 acceptedAt: input.acceptedAt,
741 resultSequence: 0,
742 status: "accepted",
743 error: null,
744 });
745 if (!reserved) {
746 const existing = yield* existingCommandResult(input.commandId);
747 return { ...existing, committed: false as const, cancelledEffectIds: [] };
748 }
749
750 const runId = input.updateDelegatedCompletionOnlyRunId;
751 const normalized = yield* normalizeEvents(
752 runId === undefined
753 ? input.events
754 : yield* updateDelegatedCompletionOnly(input.events, (id) => id === runId),
755 );
756 const storedEvents = yield* eventStore.append({
757 commandId: input.commandId,
758 events: normalized,
759 });
760 // A Stop that changes nothing can still queue the stop of its delegated tasks.
761 const sequence =
762 storedEvents.at(-1)?.sequence ??
763 (input.effects.length === 0
764 ? undefined
765 : yield* eventStore.latestSequence({ threadId: input.threadId }));
766 if (sequence === undefined) {
767 return yield* Effect.die(
768 new Error(`Command ${input.commandId} produced no orchestration events.`),
769 );
770 }
771 const outcomes = yield* applyStoredEvents(storedEvents);
772 yield* effectOutbox.enqueue(input.effects);
773 const receipt: CommandReceiptStore.CommandReceiptV2 = {
774 commandId: input.commandId,
775 threadId: input.threadId,
776 commandType: input.commandType,
777 acceptedAt: input.acceptedAt,
778 resultSequence: sequence,
779 status: "accepted",
780 error: null,
781 };
782 yield* commandReceipts.upsert(receipt);
783 const cancelledEffectIds =
784 input.cancelUnsettledEffects === undefined
785 ? []
786 : yield* effectOutbox.cancelUnsettled({
787 threadId: input.threadId,
788 ...input.cancelUnsettledEffects,
789 });
790 return { receipt, storedEvents, outcomes, committed: true as const, cancelledEffectIds };
791 }),
792 (result) =>
793 Effect.gen(function* () {
794 yield* effectOutbox.signalCancellations(result.cancelledEffectIds);
795 if (result.committed && input.effects.length > 0) {
796 yield* effectOutbox.notifyAvailable(input.effects.length);
797 }
798 if (result.committed) yield* publishStoredEvents(result.storedEvents, result.outcomes);
799 }),
800 );
801 return {
802 receipt: result.receipt,
803 storedEvents: result.storedEvents,
804 committed: result.committed,
805 cancelledEffectCount: result.cancelledEffectIds.length,
806 };
807 });
808
809 const commitRejectedCommandEffect = Effect.fn(
810 "orchestrationV2.EventSink.commitRejectedCommand",
811 )(function* (input: Parameters<EventSinkV2Shape["commitRejectedCommand"]>[0]) {
812 return yield* sql.withTransaction(
813 Effect.gen(function* () {
814 const sequence = yield* eventStore.latestSequence({ threadId: input.threadId });
815 const receipt: CommandReceiptStore.CommandReceiptV2 = {
816 commandId: input.commandId,
817 threadId: input.threadId,
818 commandType: input.commandType,
819 acceptedAt: input.rejectedAt,
820 resultSequence: sequence,
821 status: "rejected",
822 error: input.error,
823 };
824 const inserted = yield* commandReceipts.insertIfAbsent(receipt);
825 if (inserted) {
826 return receipt;
827 }
828 const existing = yield* commandReceipts.getByCommandId(input.commandId);
829 return Option.getOrElse(existing, () => receipt);
830 }),
831 );
832 });
833
834 const existingProjectReceipt = (commandId: CommandId) =>
835 commandReceipts.getProjectByCommandId(commandId).pipe(
836 Effect.flatMap(
837 Option.match({
838 onNone: () =>
839 Effect.fail(`Command ${commandId} was already used by a thread command.` as const),
840 onSome: Effect.succeed,
841 }),
842 ),
843 );
844
845 const commitProjectCommandEffect = Effect.fn("orchestrationV2.EventSink.commitProjectCommand")(
846 function* (input: Parameters<EventSinkV2Shape["commitProjectCommand"]>[0]) {
847 const result = yield* commitThenPublish(
848 Effect.gen(function* () {
849 const reserved: CommandReceiptStore.ProjectCommandReceiptV2 = {
850 commandId: input.commandId,
851 projectId: input.projectId,
852 commandType: input.commandType,
853 acceptedAt: input.acceptedAt,
854 resultSequence: 0,
855 status: "accepted",
856 error: null,
857 };
858 if (!(yield* commandReceipts.insertIfAbsent(reserved))) {
859 return { receipt: yield* existingProjectReceipt(input.commandId), event: undefined };
860 }
861 const event = yield* eventStore.appendProjectEvent(input.event);
862 yield* projectStore.apply(event);
863 const receipt = { ...reserved, resultSequence: event.sequence };
864 yield* commandReceipts.upsert(receipt);
865 return { receipt, event };
866 }),
867 (result) =>
868 result.event === undefined ? Effect.void : eventStore.publishCommitted([result.event]),
869 );
870 return { receipt: result.receipt, committed: result.event !== undefined };
871 },
872 );
873
874 const commitRejectedProjectCommandEffect = Effect.fn(
875 "orchestrationV2.EventSink.commitRejectedProjectCommand",
876 )(function* (input: Parameters<EventSinkV2Shape["commitRejectedProjectCommand"]>[0]) {
877 return yield* sql.withTransaction(
878 Effect.gen(function* () {
879 const receipt: CommandReceiptStore.ProjectCommandReceiptV2 = {
880 commandId: input.commandId,
881 projectId: input.projectId,
882 commandType: input.commandType,
883 acceptedAt: input.rejectedAt,
884 resultSequence: yield* eventStore.latestApplicationSequence,
885 status: "rejected",
886 error: input.error,
887 };
888 return (yield* commandReceipts.insertIfAbsent(receipt))
889 ? receipt
890 : yield* existingProjectReceipt(input.commandId);
891 }),
892 );
893 });
894
895 const catchUp = (input: {
896 readonly afterSequence: number;
897 readonly throughSequence: number;
898 readonly threadId?: ThreadId;
899 readonly eventType?: OrchestrationV2DomainEvent["type"];
900 }): Stream.Stream<OrchestrationV2StoredEvent, unknown> => {
901 const pageSize = 256;
902 const loop = (afterSequence: number): Stream.Stream<OrchestrationV2StoredEvent, unknown> =>
903 Stream.unwrap(
904 eventStore
905 .read({
906 afterSequence,
907 throughSequence: input.throughSequence,
908 ...(input.threadId === undefined ? {} : { threadId: input.threadId }),
909 ...(input.eventType === undefined ? {} : { eventType: input.eventType }),
910 limit: pageSize,
911 })
912 .pipe(
913 Stream.runCollect,
914 Effect.map((chunk) => Array.from(chunk)),
915 Effect.map((events) => {
916 if (events.length === 0) {
917 return Stream.empty;
918 }
919 const current = Stream.fromIterable(events);
920 const last = events.at(-1)?.sequence ?? input.throughSequence;
921 return events.length < pageSize || last >= input.throughSequence
922 ? current
923 : Stream.concat(current, loop(last));
924 }),
925 ),
926 );
927 return loop(input.afterSequence);
928 };
929
930 const stream = (input?: Parameters<EventSinkV2Shape["stream"]>[0]) => {
931 const afterSequence = input?.afterSequence ?? 0;
932 const matches = (stored: OrchestrationV2StoredEvent) =>
933 (input?.threadId === undefined || stored.event.threadId === input.threadId) &&
934 (input?.eventType === undefined || stored.event.type === input.eventType);
935 const replay = (throughSequence: number) =>
936 catchUp({
937 afterSequence,
938 throughSequence,
939 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
940 ...(input?.eventType === undefined ? {} : { eventType: input.eventType }),
941 }).pipe(Stream.filter(matches));
942 return Stream.unwrap(
943 Effect.gen(function* () {
944 let pubsub = liveEvents;
945 if (input?.eventType !== undefined) {
946 const existing = liveEventsByType.get(input.eventType);
947 if (existing !== undefined) {
948 pubsub = existing;
949 } else {
950 const created = yield* PubSub.unbounded<OrchestrationV2StoredEvent>();
951 pubsub = liveEventsByType.get(input.eventType) ?? created;
952 liveEventsByType.set(input.eventType, pubsub);
953 }
954 }
955 if (input?.bounded === true) {
956 return replayAndBufferProjectedLiveEvents({
957 subscribe: PubSub.subscribe(pubsub),
958 latestSequence: eventStore.latestSequence(),
959 afterSequence,
960 filter: matches,
961 replay,
962 project: projectStoredEventForWire,
963 });
964 }
965 const subscription = yield* PubSub.subscribe(pubsub);
966 const highWater = yield* eventStore.latestSequence();
967 const live = Stream.fromSubscription(subscription).pipe(
968 Stream.filter((stored) => stored.sequence > Math.max(highWater, afterSequence)),
969 Stream.filter(matches),
970 );
971 return Stream.concat(replay(highWater), live);
972 }),
973 );
974 };
975
976 const commitThenPublish = <A, E, R>(
977 transaction: Effect.Effect<A, E, R>,
978 publish: (committed: A) => Effect.Effect<void>,
979 ) =>
980 Effect.flatMap(CurrentProviderBatch, (batch) =>
981 batch === undefined
982 ? Effect.flatMap(Effect.serviceOption(sql.transactionService), (existing) =>
983 Option.isSome(existing)
984 ? commitThenPublishNow(transaction, publish)
985 : coordinator.barrier(commitThenPublishNow(transaction, publish)),
986 )
987 : transaction.pipe(
988 Effect.tap((committed) =>
989 Effect.sync(() => {
990 batch.publish.push(publish(committed));
991 }),
992 ),
993 ),
994 );
995
996 const withProviderBatch: EventSinkV2Shape["withProviderBatch"] = Effect.fn(
997 "orchestrationV2.EventSink.providerBatch",
998 )(function* <A, E, R>(effect: Effect.Effect<A, E, R>) {
999 if ((yield* CurrentProviderBatch) !== undefined) return yield* effect;
1000 const batch: ProviderBatchPublication = { publish: [], eventCount: 0 };
1001 let started = false;
1002 const result = yield* commitThenPublishNow(
1003 Effect.sync(() => {
1004 started = true;
1005 }).pipe(Effect.andThen(Effect.provideService(effect, CurrentProviderBatch, batch))),
1006 () => Effect.forEach(batch.publish, (publish) => publish, { discard: true }),
1007 ).pipe(
1008 // Normalization mutates run-local state, so only retry before its transaction body starts.
1009 Effect.retry({
1010 while: (cause) =>
1011 !started && SqlError.isSqlError(cause) && cause.reason._tag === "LockTimeoutError",
1012 times: 2,
1013 }),
1014 Effect.mapError((cause) =>
1015 SqlError.isSqlError(cause)
1016 ? new EventSinkWriteError({ eventCount: batch.eventCount, cause })
1017 : cause,
1018 ),
1019 );
1020 yield* Effect.annotateCurrentSpan({
1021 "orchestration_v2.event_count": batch.eventCount,
1022 "orchestration_v2.batch_writes": batch.publish.length,
1023 });
1024 return result;
1025 });
1026
1027 const coordinator = yield* ProviderCommitBatch.make(
1028 (work) =>
1029 withProviderBatch(work).pipe(
1030 Effect.mapError((cause) =>
1031 ProviderCommitBatch.isCommitBatchError(cause)
1032 ? cause
1033 : new ProviderCommitBatch.CommitBatchError({ cause }),
1034 ),
1035 ),
1036 yield* ProviderCommitBatch.enabled,
1037 // Each snapshot owns a savepoint and its deferred publications within the shared commit.
1038 Effect.fnUntraced(function* (
1039 work: Effect.Effect<void, ProviderCommitBatch.CommitBatchError>,
1040 ) {
1041 const batch = yield* CurrentProviderBatch;
1042 const publicationCount = batch?.publish.length ?? 0;
1043 const eventCount = batch?.eventCount ?? 0;
1044 return yield* sql.withTransaction(work).pipe(
1045 Effect.onError(() =>
1046 Effect.sync(() => {
1047 if (batch !== undefined) {
1048 batch.publish.length = publicationCount;
1049 batch.eventCount = eventCount;
1050 }
1051 }),
1052 ),
1053 Effect.mapError((cause) =>
1054 ProviderCommitBatch.isCommitBatchError(cause)
1055 ? cause
1056 : new ProviderCommitBatch.CommitBatchError({ cause }),
1057 ),
1058 );
1059 }),
1060 );
1061
1062 return EventSinkV2.of({
1063 consumeProviderEvents: coordinator.consume,
1064 withProviderBatch,
1065 write: (input) =>
1066 writeEffect({ ...input, effects: [] }).pipe(
1067 Effect.mapError(
1068 (cause) =>
1069 new EventSinkWriteError({
1070 eventCount: input.events.length,
1071 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
1072 cause,
1073 }),
1074 ),
1075 ),
1076 writeWithEffects: (input) =>
1077 writeEffect(input).pipe(
1078 Effect.mapError(
1079 (cause) =>
1080 new EventSinkWriteError({
1081 eventCount: input.events.length,
1082 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
1083 cause,
1084 }),
1085 ),
1086 ),
1087 writeIfRunCurrent: (input) =>
1088 writeIfRunCurrentEffect(input).pipe(
1089 Effect.mapError(
1090 (cause) =>
1091 new EventSinkWriteError({
1092 eventCount: input.events.length,
1093 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
1094 cause,
1095 }),
1096 ),
1097 ),
1098 writeIfProviderThreadOwner: (input) =>
1099 writeIfProviderThreadOwnerEffect(input).pipe(
1100 Effect.mapError(
1101 (cause) =>
1102 new EventSinkWriteError({
1103 eventCount: input.events.length,
1104 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
1105 cause,
1106 }),
1107 ),
1108 ),
1109 writeIfProviderThreadUnchanged: (input) =>
1110 writeIfProviderThreadUnchangedEffect(input).pipe(
1111 Effect.mapError(
1112 (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }),
1113 ),
1114 ),
1115 writeIfTurnItemUnchanged: (input) =>
1116 writeIfTurnItemUnchangedEffect(input).pipe(
1117 Effect.mapError(
1118 (cause) => new EventSinkWriteError({ eventCount: input.events.length, cause }),
1119 ),
1120 ),
1121 commitCommand: (input) =>
1122 commitCommandEffect(input).pipe(
1123 Effect.mapError(
1124 (cause) =>
1125 new EventSinkWriteError({
1126 commandId: input.commandId,
1127 eventCount: input.events.length,
1128 cause,
1129 }),
1130 ),
1131 ),
1132 commitRejectedCommand: (input) =>
1133 commitRejectedCommandEffect(input).pipe(
1134 Effect.mapError(
1135 (cause) =>
1136 new EventSinkWriteError({
1137 commandId: input.commandId,
1138 eventCount: 0,
1139 cause,
1140 }),
1141 ),
1142 ),
1143 commitProjectCommand: (input) =>
1144 commitProjectCommandEffect(input).pipe(
1145 Effect.mapError(
1146 (cause) =>
1147 new EventSinkWriteError({ commandId: input.commandId, eventCount: 1, cause }),
1148 ),
1149 ),
1150 commitRejectedProjectCommand: (input) =>
1151 commitRejectedProjectCommandEffect(input).pipe(
1152 Effect.mapError(
1153 (cause) =>
1154 new EventSinkWriteError({ commandId: input.commandId, eventCount: 0, cause }),
1155 ),
1156 ),
1157 stream: (input) =>
1158 stream(input).pipe(
1159 Stream.mapError(
1160 (cause) =>
1161 new EventSinkStreamError({
1162 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
1163 ...(input?.afterSequence === undefined
1164 ? {}
1165 : { afterSequence: input.afterSequence }),
1166 cause,
1167 }),
1168 ),
1169 ),
1170 latestSequence: (input) =>
1171 eventStore.latestSequence(input).pipe(
1172 Effect.mapError(
1173 (cause) =>
1174 new EventSinkStreamError({
1175 ...(input?.threadId === undefined ? {} : { threadId: input.threadId }),
1176 cause,
1177 }),
1178 ),
1179 ),
1180 readByCommandId: (input) =>
1181 eventStore.readByCommandId(input).pipe(
1182 Stream.mapError(
1183 (cause) =>
1184 new EventSinkStreamError({
1185 cause,
1186 }),
1187 ),
1188 ),
1189 } satisfies EventSinkV2Shape);
1190 }),
1191);
1192
1193/** Share the worker's outbox instance: enqueue wakeups are in-memory. */
1194export const layerFromStores = layerBase;
1195
1196export const layer: Layer.Layer<
1197 EventSinkV2,
1198 never,
1199 EventStore.EventStoreV2 | ProjectionStore.ProjectionStoreV2 | SqlClient.SqlClient
1200> = layerBase.pipe(
1201 Layer.provide(
1202 Layer.mergeAll(
1203 CommandReceiptStore.layer,
1204 EffectOutbox.layer,
1205 ProjectStore.layer,
1206 TurnItemPositionStore.layer,
1207 ),
1208 ),
1209);