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