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