clover's git

the-knife

1	import { makeAssistantStreamingFilter } from "./assistantStreaming.ts";
2	import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
3	import {
4	isOrchestrationV2WorkActive,
5	CommandId,
6	type EventId,
7	type ModelSelection,
8	type NodeId,
9	type OrchestrationV2AppThread,
10	type OrchestrationV2CheckpointScope,
11	type OrchestrationV2DomainEvent,
12	type OrchestrationV2ExecutionNode,
13	type OrchestrationV2ProviderFailure,
14	type OrchestrationV2ProviderThread,
15	type OrchestrationV2ProviderTurn,
16	type OrchestrationV2Run,
17	type OrchestrationV2RunAttempt,
18	type OrchestrationV2Subagent,
19	type OrchestrationV2TurnItem,
20	type ProviderSessionId,
21	type ProviderThreadId,
22	type ProviderTurnId,
23	type RunAttemptId,
24	type ThreadId,
25	type TurnItemId,
26	} from "@t3tools/contracts";
27	import * as Context from "effect/Context";
28	import * as Cause from "effect/Cause";
29	import * as DateTime from "effect/DateTime";
30	import * as Effect from "effect/Effect";
31	import * as Exit from "effect/Exit";
32	import * as Fiber from "effect/Fiber";
33	import * as Layer from "effect/Layer";
34	import * as Ref from "effect/Ref";
35	import * as Schema from "effect/Schema";
36	import * as Stream from "effect/Stream";
37
38	import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
39	import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts";
40	import * as ServerSettings from "../serverSettings.ts";
41	import * as CheckpointService from "./CheckpointService.ts";
42	import * as EventSink from "./EventSink.ts";
43	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
44	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
45	import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
46	import type { ProjectionStoreV2Error } from "./ProjectionStore.ts";
47	import {
48	makeProviderFailure,
49	makeProviderFailureTurnItem,
50	} from "@t3tools/provider-core/server/failure";
51	import * as RunFinalizationService from "./RunFinalizationService.ts";
52
53	export interface ProviderEventRoutingState {
54	readonly ownedThreadIds: ReadonlySet<ThreadId>;
55	// Set once this run's root turn ended. A child thread created after that
56	// belongs to the run that is live then, so this one no longer adopts it,
57	// except a workflow member enrolled below: its coordinator keeps this run
58	// open, so the member is this run's to follow.
59	readonly rootTurnEnded: boolean;
60	/** Member thread id to the coordinator child thread that spawned it. */
61	readonly workflowMemberParents: ReadonlyMap<ThreadId, ThreadId>;
62	readonly ownedProviderThreadIds: ReadonlySet<ProviderThreadId>;
63	readonly ownedProviderTurnIds: ReadonlySet<ProviderTurnId>;
64	readonly inheritedBackgroundTurnItems: ReadonlyMap<TurnItemId, OrchestrationV2Run["id"]>;
65	readonly rootProviderTurnId: ProviderTurnId | null;
66	}
67
68	export interface ProviderEventRouteIdentity {
69	readonly threadId: ThreadId;
70	readonly runId: OrchestrationV2Run["id"];
71	readonly attemptId: RunAttemptId;
72	readonly providerThreadId: ProviderThreadId;
73	}
74
75	export interface InheritedBackgroundTurnItemRoute {
76	readonly id: TurnItemId;
77	readonly runId: OrchestrationV2Run["id"];
78	}
79
80	type ProviderTerminalEvent = Extract<
81	ProviderAdapter.ProviderAdapterV2Event,
82	{ readonly type: "turn.terminal" }
83	>;
84
85	function isTerminalProviderTurnStatus(status: OrchestrationV2ProviderTurn["status"]): boolean {
86	return (
87	status === "completed" ||
88	status === "interrupted" ||
89	status === "failed" ||
90	status === "cancelled"
91	);
92	}
93
94	function isSettledSubagentStatus(status: OrchestrationV2Subagent["status"]): boolean {
95	return !isOrchestrationV2WorkActive(status);
96	}
97
98	// Turn item types whose lifecycle can outlive the root turn (background
99	// commands, monitors/dynamic tools, subagent rows). Ingestion must not stop
100	// while one of these is still non-terminal, or the late completion event is
101	// dropped and the item spins forever in the projection.
102	const backgroundCapableTurnItemTypes: ReadonlySet<OrchestrationV2TurnItem["type"]> = new Set([
103	"command_execution",
104	"dynamic_tool",
105	"subagent",
106	]);
107
108	function isSettledTurnItemStatus(status: OrchestrationV2TurnItem["status"]): boolean {
109	return !isOrchestrationV2WorkActive(status);
110	}
111
112	function isSettledRunEligibleForInheritedBackground(status: OrchestrationV2Run["status"]): boolean {
113	return status === "interrupted" || status === "failed" || status === "cancelled";
114	}
115
116	/**
117	* Transfer delivery permission for exact live background items whose original
118	* run no longer has a subscriber. Provider sessions are runtime containers and
119	* can host multiple provider threads, so the durable provider-thread lineage is
120	* the discriminator. Completed, rolled-back, and already-terminal items remain
121	* excluded.
122	*/
123	export function selectInheritedBackgroundTurnItems(input: {
124	readonly threadId: ThreadId;
125	readonly currentProviderThreadId: ProviderThreadId;
126	readonly currentRunOrdinal: number;
127	readonly runs: ReadonlyArray<OrchestrationV2Run>;
128	readonly turnItems: ReadonlyArray<OrchestrationV2TurnItem>;
129	}): ReadonlyArray<InheritedBackgroundTurnItemRoute> {
130	const settledPriorRunIds = new Set(
131	input.runs
132	.filter(
133	(run) =>
134	run.threadId === input.threadId &&
135	run.ordinal < input.currentRunOrdinal &&
136	isSettledRunEligibleForInheritedBackground(run.status),
137	)
138	.map((run) => run.id),
139	);
140	return input.turnItems.flatMap((turnItem) =>
141	turnItem.threadId === input.threadId &&
142	turnItem.providerThreadId === input.currentProviderThreadId &&
143	turnItem.runId !== null &&
144	settledPriorRunIds.has(turnItem.runId) &&
145	backgroundCapableTurnItemTypes.has(turnItem.type) &&
146	!isSettledTurnItemStatus(turnItem.status)
147	? [{ id: turnItem.id, runId: turnItem.runId }]
148	: [],
149	);
150	}
151
152	type SubagentTurnItem = Extract<OrchestrationV2TurnItem, { readonly type: "subagent" }>;
153
154	type OpenRunOwnedSubagentProjection = {
155	readonly subagents: ReadonlyMap<NodeId, OrchestrationV2Subagent>;
156	readonly turnItems: ReadonlyMap<NodeId, SubagentTurnItem>;
157	readonly childTurnItems: ReadonlyMap<TurnItemId, OrchestrationV2TurnItem>;
158	readonly nodes: ReadonlyMap<NodeId, OrchestrationV2ExecutionNode>;
159	/** Child threads once linked by a root-run subagent row; kept for cascade. */
160	readonly linkedChildThreadIds: ReadonlySet<ThreadId>;
161	};
162
163	type RunOwnedSubagentTerminalStatus = Extract<
164	OrchestrationV2Subagent["status"],
165	"interrupted" | "failed" | "cancelled"
166	>;
167
168	function isOpenExecutionNodeStatus(status: OrchestrationV2ExecutionNode["status"]): boolean {
169	return status === "pending" || status === "running" || status === "waiting";
170	}
171
172	function isRunOwnedSubagentTerminalStatus(
173	status: ProviderTerminalEvent["status"],
174	): status is RunOwnedSubagentTerminalStatus {
175	return status === "interrupted" || status === "failed" || status === "cancelled";
176	}
177
178	/**
179	* Whether a new run takes over a subagent's child thread, so a later message
180	* can resume it there. A running subagent stays with the run that launched it,
181	* which keeps ingesting until it ends; taking it over too would store its
182	* events twice. An interrupted, failed or cancelled one is never resumed.
183	*/
184	export function canRouteRelatedSubagent(status: OrchestrationV2Subagent["status"]): boolean {
185	return status === "completed";
186	}
187
188	function emptyOpenRunOwnedSubagentProjection(): OpenRunOwnedSubagentProjection {
189	return {
190	subagents: new Map(),
191	turnItems: new Map(),
192	childTurnItems: new Map(),
193	nodes: new Map(),
194	linkedChildThreadIds: new Set(),
195	};
196	}
197
198	function withLinkedChildThreadId(
199	current: OpenRunOwnedSubagentProjection,
200	childThreadId: ThreadId | null,
201	): OpenRunOwnedSubagentProjection {
202	if (childThreadId === null || current.linkedChildThreadIds.has(childThreadId)) {
203	return current;
204	}
205	const linkedChildThreadIds = new Set(current.linkedChildThreadIds);
206	linkedChildThreadIds.add(childThreadId);
207	return { ...current, linkedChildThreadIds };
208	}
209
210	export function cascadeTerminalizeRunOwnedSubagents(input: {
211	readonly run: OrchestrationV2Run;
212	readonly open: OpenRunOwnedSubagentProjection;
213	readonly status: RunOwnedSubagentTerminalStatus;
214	readonly completedAt: DateTime.Utc;
215	readonly allocateEventId: () => Effect.Effect<EventId, IdAllocator.IdAllocatorV2AllocationError>;
216	}): Effect.Effect<
217	ReadonlyArray<OrchestrationV2DomainEvent>,
218	IdAllocator.IdAllocatorV2AllocationError
219	> {
220	return Effect.gen(function* () {
221	const events: Array<OrchestrationV2DomainEvent> = [];
222	// Prefer lifetime linkage over currently-open rows: subagent/turn-item
223	// snapshots may terminalize before the linked child-thread node settles.
224	const childThreadIds = new Set(input.open.linkedChildThreadIds);
225	for (const item of [...input.open.subagents.values(), ...input.open.turnItems.values()]) {
226	if (item.childThreadId !== null) {
227	childThreadIds.add(item.childThreadId);
228	}
229	}
230	const keys = new Set<NodeId>([
231	...input.open.subagents.keys(),
232	...input.open.turnItems.keys(),
233	...input.open.nodes.keys(),
234	]);
235	for (const key of keys) {
236	const subagent = input.open.subagents.get(key);
237	if (subagent !== undefined && !isSettledSubagentStatus(subagent.status)) {
238	events.push({
239	id: yield* input.allocateEventId(),
240	type: "subagent.updated",
241	threadId: subagent.threadId,
242	runId: input.run.id,
243	nodeId: subagent.id,
244	driver: subagent.driver,
245	providerInstanceId: subagent.providerInstanceId,
246	occurredAt: input.completedAt,
247	payload: {
248	...subagent,
249	status: input.status,
250	completedAt: input.completedAt,
251	updatedAt: input.completedAt,
252	},
253	});
254	}
255	const node = input.open.nodes.get(key);
256	if (
257	node !== undefined &&
258	((node.threadId === input.run.threadId && node.runId === input.run.id) ||
259	childThreadIds.has(node.threadId)) &&
260	isOpenExecutionNodeStatus(node.status)
261	) {
262	events.push({
263	id: yield* input.allocateEventId(),
264	type: "node.updated",
265	threadId: node.threadId,
266	runId: node.runId ?? input.run.id,
267	nodeId: node.id,
268	providerInstanceId: input.run.providerInstanceId,
269	occurredAt: input.completedAt,
270	payload: {
271	...node,
272	status: input.status,
273	completedAt: input.completedAt,
274	},
275	});
276	}
277	const turnItem = input.open.turnItems.get(key);
278	if (
279	turnItem !== undefined &&
280	turnItem.runId === input.run.id &&
281	!isSettledTurnItemStatus(turnItem.status)
282	) {
283	events.push({
284	id: yield* input.allocateEventId(),
285	type: "turn-item.updated",
286	threadId: turnItem.threadId,
287	runId: input.run.id,
288	...(turnItem.nodeId === null ? {} : { nodeId: turnItem.nodeId }),
289	providerInstanceId: input.run.providerInstanceId,
290	occurredAt: input.completedAt,
291	payload: {
292	...turnItem,
293	status: input.status,
294	completedAt: input.completedAt,
295	updatedAt: input.completedAt,
296	},
297	});
298	}
299	}
300	for (const turnItem of input.open.childTurnItems.values()) {
301	if (!childThreadIds.has(turnItem.threadId) || isSettledTurnItemStatus(turnItem.status)) {
302	continue;
303	}
304	events.push({
305	id: yield* input.allocateEventId(),
306	type: "turn-item.updated",
307	threadId: turnItem.threadId,
308	runId: turnItem.runId ?? input.run.id,
309	...(turnItem.nodeId === null ? {} : { nodeId: turnItem.nodeId }),
310	providerInstanceId: input.run.providerInstanceId,
311	occurredAt: input.completedAt,
312	payload: {
313	...turnItem,
314	...("streaming" in turnItem ? { streaming: false } : {}),
315	status: input.status,
316	completedAt: input.completedAt,
317	updatedAt: input.completedAt,
318	},
319	});
320	}
321	return events;
322	});
323	}
324
325	export function finalProviderThreadStatus(
326	disposition: ProviderTerminalEvent["threadDisposition"],
327	): OrchestrationV2ProviderThread["status"] {
328	return disposition === "broken" ? "error" : "idle";
329	}
330
331	export function makeProviderEventRoutingState(input: {
332	readonly identity: ProviderEventRouteIdentity;
333	readonly inheritedBackgroundTurnItems?: ReadonlyArray<InheritedBackgroundTurnItemRoute>;
334	readonly providerTurnId: ProviderTurnId | null;
335	readonly relatedThreadIds?: ReadonlyArray<ThreadId>;
336	readonly relatedProviderThreadIds?: ReadonlyArray<ProviderThreadId>;
337	}): ProviderEventRoutingState {
338	return {
339	ownedThreadIds: new Set([input.identity.threadId, ...(input.relatedThreadIds ?? [])]),
340	rootTurnEnded: false,
341	workflowMemberParents: new Map(),
342	ownedProviderThreadIds: new Set([
343	input.identity.providerThreadId,
344	...(input.relatedProviderThreadIds ?? []),
345	]),
346	ownedProviderTurnIds:
347	input.providerTurnId === null ? new Set() : new Set([input.providerTurnId]),
348	inheritedBackgroundTurnItems: new Map(
349	(input.inheritedBackgroundTurnItems ?? []).map((item) => [item.id, item.runId]),
350	),
351	rootProviderTurnId: input.providerTurnId,
352	};
353	}
354
355	export function routeProviderEvent(
356	event: ProviderAdapter.ProviderAdapterV2Event,
357	input: ProviderEventRouteIdentity,
358	state: ProviderEventRoutingState,
359	): readonly [boolean, ProviderEventRoutingState] {
360	const ownsThread = (threadId: ThreadId): boolean => state.ownedThreadIds.has(threadId);
361	const ownsChildThread = (threadId: ThreadId): boolean =>
362	threadId !== input.threadId && ownsThread(threadId);
363	const ownsRun = (runId: string | null): boolean => runId === input.runId;
364	const addProviderThread = (providerThreadId: ProviderThreadId): ProviderEventRoutingState => ({
365	...state,
366	ownedProviderThreadIds: new Set([...state.ownedProviderThreadIds, providerThreadId]),
367	});
368	const addProviderTurn = (
369	providerTurnId: ProviderTurnId,
370	root: boolean,
371	): ProviderEventRoutingState => ({
372	...state,
373	ownedProviderTurnIds: new Set([...state.ownedProviderTurnIds, providerTurnId]),
374	rootProviderTurnId: root ? providerTurnId : state.rootProviderTurnId,
375	});
376
377	switch (event.type) {
378	case "provider_session.updated":
379	// The session manager persists process-wide status once for every
380	// attached app thread before broadcasting the adapter event.
381	return [false, state];
382	case "app_thread.created": {
383	if (event.appThread.id === input.threadId) {
384	return [true, state];
385	}
386	const isOwnedSubagent =
387	event.appThread.lineage.relationshipToParent === "subagent" &&
388	event.appThread.lineage.parentThreadId !== null &&
389	ownsThread(event.appThread.lineage.parentThreadId) &&
390	(!state.rootTurnEnded ||
391	state.workflowMemberParents.get(event.appThread.id) ===
392	event.appThread.lineage.parentThreadId);
393	if (!isOwnedSubagent) {
394	return [false, state];
395	}
396	return [
397	true,
398	{
399	...state,
400	ownedThreadIds: new Set([...state.ownedThreadIds, event.appThread.id]),
401	},
402	];
403	}
404	case "provider_thread.updated": {
405	const belongs =
406	state.ownedProviderThreadIds.has(event.providerThread.id) ||
407	(event.providerThread.appThreadId !== null && ownsThread(event.providerThread.appThreadId));
408	return belongs ? [true, addProviderThread(event.providerThread.id)] : [false, state];
409	}
410	case "provider_turn.updated": {
411	const isRoot = event.providerTurn.runAttemptId === input.attemptId;
412	const belongs =
413	isRoot ||
414	(event.providerTurn.providerThreadId !== input.providerThreadId &&
415	state.ownedProviderThreadIds.has(event.providerTurn.providerThreadId)) ||
416	state.ownedProviderTurnIds.has(event.providerTurn.id) ||
417	(event.threadId !== undefined && ownsChildThread(event.threadId));
418	return belongs ? [true, addProviderTurn(event.providerTurn.id, isRoot)] : [false, state];
419	}
420	case "node.updated": {
421	const belongs = ownsRun(event.node.runId) || ownsChildThread(event.node.threadId);
422	if (!belongs || event.node.providerThreadId === null) {
423	return [belongs, state];
424	}
425	return [true, addProviderThread(event.node.providerThreadId)];
426	}
427	case "subagent.updated": {
428	const belongs = ownsRun(event.subagent.runId) || ownsChildThread(event.subagent.threadId);
429	const coordinator = event.subagent;
430	if (
431	!ownsRun(coordinator.runId) ||
432	coordinator.workflow === undefined ||
433	coordinator.childThreadId === null
434	) {
435	return [belongs, state];
436	}
437	const workflowMemberParents = new Map(state.workflowMemberParents);
438	for (const member of coordinator.workflow.agents) {
439	if (member.childThreadId !== undefined) {
440	workflowMemberParents.set(member.childThreadId, coordinator.childThreadId);
441	}
442	}
443	return [true, { ...state, workflowMemberParents }];
444	}
445	case "message.updated":
446	return [ownsRun(event.message.runId) || ownsChildThread(event.message.threadId), state];
447	case "turn_item.updated": {
448	if (ownsRun(event.turnItem.runId) || ownsChildThread(event.turnItem.threadId)) {
449	return [true, state];
450	}
451	const inheritedRunId = state.inheritedBackgroundTurnItems.get(event.turnItem.id);
452	// Preserve the item's original ownership while allowing the one live run
453	// to deliver an exact carryover identity selected from the projection.
454	const isInheritedBackgroundItem =
455	event.turnItem.threadId === input.threadId &&
456	event.turnItem.runId !== null &&
457	event.turnItem.runId === inheritedRunId &&
458	backgroundCapableTurnItemTypes.has(event.turnItem.type);
459	if (!isInheritedBackgroundItem) {
460	return [false, state];
461	}
462	if (!isSettledTurnItemStatus(event.turnItem.status)) {
463	return [true, state];
464	}
465	const inheritedBackgroundTurnItems = new Map(state.inheritedBackgroundTurnItems);
466	inheritedBackgroundTurnItems.delete(event.turnItem.id);
467	return [true, { ...state, inheritedBackgroundTurnItems }];
468	}
469	case "plan.updated":
470	return [ownsRun(event.plan.runId) || ownsChildThread(event.plan.threadId), state];
471	case "runtime_request.updated":
472	return [
473	(event.threadId !== undefined && ownsChildThread(event.threadId)) ||
474	(event.runtimeRequest.providerTurnId !== null &&
475	state.ownedProviderTurnIds.has(event.runtimeRequest.providerTurnId)),
476	state,
477	];
478	case "turn.terminal":
479	return event.providerTurnId === state.rootProviderTurnId
480	? [true, { ...state, rootTurnEnded: true }]
481	: [false, state];
482	}
483	}
484
485	/**
486	* ERRORS
487	*/
488	export class RunExecutionStartError extends Schema.TaggedError<RunExecutionStartError>()(
489	"RunExecutionStartError",
490	{
491	commandId: CommandId,
492	runId: Schema.String,
493	cause: Schema.optional(Schema.Defect()),
494	},
495	) {
496	override get message(): string {
497	return `Failed to start orchestration V2 run execution ${this.runId}.`;
498	}
499	}
500
501	export class RunExecutionIngestError extends Schema.TaggedError<RunExecutionIngestError>()(
502	"RunExecutionIngestError",
503	{
504	runId: Schema.String,
505	cause: Schema.optional(Schema.Defect()),
506	},
507	) {
508	override get message(): string {
509	return `Failed while ingesting orchestration V2 run execution ${this.runId}.`;
510	}
511	}
512
513	export const RunExecutionServiceV2Error = Schema.Union([
514	RunExecutionStartError,
515	RunExecutionIngestError,
516	]);
517	export type RunExecutionServiceV2Error = typeof RunExecutionServiceV2Error.Type;
518
519	/**
520	* SERVICE DEFINITION
521	*/
522	export interface RunExecutionServiceV2StartRootRunInput {
523	readonly commandId: CommandId;
524	readonly appThread: OrchestrationV2AppThread;
525	readonly providerSessionId: ProviderSessionId;
526	readonly session: ProviderAdapter.ProviderAdapterV2SessionRuntime;
527	readonly run: OrchestrationV2Run;
528	readonly rootNode: OrchestrationV2ExecutionNode;
529	readonly checkpointScope: OrchestrationV2CheckpointScope;
530	readonly providerThread: OrchestrationV2ProviderThread;
531	readonly attempt: OrchestrationV2RunAttempt;
532	readonly attemptId: RunAttemptId;
533	readonly providerTurnOrdinal: number;
534	readonly nativeThreadHasTurns?: boolean;
535	readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect<
536	ReadonlyArray<InheritedBackgroundTurnItemRoute>,
537	unknown
538	>;
539	readonly relatedThreadIds?: ReadonlyArray<ThreadId>;
540	readonly relatedProviderThreadIds?: ReadonlyArray<ProviderThreadId>;
541	readonly shouldStartProviderTurn?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
542	readonly shouldFinalizeRun?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
543	readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
544	readonly message: ProviderAdapter.ProviderAdapterV2TurnMessage;
545	readonly modelSelection: ModelSelection;
546	readonly runtimePolicy: ProviderAdapter.ProviderAdapterV2RuntimePolicy;
547	}
548
549	export interface RunExecutionServiceV2Shape {
550	readonly startRootRun: (
551	input: RunExecutionServiceV2StartRootRunInput,
552	) => Effect.Effect<void, RunExecutionServiceV2Error>;
553	}
554
555	export class RunExecutionServiceV2 extends Context.Service<
556	RunExecutionServiceV2,
557	RunExecutionServiceV2Shape
558	>()("t3/orchestration-v2/RunExecutionService/RunExecutionServiceV2") {}
559
560	/**
561	* IMPLEMENTATIONS
562	*/
563	export const layer: Layer.Layer<
564	RunExecutionServiceV2,
565	never,
566	| CheckpointService.CheckpointServiceV2
567	| EventSink.EventSinkV2
568	| IdAllocator.IdAllocatorV2
569	| ProviderEventIngestor.ProviderEventIngestorV2
570	| ServerSettings.ServerSettingsService
571	| McpAppModelContext.McpAppModelContext
572	> = Layer.effect(
573	RunExecutionServiceV2,
574	Effect.gen(function* () {
575	const checkpointService = yield* CheckpointService.CheckpointServiceV2;
576	const eventSink = yield* EventSink.EventSinkV2;
577	const idAllocator = yield* IdAllocator.IdAllocatorV2;
578	const providerEventIngestor = yield* ProviderEventIngestor.ProviderEventIngestorV2;
579	const serverSettings = yield* ServerSettings.ServerSettingsService;
580	const mcpAppModelContext = yield* McpAppModelContext.McpAppModelContext;
581	const finalizationObserver = yield* RunFinalizationService.RunFinalizationObserver;
582
583	const writeFinalRunEvents = (input: {
584	readonly run: OrchestrationV2Run;
585	readonly rootNode: OrchestrationV2ExecutionNode;
586	readonly checkpointScope: OrchestrationV2CheckpointScope;
587	readonly providerThread: OrchestrationV2ProviderThread;
588	readonly attempt: OrchestrationV2RunAttempt;
589	readonly shouldFinalizeRun?: () => Effect.Effect<boolean, ProjectionStoreV2Error>;
590	readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
591	readonly openRunOwnedSubagents?: OpenRunOwnedSubagentProjection;
592	readonly terminal: ProviderTerminalEvent;
593	readonly failureItemPersisted: boolean;
594	readonly refreshAfterTurn: Effect.Effect<void>;
595	readonly writeIfRunCurrent?: {
596	readonly activeAttemptId: RunAttemptId;
597	readonly expectedStatus: OrchestrationV2Run["status"];
598	};
599	}) =>
600	Effect.gen(function* () {
601	const completedAt = yield* DateTime.now;
602	const finalizedAttempt: OrchestrationV2RunAttempt | null = {
603	...input.attempt,
604	status: input.terminal.status,
605	completedAt,
606	};
607	const shouldFinalizeRun =
608	input.shouldFinalizeRun === undefined ? true : yield* input.shouldFinalizeRun();
609	if (!shouldFinalizeRun) {
610	// Superseded attempt (steer / selection restart). Emit
611	// run_interrupt_result only when hard Stop left an unpaired request
612	// for this run; plain steers and already-paired stops emit nothing.
613	if (input.terminal.status === "interrupted") {
614	const hasUnpairedRequest =
615	input.hasUnpairedRunInterruptRequest === undefined
616	? false
617	: yield* input.hasUnpairedRunInterruptRequest();
618	if (hasUnpairedRequest) {
619	yield* eventSink.writeWithEffects({
620	effects: [],
621	events: [
622	{
623	id: yield* idAllocator.allocate.event({ threadId: input.run.threadId }),
624	type: "turn-item.updated" as const,
625	threadId: input.run.threadId,
626	runId: input.run.id,
627	nodeId: input.rootNode.id,
628	providerInstanceId: input.run.providerInstanceId,
629	occurredAt: completedAt,
630	payload: makeInterruptResultTurnItem({
631	idAllocator,
632	run: input.run,
633	rootNode: input.rootNode,
634	providerThread: input.providerThread,
635	completedAt,
636	}),
637	},
638	],
639	});
640	yield* input.refreshAfterTurn;
641	}
642	}
643	return;
644	}
645	const allocateEventId = () => idAllocator.allocate.event({ threadId: input.run.threadId });
646	const open = input.openRunOwnedSubagents ?? emptyOpenRunOwnedSubagentProjection();
647	const hasOpenSubagentProjection =
648	open.subagents.size > 0 ||
649	open.turnItems.size > 0 ||
650	open.childTurnItems.size > 0 ||
651	open.nodes.size > 0;
652	const cascadedSubagentEvents =
653	isRunOwnedSubagentTerminalStatus(input.terminal.status) && hasOpenSubagentProjection
654	? yield* cascadeTerminalizeRunOwnedSubagents({
655	run: input.run,
656	open,
657	status: input.terminal.status,
658	completedAt,
659	allocateEventId,
660	})
661	: [];
662	const persistedStatus =
663	input.terminal.status === "completed" ? "waiting" : input.terminal.status;
664	// Completion cohorts are advanced by Orchestrator while a provider
665	// turn is in flight. Do not replay the run snapshot captured at start
666	// over a newer acknowledgement, successor, or Stop barrier.
667	const { delegatedCompletion: _delegatedCompletion, ...runWithoutDelegatedCompletion } =
668	input.run;
669	const finalizedRun: OrchestrationV2Run = {
670	...runWithoutDelegatedCompletion,
671	status: persistedStatus,
672	completedAt: input.terminal.status === "completed" ? null : completedAt,
673	};
674	const finalizedRootNode: OrchestrationV2ExecutionNode = {
675	...input.rootNode,
676	status: persistedStatus,
677	completedAt: input.terminal.status === "completed" ? null : completedAt,
678	checkpointScopeId: input.checkpointScope.id,
679	};
680	const finalizedProviderThread: OrchestrationV2ProviderThread = {
681	...input.providerThread,
682	status: finalProviderThreadStatus(input.terminal.threadDisposition),
683	updatedAt: completedAt,
684	};
685	const runEventId = yield* allocateEventId();
686	const nodeEventId = yield* allocateEventId();
687	const providerThreadEventId = yield* allocateEventId();
688	const checkpointCaptureCommandId = CommandId.make(
689	`command:effect:checkpoint.capture:${input.run.id}`,
690	);
691	// Stopped runs capture too: their checkpoint is the rollback point for
692	// the next message. The capture is enqueued with these terminal events,
693	// ahead of any later run's start on this thread's effect lane.
694	const finalization = {
695	effects:
696	input.terminal.status === "completed" ||
697	input.terminal.status === "interrupted" ||
698	input.terminal.status === "cancelled"
699	? [
700	{
701	id: `effect:checkpoint.capture:${input.run.id}`,
702	commandId: checkpointCaptureCommandId,
703	threadId: input.run.threadId,
704	request: {
705	type: "checkpoint.capture" as const,
706	runId: input.run.id,
707	scopeId: input.checkpointScope.id,
708	},
709	},
710	]
711	: [],
712	events: [
713	// Terminalize open run-owned subagent rows before the root run
714	// settles so projections never keep a forever-running subagent card.
715	...cascadedSubagentEvents,
716	...(finalizedAttempt === null
717	? []
718	: [
719	{
720	id: yield* allocateEventId(),
721	type: "run-attempt.updated" as const,
722	threadId: input.run.threadId,
723	runId: input.run.id,
724	nodeId: input.rootNode.id,
725	providerInstanceId: input.run.providerInstanceId,
726	occurredAt: completedAt,
727	payload: finalizedAttempt,
728	},
729	]),
730	...(input.terminal.status === "interrupted"
731	? [
732	{
733	id: yield* allocateEventId(),
734	type: "turn-item.updated" as const,
735	threadId: input.run.threadId,
736	runId: input.run.id,
737	nodeId: input.rootNode.id,
738	providerInstanceId: input.run.providerInstanceId,
739	occurredAt: completedAt,
740	payload: makeInterruptResultTurnItem({
741	idAllocator,
742	run: input.run,
743	rootNode: input.rootNode,
744	providerThread: input.providerThread,
745	completedAt,
746	}),
747	},
748	]
749	: []),
750	...(input.terminal.status === "failed" && !input.failureItemPersisted
751	? [
752	{
753	id: yield* allocateEventId(),
754	type: "turn-item.updated" as const,
755	threadId: input.run.threadId,
756	runId: input.run.id,
757	nodeId: input.rootNode.id,
758	providerInstanceId: input.run.providerInstanceId,
759	occurredAt: completedAt,
760	payload: makeProviderFailureTurnItem({
761	driver: input.terminal.driver,
762	threadId: input.run.threadId,
763	runId: input.run.id,
764	nodeId: input.rootNode.id,
765	providerThreadId: input.terminal.providerThreadId,
766	providerTurnId: input.terminal.providerTurnId,
767	itemOrdinal: input.terminal.failureItemOrdinal,
768	failure: input.terminal.failure,
769	occurredAt: completedAt,
770	}),
771	},
772	]
773	: []),
774	{
775	id: runEventId,
776	type: "run.updated",
777	threadId: input.run.threadId,
778	runId: input.run.id,
779	nodeId: input.rootNode.id,
780	providerInstanceId: input.run.providerInstanceId,
781	occurredAt: completedAt,
782	payload: finalizedRun,
783	},
784	{
785	id: nodeEventId,
786	type: "node.updated",
787	threadId: input.run.threadId,
788	runId: input.run.id,
789	nodeId: input.rootNode.id,
790	providerInstanceId: input.run.providerInstanceId,
791	occurredAt: completedAt,
792	payload: finalizedRootNode,
793	},
794	{
795	id: providerThreadEventId,
796	type: "provider-thread.updated",
797	threadId: input.run.threadId,
798	providerInstanceId: input.run.providerInstanceId,
799	occurredAt: completedAt,
800	payload: finalizedProviderThread,
801	},
802	],
803	} satisfies Parameters<typeof eventSink.writeWithEffects>[0];
804	if (input.writeIfRunCurrent !== undefined) {
805	const result = yield* eventSink.writeIfRunCurrent({
806	threadId: input.run.threadId,
807	runId: input.run.id,
808	activeAttemptId: input.writeIfRunCurrent.activeAttemptId,
809	expectedStatus: input.writeIfRunCurrent.expectedStatus,
810	events: finalization.events,
811	effects: finalization.effects,
812	});
813	if (!result.committed) {
814	return;
815	}
816	} else {
817	yield* eventSink.writeWithEffects(finalization);
818	}
819	yield* input.refreshAfterTurn;
820	});
821
822	return RunExecutionServiceV2.of({
823	startRootRun: (input) =>
824	Effect.gen(function* () {
825	// Startup failure and stream shutdown can report the same attempt.
826	const refreshAfterTurn = yield* Effect.cached(
827	finalizationObserver.refreshAfterTurn(input.appThread.projectId).pipe(
828	Effect.catchCause((cause) =>
829	Effect.logWarning("failed to refresh pull requests after run termination", {
830	threadId: input.run.threadId,
831	runId: input.run.id,
832	cause,
833	}),
834	),
835	),
836	);
837	const makeFailedTerminalEvent = (
838	failure: OrchestrationV2ProviderFailure,
839	failureItemOrdinal: number,
840	): ProviderTerminalEvent => ({
841	type: "turn.terminal",
842	driver: input.providerThread.driver,
843	providerThreadId: input.providerThread.id,
844	providerTurnId:
845	input.attempt.providerTurnId ??
846	idAllocator.derive.providerTurn({
847	driver: input.providerThread.driver,
848	nativeTurnId: `failed:${input.attempt.id}`,
849	}),
850	runOrdinal: input.run.ordinal,
851	failureItemOrdinal,
852	status: "failed",
853	failure,
854	threadDisposition: "reusable",
855	});
856	const responseStreamingMode = yield* Effect.gen(function* () {
857	const responseStreamingMode = yield* serverSettings.getSettings.pipe(
858	Effect.map(
859	(settings) =>
860	resolveProjectSettings(settings, input.appThread.projectId).settings
861	.responseStreamingMode,
862	),
863	);
864	yield* checkpointService
865	.captureBaseline({
866	scope: input.checkpointScope,
867	ordinalWithinScope: Math.max(0, input.run.ordinal - 1),
868	})
869	.pipe(
870	Effect.catchCause((cause) =>
871	Cause.hasInterruptsOnly(cause)
872	? Effect.failCause(cause)
873	: Effect.logWarning(
874	"orchestration V2 checkpoint baseline capture failed; starting provider without a baseline",
875	{ runId: input.run.id },
876	),
877	),
878	);
879	if (
880	input.shouldStartProviderTurn !== undefined &&
881	!(yield* input.shouldStartProviderTurn())
882	) {
883	return null;
884	}
885	return responseStreamingMode;
886	}).pipe(
887	Effect.catchCause((cause) =>
888	Effect.gen(function* () {
889	if (Cause.hasInterruptsOnly(cause)) {
890	return yield* Effect.failCause(cause);
891	}
892	yield* Effect.logError("orchestration V2 run preparation failed", {
893	runId: input.run.id,
894	cause,
895	});
896	yield* writeFinalRunEvents({
897	run: input.run,
898	rootNode: input.rootNode,
899	checkpointScope: input.checkpointScope,
900	providerThread: input.providerThread,
901	attempt: input.attempt,
902	terminal: makeFailedTerminalEvent(
903	makeProviderFailure({
904	cause: Cause.squash(cause),
905	// Keep exact underlying text in the logged cause only;
906	// the persisted turn item gets a bounded curated message.
907	message: "Run preparation failed.",
908	class: "unknown",
909	}),
910	input.providerTurnOrdinal * 100 + 1,
911	),
912	failureItemPersisted: false,
913	refreshAfterTurn,
914	writeIfRunCurrent: {
915	activeAttemptId: input.attemptId,
916	expectedStatus: "running",
917	},
918	});
919	return null;
920	}),
921	),
922	Effect.mapError(
923	(cause) =>
924	new RunExecutionStartError({
925	commandId: input.commandId,
926	runId: input.run.id,
927	cause,
928	}),
929	),
930	);
931	if (responseStreamingMode === null) {
932	return;
933	}
934	const terminalEvent = yield* Ref.make<ProviderTerminalEvent | null>(null);
935	const latestTurnItemOrdinal = yield* Ref.make(input.providerTurnOrdinal * 100);
936	const latestProviderThread = yield* Ref.make(input.providerThread);
937	const routeIdentity: ProviderEventRouteIdentity = {
938	threadId: input.run.threadId,
939	runId: input.run.id,
940	attemptId: input.attempt.id,
941	providerThreadId: input.providerThread.id,
942	};
943	const eventSubscription =
944	input.session.subscribeEvents === undefined
945	? { events: input.session.events, close: Effect.void }
946	: yield* input.session.subscribeEvents;
947	const inheritedBackgroundTurnItems = yield* (
948	input.loadInheritedBackgroundTurnItems?.() ?? Effect.succeed([])
949	).pipe(
950	Effect.onError(() => eventSubscription.close),
951	Effect.mapError(
952	(cause) =>
953	new RunExecutionStartError({
954	commandId: input.commandId,
955	runId: input.run.id,
956	cause,
957	}),
958	),
959	);
960	const inheritedBackgroundTurnItemsById = new Map(
961	inheritedBackgroundTurnItems.map((item) => [item.id, item.runId]),
962	);
963	const eventRouting = yield* Ref.make<ProviderEventRoutingState>(
964	makeProviderEventRoutingState({
965	identity: routeIdentity,
966	inheritedBackgroundTurnItems,
967	providerTurnId: input.attempt.providerTurnId,
968	...(input.relatedThreadIds === undefined
969	? {}
970	: { relatedThreadIds: input.relatedThreadIds }),
971	...(input.relatedProviderThreadIds === undefined
972	? {}
973	: { relatedProviderThreadIds: input.relatedProviderThreadIds }),
974	}),
975	);
976	const rootTerminalSeen = yield* Ref.make(false);
977	const rootRunFinalized = yield* Ref.make(false);
978	const providerThreadOwnerLost = yield* Ref.make(false);
979	const activeChildProviderTurns = yield* Ref.make<ReadonlySet<ProviderTurnId>>(new Set());
980	const activeChildSubagents = yield* Ref.make<ReadonlySet<NodeId>>(new Set());
981	const activeBackgroundTurnItems = yield* Ref.make<
982	ReadonlySet<OrchestrationV2TurnItem["id"]>
983	>(new Set(inheritedBackgroundTurnItemsById.keys()));
984	const openRunOwnedSubagents = yield* Ref.make(emptyOpenRunOwnedSubagentProjection());
985	const finalizeRootRun = (terminal: ProviderTerminalEvent) =>
986	Effect.gen(function* () {
987	if (yield* Ref.get(rootRunFinalized)) {
988	return;
989	}
990	const providerThread = yield* Ref.get(latestProviderThread);
991	const openSubagents = yield* Ref.get(openRunOwnedSubagents);
992	yield* writeFinalRunEvents({
993	run: input.run,
994	rootNode: input.rootNode,
995	checkpointScope: input.checkpointScope,
996	providerThread,
997	attempt: input.attempt,
998	...(input.shouldFinalizeRun === undefined
999	? {}
1000	: {
1001	shouldFinalizeRun: input.shouldFinalizeRun,
1002	// Stop may commit between the ownership read and this
1003	// terminal write. Gate the events and checkpoint together.
1004	writeIfRunCurrent: {
1005	activeAttemptId: input.attempt.id,
1006	expectedStatus: "running" as const,
1007	},
1008	}),
1009	...(input.hasUnpairedRunInterruptRequest === undefined
1010	? {}
1011	: {
1012	hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest,
1013	}),
1014	openRunOwnedSubagents: openSubagents,
1015	terminal,
1016	failureItemPersisted: terminal.status === "failed",
1017	refreshAfterTurn,
1018	}).pipe(
1019	Effect.mapError(
1020	(cause) => new RunExecutionIngestError({ runId: input.run.id, cause }),
1021	),
1022	);
1023	if (isRunOwnedSubagentTerminalStatus(terminal.status)) {
1024	yield* Ref.set(openRunOwnedSubagents, emptyOpenRunOwnedSubagentProjection());
1025	}
1026	yield* Ref.set(rootRunFinalized, true);
1027	});
1028	const trackChildLifecycle = (
1029	event: ProviderAdapter.ProviderAdapterV2Event,
1030	deliverable: boolean,
1031	) =>
1032	Effect.gen(function* () {
1033	const routing = yield* Ref.get(eventRouting);
1034	if (event.type === "provider_turn.updated") {
1035	const isRoot =
1036	event.providerTurn.runAttemptId === input.attempt.id ||
1037	event.providerTurn.id === routing.rootProviderTurnId;
1038	if (!isRoot) {
1039	yield* Ref.update(activeChildProviderTurns, (current) => {
1040	const next = new Set(current);
1041	if (isTerminalProviderTurnStatus(event.providerTurn.status)) {
1042	next.delete(event.providerTurn.id);
1043	} else {
1044	next.add(event.providerTurn.id);
1045	}
1046	return next;
1047	});
1048	}
1049	}
1050	if (event.type === "subagent.updated") {
1051	const belongsToRootRun = event.subagent.runId === input.run.id;
1052	const belongsToOwnedChildThread =
1053	event.subagent.threadId !== input.run.threadId &&
1054	routing.ownedThreadIds.has(event.subagent.threadId);
1055	if (belongsToRootRun || belongsToOwnedChildThread) {
1056	yield* Ref.update(activeChildSubagents, (current) => {
1057	const next = new Set(current);
1058	if (isSettledSubagentStatus(event.subagent.status)) {
1059	next.delete(event.subagent.id);
1060	} else {
1061	next.add(event.subagent.id);
1062	}
1063	return next;
1064	});
1065	}
1066	// Snapshot run-owned subagents for interrupt cascade.
1067	// Preserve childThreadId linkage for the root-run lifetime even
1068	// after the subagent row terminalizes, so open child-thread
1069	// nodes can still be proven linked on a later root interrupt.
1070	// A provider records nested native subagents on the child
1071	// thread, runless; they die with this run too.
1072	if (
1073	belongsToRootRun ||
1074	(belongsToOwnedChildThread &&
1075	event.subagent.runId === null &&
1076	event.subagent.origin === "provider_native")
1077	) {
1078	yield* Ref.update(openRunOwnedSubagents, (current) => {
1079	const withLink = withLinkedChildThreadId(current, event.subagent.childThreadId);
1080	const subagents = new Map(withLink.subagents);
1081	if (isSettledSubagentStatus(event.subagent.status)) {
1082	subagents.delete(event.subagent.id);
1083	} else {
1084	subagents.set(event.subagent.id, event.subagent);
1085	}
1086	return { ...withLink, subagents };
1087	});
1088	}
1089	}
1090	if (event.type === "node.updated") {
1091	const belongsToRootSubagent =
1092	event.node.kind === "subagent" && event.node.runId === input.run.id;
1093	const belongsToOwnedChildThread =
1094	event.node.threadId !== input.run.threadId &&
1095	routing.ownedThreadIds.has(event.node.threadId);
1096	if (
1097	!belongsToRootSubagent &&
1098	!(
1099	belongsToOwnedChildThread &&
1100	(event.node.runId === null || event.node.runId === input.run.id)
1101	)
1102	) {
1103	return;
1104	}
1105	yield* Ref.update(openRunOwnedSubagents, (current) => {
1106	const nodes = new Map(current.nodes);
1107	if (isOpenExecutionNodeStatus(event.node.status)) {
1108	nodes.set(event.node.id, event.node);
1109	} else {
1110	nodes.delete(event.node.id);
1111	}
1112	return { ...current, nodes };
1113	});
1114	}
1115	if (event.type === "turn_item.updated") {
1116	const belongsToRootRun = event.turnItem.runId === input.run.id;
1117	const belongsToOwnedChildThread =
1118	event.turnItem.threadId !== input.run.threadId &&
1119	routing.ownedThreadIds.has(event.turnItem.threadId);
1120	const belongsToInheritedBackgroundItem =
1121	event.turnItem.threadId === input.run.threadId &&
1122	event.turnItem.runId !== null &&
1123	inheritedBackgroundTurnItemsById.get(event.turnItem.id) === event.turnItem.runId;
1124	if (
1125	backgroundCapableTurnItemTypes.has(event.turnItem.type) &&
1126	(belongsToRootRun ||
1127	belongsToOwnedChildThread ||
1128	belongsToInheritedBackgroundItem)
1129	) {
1130	yield* Ref.update(activeBackgroundTurnItems, (current) => {
1131	const next = new Set(current);
1132	if (isSettledTurnItemStatus(event.turnItem.status)) {
1133	next.delete(event.turnItem.id);
1134	} else {
1135	next.add(event.turnItem.id);
1136	}
1137	return next;
1138	});
1139	}
1140	if (
1141	belongsToOwnedChildThread &&
1142	(event.turnItem.runId === null || belongsToRootRun) &&
1143	deliverable
1144	) {
1145	yield* Ref.update(openRunOwnedSubagents, (current) => {
1146	const childTurnItems = new Map(current.childTurnItems);
1147	if (isSettledTurnItemStatus(event.turnItem.status)) {
1148	childTurnItems.delete(event.turnItem.id);
1149	} else {
1150	childTurnItems.set(event.turnItem.id, event.turnItem);
1151	}
1152	return { ...current, childTurnItems };
1153	});
1154	}
1155	if (belongsToRootRun && event.turnItem.type === "subagent") {
1156	const subagentItem = event.turnItem;
1157	yield* Ref.update(openRunOwnedSubagents, (current) => {
1158	const withLink = withLinkedChildThreadId(current, subagentItem.childThreadId);
1159	const turnItems = new Map(withLink.turnItems);
1160	if (isSettledTurnItemStatus(subagentItem.status)) {
1161	turnItems.delete(subagentItem.subagentId);
1162	} else {
1163	turnItems.set(subagentItem.subagentId, subagentItem);
1164	}
1165	return { ...withLink, turnItems };
1166	});
1167	}
1168	}
1169	});
1170	const shouldStopProviderEventIngestion = Effect.gen(function* () {
1171	if (!(yield* Ref.get(rootTerminalSeen))) {
1172	return false;
1173	}
1174	const terminal = yield* Ref.get(terminalEvent);
1175	// Non-completed terminals drop background tracking immediately.
1176	if (terminal !== null && terminal.status !== "completed") {
1177	return true;
1178	}
1179	const childProviderTurns = yield* Ref.get(activeChildProviderTurns);
1180	if (childProviderTurns.size > 0) {
1181	return false;
1182	}
1183	const childSubagents = yield* Ref.get(activeChildSubagents);
1184	if (childSubagents.size > 0) {
1185	return false;
1186	}
1187	// Keep ingesting past root settlement while background-capable
1188	// items owned by this run (or an owned child thread) are still
1189	// non-terminal, so their late completion events reach the
1190	// projection (stuck-spinner fix). Only for completed runs:
1191	// interrupted/failed turns intentionally drop background tracking
1192	// rather than pinning the stream open. Newly owned items depend on
1193	// adapters emitting a non-terminal event before the root terminal.
1194	// Exact inherited items are seeded from their selected durable rows.
1195	//
1196	// Owner loss (a newer run claimed lastRunOrdinal) must not close
1197	// this stream while these sets are non-empty: turn_item.updated
1198	// writes are not ownership-gated, so late completions still land.
1199	const backgroundItems = yield* Ref.get(activeBackgroundTurnItems);
1200	if (backgroundItems.size > 0) {
1201	return false;
1202	}
1203	// Owner loss means do not hold the stream open solely for the
1204	// roster probe; once background sets are empty, release.
1205	if (yield* Ref.get(providerThreadOwnerLost)) {
1206	return true;
1207	}
1208	// Claude background Bash has no turn-item projection. Keep the
1209	// stream open while this root's provider thread still reports
1210	// pending roster work so late empty updates can clear Waiting.
1211	// Use only the thread-scoped probe: session-wide pending work
1212	// (siblings, wake buffers, session subagents) must not pin this
1213	// root subscription. Session idle release still uses
1214	// hasPendingBackgroundWork via ProviderSessionManager.
1215	const latestProviderThreadSnapshot = yield* Ref.get(latestProviderThread);
1216	if (input.session.hasPendingBackgroundWorkForThread !== undefined) {
1217	const hasPendingWork = yield* input.session
1218	.hasPendingBackgroundWorkForThread(latestProviderThreadSnapshot)
1219	.pipe(Effect.catchCause(() => Effect.succeed(false)));
1220	if (hasPendingWork) {
1221	return false;
1222	}
1223	}
1224	return true;
1225	});
1226	const filterAssistantEvent = makeAssistantStreamingFilter(responseStreamingMode);
1227	const providerEventFiber = yield* eventSubscription.events.pipe(
1228	Stream.filterEffect((event) =>
1229	Ref.modify(eventRouting, (state) => routeProviderEvent(event, routeIdentity, state)),
1230	),
1231	Stream.tap((event) =>
1232	Effect.gen(function* () {
1233	let storedEventCount = 0;
1234	const deliveredEvent = filterAssistantEvent(
1235	event,
1236	DateTime.toEpochMillis(yield* DateTime.now),
1237	);
1238	if (deliveredEvent) {
1239	// Root provider_thread.updated always uses an ownership gate:
1240	// pre-terminal writeIfRunCurrent (attempt still running), or
1241	// post-terminal writeIfProviderThreadOwner so late roster
1242	// clears still land while this attempt owns the run and this
1243	// run owns lastRunOrdinal.
1244	const rootTerminalAlreadySeen = yield* Ref.get(rootTerminalSeen);
1245	const isRootProviderThreadUpdate =
1246	event.type === "provider_thread.updated" &&
1247	event.providerThread.id === input.providerThread.id;
1248	const storedEvents = yield* providerEventIngestor.ingestNormalized({
1249	providerSessionId: input.providerSessionId,
1250	providerInstanceId: input.run.providerInstanceId,
1251	threadId: input.run.threadId,
1252	runId: input.run.id,
1253	nodeId: input.rootNode.id,
1254	event: deliveredEvent,
1255	...(isRootProviderThreadUpdate
1256	? rootTerminalAlreadySeen
1257	? {
1258	writeIfProviderThreadOwner: {
1259	providerThreadId: input.providerThread.id,
1260	runId: input.run.id,
1261	activeAttemptId: input.attempt.id,
1262	expectedLastRunOrdinal: input.run.ordinal,
1263	},
1264	}
1265	: {
1266	writeIfRunCurrent: {
1267	runId: input.run.id,
1268	activeAttemptId: input.attempt.id,
1269	expectedStatus: "running" as const,
1270	},
1271	}
1272	: {}),
1273	});
1274	storedEventCount = storedEvents.length;
1275	if (
1276	isRootProviderThreadUpdate &&
1277	rootTerminalAlreadySeen &&
1278	storedEventCount === 0
1279	) {
1280	// Ownership lost (or thread row missing). Stop pinning the
1281	// stream on this run's background probe.
1282	yield* Ref.set(providerThreadOwnerLost, true);
1283	}
1284	}
1285	if (event.type === "provider_thread.updated") {
1286	if (event.providerThread.id === input.providerThread.id && storedEventCount > 0) {
1287	yield* Ref.set(latestProviderThread, event.providerThread);
1288	}
1289	}
1290	if (
1291	event.type === "turn_item.updated" &&
1292	event.turnItem.providerTurnId ===
1293	(yield* Ref.get(eventRouting)).rootProviderTurnId
1294	) {
1295	yield* Ref.update(latestTurnItemOrdinal, (current) =>
1296	Math.max(current, event.turnItem.ordinal),
1297	);
1298	}
1299	if (event.type === "turn.terminal") {
1300	yield* Ref.set(terminalEvent, event);
1301	yield* Ref.set(rootTerminalSeen, true);
1302	yield* finalizeRootRun(event);
1303	}
1304	yield* trackChildLifecycle(event, deliveredEvent !== null);
1305	}),
1306	),
1307	Stream.takeUntilEffect(() => shouldStopProviderEventIngestion),
1308	Stream.runDrain,
1309	Effect.mapError((cause) => new RunExecutionIngestError({ runId: input.run.id, cause })),
1310	Effect.flatMap(() =>
1311	Effect.gen(function* () {
1312	const terminal = yield* Ref.get(terminalEvent);
1313	if (terminal === null) {
1314	return;
1315	}
1316	yield* finalizeRootRun(terminal);
1317	}),
1318	),
1319	Effect.catchCause((cause) =>
1320	Ref.get(rootRunFinalized).pipe(
1321	Effect.flatMap((finalized) =>
1322	Effect.logWarning("orchestration V2 provider event ingestion failed", {
1323	runId: input.run.id,
1324	cause,
1325	}).pipe(
1326	Effect.andThen(
1327	finalized
1328	? Effect.void
1329	: Ref.get(latestProviderThread).pipe(
1330	Effect.flatMap((providerThread) =>
1331	Ref.get(latestTurnItemOrdinal).pipe(
1332	Effect.flatMap((latestItemOrdinal) =>
1333	Ref.get(openRunOwnedSubagents).pipe(
1334	Effect.flatMap((openSubagents) =>
1335	writeFinalRunEvents({
1336	run: input.run,
1337	rootNode: input.rootNode,
1338	checkpointScope: input.checkpointScope,
1339	providerThread,
1340	attempt: input.attempt,
1341	// The failure may be the ownership
1342	// read itself, so check in the write.
1343	writeIfRunCurrent: {
1344	activeAttemptId: input.attempt.id,
1345	expectedStatus: "running",
1346	},
1347	openRunOwnedSubagents: openSubagents,
1348	terminal: makeFailedTerminalEvent(
1349	makeProviderFailure({
1350	cause: Cause.squash(cause),
1351	class: "unknown",
1352	}),
1353	latestItemOrdinal + 1,
1354	),
1355	failureItemPersisted: false,
1356	refreshAfterTurn,
1357	}),
1358	),
1359	),
1360	),
1361	),
1362	),
1363	),
1364	),
1365	Effect.mapError(
1366	(writeCause) =>
1367	new RunExecutionIngestError({
1368	runId: input.run.id,
1369	cause: { ingest: cause, write: writeCause },
1370	}),
1371	),
1372	),
1373	),
1374	),
1375	),
1376	Effect.ensuring(eventSubscription.close),
1377	Effect.forkDetach,
1378	);
1379
1380	// A failed read fails the start below, so the run is recorded as
1381	// failed instead of staying active with no provider turn.
1382	const shouldStart =
1383	input.shouldStartProviderTurn === undefined
1384	? Exit.succeed(true)
1385	: yield* Effect.exit(input.shouldStartProviderTurn());
1386	if (Exit.isSuccess(shouldStart) && !shouldStart.value) {
1387	yield* Fiber.interrupt(providerEventFiber);
1388	return;
1389	}
1390
1391	// A provider turn is a sign that the session is still alive. Keep
1392	// its already-issued MCP credential valid even when the agent goes
1393	// a long time between browser-tool calls.
1394	yield* McpSessionRegistry.touchActiveMcpThread(input.run.threadId);
1395	// A context read that fails costs the agent the apps' notes for
1396	// this turn, not the turn itself.
1397	const appContext = (yield* mcpAppModelContext
1398	.forThread(input.run.threadId)
1399	.pipe(
1400	Effect.catch((cause) =>
1401	Effect.logWarning("Failed to read MCP app model context.", { cause }).pipe(
1402	Effect.as([]),
1403	),
1404	),
1405	)).map((entry) => ({
1406	// The item id alone is unique and needs no escaping; server and
1407	// tool names are free text that would break the tag Codex wraps
1408	// the context in.
1409	key: `mcp_app_${entry.itemId.replace(/[^\w.-]/g, "_")}`,
1410	text: entry.text,
1411	}));
1412	const turnInput = {
1413	appThread: input.appThread,
1414	threadId: input.run.threadId,
1415	runId: input.run.id,
1416	runOrdinal: input.run.ordinal,
1417	providerTurnOrdinal: input.providerTurnOrdinal,
1418	...(input.nativeThreadHasTurns === undefined
1419	? {}
1420	: { nativeThreadHasTurns: input.nativeThreadHasTurns }),
1421	...(input.run.restartContinuationOfRunId === undefined
1422	? {}
1423	: {
1424	restartContinuationOfRunId: input.run.restartContinuationOfRunId,
1425	}),
1426	attemptId: input.attemptId,
1427	rootNodeId: input.rootNode.id,
1428	providerThread: input.providerThread,
1429	message: input.message,
1430	modelSelection: input.modelSelection,
1431	runtimePolicy: input.runtimePolicy,
1432	...(appContext.length === 0 ? {} : { appContext }),
1433	};
1434	const compact =
1435	input.message.attachments.length === 0 &&
1436	input.message.text.trim().toLowerCase() === "/compact";
1437	const startTurn = compact
1438	? (input.session.compactThread?.(turnInput) ??
1439	Effect.fail(
1440	new ProviderAdapter.ProviderAdapterTurnStartError({
1441	driver: input.session.driver,
1442	threadId: input.run.threadId,
1443	providerThreadId: input.providerThread.id,
1444	runId: input.run.id,
1445	cause: "This provider does not support context compaction.",
1446	}),
1447	))
1448	: input.session.startTurn(turnInput);
1449	yield* Effect.andThen(shouldStart, startTurn).pipe(
1450	Effect.catchCause((cause) =>
1451	Effect.logError("orchestration V2 provider turn start failed", {
1452	runId: input.run.id,
1453	cause,
1454	}).pipe(
1455	Effect.andThen(Fiber.interrupt(providerEventFiber)),
1456	Effect.andThen(Ref.get(latestProviderThread)),
1457	Effect.flatMap((providerThread) =>
1458	Ref.get(latestTurnItemOrdinal).pipe(
1459	Effect.flatMap((latestItemOrdinal) =>
1460	Ref.get(openRunOwnedSubagents).pipe(
1461	Effect.flatMap((openSubagents) =>
1462	writeFinalRunEvents({
1463	run: input.run,
1464	rootNode: input.rootNode,
1465	checkpointScope: input.checkpointScope,
1466	providerThread,
1467	attempt: input.attempt,
1468	// Checked in the write transaction, not by another
1469	// read that can fail like the one before the start.
1470	writeIfRunCurrent: {
1471	activeAttemptId: input.attempt.id,
1472	expectedStatus: "running",
1473	},
1474	openRunOwnedSubagents: openSubagents,
1475	terminal: makeFailedTerminalEvent(
1476	makeProviderFailure({
1477	cause: Cause.squash(cause),
1478	// A failed ownership read is not the provider's fault.
1479	class: Exit.isFailure(shouldStart) ? "unknown" : "provider_error",
1480	}),
1481	latestItemOrdinal + 1,
1482	),
1483	failureItemPersisted: false,
1484	refreshAfterTurn,
1485	}),
1486	),
1487	),
1488	),
1489	),
1490	),
1491	Effect.mapError(
1492	(writeCause) =>
1493	new RunExecutionStartError({
1494	commandId: input.commandId,
1495	runId: input.run.id,
1496	cause: { start: cause, write: writeCause },
1497	}),
1498	),
1499	),
1500	),
1501	);
1502	}),
1503	} satisfies RunExecutionServiceV2Shape);
1504	}),
1505	);
1506
1507	export function makeInterruptResultTurnItem(input: {
1508	readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
1509	readonly run: OrchestrationV2Run;
1510	readonly rootNode: OrchestrationV2ExecutionNode;
1511	readonly providerThread: OrchestrationV2ProviderThread;
1512	readonly completedAt: DateTime.Utc;
1513	}): OrchestrationV2TurnItem {
1514	return {
1515	id: input.idAllocator.derive.runSignalTurnItem({
1516	runId: input.run.id,
1517	signal: "interrupt-result",
1518	}),
1519	threadId: input.run.threadId,
1520	runId: input.run.id,
1521	nodeId: input.rootNode.id,
1522	providerThreadId: input.providerThread.id,
1523	providerTurnId: input.rootNode.providerTurnId,
1524	nativeItemRef: null,
1525	parentItemId: input.idAllocator.derive.runSignalTurnItem({
1526	runId: input.run.id,
1527	signal: "interrupt-request",
1528	}),
1529	ordinal: input.run.ordinal * 100 + 98,
1530	status: "interrupted",
1531	title: "Interrupted",
1532	startedAt: input.completedAt,
1533	completedAt: input.completedAt,
1534	updatedAt: input.completedAt,
1535	type: "run_interrupt_result",
1536	message: "Run interrupted by user",
1537	};
1538	}