1import {
2 NodeId,
3 CommandId,
4 OrchestrationV2DomainEvent,
5 OrchestrationV2StoredEvent,
6 type OrchestrationV2PlanArtifact,
7 type OrchestrationV2Run,
8 type OrchestrationV2Subagent,
9 type OrchestrationV2ThreadProjection,
10 type ModelSelection,
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";
25import * as SqlError from "effect/sql/SqlError";
26
27import { AgentScope } from "@t3tools/shared/AgentScope";
28import { isTemporaryWorktreeBranch } from "@t3tools/shared/git";
29import { modelSelectionsEqual } from "@t3tools/shared/model";
30import {
31 ProviderTurnAnalytics,
32 type ProviderTurnAnalyticsContext,
33} from "../telemetry/ProviderUsageAnalytics.ts";
34import * as GitVcsDriver from "../vcs/GitVcsDriver.ts";
35import * as ProviderInstanceRegistry from "../provider/ProviderInstanceRegistry.ts";
36import * as EventSink from "./EventSink.ts";
37import * as ProjectionStore from "./ProjectionStore.ts";
38import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
39import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
40import { makeProviderFailureTurnItem } from "@t3tools/provider-core/server/failure";
41import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
42import { stripUnservedToolOutputImageBytes } from "./toolOutputImageBytes.ts";
43import { boundFileChangeForStorage } from "@t3tools/shared/wireProjection";
44
45export class ProviderEventNormalizeError extends Schema.TaggedError<ProviderEventNormalizeError>()(
46 "ProviderEventNormalizeError",
47 {
48 providerSessionId: ProviderSessionId,
49 threadId: ThreadId,
50 providerEvent: ProviderAdapter.ProviderAdapterV2Event,
51 cause: Schema.optional(Schema.Defect()),
52 },
53) {
54 override get message(): string {
55 return `Failed to normalize provider event ${this.providerEvent.type} for thread ${this.threadId}.`;
56 }
57}
58
59export class ProviderEventPublishError extends Schema.TaggedError<ProviderEventPublishError>()(
60 "ProviderEventPublishError",
61 {
62 providerSessionId: ProviderSessionId,
63 eventCount: Schema.Number,
64 cause: Schema.optional(Schema.Defect()),
65 },
66) {
67 override get message(): string {
68 return `Failed to publish ${this.eventCount} normalized provider event(s).`;
69 }
70}
71
72// A write that outlasts busy_timeout at BEGIN IMMEDIATE fails before anything
73// commits, and the sink publishes only after a commit, so repeating the whole
74// write is safe. Failing it would fail the run over a brief lock.
75const isLockTimeout = (error: EventSink.EventSinkV2Error) =>
76 SqlError.isSqlError(error.cause) && error.cause.reason._tag === "LockTimeoutError";
77const retryLockTimeout = <A, R>(write: Effect.Effect<A, EventSink.EventSinkV2Error, R>) =>
78 Effect.retry(write, { while: isLockTimeout, times: 2 });
79
80export const ProviderEventIngestorV2Error = Schema.Union([
81 ProviderEventNormalizeError,
82 ProviderEventPublishError,
83]);
84export type ProviderEventIngestorV2Error = typeof ProviderEventIngestorV2Error.Type;
85
86type TodoListPlan = Extract<OrchestrationV2PlanArtifact, { readonly kind: "todo_list" }>;
87
88function withPlanStepDurations(
89 plan: TodoListPlan,
90 previous: TodoListPlan | undefined,
91 occurredAt: DateTime.Utc,
92): TodoListPlan {
93 const occurredAtIso = DateTime.formatIso(occurredAt);
94 const occurredAtMs = DateTime.toEpochMillis(occurredAt);
95 const previousById = new Map(previous?.steps.map((step) => [step.id, step]));
96 // Provider step IDs may be positional. Changed text must not inherit another task's timing.
97 const previousStep = (step: TodoListPlan["steps"][number]) => {
98 const prior = previousById.get(step.id);
99 return prior?.text === step.text ? prior : undefined;
100 };
101 const hasNewCompletion = plan.steps.some((step) => {
102 const prior = previousStep(step);
103 return step.status === "completed" && prior?.status !== "completed";
104 });
105 let fallbackCompletionConsumed = false;
106
107 return {
108 ...plan,
109 steps: plan.steps.map((step) => {
110 const prior = previousStep(step);
111 const baseStep = { id: step.id, text: step.text, status: step.status };
112 if (step.status === "completed") {
113 if (prior?.status === "completed") {
114 return {
115 ...baseStep,
116 ...(prior.durationMs === undefined ? {} : { durationMs: prior.durationMs }),
117 };
118 }
119 const durationAnchorAt =
120 prior?.status === "running" || !fallbackCompletionConsumed
121 ? prior?.durationAnchorAt
122 : occurredAtIso;
123 fallbackCompletionConsumed = true;
124 const anchorMs =
125 durationAnchorAt === undefined ? occurredAtMs : Date.parse(durationAnchorAt);
126 const durationMs = Number.isFinite(anchorMs) ? Math.max(0, occurredAtMs - anchorMs) : 0;
127 return {
128 ...baseStep,
129 ...(durationMs > 0 ? { durationMs } : {}),
130 };
131 }
132 if (step.status === "running") {
133 return {
134 ...baseStep,
135 durationAnchorAt:
136 prior?.status === "running" ? (prior.durationAnchorAt ?? occurredAtIso) : occurredAtIso,
137 };
138 }
139 return {
140 ...baseStep,
141 durationAnchorAt:
142 hasNewCompletion || prior?.status !== "pending"
143 ? occurredAtIso
144 : (prior.durationAnchorAt ?? occurredAtIso),
145 };
146 }),
147 };
148}
149
150const CONVERSATION_RUN_STATUSES = new Set<OrchestrationV2Run["status"]>([
151 "completed",
152 "interrupted",
153 "failed",
154]);
155
156function runsOffNativeBranch(input: {
157 readonly projection: Pick<OrchestrationV2ThreadProjection, "runs" | "attempts" | "providerTurns">;
158 readonly providerThreadId: ProviderThreadId;
159 readonly retainedNativeTurnIds: ReadonlySet<string>;
160 readonly reportingRunId: RunId | undefined;
161}): ReadonlyArray<OrchestrationV2Run> {
162 const { attempts, providerTurns, runs } = input.projection;
163 const nativeTurnId = (run: OrchestrationV2Run) => {
164 const attempt = attempts.find((candidate) => candidate.id === run.activeAttemptId);
165 if (attempt === undefined) return null;
166 const ref = providerTurns.find(
167 (turn) => turn.id === attempt.providerTurnId || turn.runAttemptId === attempt.id,
168 )?.nativeTurnRef;
169 return ref?.strength === "strong" ? ref.nativeId : null;
170 };
171 const settled = runs.flatMap((run) =>
172 run.providerThreadId === input.providerThreadId &&
173 run.id !== input.reportingRunId &&
174 CONVERSATION_RUN_STATUSES.has(run.status)
175 ? [{ run, nativeTurnId: nativeTurnId(run) }]
176 : [],
177 );
178 const lastRetainedOrdinal = settled.reduce(
179 (last, { run, nativeTurnId }) =>
180 nativeTurnId !== null && input.retainedNativeTurnIds.has(nativeTurnId)
181 ? Math.max(last, run.ordinal)
182 : last,
183 0,
184 );
185 return settled
186 .filter(({ run, nativeTurnId }) =>
187 nativeTurnId === null
188 ? run.ordinal > lastRetainedOrdinal
189 : !input.retainedNativeTurnIds.has(nativeTurnId),
190 )
191 .map(({ run }) => run);
192}
193
194export interface ProviderEventIngestInput {
195 readonly providerSessionId: ProviderSessionId;
196 readonly providerInstanceId: ProviderInstanceId;
197 readonly commandId?: CommandId;
198 readonly threadId: ThreadId;
199 readonly runId?: RunId;
200 readonly nodeId?: NodeId;
201 readonly rawEventId?: RawEventId;
202 readonly event: ProviderAdapter.ProviderAdapterV2Event;
203 readonly analyticsContext?: ProviderTurnAnalyticsContext;
204}
205
206export interface ProviderEventIngestorV2Shape {
207 readonly normalize: (
208 input: ProviderEventIngestInput,
209 ) => Effect.Effect<ReadonlyArray<OrchestrationV2DomainEvent>, ProviderEventIngestorV2Error>;
210 readonly ingestNormalized: (
211 input: ProviderEventIngestInput & {
212 readonly writeIfRunCurrent?: {
213 readonly runId: RunId;
214 readonly activeAttemptId: RunAttemptId;
215 readonly expectedStatus: OrchestrationV2Run["status"];
216 };
217 readonly writeIfProviderThreadOwner?: {
218 readonly providerThreadId: ProviderThreadId;
219 readonly runId: RunId;
220 readonly activeAttemptId: RunAttemptId;
221 readonly expectedLastRunOrdinal: number;
222 };
223 },
224 ) => Effect.Effect<ReadonlyArray<OrchestrationV2StoredEvent>, ProviderEventIngestorV2Error>;
225}
226
227export class ProviderEventIngestorV2 extends Context.Service<
228 ProviderEventIngestorV2,
229 ProviderEventIngestorV2Shape
230>()("t3/orchestration-v2/ProviderEventIngestor/ProviderEventIngestorV2") {}
231
232function compactUndefined<T extends Record<string, unknown>>(record: T): T {
233 return Object.fromEntries(Object.entries(record).filter(([, value]) => value !== undefined)) as T;
234}
235
236const decodeDomainEvent = Schema.decodeUnknownEffect(OrchestrationV2DomainEvent);
237
238export const layer: Layer.Layer<
239 ProviderEventIngestorV2,
240 never,
241 | EventSink.EventSinkV2
242 | GitVcsDriver.GitVcsDriver
243 | IdAllocator.IdAllocatorV2
244 | ProjectionStore.ProjectionStoreV2
245 | ThreadCommandExecutor.ThreadCommandExecutor
246 | ProviderInstanceRegistry.ProviderInstanceRegistry
247> = Layer.effect(
248 ProviderEventIngestorV2,
249 Effect.gen(function* () {
250 const eventSink = yield* EventSink.EventSinkV2;
251 const git = yield* GitVcsDriver.GitVcsDriver;
252 const projections = yield* ProjectionStore.ProjectionStoreV2;
253 const idAllocator = yield* IdAllocator.IdAllocatorV2;
254 const threadCommands = yield* ThreadCommandExecutor.ThreadCommandExecutor;
255 const analytics = yield* ProviderTurnAnalytics;
256 const agentScope = yield* AgentScope;
257 const instances = yield* ProviderInstanceRegistry.ProviderInstanceRegistry;
258 const scope = yield* Effect.scope;
259 const completedTurnAnalytics = new Set<string>();
260
261 const makeDomainEvent =
262 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
263 Effect.fnUntraced(function* (
264 input: ProviderEventIngestInput,
265 payloadInput: {
266 readonly type: OrchestrationV2DomainEvent["type"];
267 readonly payload: OrchestrationV2DomainEvent["payload"];
268 readonly threadId?: ThreadId;
269 readonly runId?: RunId | null;
270 readonly nodeId?: NodeId | null;
271 readonly occurredAt?: DateTime.Utc;
272 },
273 ) {
274 const threadId = payloadInput.threadId ?? input.threadId;
275 const eventId = yield* idAllocator.allocate.event({
276 threadId,
277 providerSessionId: input.providerSessionId,
278 });
279 const occurredAt = payloadInput.occurredAt ?? (yield* DateTime.now);
280 return yield* decodeDomainEvent(
281 compactUndefined({
282 id: eventId,
283 type: payloadInput.type,
284 threadId,
285 runId: payloadInput.runId ?? input.runId,
286 nodeId: payloadInput.nodeId ?? input.nodeId,
287 driver: input.event.driver,
288 providerInstanceId: input.providerInstanceId,
289 rawEventId: input.rawEventId,
290 occurredAt,
291 payload: payloadInput.payload,
292 }),
293 );
294 });
295
296 const dismissNativeUserInputs = Effect.fn("ProviderEventIngestor.dismissNativeUserInputs")(
297 function* (
298 input: ProviderEventIngestInput,
299 providerTurnId: ProviderTurnId,
300 threadId = input.threadId,
301 ) {
302 const pending = yield* projections.getPendingNativeUserInputs(threadId, providerTurnId);
303 const now = yield* DateTime.now;
304 const events: Array<OrchestrationV2DomainEvent> = [];
305 for (const request of pending.runtimeRequests) {
306 events.push(
307 yield* makeDomainEvent(input, {
308 type: "runtime-request.updated",
309 threadId,
310 nodeId: request.nodeId,
311 payload: { ...request, status: "cancelled", resolvedAt: now },
312 }),
313 );
314 }
315 for (const node of pending.nodes) {
316 events.push(
317 yield* makeDomainEvent(input, {
318 type: "node.updated",
319 threadId,
320 nodeId: node.id,
321 runId: node.runId,
322 payload: { ...node, status: "cancelled", completedAt: now },
323 }),
324 );
325 }
326 for (const item of pending.turnItems) {
327 events.push(
328 yield* makeDomainEvent(input, {
329 type: "turn-item.updated",
330 threadId,
331 nodeId: item.nodeId,
332 runId: item.runId,
333 payload: { ...item, status: "cancelled", completedAt: now, updatedAt: now },
334 }),
335 );
336 }
337 return events;
338 },
339 );
340
341 /** Providers may reveal native subagent models late; lock full-row updates against concurrent thread commands. */
342 const syncSubagentThreadModel = Effect.fn("ProviderEventIngestor.syncSubagentThreadModel")(
343 function* (input: ProviderEventIngestInput, subagent: OrchestrationV2Subagent) {
344 const { childThreadId, model } = subagent;
345 if (
346 subagent.origin !== "provider_native" ||
347 childThreadId === null ||
348 (model === null && subagent.modelSelection === undefined)
349 ) {
350 return [];
351 }
352 const reportedSelection = ({ modelSelection }: { modelSelection: ModelSelection }) =>
353 subagent.modelSelection ??
354 (model === null || model === modelSelection.model
355 ? modelSelection
356 : { instanceId: modelSelection.instanceId, model });
357 const staleThread = projections.getThread(childThreadId).pipe(
358 Effect.map((thread) =>
359 modelSelectionsEqual(thread.modelSelection, reportedSelection(thread)) ? null : thread,
360 ),
361 Effect.catchTags({ ProjectionStoreThreadNotFoundError: () => Effect.succeed(null) }),
362 );
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] }).pipe(retryLockTimeout);
381 }),
382 );
383 },
384 );
385
386 /** Retire rewound runs like CheckpointRollbackService; retain the reporting run and its lastRunOrdinal. */
387 const rollBackRewoundRuns = Effect.fn("ProviderEventIngestor.rollBackRewoundRuns")(function* (
388 input: ProviderEventIngestInput,
389 rewind: {
390 readonly threadId: ThreadId;
391 readonly providerThreadId: ProviderThreadId;
392 readonly retainedNativeTurnIds: ReadonlySet<string>;
393 },
394 ) {
395 const projection = yield* projections.getThreadRecords(rewind.threadId, [
396 "runs",
397 "attempts",
398 "providerTurns",
399 "nodes",
400 "checkpoints",
401 ]);
402 const runs = runsOffNativeBranch({
403 projection,
404 providerThreadId: rewind.providerThreadId,
405 retainedNativeTurnIds: rewind.retainedNativeTurnIds,
406 reportingRunId: input.runId,
407 });
408 if (runs.length === 0) return [];
409 const now = yield* DateTime.now;
410 const eventId = () =>
411 idAllocator.allocate.event({
412 threadId: rewind.threadId,
413 providerSessionId: input.providerSessionId,
414 });
415 const runIds = new Set(runs.map((run) => run.id));
416 const events: Array<OrchestrationV2DomainEvent> = [];
417 for (const checkpoint of projection.checkpoints) {
418 if (checkpoint.status !== "ready" || checkpoint.runId === null) continue;
419 if (!runIds.has(checkpoint.runId)) continue;
420 events.push({
421 id: yield* eventId(),
422 type: "checkpoint.captured",
423 threadId: rewind.threadId,
424 runId: checkpoint.runId,
425 nodeId: checkpoint.nodeId,
426 providerInstanceId: input.providerInstanceId,
427 occurredAt: now,
428 payload: { ...checkpoint, status: "stale" },
429 });
430 }
431 for (const run of runs) {
432 const rootNode = projection.nodes.find((node) => node.id === run.rootNodeId);
433 events.push({
434 id: yield* eventId(),
435 type: "run.updated",
436 threadId: rewind.threadId,
437 runId: run.id,
438 ...(rootNode === undefined ? {} : { nodeId: rootNode.id }),
439 providerInstanceId: run.providerInstanceId,
440 occurredAt: now,
441 payload: { ...run, status: "rolled_back", completedAt: now },
442 });
443 if (rootNode !== undefined) {
444 events.push({
445 id: yield* eventId(),
446 type: "node.updated",
447 threadId: rewind.threadId,
448 runId: run.id,
449 nodeId: rootNode.id,
450 providerInstanceId: run.providerInstanceId,
451 occurredAt: now,
452 payload: { ...rootNode, status: "rolled_back", completedAt: now },
453 });
454 }
455 }
456 return events;
457 });
458
459 const normalize: ProviderEventIngestorV2Shape["normalize"] =
460 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
461 Effect.fnUntraced(
462 function* (input) {
463 switch (input.event.type) {
464 case "app_thread.created": {
465 const offered = input.event.appThread;
466 const existing = yield* projections.getThread(offered.id).pipe(
467 Effect.catchTags({
468 ProjectionStoreThreadNotFoundError: () => Effect.succeed(null),
469 }),
470 );
471 if (
472 existing !== null &&
473 (input.event.createOnly === true ||
474 (existing.title === offered.title &&
475 modelSelectionsEqual(existing.modelSelection, offered.modelSelection)))
476 )
477 return [];
478 const now = yield* DateTime.now;
479 return [
480 yield* makeDomainEvent(input, {
481 type: existing === null ? "thread.created" : "thread.metadata-updated",
482 threadId: offered.id,
483 payload:
484 existing === null
485 ? offered
486 : {
487 ...existing,
488 title: offered.title,
489 modelSelection: offered.modelSelection,
490 updatedAt: now,
491 },
492 occurredAt: now,
493 }),
494 ];
495 }
496 case "provider_session.updated":
497 return [
498 yield* makeDomainEvent(input, {
499 type: "provider-session.updated",
500 payload: input.event.providerSession,
501 }),
502 ];
503 case "provider_thread.updated": {
504 const threadId = input.event.providerThread.appThreadId ?? input.threadId;
505 const updated = yield* makeDomainEvent(input, {
506 type: "provider-thread.updated",
507 threadId,
508 payload: input.event.providerThread,
509 });
510 return input.event.retainedNativeTurnIds === undefined
511 ? [updated]
512 : [
513 updated,
514 ...(yield* rollBackRewoundRuns(input, {
515 threadId,
516 providerThreadId: input.event.providerThread.id,
517 retainedNativeTurnIds: new Set(input.event.retainedNativeTurnIds),
518 })),
519 ];
520 }
521 case "provider_turn.updated":
522 return [
523 ...(["completed", "interrupted", "failed", "cancelled"].includes(
524 input.event.providerTurn.status,
525 )
526 ? yield* dismissNativeUserInputs(
527 input,
528 input.event.providerTurn.id,
529 input.event.threadId,
530 )
531 : []),
532 yield* makeDomainEvent(input, {
533 type: "provider-turn.updated",
534 ...(input.event.threadId === undefined ? {} : { threadId: input.event.threadId }),
535 payload: input.event.providerTurn,
536 nodeId: input.event.providerTurn.nodeId,
537 }),
538 ];
539 case "node.updated":
540 // Adapters send an unchanged node again with each text delta. Compare with the
541 // stored node, not the last one sent, because the server also writes nodes.
542 if (yield* projections.isNodeStored(input.event.node)) return [];
543 return [
544 yield* makeDomainEvent(input, {
545 type: "node.updated",
546 threadId: input.event.node.threadId,
547 payload: input.event.node,
548 runId: input.event.node.runId,
549 nodeId: input.event.node.id,
550 }),
551 ];
552 case "subagent.updated":
553 return [
554 yield* makeDomainEvent(input, {
555 type: "subagent.updated",
556 threadId: input.event.subagent.threadId,
557 payload: input.event.subagent,
558 runId: input.event.subagent.runId,
559 nodeId: input.event.subagent.id,
560 }),
561 ];
562 case "message.updated":
563 return [
564 yield* makeDomainEvent(input, {
565 type: "message.updated",
566 threadId: input.event.message.threadId,
567 payload: input.event.message,
568 runId: input.event.message.runId,
569 nodeId: input.event.message.nodeId,
570 }),
571 ];
572 case "turn_item.updated":
573 return [
574 yield* makeDomainEvent(input, {
575 type: "turn-item.updated",
576 threadId: input.event.turnItem.threadId,
577 payload: boundFileChangeForStorage(
578 stripUnservedToolOutputImageBytes(input.event.turnItem),
579 ),
580 runId: input.event.turnItem.runId,
581 nodeId: input.event.turnItem.nodeId,
582 }),
583 ];
584 case "runtime_request.updated":
585 return [
586 yield* makeDomainEvent(input, {
587 type: "runtime-request.updated",
588 ...(input.event.threadId === undefined ? {} : { threadId: input.event.threadId }),
589 payload: input.event.runtimeRequest,
590 nodeId: input.event.runtimeRequest.nodeId,
591 }),
592 ];
593 case "plan.updated": {
594 const occurredAt = yield* DateTime.now;
595 const plan = input.event.plan;
596 const previous =
597 plan.kind === "todo_list"
598 ? yield* projections.getPlan(plan.threadId, plan.id)
599 : undefined;
600 const payload =
601 plan.kind === "todo_list"
602 ? withPlanStepDurations(
603 plan,
604 previous?.kind === "todo_list" ? previous : undefined,
605 occurredAt,
606 )
607 : plan;
608 return [
609 yield* makeDomainEvent(input, {
610 type: "plan.updated",
611 threadId: plan.threadId,
612 payload,
613 runId: plan.runId,
614 nodeId: plan.nodeId,
615 occurredAt,
616 }),
617 ];
618 }
619 case "steer.updated": {
620 const item = yield* projections.getTurnItem({
621 threadId: input.event.threadId,
622 itemId: idAllocator.derive.userTurnItem({ messageId: input.event.messageId }),
623 });
624 if (
625 item?.type !== "user_message" ||
626 item.status === input.event.status ||
627 (item.status !== "pending" && item.status !== "completed")
628 ) {
629 return [];
630 }
631 const occurredAt = yield* DateTime.now;
632 return [
633 yield* makeDomainEvent(input, {
634 type: "turn-item.updated",
635 threadId: item.threadId,
636 payload: {
637 ...item,
638 status: input.event.status,
639 completedAt: input.event.status === "pending" ? null : occurredAt,
640 updatedAt: occurredAt,
641 },
642 runId: item.runId,
643 nodeId: item.nodeId,
644 occurredAt,
645 }),
646 ];
647 }
648 case "background_work.updated":
649 return [];
650 case "command_output.delta":
651 case "prompt_suggestion.updated":
652 // The session manager serves these to clients before ingestion.
653 return [];
654 case "turn.terminal":
655 const dismissed = yield* dismissNativeUserInputs(input, input.event.providerTurnId);
656 if (input.event.status !== "failed") {
657 return dismissed;
658 }
659 const occurredAt = yield* DateTime.now;
660 const failure = (yield* agentScope.oomKilled(input.threadId))
661 ? { ...input.event.failure, message: "Killed: out of memory.", code: "oom_kill" }
662 : input.event.failure;
663 return [
664 ...dismissed,
665 yield* makeDomainEvent(input, {
666 type: "turn-item.updated",
667 payload: makeProviderFailureTurnItem({
668 driver: input.event.driver,
669 threadId: input.threadId,
670 runId: input.runId ?? null,
671 nodeId: input.nodeId ?? null,
672 providerThreadId: input.event.providerThreadId,
673 providerTurnId: input.event.providerTurnId,
674 itemOrdinal: input.event.failureItemOrdinal,
675 failure,
676 ...(input.event.retry === undefined ? {} : { retry: input.event.retry }),
677 ...(input.event.retryStartedAt === undefined
678 ? {}
679 : { retryStartedAt: input.event.retryStartedAt }),
680 occurredAt,
681 }),
682 }),
683 ];
684 }
685 },
686 (effect, input) =>
687 effect.pipe(
688 Effect.mapError(
689 (cause) =>
690 new ProviderEventNormalizeError({
691 providerSessionId: input.providerSessionId,
692 threadId: input.threadId,
693 providerEvent: input.event,
694 cause,
695 }),
696 ),
697 ),
698 );
699
700 return ProviderEventIngestorV2.of({
701 normalize,
702 ingestNormalized:
703 /* This helper runs for individual orchestration events; the enclosing operation owns the span. */
704 Effect.fnUntraced(
705 function* (input) {
706 const events = yield* normalize(input);
707 if (events.length === 0) {
708 return [];
709 }
710 const mapWriteError = (cause: unknown) =>
711 new ProviderEventPublishError({
712 providerSessionId: input.providerSessionId,
713 eventCount: events.length,
714 cause,
715 });
716 if (input.writeIfProviderThreadOwner !== undefined) {
717 const ownerResult = yield* eventSink
718 .writeIfProviderThreadOwner({
719 guardPendingUserInputCancellations: true,
720 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
721 ...input.writeIfProviderThreadOwner,
722 events,
723 })
724 .pipe(retryLockTimeout, Effect.mapError(mapWriteError));
725 return ownerResult.storedEvents;
726 }
727 if (input.writeIfRunCurrent === undefined) {
728 return yield* eventSink
729 .write({
730 guardPendingUserInputCancellations: true,
731 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
732 events,
733 })
734 .pipe(retryLockTimeout, Effect.mapError(mapWriteError));
735 }
736 const result = yield* eventSink
737 .writeIfRunCurrent({
738 guardPendingUserInputCancellations: true,
739 ...(input.commandId === undefined ? {} : { commandId: input.commandId }),
740 threadId: input.threadId,
741 ...input.writeIfRunCurrent,
742 events,
743 })
744 .pipe(retryLockTimeout, Effect.mapError(mapWriteError));
745 return result.storedEvents;
746 },
747 (effect, input) =>
748 effect.pipe(
749 Effect.tap((storedEvents) =>
750 Effect.gen(function* () {
751 if (
752 storedEvents.length === 0 ||
753 input.event.type !== "turn.terminal" ||
754 input.event.status !== "failed" ||
755 input.event.failure.class !== "usage_limit"
756 )
757 return;
758 const instance = yield* instances.getInstance(input.providerInstanceId);
759 if (!instance?.snapshot.refreshUsage) return;
760 yield* instance.snapshot.refreshUsage.pipe(
761 Effect.ignoreCause({ log: true }),
762 Effect.forkIn(scope, { startImmediately: true }),
763 );
764 }),
765 ),
766 Effect.flatMap((storedEvents) =>
767 Effect.gen(function* () {
768 if (input.event.type !== "app_thread.created") return storedEvents;
769 const created = storedEvents.find(
770 (stored) => stored.event.type === "thread.created",
771 );
772 if (created?.event.type !== "thread.created") return storedEvents;
773 const child = created.event.payload;
774 if (
775 child.worktreePath === null ||
776 child.branch === null ||
777 !isTemporaryWorktreeBranch(child.branch)
778 )
779 return storedEvents;
780 // Git renames before scanning committed threads. If that scan missed this
781 // creation, HEAD already has the new name, even if the parent moved away.
782 const branch = yield* git.resolveBranchName(child.worktreePath, "HEAD");
783 if (branch === null || branch === child.branch) return storedEvents;
784 const now = yield* DateTime.now;
785 const event = yield* makeDomainEvent(input, {
786 type: "thread.metadata-updated",
787 threadId: child.id,
788 payload: { ...child, branch, updatedAt: now },
789 occurredAt: now,
790 });
791 const synced = yield* eventSink.write({ events: [event] });
792 return [...storedEvents, ...synced];
793 }).pipe(
794 Effect.catchCause((cause) =>
795 Effect.logWarning("Native thread worktree branch recheck failed", {
796 threadId:
797 input.event.type === "app_thread.created"
798 ? input.event.appThread.id
799 : input.threadId,
800 cause,
801 }).pipe(Effect.as(storedEvents)),
802 ),
803 ),
804 ),
805 // @effect-diagnostics-next-line flatMapConditionalToFilterOrFail:off - The predicate narrows the captured event for the subagent update.
806 Effect.flatMap((storedEvents) =>
807 storedEvents.length === 0 || input.event.type !== "subagent.updated"
808 ? Effect.succeed(storedEvents)
809 : syncSubagentThreadModel(input, input.event.subagent).pipe(
810 Effect.map((synced) => [...storedEvents, ...synced]),
811 Effect.mapError(
812 (cause) =>
813 new ProviderEventPublishError({
814 providerSessionId: input.providerSessionId,
815 eventCount: 1,
816 cause,
817 }),
818 ),
819 ),
820 ),
821 Effect.tap((storedEvents) =>
822 Effect.gen(function* () {
823 if (storedEvents.length === 0 || input.event.type !== "provider_turn.updated")
824 return;
825 const providerTurn = input.event.providerTurn;
826 if (
827 providerTurn.status !== "completed" &&
828 providerTurn.status !== "failed" &&
829 providerTurn.status !== "interrupted" &&
830 providerTurn.status !== "cancelled"
831 )
832 return;
833 const key = `${input.providerInstanceId}:${providerTurn.id}`;
834 if (completedTurnAnalytics.has(key)) return;
835 completedTurnAnalytics.add(key);
836 if (completedTurnAnalytics.size > 4096) {
837 const oldest = completedTurnAnalytics.values().next().value;
838 if (oldest !== undefined) completedTurnAnalytics.delete(oldest);
839 }
840 yield* analytics.turnCompleted({
841 driver: input.event.driver,
842 providerTurn,
843 ...(input.analyticsContext === undefined
844 ? {}
845 : { context: input.analyticsContext }),
846 });
847 }),
848 ),
849 (effect) =>
850 input.event.type === "app_thread.created"
851 ? threadCommands.withLock(input.event.appThread.id, effect)
852 : effect,
853 ),
854 ),
855 });
856 }),
857);