1import {
2 NodeId,
3 CommandId,
4 OrchestrationV2DomainEvent,
5 OrchestrationV2StoredEvent,
6 type OrchestrationV2PlanArtifact,
7 type OrchestrationV2Run,
8 type ModelSelection,
9 type OrchestrationV2Subagent,
10 type OrchestrationV2ThreadProjection,
11 ProviderInstanceId,
12 ProviderSessionId,
13 ProviderThreadId,
14 ProviderTurnId,
15 RawEventId,
16 RunAttemptId,
17 RunId,
18 ThreadId,
19} from "@t3tools/contracts";
20import * as Context from "effect/Context";
21import * as DateTime from "effect/DateTime";
22import * as Effect from "effect/Effect";
23import * as Layer from "effect/Layer";
24import * as Schema from "effect/Schema";
25
26import { AgentScope } from "@t3tools/shared/AgentScope";
27import { modelSelectionsEqual } from "@t3tools/shared/model";
28import * as EventSink from "./EventSink.ts";
29import * as ProjectionStore from "./ProjectionStore.ts";
30import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
31import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
32import { makeProviderFailureTurnItem } from "@t3tools/provider-core/server/failure";
33import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
34import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts";
35import { boundFileChangeForStorage } from "./WireProjection.ts";
36
37export class ProviderEventNormalizeError extends Schema.TaggedError<ProviderEventNormalizeError>()(
38 "ProviderEventNormalizeError",
39 {
40 providerSessionId: ProviderSessionId,
41 threadId: ThreadId,
42 providerEvent: ProviderAdapter.ProviderAdapterV2Event,
43 cause: Schema.optional(Schema.Defect()),
44 },
45) {
46 override get message(): string {
47 return `Failed to normalize provider event ${this.providerEvent.type} for thread ${this.threadId}.`;
48 }
49}
50
51export class ProviderEventPublishError extends Schema.TaggedError<ProviderEventPublishError>()(
52 "ProviderEventPublishError",
53 {
54 providerSessionId: ProviderSessionId,
55 eventCount: Schema.Number,
56 cause: Schema.optional(Schema.Defect()),
57 },
58) {
59 override get message(): string {
60 return `Failed to publish ${this.eventCount} normalized provider event(s).`;
61 }
62}
63
64export const ProviderEventIngestorV2Error = Schema.Union([
65 ProviderEventNormalizeError,
66 ProviderEventPublishError,
67]);
68export type ProviderEventIngestorV2Error = typeof ProviderEventIngestorV2Error.Type;
69
70type TodoListPlan = Extract<OrchestrationV2PlanArtifact, { readonly kind: "todo_list" }>;
71
72function withPlanStepDurations(
73 plan: TodoListPlan,
74 previous: TodoListPlan | undefined,
75 occurredAt: DateTime.Utc,
76): TodoListPlan {
77 const occurredAtIso = DateTime.formatIso(occurredAt);
78 const occurredAtMs = DateTime.toEpochMillis(occurredAt);
79 const previousById = new Map(previous?.steps.map((step) => [step.id, step]));
80 // Provider step IDs may be positional. Changed text must not inherit another task's timing.
81 const previousStep = (step: TodoListPlan["steps"][number]) => {
82 const prior = previousById.get(step.id);
83 return prior?.text === step.text ? prior : undefined;
84 };
85 const hasNewCompletion = plan.steps.some((step) => {
86 const prior = previousStep(step);
87 return step.status === "completed" && prior?.status !== "completed";
88 });
89 let fallbackCompletionConsumed = false;
90
91 return {
92 ...plan,
93 steps: plan.steps.map((step) => {
94 const prior = previousStep(step);
95 const baseStep = { id: step.id, text: step.text, status: step.status };
96 if (step.status === "completed") {
97 if (prior?.status === "completed") {
98 return {
99 ...baseStep,
100 ...(prior.durationMs === undefined ? {} : { durationMs: prior.durationMs }),
101 };
102 }
103 const durationAnchorAt =
104 prior?.status === "running" || !fallbackCompletionConsumed
105 ? prior?.durationAnchorAt
106 : occurredAtIso;
107 fallbackCompletionConsumed = true;
108 const anchorMs =
109 durationAnchorAt === undefined ? occurredAtMs : Date.parse(durationAnchorAt);
110 const durationMs = Number.isFinite(anchorMs) ? Math.max(0, occurredAtMs - anchorMs) : 0;
111 return {
112 ...baseStep,
113 ...(durationMs > 0 ? { durationMs } : {}),
114 };
115 }
116 if (step.status === "running") {
117 return {
118 ...baseStep,
119 durationAnchorAt:
120 prior?.status === "running" ? (prior.durationAnchorAt ?? occurredAtIso) : occurredAtIso,
121 };
122 }
123 return {
124 ...baseStep,
125 durationAnchorAt:
126 hasNewCompletion || prior?.status !== "pending"
127 ? occurredAtIso
128 : (prior.durationAnchorAt ?? occurredAtIso),
129 };
130 }),
131 };
132}
133
134/** Settled runs that carried conversation to the provider. */
135const CONVERSATION_RUN_STATUSES = new Set<OrchestrationV2Run["status"]>([
136 "completed",
137 "interrupted",
138 "failed",
139]);
140
141/**
142 * Settled runs on a provider thread that a native rewind left off the active
143 * branch. A run whose provider turn has a strong native ref is judged by that
144 * ref. A run without one cannot be located, so it falls with everything after
145 * the last run still on the branch.
146 */
147function runsOffNativeBranch(input: {
148 readonly projection: Pick<OrchestrationV2ThreadProjection, "runs" | "attempts" | "providerTurns">;
149 readonly providerThreadId: ProviderThreadId;
150 readonly retainedNativeTurnIds: ReadonlySet<string>;
151 readonly reportingRunId: RunId | undefined;
152}): ReadonlyArray<OrchestrationV2Run> {
153 const { attempts, providerTurns, runs } = input.projection;
154 const nativeTurnId = (run: OrchestrationV2Run) => {
155 const attempt = attempts.find((candidate) => candidate.id === run.activeAttemptId);
156 if (attempt === undefined) return null;
157 const ref = providerTurns.find(
158 (turn) => turn.id === attempt.providerTurnId || turn.runAttemptId === attempt.id,
159 )?.nativeTurnRef;
160 return ref?.strength === "strong" ? ref.nativeId : null;
161 };
162 const settled = runs.flatMap((run) =>
163 run.providerThreadId === input.providerThreadId &&
164 run.id !== input.reportingRunId &&
165 CONVERSATION_RUN_STATUSES.has(run.status)
166 ? [{ run, nativeTurnId: nativeTurnId(run) }]
167 : [],
168 );
169 const lastRetainedOrdinal = settled.reduce(
170 (last, { run, nativeTurnId }) =>
171 nativeTurnId !== null && input.retainedNativeTurnIds.has(nativeTurnId)
172 ? Math.max(last, run.ordinal)
173 : last,
174 0,
175 );
176 return settled
177 .filter(({ run, nativeTurnId }) =>
178 nativeTurnId === null
179 ? run.ordinal > lastRetainedOrdinal
180 : !input.retainedNativeTurnIds.has(nativeTurnId),
181 )
182 .map(({ run }) => run);
183}
184
185export interface ProviderEventIngestInput {
186 readonly providerSessionId: ProviderSessionId;
187 readonly providerInstanceId: ProviderInstanceId;
188 readonly commandId?: CommandId;
189 readonly threadId: ThreadId;
190 readonly runId?: RunId;
191 readonly nodeId?: NodeId;
192 readonly rawEventId?: RawEventId;
193 readonly event: ProviderAdapter.ProviderAdapterV2Event;
194}
195
196export interface ProviderEventIngestorV2Shape {
197 readonly normalize: (
198 input: ProviderEventIngestInput,
199 ) => Effect.Effect<ReadonlyArray<OrchestrationV2DomainEvent>, ProviderEventIngestorV2Error>;
200 readonly ingestNormalized: (
201 input: ProviderEventIngestInput & {
202 /**
203 * Atomically reject mutable provider state emitted by an attempt that
204 * lost ownership while the adapter event was in flight.
205 */
206 readonly writeIfRunCurrent?: {
207 readonly runId: RunId;
208 readonly activeAttemptId: RunAttemptId;
209 readonly expectedStatus: OrchestrationV2Run["status"];
210 };
211 /**
212 * Atomically reject provider-thread snapshots from an attempt that no
213 * longer owns the run or from a run that no longer owns the thread.
214 */
215 readonly writeIfProviderThreadOwner?: {
216 readonly providerThreadId: ProviderThreadId;
217 readonly runId: RunId;
218 readonly activeAttemptId: RunAttemptId;
219 readonly expectedLastRunOrdinal: number;
220 };
221 },
222 ) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, ProviderEventIngestorV2Error>;
223}
224
225export class ProviderEventIngestorV2 extends Context.Service<
226 ProviderEventIngestorV2,
227 ProviderEventIngestorV2Shape
228>()("t3/orchestration-v2/ProviderEventIngestor/ProviderEventIngestorV2") {}
229
230function compactUndefined<T extends Record<string, unknown>>(record: T): T {
231 return Object.fromEntries(Object.entries(record).filter(([, value]) => value !== undefined)) as T;
232}
233
234const decodeDomainEvent = Schema.decodeUnknownEffect(OrchestrationV2DomainEvent);
235
236export const layer: Layer.Layer<
237 ProviderEventIngestorV2,
238 never,
239 | EventSink.EventSinkV2
240 | IdAllocator.IdAllocatorV2
241 | ProjectionStore.ProjectionStoreV2
242 | ThreadCommandExecutor.ThreadCommandExecutor
243> = Layer.effect(
244 ProviderEventIngestorV2,
245 Effect.gen(function* () {
246 const eventSink = yield* EventSink.EventSinkV2;
247 const projections = yield* ProjectionStore.ProjectionStoreV2;
248 const idAllocator = yield* IdAllocator.IdAllocatorV2;
249 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
250 const agentScope = yield* AgentScope;
251
252 const makeDomainEvent = (
253 input: ProviderEventIngestInput,
254 payloadInput: {
255 readonly type: OrchestrationV2DomainEvent["type"];
256 readonly payload: OrchestrationV2DomainEvent["payload"];
257 readonly threadId?: ThreadId;
258 readonly runId?: RunId | null;
259 readonly nodeId?: NodeId | null;
260 readonly occurredAt?: DateTime.Utc;
261 },
262 ) =>
263 Effect.gen(function* () {
264 const threadId = payloadInput.threadId ?? input.threadId;
265 const eventId = yield* idAllocator.allocate.event({
266 threadId,
267 providerSessionId: input.providerSessionId,
268 });
269 const occurredAt = payloadInput.occurredAt ?? (yield* DateTime.now);
270 return yield* decodeDomainEvent(
271 compactUndefined({
272 id: eventId,
273 type: payloadInput.type,
274 threadId,
275 runId: payloadInput.runId ?? input.runId,
276 nodeId: payloadInput.nodeId ?? input.nodeId,
277 driver: input.event.driver,
278 providerInstanceId: input.providerInstanceId,
279 rawEventId: input.rawEventId,
280 occurredAt,
281 payload: payloadInput.payload,
282 }),
283 );
284 });
285
286 const dismissNativeUserInputs = Effect.fn("ProviderEventIngestor.dismissNativeUserInputs")(
287 function* (
288 input: ProviderEventIngestInput,
289 providerTurnId: ProviderTurnId,
290 threadId = input.threadId,
291 ) {
292 const pending = yield* projections.getPendingNativeUserInputs(threadId, providerTurnId);
293 const now = yield* DateTime.now;
294 const events: Array<OrchestrationV2DomainEvent> = [];
295 for (const request of pending.runtimeRequests) {
296 events.push(
297 yield* makeDomainEvent(input, {
298 type: "runtime-request.updated",
299 threadId,
300 nodeId: request.nodeId,
301 payload: { ...request, status: "cancelled", resolvedAt: now },
302 }),
303 );
304 }
305 for (const node of pending.nodes) {
306 events.push(
307 yield* makeDomainEvent(input, {
308 type: "node.updated",
309 threadId,
310 nodeId: node.id,
311 runId: node.runId,
312 payload: { ...node, status: "cancelled", completedAt: now },
313 }),
314 );
315 }
316 for (const item of pending.turnItems) {
317 events.push(
318 yield* makeDomainEvent(input, {
319 type: "turn-item.updated",
320 threadId,
321 nodeId: item.nodeId,
322 runId: item.runId,
323 payload: { ...item, status: "cancelled", completedAt: now, updatedAt: now },
324 }),
325 );
326 }
327 return events;
328 },
329 );
330
331 /**
332 * A native subagent's thread starts on the parent's model when the
333 * provider names the real one later (a Claude agent file's model arrives
334 * with the subagent's first reply). Clients read the thread's model and
335 * effort, so move the thread to the reported selection, or to the reported
336 * model alone when the provider reports no selection. Thread commands
337 * rewrite the whole thread row under the thread's lock, so this read and
338 * write take it too.
339 */
340 const syncSubagentThreadModel = Effect.fn("ProviderEventIngestor.syncSubagentThreadModel")(
341 function* (input: ProviderEventIngestInput, subagent: OrchestrationV2Subagent) {
342 const { childThreadId, model } = subagent;
343 if (
344 subagent.origin !== "provider_native" ||
345 childThreadId === null ||
346 (model === null && subagent.modelSelection === undefined)
347 ) {
348 return [];
349 }
350 const reportedSelection = ({ modelSelection }: { modelSelection: ModelSelection }) =>
351 subagent.modelSelection ??
352 (model === null || model === modelSelection.model
353 ? modelSelection
354 : // The parent's options belong to the parent's model.
355 { instanceId: modelSelection.instanceId, model });
356 const staleThread = projections.getThread(childThreadId).pipe(
357 Effect.map((thread) =>
358 modelSelectionsEqual(thread.modelSelection, reportedSelection(thread)) ? null : thread,
359 ),
360 Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
361 );
362 // Nearly every update already matches; only a mismatch takes the lock.
363 if ((yield* staleThread) === null) return [];
364 return yield* threadCommands.withLock(
365 childThreadId,
366 Effect.gen(function* () {
367 const thread = yield* staleThread;
368 if (thread === null) return [];
369 const now = yield* DateTime.now;
370 const event = yield* makeDomainEvent(input, {
371 type: "thread.model-selection-updated",
372 threadId: thread.id,
373 payload: {
374 ...thread,
375 modelSelection: reportedSelection(thread),
376 updatedAt: now,
377 },
378 occurredAt: now,
379 });
380 return yield* eventSink.write({ events: [event] });
381 }),
382 );
383 },
384 );
385
386 /**
387 * The provider rewound its conversation outside T3. Retire the runs that
388 * fell off its active branch with the same run, root-node, and checkpoint
389 * updates CheckpointRollbackService writes for a T3 revert. The run
390 * reporting the rewind performed it and stays, so it keeps lastRunOrdinal.
391 */
392 const rollBackRewoundRuns = Effect.fn("ProviderEventIngestor.rollBackRewoundRuns")(function* (
393 input: ProviderEventIngestInput,
394 rewind: {
395 readonly threadId: ThreadId;
396 readonly providerThreadId: ProviderThreadId;
397 readonly retainedNativeTurnIds: ReadonlySet<string>;
398 },
399 ) {
400 const projection = yield* projections.getThreadRecords(rewind.threadId, [
401 "runs",
402 "attempts",
403 "providerTurns",
404 "nodes",
405 "checkpoints",
406 ]);
407 const runs = runsOffNativeBranch({
408 projection,
409 providerThreadId: rewind.providerThreadId,
410 retainedNativeTurnIds: rewind.retainedNativeTurnIds,
411 reportingRunId: input.runId,
412 });
413 if (runs.length === 0) return [];
414 const now = yield* DateTime.now;
415 const eventId = () =>
416 idAllocator.allocate.event({
417 threadId: rewind.threadId,
418 providerSessionId: input.providerSessionId,
419 });
420 const runIds = new Set(runs.map((run) => run.id));
421 const events: Array<OrchestrationV2DomainEvent> = [];
422 for (const checkpoint of projection.checkpoints) {
423 if (checkpoint.status !== "ready" || checkpoint.runId === null) continue;
424 if (!runIds.has(checkpoint.runId)) continue;
425 events.push({
426 id: yield* eventId(),
427 type: "checkpoint.captured",
428 threadId: rewind.threadId,
429 runId: checkpoint.runId,
430 nodeId: checkpoint.nodeId,
431 providerInstanceId: input.providerInstanceId,
432 occurredAt: now,
433 payload: { ...checkpoint, status: "stale" },
434 });
435 }
436 for (const run of runs) {
437 const rootNode = projection.nodes.find((node) => node.id === run.rootNodeId);
438 events.push({
439 id: yield* eventId(),
440 type: "run.updated",
441 threadId: rewind.threadId,
442 runId: run.id,
443 ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
444 providerInstanceId: run.providerInstanceId,
445 occurredAt: now,
446 payload: { ...run, status: "rolled_back", completedAt: now },
447 });
448 if (rootNode !== undefined) {
449 events.push({
450 id: yield* eventId(),
451 type: "node.updated",
452 threadId: rewind.threadId,
453 runId: run.id,
454 nodeId: rootNode.id,
455 providerInstanceId: run.providerInstanceId,
456 occurredAt: now,
457 payload: { ...rootNode, status: "rolled_back", completedAt: now },
458 });
459 }
460 }
461 return events;
462 });
463
464 const normalize: ProviderEventIngestorV2Shape["normalize"] = (input) =>
465 Effect.gen(function* () {
466 switch (input.event.type) {
467 case "app_thread.created": {
468 const offered = input.event.appThread;
469 const existing = yield* projections.getThread(offered.id).pipe(
470 Effect.catchTags({
471 ProjectionStoreThreadNotFoundError: () => Effect.succeed(null),
472 }),
473 );
474 if (
475 existing !== null &&
476 existing.title === offered.title &&
477 modelSelectionsEqual(existing.modelSelection, offered.modelSelection)
478 )
479 return [];
480 const now = yield* DateTime.now;
481 return [
482 yield* makeDomainEvent(input, {
483 type: existing === null ? "thread.created" : "thread.metadata-updated",
484 threadId: offered.id,
485 payload:
486 existing === null
487 ? offered
488 : {
489 ...existing,
490 title: offered.title,
491 modelSelection: offered.modelSelection,
492 updatedAt: now,
493 },
494 occurredAt: now,
495 }),
496 ];
497 }
498 case "provider_session.updated":
499 return [
500 yield* makeDomainEvent(input, {
501 type: "provider-session.updated",
502 payload: input.event.providerSession,
503 }),
504 ];
505 case "provider_thread.updated": {
506 const threadId = input.event.providerThread.appThreadId ?? input.threadId;
507 const updated = yield* makeDomainEvent(input, {
508 type: "provider-thread.updated",
509 threadId,
510 payload: input.event.providerThread,
511 });
512 return input.event.retainedNativeTurnIds === undefined
513 ? [updated]
514 : [
515 updated,
516 ...(yield* rollBackRewoundRuns(input, {
517 threadId,
518 providerThreadId: input.event.providerThread.id,
519 retainedNativeTurnIds: new Set(input.event.retainedNativeTurnIds),
520 })),
521 ];
522 }
523 case "provider_turn.updated":
524 return [
525 ...(["completed", "interrupted", "failed", "cancelled"].includes(
526 input.event.providerTurn.status,
527 )
528 ? yield* dismissNativeUserInputs(
529 input,
530 input.event.providerTurn.id,
531 input.event.threadId,
532 )
533 : []),
534 yield* makeDomainEvent(input, {
535 type: "provider-turn.updated",
536 ...(input.event.threadId === undefined ? {} : { threadId: input.event.threadId }),
537 payload: input.event.providerTurn,
538 nodeId: input.event.providerTurn.nodeId,
539 }),
540 ];
541 case "node.updated":
542 // Adapters send an unchanged node again with each text delta. Compare with the
543 // stored node, not the last one sent, because the server also writes nodes.
544 if (yield* projections.isNodeStored(input.event.node)) return [];
545 return [
546 yield* makeDomainEvent(input, {
547 type: "node.updated",
548 threadId: input.event.node.threadId,
549 payload: input.event.node,
550 runId: input.event.node.runId,
551 nodeId: input.event.node.id,
552 }),
553 ];
554 case "subagent.updated":
555 return [
556 yield* makeDomainEvent(input, {
557 type: "subagent.updated",
558 threadId: input.event.subagent.threadId,
559 payload: input.event.subagent,
560 runId: input.event.subagent.runId,
561 nodeId: input.event.subagent.id,
562 }),
563 ];
564 case "message.updated":
565 return [
566 yield* makeDomainEvent(input, {
567 type: "message.updated",
568 threadId: input.event.message.threadId,
569 payload: input.event.message,
570 runId: input.event.message.runId,
571 nodeId: input.event.message.nodeId,
572 }),
573 ];
574 case "turn_item.updated":
575 return [
576 yield* makeDomainEvent(input, {
577 type: "turn-item.updated",
578 threadId: input.event.turnItem.threadId,
579 payload: boundFileChangeForStorage(
580 stripUnservedToolOutputImageBytes(input.event.turnItem),
581 ),
582 runId: input.event.turnItem.runId,
583 nodeId: input.event.turnItem.nodeId,
584 }),
585 ];
586 case "runtime_request.updated":
587 return [
588 yield* makeDomainEvent(input, {
589 type: "runtime-request.updated",
590 ...(input.event.threadId === undefined ? {} : { threadId: input.event.threadId }),
591 payload: input.event.runtimeRequest,
592 nodeId: input.event.runtimeRequest.nodeId,
593 }),
594 ];
595 case "plan.updated": {
596 const occurredAt = yield* DateTime.now;
597 const plan = input.event.plan;
598 const previous =
599 plan.kind === "todo_list"
600 ? yield* projections.getPlan(plan.threadId, plan.id)
601 : undefined;
602 const payload =
603 plan.kind === "todo_list"
604 ? withPlanStepDurations(
605 plan,
606 previous?.kind === "todo_list" ? previous : undefined,
607 occurredAt,
608 )
609 : plan;
610 return [
611 yield* makeDomainEvent(input, {
612 type: "plan.updated",
613 threadId: plan.threadId,
614 payload,
615 runId: plan.runId,
616 nodeId: plan.nodeId,
617 occurredAt,
618 }),
619 ];
620 }
621 case "turn.terminal":
622 const dismissed = yield* dismissNativeUserInputs(input, input.event.providerTurnId);
623 if (input.event.status !== "failed") {
624 return dismissed;
625 }
626 const occurredAt = yield* DateTime.now;
627 // A turn that failed because systemd killed the agent's scope for
628 // memory would otherwise read as a plain provider crash.
629 const failure = (yield* agentScope.oomKilled(input.threadId))
630 ? { ...input.event.failure, message: "Killed: out of memory.", code: "oom_kill" }
631 : input.event.failure;
632 return [
633 ...dismissed,
634 yield* makeDomainEvent(input, {
635 type: "turn-item.updated",
636 payload: makeProviderFailureTurnItem({
637 driver: input.event.driver,
638 threadId: input.threadId,
639 runId: input.runId ?? null,
640 nodeId: input.nodeId ?? null,
641 providerThreadId: input.event.providerThreadId,
642 providerTurnId: input.event.providerTurnId,
643 itemOrdinal: input.event.failureItemOrdinal,
644 failure,
645 ...(input.event.retry === undefined ? {} : { retry: input.event.retry }),
646 ...(input.event.retryStartedAt === undefined
647 ? {}
648 : { retryStartedAt: input.event.retryStartedAt }),
649 occurredAt,
650 }),
651 }),
652 ];
653 }
654 }).pipe(
655 Effect.mapError(
656 (cause) =>
657 new ProviderEventNormalizeError({
658 providerSessionId: input.providerSessionId,
659 threadId: input.threadId,
660 providerEvent: input.event,
661 cause,
662 }),
663 ),
664 );
665
666 return ProviderEventIngestorV2.of({
667 normalize,
668 ingestNormalized: (input) =>
669 Effect.gen(function* () {
670 const events = yield* normalize(input);
671 if (events.length === 0) {
672 return [];
673 }
674 const mapWriteError = (cause: unknown) =>
675 new ProviderEventPublishError({
676 providerSessionId: input.providerSessionId,
677 eventCount: events.length,
678 cause,
679 });
680 if (input.writeIfProviderThreadOwner !== undefined) {
681 const ownerResult = yield* eventSink
682 .writeIfProviderThreadOwner({
683 guardPendingUserInputCancellations: true,
684 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
685 ...input.writeIfProviderThreadOwner,
686 events,
687 })
688 .pipe(Effect.mapError(mapWriteError));
689 return ownerResult.storedEvents;
690 }
691 if (input.writeIfRunCurrent === undefined) {
692 return yield* eventSink
693 .write({
694 guardPendingUserInputCancellations: true,
695 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
696 events,
697 })
698 .pipe(Effect.mapError(mapWriteError));
699 }
700 const result = yield* eventSink
701 .writeIfRunCurrent({
702 guardPendingUserInputCancellations: true,
703 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
704 threadId: input.threadId,
705 ...input.writeIfRunCurrent,
706 events,
707 })
708 .pipe(Effect.mapError(mapWriteError));
709 return result.storedEvents;
710 }).pipe(
711 Effect.flatMap((storedEvents) =>
712 storedEvents.length === 0 || input.event.type !== "subagent.updated"
713 ? Effect.succeed(storedEvents)
714 : syncSubagentThreadModel(input, input.event.subagent).pipe(
715 Effect.map((synced) => [...storedEvents, ...synced]),
716 Effect.mapError(
717 (cause) =>
718 new ProviderEventPublishError({
719 providerSessionId: input.providerSessionId,
720 eventCount: 1,
721 cause,
722 }),
723 ),
724 ),
725 ),
726 (effect) =>
727 input.event.type === "app_thread.created"
728 ? threadCommands.withLock(input.event.appThread.id, effect)
729 : effect,
730 ),
731 });
732 }),
733);