clover's git

the-knife

1	import { assert, it, vi } from "@effect/vitest";
2	import {
3	CheckpointScopeId,
4	ChatAttachmentId,
5	ChatFileAttachment,
6	CommandId,
7	EventId,
8	MessageId,
9	NodeId,
10	type OrchestrationV2AppThread,
11	type OrchestrationV2CheckpointScope,
12	type OrchestrationV2DomainEvent,
13	type OrchestrationV2ExecutionNode,
14	type OrchestrationV2ProviderThread,
15	type OrchestrationV2Run,
16	type OrchestrationV2RunAttempt,
17	type OrchestrationV2Subagent,
18	type OrchestrationV2TurnItem,
19	ProviderDriverKind,
20	ProviderInstanceId,
21	ProviderSessionId,
22	ProviderThreadId,
23	ProviderTurnId,
24	RunAttemptId,
25	RunId,
26	ServerSettingsError,
27	ThreadId,
28	TurnItemId,
29	} from "@t3tools/contracts";
30	import * as Cause from "effect/Cause";
31	import * as Exit from "effect/Exit";
32	import * as DateTime from "effect/DateTime";
33	import * as Deferred from "effect/Deferred";
34	import * as Effect from "effect/Effect";
35	import * as Layer from "effect/Layer";
36	import * as Option from "effect/Option";
37	import * as Ref from "effect/Ref";
38	import * as Stream from "effect/Stream";
39
40	import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
41	import * as ServerSettings from "../serverSettings.ts";
42	import * as CheckpointService from "./CheckpointService.ts";
43	import type { PendingOrchestrationEffectV2 } from "./EffectOutbox.ts";
44	import * as EventSink from "./EventSink.ts";
45	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
46	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
47	import * as ProjectionStore from "./ProjectionStore.ts";
48	import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
49	import * as McpAppModelContext from "../mcpApps/McpAppModelContext.ts";
50	import * as RunExecutionService from "./RunExecutionService.ts";
51	import * as RunFinalizationService from "./RunFinalizationService.ts";
52
53	const driver = ProviderDriverKind.make("codex");
54
55	const layerRunExecutionTest = RunExecutionService.layer.pipe(
56	Layer.provide(
57	Layer.mergeAll(
58	McpAppModelContext.layerEmpty,
59	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
60	Layer.mock(EventSink.EventSinkV2)({}),
61	IdAllocator.layer,
62	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
63	ingestNormalized: () => Effect.succeed([]),
64	}),
65	ServerSettings.layerTest(),
66	),
67	),
68	);
69
70	it("keeps recoverable turn failures reusable and reserves error for broken threads", () => {
71	assert.equal(RunExecutionService.finalProviderThreadStatus("reusable"), "idle");
72	assert.equal(RunExecutionService.finalProviderThreadStatus("broken"), "error");
73	});
74
75	it.effect("routes shared-runtime events only to their owning root run", () =>
76	Effect.gen(function* () {
77	const now = yield* DateTime.now;
78	const first: RunExecutionService.ProviderEventRouteIdentity = {
79	threadId: ThreadId.make("thread:shared-runtime:first"),
80	runId: RunId.make("run:shared-runtime:first"),
81	attemptId: RunAttemptId.make("attempt:shared-runtime:first"),
82	providerThreadId: ProviderThreadId.make("provider-thread:shared-runtime:first"),
83	};
84	const second: RunExecutionService.ProviderEventRouteIdentity = {
85	threadId: ThreadId.make("thread:shared-runtime:second"),
86	runId: RunId.make("run:shared-runtime:second"),
87	attemptId: RunAttemptId.make("attempt:shared-runtime:second"),
88	providerThreadId: ProviderThreadId.make("provider-thread:shared-runtime:second"),
89	};
90	const firstTurnId = ProviderTurnId.make("provider-turn:shared-runtime:first");
91	const turnEvent: ProviderAdapter.ProviderAdapterV2Event = {
92	type: "provider_turn.updated",
93	driver,
94	threadId: first.threadId,
95	providerTurn: {
96	id: firstTurnId,
97	providerThreadId: first.providerThreadId,
98	nodeId: NodeId.make("node:shared-runtime:first"),
99	runAttemptId: first.attemptId,
100	nativeTurnRef: null,
101	ordinal: 1,
102	status: "running",
103	startedAt: now,
104	completedAt: null,
105	},
106	};
107	const messageEvent: ProviderAdapter.ProviderAdapterV2Event = {
108	type: "message.updated",
109	driver,
110	message: {
111	createdBy: "agent",
112	creationSource: "provider",
113	id: MessageId.make("message:shared-runtime:first"),
114	threadId: first.threadId,
115	runId: first.runId,
116	nodeId: NodeId.make("node:shared-runtime:first"),
117	role: "assistant",
118	text: "first only",
119	attachments: [],
120	streaming: false,
121	createdAt: now,
122	updatedAt: now,
123	},
124	};
125	const terminalEvent: ProviderAdapter.ProviderAdapterV2Event = {
126	type: "turn.terminal",
127	driver,
128	providerThreadId: first.providerThreadId,
129	providerTurnId: firstTurnId,
130	runOrdinal: 1,
131	status: "completed",
132	failure: null,
133	threadDisposition: "reusable",
134	};
135
136	const firstInitial = RunExecutionService.makeProviderEventRoutingState({
137	identity: first,
138	providerTurnId: null,
139	});
140	const secondInitial = RunExecutionService.makeProviderEventRoutingState({
141	identity: second,
142	providerTurnId: null,
143	});
144	const [firstTurnAccepted, firstAfterTurn] = RunExecutionService.routeProviderEvent(
145	turnEvent,
146	first,
147	firstInitial,
148	);
149	const [secondTurnAccepted, secondAfterTurn] = RunExecutionService.routeProviderEvent(
150	turnEvent,
151	second,
152	secondInitial,
153	);
154
155	assert.isTrue(firstTurnAccepted);
156	assert.isFalse(secondTurnAccepted);
157	assert.isTrue(RunExecutionService.routeProviderEvent(messageEvent, first, firstAfterTurn)[0]);
158	assert.isFalse(
159	RunExecutionService.routeProviderEvent(messageEvent, second, secondAfterTurn)[0],
160	);
161	assert.isTrue(RunExecutionService.routeProviderEvent(terminalEvent, first, firstAfterTurn)[0]);
162	assert.isFalse(
163	RunExecutionService.routeProviderEvent(terminalEvent, second, secondAfterTurn)[0],
164	);
165	}),
166	);
167
168	it("routes only enrolled workflow members after the root turn ends", () => {
169	const threadId = ThreadId.make("thread:late-child");
170	const rootProviderTurnId = ProviderTurnId.make("provider-turn:late-child");
171	const identity: RunExecutionService.ProviderEventRouteIdentity = {
172	threadId,
173	runId: RunId.make("run:late-child"),
174	attemptId: RunAttemptId.make("attempt:late-child"),
175	providerThreadId: ProviderThreadId.make("provider-thread:late-child"),
176	};
177	const childCreated = (
178	childThreadId: ThreadId,
179	parentThreadId = threadId,
180	): ProviderAdapter.ProviderAdapterV2Event =>
181	({
182	type: "app_thread.created",
183	driver,
184	appThread: {
185	id: childThreadId,
186	lineage: {
187	parentThreadId,
188	relationshipToParent: "subagent",
189	rootThreadId: threadId,
190	},
191	},
192	}) as ProviderAdapter.ProviderAdapterV2Event;
193	const earlyChild = ThreadId.make("thread:late-child:early");
194	const lateChild = ThreadId.make("thread:late-child:late");
195
196	const initial = RunExecutionService.makeProviderEventRoutingState({
197	identity,
198	providerTurnId: rootProviderTurnId,
199	});
200	const [earlyAccepted, live] = RunExecutionService.routeProviderEvent(
201	childCreated(earlyChild),
202	identity,
203	initial,
204	);
205	assert.isTrue(earlyAccepted);
206	const existingDescendant = ThreadId.make("thread:late-child:existing-descendant");
207	const [descendantAccepted, withDescendant] = RunExecutionService.routeProviderEvent(
208	childCreated(existingDescendant, earlyChild),
209	identity,
210	live,
211	);
212	assert.isTrue(descendantAccepted);
213	const [terminalAccepted, ended] = RunExecutionService.routeProviderEvent(
214	{
215	type: "turn.terminal",
216	driver,
217	providerThreadId: identity.providerThreadId,
218	providerTurnId: rootProviderTurnId,
219	runOrdinal: 1,
220	status: "completed",
221	failure: null,
222	threadDisposition: "reusable",
223	},
224	identity,
225	withDescendant,
226	);
227	assert.isTrue(terminalAccepted);
228	// A child the root launched before it ended stays with this run.
229	assert.isTrue(ended.ownedThreadIds.has(earlyChild));
230	const [lateAccepted, afterLate] = RunExecutionService.routeProviderEvent(
231	childCreated(lateChild),
232	identity,
233	ended,
234	);
235	assert.isFalse(lateAccepted);
236	assert.isFalse(afterLate.ownedThreadIds.has(lateChild));
237
238	const memberId = ThreadId.make("thread:late-child:workflow-member");
239	assert.isFalse(
240	RunExecutionService.routeProviderEvent(childCreated(memberId, earlyChild), identity, ended)[0],
241	);
242	const coordinator: OrchestrationV2Subagent = {
243	id: NodeId.make("node:late-child:workflow"),
244	threadId,
245	runId: identity.runId,
246	parentNodeId: NodeId.make("node:late-child"),
247	origin: "provider_native",
248	createdBy: "agent",
249	driver,
250	providerInstanceId: ProviderInstanceId.make("codex"),
251	providerThreadId: null,
252	childThreadId: earlyChild,
253	nativeTaskRef: null,
254	prompt: "Review the code",
255	title: "Review",
256	model: null,
257	status: "running",
258	result: null,
259	startedAt: null,
260	completedAt: null,
261	updatedAt: DateTime.makeUnsafe(0),
262	workflow: {
263	phases: [],
264	agents: [{ index: 0, label: "Reviewer", state: "running", childThreadId: memberId }],
265	},
266	};
267	const [foreignAccepted, afterForeign] = RunExecutionService.routeProviderEvent(
268	{
269	type: "subagent.updated",
270	driver,
271	subagent: {
272	...coordinator,
273	threadId: earlyChild,
274	childThreadId: existingDescendant,
275	runId: RunId.make("run:late-child:new"),
276	},
277	},
278	identity,
279	ended,
280	);
281	assert.isTrue(foreignAccepted);
282	assert.isFalse(
283	RunExecutionService.routeProviderEvent(
284	childCreated(memberId, existingDescendant),
285	identity,
286	afterForeign,
287	)[0],
288	);
289	const [workflowAccepted, afterWorkflow] = RunExecutionService.routeProviderEvent(
290	{ type: "subagent.updated", driver, subagent: coordinator },
291	identity,
292	ended,
293	);
294	assert.isTrue(workflowAccepted);
295	assert.isFalse(
296	RunExecutionService.routeProviderEvent(childCreated(memberId), identity, afterWorkflow)[0],
297	);
298	const [memberAccepted, afterMember] = RunExecutionService.routeProviderEvent(
299	childCreated(memberId, earlyChild),
300	identity,
301	afterWorkflow,
302	);
303	assert.isTrue(memberAccepted);
304	assert.isTrue(afterMember.ownedThreadIds.has(memberId));
305	const [unrelatedAccepted] = RunExecutionService.routeProviderEvent(
306	childCreated(ThreadId.make("thread:late-child:unrelated-member"), lateChild),
307	identity,
308	ended,
309	);
310	assert.isFalse(unrelatedAccepted);
311	const [unrelatedDescendantAccepted, afterUnrelatedDescendant] =
312	RunExecutionService.routeProviderEvent(
313	childCreated(ThreadId.make("thread:late-child:unrelated-descendant"), earlyChild),
314	identity,
315	afterWorkflow,
316	);
317	assert.isFalse(unrelatedDescendantAccepted);
318	assert.isFalse(
319	afterUnrelatedDescendant.ownedThreadIds.has(
320	ThreadId.make("thread:late-child:unrelated-descendant"),
321	),
322	);
323	});
324
325	it("does not route a superseded attempt through a reused provider thread", () => {
326	const threadId = ThreadId.make("thread:shared-runtime:restart");
327	const providerThreadId = ProviderThreadId.make("provider-thread:shared-runtime:restart");
328	const oldAttempt: RunExecutionService.ProviderEventRouteIdentity = {
329	threadId,
330	runId: RunId.make("run:shared-runtime:restart"),
331	attemptId: RunAttemptId.make("attempt:shared-runtime:restart:old"),
332	providerThreadId,
333	};
334	const newAttempt: RunExecutionService.ProviderEventRouteIdentity = {
335	...oldAttempt,
336	attemptId: RunAttemptId.make("attempt:shared-runtime:restart:new"),
337	};
338	const oldTurnEvent: ProviderAdapter.ProviderAdapterV2Event = {
339	type: "provider_turn.updated",
340	driver,
341	threadId,
342	providerTurn: {
343	id: ProviderTurnId.make("provider-turn:shared-runtime:restart:old"),
344	providerThreadId,
345	nodeId: NodeId.make("node:shared-runtime:restart:old"),
346	runAttemptId: oldAttempt.attemptId,
347	nativeTurnRef: null,
348	ordinal: 1,
349	status: "interrupted",
350	startedAt: null,
351	completedAt: null,
352	},
353	};
354
355	const newState = RunExecutionService.makeProviderEventRoutingState({
356	identity: newAttempt,
357	providerTurnId: null,
358	});
359	assert.isFalse(RunExecutionService.routeProviderEvent(oldTurnEvent, newAttempt, newState)[0]);
360	});
361
362	it("routes only exact same-thread background items inherited from settled runs", () => {
363	const threadId = ThreadId.make("thread:inherited-background-routing");
364	const otherThreadId = ThreadId.make("thread:inherited-background-routing:other");
365	const priorRunId = RunId.make("run:inherited-background-routing:prior");
366	const currentRunId = RunId.make("run:inherited-background-routing:current");
367	const itemId = TurnItemId.make("turn-item:inherited-background-routing");
368	const identity: RunExecutionService.ProviderEventRouteIdentity = {
369	threadId,
370	runId: currentRunId,
371	attemptId: RunAttemptId.make("attempt:inherited-background-routing:current"),
372	providerThreadId: ProviderThreadId.make("provider-thread:inherited-background-routing:current"),
373	};
374	const initial = RunExecutionService.makeProviderEventRoutingState({
375	identity,
376	inheritedBackgroundTurnItems: [{ id: itemId, runId: priorRunId }],
377	providerTurnId: null,
378	});
379	const inheritedRunning = {
380	type: "turn_item.updated",
381	driver,
382	turnItem: {
383	id: itemId,
384	threadId,
385	runId: priorRunId,
386	providerTurnId: null,
387	ordinal: 1,
388	type: "subagent",
389	status: "running",
390	},
391	} as Extract<ProviderAdapter.ProviderAdapterV2Event, { type: "turn_item.updated" }>;
392	const unrelatedRunItem = {
393	...inheritedRunning,
394	turnItem: {
395	...inheritedRunning.turnItem,
396	runId: RunId.make("run:inherited-background-routing:unrelated"),
397	},
398	} as ProviderAdapter.ProviderAdapterV2Event;
399	const unlistedPriorItem = {
400	...inheritedRunning,
401	turnItem: {
402	...inheritedRunning.turnItem,
403	id: TurnItemId.make("turn-item:inherited-background-routing:unrelated"),
404	},
405	} as ProviderAdapter.ProviderAdapterV2Event;
406	const unrelatedThreadItem = {
407	...inheritedRunning,
408	turnItem: { ...inheritedRunning.turnItem, threadId: otherThreadId },
409	} as ProviderAdapter.ProviderAdapterV2Event;
410	const ordinaryItem = {
411	...inheritedRunning,
412	turnItem: { ...inheritedRunning.turnItem, type: "reasoning" as const },
413	} as ProviderAdapter.ProviderAdapterV2Event;
414	const inheritedTerminal = {
415	...inheritedRunning,
416	turnItem: { ...inheritedRunning.turnItem, status: "completed" as const },
417	} as ProviderAdapter.ProviderAdapterV2Event;
418
419	const [runningAccepted, afterRunning] = RunExecutionService.routeProviderEvent(
420	inheritedRunning,
421	identity,
422	initial,
423	);
424	assert.isTrue(runningAccepted);
425	assert.isFalse(
426	RunExecutionService.routeProviderEvent(unrelatedRunItem, identity, afterRunning)[0],
427	);
428	assert.isFalse(
429	RunExecutionService.routeProviderEvent(unlistedPriorItem, identity, afterRunning)[0],
430	);
431	assert.isFalse(
432	RunExecutionService.routeProviderEvent(unrelatedThreadItem, identity, afterRunning)[0],
433	);
434	assert.isFalse(RunExecutionService.routeProviderEvent(ordinaryItem, identity, afterRunning)[0]);
435
436	const [terminalAccepted, afterTerminal] = RunExecutionService.routeProviderEvent(
437	inheritedTerminal,
438	identity,
439	afterRunning,
440	);
441	assert.isTrue(terminalAccepted);
442	assert.isFalse(
443	RunExecutionService.routeProviderEvent(inheritedRunning, identity, afterTerminal)[0],
444	"a nonterminal replay must not resurrect an inherited terminal",
445	);
446	});
447
448	it("selects only live background items from non-completed settled prior runs", () => {
449	const threadId = ThreadId.make("thread:inherited-background-selection");
450	const otherThreadId = ThreadId.make("thread:inherited-background-selection:other");
451	const currentProviderThreadId = ProviderThreadId.make(
452	"provider-thread:inherited-background-selection:current",
453	);
454	const foreignProviderThreadId = ProviderThreadId.make(
455	"provider-thread:inherited-background-selection:foreign-session",
456	);
457	const interruptedRunId = RunId.make("run:inherited-background-selection:interrupted");
458	const failedRunId = RunId.make("run:inherited-background-selection:failed");
459	const cancelledRunId = RunId.make("run:inherited-background-selection:cancelled");
460	const completedRunId = RunId.make("run:inherited-background-selection:completed");
461	const rolledBackRunId = RunId.make("run:inherited-background-selection:rolled-back");
462	const currentRunId = RunId.make("run:inherited-background-selection:current");
463	const inheritedItemId = TurnItemId.make("turn-item:inherited-background-selection:live");
464	const failedItemId = TurnItemId.make("turn-item:inherited-background-selection:failed");
465	const cancelledItemId = TurnItemId.make("turn-item:inherited-background-selection:cancelled");
466	const makeRun = (
467	id: RunId,
468	ordinal: number,
469	status: OrchestrationV2Run["status"],
470	runThreadId = threadId,
471	) =>
472	({
473	id,
474	threadId: runThreadId,
475	ordinal,
476	status,
477	}) as OrchestrationV2Run;
478	const makeItem = (
479	id: TurnItemId,
480	runId: RunId,
481	status: OrchestrationV2TurnItem["status"],
482	type: OrchestrationV2TurnItem["type"] = "subagent",
483	itemThreadId = threadId,
484	providerThreadId = currentProviderThreadId,
485	) =>
486	({
487	id,
488	threadId: itemThreadId,
489	runId,
490	providerThreadId,
491	type,
492	status,
493	}) as OrchestrationV2TurnItem;
494
495	const selected = RunExecutionService.selectInheritedBackgroundTurnItems({
496	threadId,
497	currentProviderThreadId,
498	currentRunOrdinal: 6,
499	runs: [
500	makeRun(interruptedRunId, 1, "interrupted"),
501	makeRun(failedRunId, 2, "failed"),
502	makeRun(cancelledRunId, 3, "cancelled"),
503	makeRun(completedRunId, 4, "completed"),
504	makeRun(rolledBackRunId, 5, "rolled_back"),
505	makeRun(currentRunId, 6, "running"),
506	makeRun(
507	RunId.make("run:inherited-background-selection:other"),
508	1,
509	"interrupted",
510	otherThreadId,
511	),
512	],
513	turnItems: [
514	makeItem(inheritedItemId, interruptedRunId, "running"),
515	makeItem(failedItemId, failedRunId, "running"),
516	makeItem(cancelledItemId, cancelledRunId, "running"),
517	makeItem(
518	TurnItemId.make("turn-item:inherited-background-selection:completed-run"),
519	completedRunId,
520	"running",
521	),
522	makeItem(
523	TurnItemId.make("turn-item:inherited-background-selection:terminal"),
524	interruptedRunId,
525	"completed",
526	),
527	makeItem(
528	TurnItemId.make("turn-item:inherited-background-selection:ordinary"),
529	interruptedRunId,
530	"running",
531	"reasoning",
532	),
533	makeItem(
534	TurnItemId.make("turn-item:inherited-background-selection:rolled-back"),
535	rolledBackRunId,
536	"running",
537	),
538	makeItem(
539	TurnItemId.make("turn-item:inherited-background-selection:other-thread"),
540	interruptedRunId,
541	"running",
542	"subagent",
543	otherThreadId,
544	),
545	makeItem(
546	TurnItemId.make("turn-item:inherited-background-selection:foreign-provider"),
547	interruptedRunId,
548	"running",
549	"subagent",
550	threadId,
551	foreignProviderThreadId,
552	),
553	],
554	});
555
556	assert.deepEqual(selected, [
557	{ id: inheritedItemId, runId: interruptedRunId },
558	{ id: failedItemId, runId: failedRunId },
559	{ id: cancelledItemId, runId: cancelledRunId },
560	]);
561	});
562
563	it("does not carry interrupted or still-running child ownership into later attempts", () => {
564	assert.isFalse(RunExecutionService.canRouteRelatedSubagent("interrupted"));
565	assert.isFalse(RunExecutionService.canRouteRelatedSubagent("failed"));
566	assert.isFalse(RunExecutionService.canRouteRelatedSubagent("cancelled"));
567	assert.isTrue(RunExecutionService.canRouteRelatedSubagent("completed"));
568	// The launching run still ingests a running subagent's child thread.
569	assert.isFalse(RunExecutionService.canRouteRelatedSubagent("running"));
570
571	const threadId = ThreadId.make("thread:related-child:next-attempt");
572	const childThreadId = ThreadId.make("thread:related-child:interrupted");
573	const identity: RunExecutionService.ProviderEventRouteIdentity = {
574	threadId,
575	runId: RunId.make("run:related-child:next-attempt"),
576	attemptId: RunAttemptId.make("attempt:related-child:next-attempt"),
577	providerThreadId: ProviderThreadId.make("provider-thread:related-child:next-attempt"),
578	};
579	const state = RunExecutionService.makeProviderEventRoutingState({
580	identity,
581	providerTurnId: null,
582	relatedThreadIds: RunExecutionService.canRouteRelatedSubagent("interrupted")
583	? [childThreadId]
584	: [],
585	});
586	const childNodeId = NodeId.make("node:related-child:interrupted");
587	const lateChildNode = {
588	type: "node.updated",
589	driver,
590	node: {
591	id: childNodeId,
592	threadId: childThreadId,
593	runId: null,
594	parentNodeId: null,
595	rootNodeId: childNodeId,
596	kind: "root_turn",
597	status: "completed",
598	countsForRun: false,
599	providerThreadId: null,
600	providerTurnId: null,
601	nativeItemRef: null,
602	runtimeRequestId: null,
603	checkpointScopeId: null,
604	startedAt: null,
605	completedAt: null,
606	},
607	} satisfies ProviderAdapter.ProviderAdapterV2Event;
608
609	assert.isFalse(RunExecutionService.routeProviderEvent(lateChildNode, identity, state)[0]);
610	});
611
612	it.effect("rechecks run ownership immediately before calling the provider", () =>
613	Effect.gen(function* () {
614	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
615	const guardCalls = yield* Ref.make(0);
616	const providerStarts = yield* Ref.make(0);
617	const threadId = ThreadId.make("thread:run-execution-start-guard");
618	const runId = RunId.make("run:run-execution-start-guard");
619	const attemptId = RunAttemptId.make("attempt:run-execution-start-guard");
620	const providerThreadId = ProviderThreadId.make("provider-thread:run-execution-start-guard");
621	const providerInstanceId = ProviderInstanceId.make("codex");
622	const providerSessionId = ProviderSessionId.make("session:run-execution-start-guard");
623	const rootNodeId = NodeId.make("node:run-execution-start-guard");
624	const run = {
625	id: runId,
626	threadId,
627	ordinal: 1,
628	providerInstanceId,
629	} as OrchestrationV2Run;
630	const rootNode = { id: rootNodeId } as OrchestrationV2ExecutionNode;
631	const providerThread = {
632	id: providerThreadId,
633	driver,
634	} as OrchestrationV2ProviderThread;
635	const attempt = {
636	id: attemptId,
637	providerTurnId: null,
638	} as OrchestrationV2RunAttempt;
639	const session = {
640	events: Stream.never,
641	startTurn: () => Ref.update(providerStarts, (count) => count + 1),
642	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime;
643
644	yield* runExecution.startRootRun({
645	commandId: CommandId.make("command:run-execution-start-guard"),
646	appThread: { id: threadId } as OrchestrationV2AppThread,
647	providerSessionId,
648	session,
649	run,
650	rootNode,
651	checkpointScope: {
652	id: CheckpointScopeId.make("checkpoint-scope:run-execution-start-guard"),
653	} as OrchestrationV2CheckpointScope,
654	providerThread,
655	attempt,
656	attemptId,
657	providerTurnOrdinal: 1,
658	shouldStartProviderTurn: () =>
659	Ref.modify(guardCalls, (calls) => [calls === 0, calls + 1] as const),
660	message: {
661	messageId: MessageId.make("message:run-execution-start-guard"),
662	text: "Do not start after ownership changes.",
663	attachments: [],
664	createdBy: "user",
665	creationSource: "web",
666	},
667	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
668	runtimePolicy: {
669	runtimeMode: "full-access",
670	interactionMode: "default",
671	cwd: process.cwd(),
672	approvalPolicy: "never",
673	sandboxPolicy: {
674	type: "readOnly",
675	access: { type: "fullAccess" },
676	networkAccess: false,
677	},
678	},
679	});
680
681	assert.equal(yield* Ref.get(guardCalls), 2);
682	assert.equal(yield* Ref.get(providerStarts), 0);
683	}).pipe(Effect.provide(layerRunExecutionTest)),
684	);
685
686	it.effect("passes the thread's MCP app context to the provider under a safe key", () =>
687	Effect.gen(function* () {
688	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
689	const started = yield* Deferred.make<ProviderAdapter.ProviderAdapterV2TurnInput>();
690	const threadId = ThreadId.make("thread:run-execution-app-context");
691	const providerInstanceId = ProviderInstanceId.make("codex");
692	const attemptId = RunAttemptId.make("attempt:run-execution-app-context");
693	const session = {
694	events: Stream.never,
695	startTurn: (input: ProviderAdapter.ProviderAdapterV2TurnInput) =>
696	Deferred.succeed(started, input),
697	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime;
698
699	yield* runExecution.startRootRun({
700	commandId: CommandId.make("command:run-execution-app-context"),
701	appThread: { id: threadId } as OrchestrationV2AppThread,
702	providerSessionId: ProviderSessionId.make("session:run-execution-app-context"),
703	session,
704	run: {
705	id: RunId.make("run:run-execution-app-context"),
706	threadId,
707	ordinal: 1,
708	providerInstanceId,
709	} as OrchestrationV2Run,
710	rootNode: {
711	id: NodeId.make("node:run-execution-app-context"),
712	} as OrchestrationV2ExecutionNode,
713	checkpointScope: {
714	id: CheckpointScopeId.make("checkpoint-scope:run-execution-app-context"),
715	} as OrchestrationV2CheckpointScope,
716	providerThread: {
717	id: ProviderThreadId.make("provider-thread:run-execution-app-context"),
718	driver,
719	} as OrchestrationV2ProviderThread,
720	attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt,
721	attemptId,
722	providerTurnOrdinal: 1,
723	message: {
724	messageId: MessageId.make("message:run-execution-app-context"),
725	text: "What is on my list?",
726	attachments: [],
727	createdBy: "user",
728	creationSource: "web",
729	},
730	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
731	runtimePolicy: { runtimeMode: "full-access", interactionMode: "default", cwd: process.cwd() },
732	});
733
734	const input = yield* Deferred.await(started);
735	// Item ids carry colons, and server names are free text; neither reaches the key.
736	assert.deepEqual(input.appContext, [
737	{ key: "mcp_app_turn-item_provider_codex_native-item_call-1", text: "2 overdue" },
738	]);
739	}).pipe(
740	Effect.provide(
741	RunExecutionService.layer.pipe(
742	Layer.provide(
743	Layer.mergeAll(
744	Layer.mock(McpAppModelContext.McpAppModelContext)({
745	forThread: () =>
746	Effect.succeed([
747	{
748	itemId: "turn-item:provider:codex:native-item:call-1",
749	server: "my tools>",
750	tool: "list",
751	text: "2 overdue",
752	},
753	]),
754	}),
755	Layer.mock(CheckpointService.CheckpointServiceV2)({
756	captureBaseline: () => Effect.void,
757	}),
758	Layer.mock(EventSink.EventSinkV2)({}),
759	IdAllocator.layer,
760	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
761	ingestNormalized: () => Effect.succeed([]),
762	}),
763	ServerSettings.layerTest(),
764	),
765	),
766	),
767	),
768	),
769	);
770
771	it.effect("fails the run when its ownership check cannot be read before calling the provider", () =>
772	Effect.gen(function* () {
773	const guardCalls = yield* Ref.make(0);
774	const providerStarts = yield* Ref.make(0);
775	const writes = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
776	const threadId = ThreadId.make("thread:run-execution-start-guard-read");
777	const runId = RunId.make("run:run-execution-start-guard-read");
778	const attemptId = RunAttemptId.make("attempt:run-execution-start-guard-read");
779	const providerInstanceId = ProviderInstanceId.make("codex");
780	const layerTest = RunExecutionService.layer.pipe(
781	Layer.provide(
782	Layer.mergeAll(
783	McpAppModelContext.layerEmpty,
784	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
785	Layer.mock(EventSink.EventSinkV2)({
786	writeIfRunCurrent: (input) =>
787	Ref.update(writes, (current) => [...current, ...input.events]).pipe(
788	Effect.as({ committed: true, storedEvents: [] }),
789	),
790	}),
791	IdAllocator.layer,
792	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
793	ingestNormalized: () => Effect.succeed([]),
794	}),
795	ServerSettings.layerTest(),
796	),
797	),
798	);
799
800	yield* Effect.gen(function* () {
801	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
802	yield* runExecution.startRootRun({
803	commandId: CommandId.make("command:run-execution-start-guard-read"),
804	appThread: { id: threadId } as OrchestrationV2AppThread,
805	providerSessionId: ProviderSessionId.make("session:run-execution-start-guard-read"),
806	session: {
807	events: Stream.never,
808	startTurn: () => Ref.update(providerStarts, (count) => count + 1),
809	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
810	run: { id: runId, threadId, ordinal: 1, providerInstanceId } as OrchestrationV2Run,
811	rootNode: {
812	id: NodeId.make("node:run-execution-start-guard-read"),
813	} as OrchestrationV2ExecutionNode,
814	checkpointScope: {
815	id: CheckpointScopeId.make("checkpoint-scope:run-execution-start-guard-read"),
816	} as OrchestrationV2CheckpointScope,
817	providerThread: {
818	id: ProviderThreadId.make("provider-thread:run-execution-start-guard-read"),
819	driver,
820	} as OrchestrationV2ProviderThread,
821	attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt,
822	attemptId,
823	providerTurnOrdinal: 1,
824	// The preparation check passes; the check right before the provider
825	// call cannot read the run.
826	shouldStartProviderTurn: () =>
827	Ref.getAndUpdate(guardCalls, (calls) => calls + 1).pipe(
828	Effect.flatMap((calls) =>
829	calls === 0
830	? Effect.succeed(true)
831	: Effect.fail(
832	new ProjectionStore.ProjectionStoreReadError({
833	threadId,
834	cause: "database unavailable",
835	}),
836	),
837	),
838	),
839	// The failure is settled by the guarded write, not by another read.
840	shouldFinalizeRun: () =>
841	Effect.fail(
842	new ProjectionStore.ProjectionStoreReadError({
843	threadId,
844	cause: "database unavailable",
845	}),
846	),
847	message: {
848	messageId: MessageId.make("message:run-execution-start-guard-read"),
849	text: "Start while the store is down.",
850	attachments: [],
851	createdBy: "user",
852	creationSource: "web",
853	},
854	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
855	runtimePolicy: {
856	runtimeMode: "full-access",
857	interactionMode: "default",
858	cwd: process.cwd(),
859	approvalPolicy: "never",
860	sandboxPolicy: {
861	type: "readOnly",
862	access: { type: "fullAccess" },
863	networkAccess: false,
864	},
865	},
866	});
867	}).pipe(Effect.provide(layerTest));
868
869	assert.equal(yield* Ref.get(guardCalls), 2);
870	assert.equal(yield* Ref.get(providerStarts), 0);
871	const runUpdate = (yield* Ref.get(writes)).find((event) => event.type === "run.updated");
872	assert.equal(
873	runUpdate?.type === "run.updated" ? runUpdate.payload.status : undefined,
874	"failed",
875	);
876	}),
877	);
878
879	it.effect(
880	"dispatches only attachment-free compact commands through the native compaction path",
881	() =>
882	Effect.gen(function* () {
883	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
884	const calls: Array<string> = [];
885	const attachment = ChatFileAttachment.make({
886	type: "file",
887	id: ChatAttachmentId.make("compact-file-12345678-1234-1234-1234-123456789abc"),
888	name: "notes.txt",
889	mimeType: "text/plain",
890	sizeBytes: 1,
891	});
892	const cases = [
893	{ text: " /compact ", attachments: [], expected: "compact" },
894	{ text: "/compact", attachments: [attachment], expected: "prompt" },
895	{ text: "Continue the work", attachments: [], expected: "prompt" },
896	];
897	for (const [index, testCase] of cases.entries()) {
898	const threadId = ThreadId.make(`thread:compact-routing:${index}`);
899	const attemptId = RunAttemptId.make(`attempt:compact-routing:${index}`);
900	yield* runExecution.startRootRun({
901	commandId: CommandId.make(`command:compact-routing:${index}`),
902	appThread: { id: threadId } as OrchestrationV2AppThread,
903	providerSessionId: ProviderSessionId.make(`session:compact-routing:${index}`),
904	session: {
905	events: Stream.never,
906	startTurn: () =>
907	Effect.sync(() => {
908	calls.push("prompt");
909	}),
910	compactThread: () =>
911	Effect.sync(() => {
912	calls.push("compact");
913	}),
914	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
915	run: {
916	id: RunId.make(`run:compact-routing:${index}`),
917	threadId,
918	ordinal: 1,
919	providerInstanceId: ProviderInstanceId.make("codex"),
920	} as OrchestrationV2Run,
921	rootNode: {
922	id: NodeId.make(`node:compact-routing:${index}`),
923	} as OrchestrationV2ExecutionNode,
924	checkpointScope: {
925	id: CheckpointScopeId.make(`checkpoint:compact-routing:${index}`),
926	} as OrchestrationV2CheckpointScope,
927	providerThread: {
928	id: ProviderThreadId.make(`provider-thread:compact-routing:${index}`),
929	driver,
930	} as OrchestrationV2ProviderThread,
931	attempt: { id: attemptId, providerTurnId: null } as OrchestrationV2RunAttempt,
932	attemptId,
933	providerTurnOrdinal: 1,
934	message: {
935	messageId: MessageId.make(`message:compact-routing:${index}`),
936	text: testCase.text,
937	attachments: testCase.attachments,
938	createdBy: "user",
939	creationSource: "web",
940	},
941	modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.4" },
942	runtimePolicy: {
943	runtimeMode: "full-access",
944	interactionMode: "default",
945	cwd: process.cwd(),
946	approvalPolicy: "never",
947	sandboxPolicy: {
948	type: "readOnly",
949	access: { type: "fullAccess" },
950	networkAccess: false,
951	},
952	},
953	});
954	}
955	assert.deepEqual(
956	calls,
957	cases.map((testCase) => testCase.expected),
958	);
959	}).pipe(Effect.provide(layerRunExecutionTest)),
960	);
961
962	it.effect("refreshes MCP credential liveness before calling the provider", () =>
963	Effect.gen(function* () {
964	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
965	const order = yield* Ref.make<ReadonlyArray<string>>([]);
966	const threadId = ThreadId.make("thread:run-execution-mcp-liveness");
967	const touchActiveMcpThread = vi
968	.spyOn(McpSessionRegistry, "touchActiveMcpThread")
969	.mockImplementation((touchedThreadId) =>
970	Ref.update(order, (entries) => [...entries, `touch:${touchedThreadId}`]),
971	);
972
973	yield* runExecution
974	.startRootRun({
975	commandId: CommandId.make("command:run-execution-mcp-liveness"),
976	appThread: { id: threadId } as OrchestrationV2AppThread,
977	providerSessionId: ProviderSessionId.make("session:run-execution-mcp-liveness"),
978	session: {
979	events: Stream.never,
980	startTurn: () => Ref.update(order, (entries) => [...entries, "start-turn"]),
981	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
982	run: {
983	id: RunId.make("run:run-execution-mcp-liveness"),
984	threadId,
985	ordinal: 1,
986	providerInstanceId: ProviderInstanceId.make("codex"),
987	} as OrchestrationV2Run,
988	rootNode: {
989	id: NodeId.make("node:run-execution-mcp-liveness"),
990	} as OrchestrationV2ExecutionNode,
991	checkpointScope: {
992	id: CheckpointScopeId.make("checkpoint-scope:run-execution-mcp-liveness"),
993	} as OrchestrationV2CheckpointScope,
994	providerThread: {
995	id: ProviderThreadId.make("provider-thread:run-execution-mcp-liveness"),
996	driver,
997	} as OrchestrationV2ProviderThread,
998	attempt: {
999	id: RunAttemptId.make("attempt:run-execution-mcp-liveness"),
1000	providerTurnId: null,
1001	} as OrchestrationV2RunAttempt,
1002	attemptId: RunAttemptId.make("attempt:run-execution-mcp-liveness"),
1003	providerTurnOrdinal: 1,
1004	message: {
1005	messageId: MessageId.make("message:run-execution-mcp-liveness"),
1006	text: "Keep the MCP credential alive.",
1007	attachments: [],
1008	createdBy: "user",
1009	creationSource: "web",
1010	},
1011	modelSelection: {
1012	instanceId: ProviderInstanceId.make("codex"),
1013	model: "gpt-5.4",
1014	},
1015	runtimePolicy: {
1016	runtimeMode: "full-access",
1017	interactionMode: "default",
1018	cwd: process.cwd(),
1019	approvalPolicy: "never",
1020	sandboxPolicy: {
1021	type: "readOnly",
1022	access: { type: "fullAccess" },
1023	networkAccess: false,
1024	},
1025	},
1026	})
1027	.pipe(Effect.ensuring(Effect.sync(() => touchActiveMcpThread.mockRestore())));
1028
1029	assert.deepEqual(yield* Ref.get(order), [`touch:${threadId}`, "start-turn"]);
1030	}).pipe(Effect.provide(layerRunExecutionTest)),
1031	);
1032
1033	it.effect("starts the provider when checkpoint baseline capture fails", () =>
1034	Effect.gen(function* () {
1035	const threadId = ThreadId.make("thread:run-execution-baseline-failure");
1036	const runId = RunId.make("run:run-execution-baseline-failure");
1037	const attemptId = RunAttemptId.make("attempt:run-execution-baseline-failure");
1038	const providerThreadId = ProviderThreadId.make(
1039	"provider-thread:run-execution-baseline-failure",
1040	);
1041	const providerInstanceId = ProviderInstanceId.make("codex");
1042	const providerSessionId = ProviderSessionId.make("session:run-execution-baseline-failure");
1043	const rootNodeId = NodeId.make("node:run-execution-baseline-failure");
1044	const checkpointScope = {
1045	id: CheckpointScopeId.make("checkpoint-scope:run-execution-baseline-failure"),
1046	} as OrchestrationV2CheckpointScope;
1047	const providerStarts = yield* Ref.make(0);
1048	const writes = yield* Ref.make<
1049	ReadonlyArray<{
1050	events: ReadonlyArray<OrchestrationV2DomainEvent>;
1051	effects: ReadonlyArray<PendingOrchestrationEffectV2>;
1052	}>
1053	>([]);
1054	const layerTest = RunExecutionService.layer.pipe(
1055	Layer.provide(
1056	Layer.mergeAll(
1057	McpAppModelContext.layerEmpty,
1058	Layer.mock(CheckpointService.CheckpointServiceV2)({
1059	captureBaseline: () =>
1060	Effect.fail(
1061	new CheckpointService.CheckpointBaselineCaptureError({
1062	scopeId: checkpointScope.id,
1063	ordinalWithinScope: 0,
1064	cause: new Error("VCS process timed out"),
1065	}),
1066	),
1067	}),
1068	Layer.mock(EventSink.EventSinkV2)({
1069	writeWithEffects: (input) =>
1070	Ref.update(writes, (current) => [
1071	...current,
1072	{ events: input.events, effects: input.effects },
1073	]).pipe(Effect.as([])),
1074	}),
1075	IdAllocator.layer,
1076	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
1077	ingestNormalized: () => Effect.succeed([]),
1078	}),
1079	ServerSettings.layerTest(),
1080	),
1081	),
1082	);
1083
1084	yield* Effect.gen(function* () {
1085	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
1086	yield* runExecution.startRootRun({
1087	commandId: CommandId.make("command:run-execution-baseline-failure"),
1088	appThread: { id: threadId } as OrchestrationV2AppThread,
1089	providerSessionId,
1090	session: {
1091	events: Stream.never,
1092	startTurn: () => Ref.update(providerStarts, (count) => count + 1),
1093	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
1094	run: {
1095	id: runId,
1096	threadId,
1097	ordinal: 1,
1098	providerInstanceId,
1099	} as OrchestrationV2Run,
1100	rootNode: { id: rootNodeId } as OrchestrationV2ExecutionNode,
1101	checkpointScope,
1102	providerThread: {
1103	id: providerThreadId,
1104	driver,
1105	} as OrchestrationV2ProviderThread,
1106	attempt: {
1107	id: attemptId,
1108	providerTurnId: null,
1109	} as OrchestrationV2RunAttempt,
1110	attemptId,
1111	providerTurnOrdinal: 1,
1112	message: {
1113	messageId: MessageId.make("message:run-execution-baseline-failure"),
1114	text: "Start the provider without a baseline.",
1115	attachments: [],
1116	createdBy: "user",
1117	creationSource: "web",
1118	},
1119	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
1120	runtimePolicy: {
1121	runtimeMode: "full-access",
1122	interactionMode: "default",
1123	cwd: process.cwd(),
1124	approvalPolicy: "never",
1125	sandboxPolicy: {
1126	type: "readOnly",
1127	access: { type: "fullAccess" },
1128	networkAccess: false,
1129	},
1130	},
1131	});
1132	}).pipe(Effect.provide(layerTest));
1133
1134	assert.equal(yield* Ref.get(providerStarts), 1);
1135
1136	const events = (yield* Ref.get(writes)).flatMap((write) => [...write.events]);
1137	const failedEvent = events.find(
1138	(event) =>
1139	(event.type === "run.updated" ||
1140	event.type === "run-attempt.updated" ||
1141	event.type === "node.updated") &&
1142	event.payload.status === "failed",
1143	);
1144	assert.isUndefined(failedEvent);
1145	const errorItem = events.find(
1146	(event) => event.type === "turn-item.updated" && event.payload.type === "error",
1147	);
1148	assert.isUndefined(errorItem);
1149	}),
1150	);
1151
1152	it.effect.each(["failure", "interruption", "stale-attempt", "start-guard"] as const)(
1153	"handles %s before the provider turn starts",
1154	(scenario) =>
1155	Effect.gen(function* () {
1156	const threadId = ThreadId.make("thread:run-execution-settings-failure");
1157	const runId = RunId.make("run:run-execution-settings-failure");
1158	const attemptId = RunAttemptId.make("attempt:run-execution-settings-failure");
1159	const providerInstanceId = ProviderInstanceId.make("codex");
1160	const providerSessionId = ProviderSessionId.make("session:run-execution-settings-failure");
1161	const providerThreadId = ProviderThreadId.make(
1162	"provider-thread:run-execution-settings-failure",
1163	);
1164	const rootNodeId = NodeId.make("node:run-execution-settings-failure");
1165	const checkpointScope = {
1166	id: CheckpointScopeId.make("checkpoint-scope:run-execution-settings-failure"),
1167	} as OrchestrationV2CheckpointScope;
1168	const providerStarts = yield* Ref.make(0);
1169	const refreshes = yield* Ref.make(0);
1170	const guardedWrites = yield* Ref.make(0);
1171	const writes = yield* Ref.make<ReadonlyArray<ReadonlyArray<OrchestrationV2DomainEvent>>>([]);
1172	const layerTest = RunExecutionService.layer.pipe(
1173	Layer.provide(
1174	Layer.mergeAll(
1175	McpAppModelContext.layerEmpty,
1176	Layer.mock(CheckpointService.CheckpointServiceV2)({
1177	captureBaseline: () =>
1178	scenario === "start-guard" ? Effect.void : Effect.die("not reached"),
1179	}),
1180	Layer.mock(EventSink.EventSinkV2)({
1181	writeIfRunCurrent: (input) =>
1182	Effect.gen(function* () {
1183	assert.equal(input.threadId, threadId);
1184	assert.equal(input.runId, runId);
1185	assert.equal(input.activeAttemptId, attemptId);
1186	assert.equal(input.expectedStatus, "running");
1187	yield* Ref.update(guardedWrites, (count) => count + 1);
1188	if (scenario === "stale-attempt") {
1189	return { committed: false, storedEvents: [] };
1190	}
1191	yield* Ref.update(writes, (current) => [...current, input.events]);
1192	return { committed: true, storedEvents: [] };
1193	}),
1194	}),
1195	IdAllocator.layer,
1196	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
1197	ingestNormalized: () => Effect.succeed([]),
1198	}),
1199	scenario === "start-guard"
1200	? ServerSettings.layerTest()
1201	: Layer.mock(ServerSettings.ServerSettingsService)({
1202	getSettings:
1203	scenario === "interruption"
1204	? Effect.interrupt
1205	: Effect.fail(
1206	new ServerSettingsError({
1207	settingsPath: "<test>",
1208	operation: "read-file",
1209	cause: new Error("settings read failed"),
1210	}),
1211	),
1212	}),
1213	Layer.succeed(RunFinalizationService.RunFinalizationObserver, {
1214	refresh: () => Effect.void,
1215	refreshAfterTurn: () => Ref.update(refreshes, (count) => count + 1),
1216	}),
1217	),
1218	),
1219	);
1220
1221	const result = yield* Effect.gen(function* () {
1222	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
1223	yield* runExecution.startRootRun({
1224	commandId: CommandId.make("command:run-execution-settings-failure"),
1225	appThread: { id: threadId } as OrchestrationV2AppThread,
1226	providerSessionId,
1227	session: {
1228	events: Stream.never,
1229	startTurn: () => Ref.update(providerStarts, (count) => count + 1),
1230	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
1231	run: {
1232	id: runId,
1233	threadId,
1234	ordinal: 1,
1235	providerInstanceId,
1236	status: "running",
1237	} as OrchestrationV2Run,
1238	rootNode: { id: rootNodeId, status: "running" } as OrchestrationV2ExecutionNode,
1239	checkpointScope,
1240	providerThread: {
1241	id: providerThreadId,
1242	driver,
1243	} as OrchestrationV2ProviderThread,
1244	attempt: {
1245	id: attemptId,
1246	providerTurnId: null,
1247	status: "running",
1248	} as OrchestrationV2RunAttempt,
1249	attemptId,
1250	providerTurnOrdinal: 1,
1251	// A declined start is a normal exit, not a preparation failure.
1252	...(scenario === "start-guard"
1253	? { shouldStartProviderTurn: () => Effect.succeed(false) }
1254	: {}),
1255	message: {
1256	messageId: MessageId.make("message:run-execution-settings-failure"),
1257	text: "Start after settings fail.",
1258	attachments: [],
1259	createdBy: "user",
1260	creationSource: "web",
1261	},
1262	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
1263	runtimePolicy: {
1264	runtimeMode: "full-access",
1265	interactionMode: "default",
1266	cwd: process.cwd(),
1267	approvalPolicy: "never",
1268	sandboxPolicy: {
1269	type: "readOnly",
1270	access: { type: "fullAccess" },
1271	networkAccess: false,
1272	},
1273	},
1274	});
1275	}).pipe(Effect.provide(layerTest), Effect.exit);
1276
1277	assert.equal(yield* Ref.get(providerStarts), 0);
1278	const events = (yield* Ref.get(writes)).flat();
1279	if (scenario === "interruption") {
1280	assert.isTrue(Exit.isFailure(result));
1281	if (Exit.isFailure(result)) assert.isTrue(Cause.hasInterruptsOnly(result.cause));
1282	assert.equal(yield* Ref.get(guardedWrites), 0);
1283	assert.equal(yield* Ref.get(refreshes), 0);
1284	assert.isEmpty(events);
1285	return;
1286	}
1287	assert.isTrue(Exit.isSuccess(result));
1288	if (scenario === "start-guard") {
1289	assert.equal(yield* Ref.get(guardedWrites), 0);
1290	assert.equal(yield* Ref.get(refreshes), 0);
1291	assert.isEmpty(events);
1292	return;
1293	}
1294	assert.equal(yield* Ref.get(guardedWrites), 1);
1295	if (scenario === "stale-attempt") {
1296	assert.equal(yield* Ref.get(refreshes), 0);
1297	assert.isEmpty(events);
1298	return;
1299	}
1300	assert.equal(yield* Ref.get(refreshes), 1);
1301	assert.deepEqual(
1302	events
1303	.filter(
1304	(event) =>
1305	event.type === "run.updated" ||
1306	event.type === "run-attempt.updated" ||
1307	event.type === "node.updated",
1308	)
1309	.map((event) => event.payload.status),
1310	["failed", "failed", "failed"],
1311	);
1312	const errorItem = events.find(
1313	(event) => event.type === "turn-item.updated" && event.payload.type === "error",
1314	);
1315	assert.isDefined(errorItem);
1316	if (errorItem?.type === "turn-item.updated" && errorItem.payload.type === "error") {
1317	// The persisted item carries a bounded curated message; the exact
1318	// underlying text stays in the logged cause.
1319	assert.equal(errorItem.payload.failure.message, "Run preparation failed.");
1320	}
1321	}),
1322	);
1323
1324	it.effect("ingests new member threads of an owned child after the root turn terminalizes", () =>
1325	Effect.gen(function* () {
1326	const threadId = ThreadId.make("thread:run-execution-late-child");
1327	const childThreadId = ThreadId.make("thread:run-execution-late-child:child");
1328	const memberThreadId = ThreadId.make("thread:run-execution-late-child:member");
1329	const runId = RunId.make("run:run-execution-late-child");
1330	const attemptId = RunAttemptId.make("attempt:run-execution-late-child");
1331	const providerInstanceId = ProviderInstanceId.make("codex");
1332	const providerSessionId = ProviderSessionId.make("session:run-execution-late-child");
1333	const providerThreadId = ProviderThreadId.make("provider-thread:run-execution-late-child");
1334	const childProviderThreadId = ProviderThreadId.make(
1335	"provider-thread:run-execution-late-child:child",
1336	);
1337	const rootProviderTurnId = ProviderTurnId.make("provider-turn:run-execution-late-child");
1338	const childProviderTurnId = ProviderTurnId.make("provider-turn:run-execution-late-child:child");
1339	const rootNodeId = NodeId.make("node:run-execution-late-child");
1340	const childNodeId = NodeId.make("node:run-execution-late-child:child");
1341	const subagentNodeId = NodeId.make("node:run-execution-late-child:subagent");
1342	const childMessageIngested = yield* Deferred.make<void>();
1343	const order = yield* Ref.make<ReadonlyArray<string>>([]);
1344	const layerTest = RunExecutionService.layer.pipe(
1345	Layer.provide(
1346	Layer.mergeAll(
1347	McpAppModelContext.layerEmpty,
1348	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
1349	Layer.mock(EventSink.EventSinkV2)({
1350	write: () => Effect.succeed([]),
1351	writeWithEffects: (input) =>
1352	Effect.gen(function* () {
1353	if (
1354	input.events.some(
1355	(event) => event.type === "run.updated" && event.runId === runId,
1356	)
1357	) {
1358	yield* Ref.update(order, (current) => [...current, "root-finalized"]);
1359	}
1360	return [];
1361	}),
1362	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
1363	}),
1364	IdAllocator.layer,
1365	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
1366	ingestNormalized: (input) =>
1367	Effect.gen(function* () {
1368	if (
1369	input.event.type === "app_thread.created" &&
1370	input.event.appThread.id === memberThreadId
1371	) {
1372	yield* Ref.update(order, (current) => [...current, "member-thread"]);
1373	}
1374	if (
1375	input.event.type === "message.updated" &&
1376	input.event.message.threadId === memberThreadId
1377	) {
1378	yield* Ref.update(order, (current) => [...current, "child-message"]);
1379	yield* Deferred.succeed(childMessageIngested, undefined).pipe(Effect.ignore);
1380	}
1381	return [];
1382	}),
1383	}),
1384	ServerSettings.layerTest(),
1385	),
1386	),
1387	);
1388	const events: ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event> = [
1389	{
1390	type: "app_thread.created",
1391	driver,
1392	appThread: {
1393	id: childThreadId,
1394	lineage: {
1395	parentThreadId: threadId,
1396	relationshipToParent: "subagent",
1397	rootThreadId: threadId,
1398	},
1399	},
1400	} as ProviderAdapter.ProviderAdapterV2Event,
1401	{
1402	type: "provider_thread.updated",
1403	driver,
1404	providerThread: {
1405	id: childProviderThreadId,
1406	appThreadId: childThreadId,
1407	},
1408	} as ProviderAdapter.ProviderAdapterV2Event,
1409	{
1410	type: "subagent.updated",
1411	driver,
1412	subagent: {
1413	...makeRunOwnedSubagentFixture({
1414	ids: backgroundScenarioIds("run-execution-late-child"),
1415	providerInstanceId,
1416	childThreadId,
1417	driver,
1418	status: "running",
1419	}),
1420	workflow: {
1421	phases: [],
1422	agents: [
1423	{ index: 0, label: "Reviewer", state: "running", childThreadId: memberThreadId },
1424	],
1425	},
1426	},
1427	},
1428	{
1429	type: "provider_turn.updated",
1430	driver,
1431	threadId: childThreadId,
1432	providerTurn: {
1433	id: childProviderTurnId,
1434	providerThreadId: childProviderThreadId,
1435	nodeId: childNodeId,
1436	runAttemptId: null,
1437	status: "running",
1438	},
1439	} as ProviderAdapter.ProviderAdapterV2Event,
1440	{
1441	type: "turn.terminal",
1442	driver,
1443	providerThreadId,
1444	providerTurnId: rootProviderTurnId,
1445	runOrdinal: 1,
1446	status: "completed",
1447	failure: null,
1448	threadDisposition: "reusable",
1449	},
1450	{
1451	type: "app_thread.created",
1452	driver,
1453	appThread: {
1454	id: memberThreadId,
1455	lineage: {
1456	parentThreadId: childThreadId,
1457	relationshipToParent: "subagent",
1458	rootThreadId: threadId,
1459	},
1460	},
1461	} as ProviderAdapter.ProviderAdapterV2Event,
1462	{
1463	type: "message.updated",
1464	driver,
1465	message: {
1466	id: MessageId.make("message:run-execution-late-child:child"),
1467	threadId: memberThreadId,
1468	runId: null,
1469	nodeId: childNodeId,
1470	role: "assistant",
1471	text: "Hello.",
1472	streaming: false,
1473	},
1474	} as ProviderAdapter.ProviderAdapterV2Event,
1475	{
1476	type: "provider_turn.updated",
1477	driver,
1478	threadId: childThreadId,
1479	providerTurn: {
1480	id: childProviderTurnId,
1481	providerThreadId: childProviderThreadId,
1482	nodeId: childNodeId,
1483	runAttemptId: null,
1484	status: "completed",
1485	},
1486	} as ProviderAdapter.ProviderAdapterV2Event,
1487	{
1488	type: "subagent.updated",
1489	driver,
1490	subagent: {
1491	id: subagentNodeId,
1492	threadId,
1493	runId,
1494	status: "completed",
1495	},
1496	} as ProviderAdapter.ProviderAdapterV2Event,
1497	];
1498
1499	yield* Effect.gen(function* () {
1500	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
1501	yield* runExecution.startRootRun({
1502	commandId: CommandId.make("command:run-execution-late-child"),
1503	appThread: { id: threadId } as OrchestrationV2AppThread,
1504	providerSessionId,
1505	session: {
1506	events: Stream.fromIterable(events),
1507	startTurn: () => Effect.void,
1508	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
1509	run: {
1510	id: runId,
1511	threadId,
1512	ordinal: 1,
1513	providerInstanceId,
1514	} as OrchestrationV2Run,
1515	rootNode: { id: rootNodeId } as OrchestrationV2ExecutionNode,
1516	checkpointScope: {
1517	id: CheckpointScopeId.make("checkpoint-scope:run-execution-late-child"),
1518	} as OrchestrationV2CheckpointScope,
1519	providerThread: {
1520	id: providerThreadId,
1521	driver,
1522	} as OrchestrationV2ProviderThread,
1523	attempt: {
1524	id: attemptId,
1525	providerTurnId: rootProviderTurnId,
1526	} as OrchestrationV2RunAttempt,
1527	attemptId,
1528	providerTurnOrdinal: 1,
1529	message: {
1530	messageId: MessageId.make("message:run-execution-late-child:user"),
1531	text: "Spawn a child and finish.",
1532	attachments: [],
1533	createdBy: "user",
1534	creationSource: "web",
1535	},
1536	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
1537	runtimePolicy: {
1538	runtimeMode: "full-access",
1539	interactionMode: "default",
1540	cwd: process.cwd(),
1541	approvalPolicy: "never",
1542	sandboxPolicy: {
1543	type: "readOnly",
1544	access: { type: "fullAccess" },
1545	networkAccess: false,
1546	},
1547	},
1548	});
1549	}).pipe(Effect.provide(layerTest));
1550
1551	const observed = yield* Deferred.await(childMessageIngested).pipe(
1552	Effect.timeoutOption("2 seconds"),
1553	);
1554	assert.isTrue(Option.isSome(observed), "child message was not ingested after root terminal");
1555	assert.deepEqual(yield* Ref.get(order), ["root-finalized", "member-thread", "child-message"]);
1556	}),
1557	);
1558
1559	it.effect("ends a nested native subagent and its thread when the root run is interrupted", () =>
1560	Effect.gen(function* () {
1561	const key = "run-execution-nested-native";
1562	const threadId = ThreadId.make(`thread:${key}`);
1563	const childThreadId = ThreadId.make(`thread:${key}:child`);
1564	const grandchildThreadId = ThreadId.make(`thread:${key}:grandchild`);
1565	const runId = RunId.make(`run:${key}`);
1566	const attemptId = RunAttemptId.make(`attempt:${key}`);
1567	const providerInstanceId = ProviderInstanceId.make("codex");
1568	const providerThreadId = ProviderThreadId.make(`provider-thread:${key}`);
1569	const rootProviderTurnId = ProviderTurnId.make(`provider-turn:${key}`);
1570	const nestedId = NodeId.make(`node:${key}:nested`);
1571	const grandchildRootId = NodeId.make(`node:${key}:grandchild-root`);
1572	const finalized = yield* Deferred.make<ReadonlyArray<OrchestrationV2DomainEvent>>();
1573	const layerTest = RunExecutionService.layer.pipe(
1574	Layer.provide(
1575	Layer.mergeAll(
1576	McpAppModelContext.layerEmpty,
1577	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
1578	Layer.mock(EventSink.EventSinkV2)({
1579	write: () => Effect.succeed([]),
1580	writeWithEffects: (input) =>
1581	Effect.as(
1582	input.events.some((event) => event.type === "run.updated")
1583	? Deferred.succeed(finalized, input.events)
1584	: Effect.void,
1585	[],
1586	),
1587	}),
1588	IdAllocator.layer,
1589	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
1590	ingestNormalized: () => Effect.succeed([]),
1591	}),
1592	ServerSettings.layerTest(),
1593	),
1594	),
1595	);
1596	const subagentThread = (id: ThreadId, parentThreadId: ThreadId) =>
1597	({
1598	type: "app_thread.created",
1599	driver,
1600	appThread: {
1601	id,
1602	lineage: { parentThreadId, relationshipToParent: "subagent", rootThreadId: threadId },
1603	},
1604	}) as ProviderAdapter.ProviderAdapterV2Event;
1605	const events: ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event> = [
1606	subagentThread(childThreadId, threadId),
1607	subagentThread(grandchildThreadId, childThreadId),
1608	{
1609	type: "subagent.updated",
1610	driver,
1611	subagent: {
1612	id: nestedId,
1613	threadId: childThreadId,
1614	runId: null,
1615	origin: "provider_native",
1616	childThreadId: grandchildThreadId,
1617	status: "running",
1618	},
1619	} as ProviderAdapter.ProviderAdapterV2Event,
1620	{
1621	type: "node.updated",
1622	driver,
1623	node: {
1624	id: grandchildRootId,
1625	threadId: grandchildThreadId,
1626	runId: null,
1627	kind: "root_turn",
1628	status: "running",
1629	},
1630	} as ProviderAdapter.ProviderAdapterV2Event,
1631	{
1632	type: "turn.terminal",
1633	driver,
1634	providerThreadId,
1635	providerTurnId: rootProviderTurnId,
1636	runOrdinal: 1,
1637	status: "interrupted",
1638	failure: null,
1639	threadDisposition: "reusable",
1640	},
1641	];
1642
1643	yield* Effect.gen(function* () {
1644	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
1645	yield* runExecution.startRootRun({
1646	commandId: CommandId.make(`command:${key}`),
1647	appThread: { id: threadId } as OrchestrationV2AppThread,
1648	providerSessionId: ProviderSessionId.make(`session:${key}`),
1649	session: {
1650	events: Stream.fromIterable(events),
1651	startTurn: () => Effect.void,
1652	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
1653	run: { id: runId, threadId, ordinal: 1, providerInstanceId } as OrchestrationV2Run,
1654	rootNode: { id: NodeId.make(`node:${key}`) } as OrchestrationV2ExecutionNode,
1655	checkpointScope: {
1656	id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
1657	} as OrchestrationV2CheckpointScope,
1658	providerThread: { id: providerThreadId, driver } as OrchestrationV2ProviderThread,
1659	attempt: { id: attemptId, providerTurnId: rootProviderTurnId } as OrchestrationV2RunAttempt,
1660	attemptId,
1661	providerTurnOrdinal: 1,
1662	message: {
1663	messageId: MessageId.make(`message:${key}:user`),
1664	text: "Spawn a nested subagent.",
1665	attachments: [],
1666	createdBy: "user",
1667	creationSource: "web",
1668	},
1669	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
1670	runtimePolicy: {
1671	runtimeMode: "full-access",
1672	interactionMode: "default",
1673	cwd: process.cwd(),
1674	approvalPolicy: "never",
1675	sandboxPolicy: { type: "readOnly", access: { type: "fullAccess" }, networkAccess: false },
1676	},
1677	});
1678	}).pipe(Effect.provide(layerTest));
1679
1680	const ended = (yield* Deferred.await(finalized)).flatMap((event) =>
1681	event.type === "subagent.updated" || event.type === "node.updated"
1682	? [[event.type, event.payload.id, event.payload.status]]
1683	: [],
1684	);
1685	assert.deepInclude(ended, ["subagent.updated", nestedId, "interrupted"]);
1686	assert.deepInclude(ended, ["node.updated", grandchildRootId, "interrupted"]);
1687	}),
1688	);
1689
1690	it.effect("keeps ingesting a late background command item completion after root terminal", () =>
1691	Effect.gen(function* () {
1692	const observed = yield* runBackgroundItemScenario("bg-command", (ids) => [
1693	backgroundTurnItemEvent(ids, "command_execution", "running", 1),
1694	rootTerminalEvent(ids, "completed"),
1695	backgroundTurnItemEvent(ids, "command_execution", "completed", 2),
1696	]);
1697	assert.deepEqual(observed, ["turn_item:running", "root-finalized", "turn_item:completed"]);
1698	}),
1699	);
1700
1701	it.effect("ingests the trailing subagent item completion after the subagent row completes", () =>
1702	Effect.gen(function* () {
1703	const observed = yield* runBackgroundItemScenario("subagent-trailing-item", (ids) => [
1704	subagentEvent(ids, "running"),
1705	backgroundTurnItemEvent(ids, "subagent", "running", 1),
1706	rootTerminalEvent(ids, "completed"),
1707	subagentEvent(ids, "completed"),
1708	backgroundTurnItemEvent(ids, "subagent", "completed", 2),
1709	]);
1710	assert.deepEqual(observed, [
1711	"subagent:running",
1712	"turn_item:running",
1713	"root-finalized",
1714	"subagent:completed",
1715	"turn_item:completed",
1716	]);
1717	}),
1718	);
1719
1720	it.effect("keeps ingesting a child thread's late background item completion", () =>
1721	Effect.gen(function* () {
1722	const observed = yield* runBackgroundItemScenario("bg-child-item", (ids) => [
1723	childThreadCreatedEvent(ids),
1724	subagentEvent(ids, "running"),
1725	childBackgroundTurnItemEvent(ids, "running", 1),
1726	rootTerminalEvent(ids, "completed"),
1727	subagentEvent(ids, "completed"),
1728	childBackgroundTurnItemEvent(ids, "completed", 2),
1729	]);
1730	assert.deepEqual(observed, [
1731	"subagent:running",
1732	"turn_item:running",
1733	"root-finalized",
1734	"subagent:completed",
1735	"turn_item:completed",
1736	]);
1737	}),
1738	);
1739
1740	it.effect("keeps ingesting until the last of several background items terminalizes", () =>
1741	Effect.gen(function* () {
1742	const secondItemId = TurnItemId.make("turn-item:bg-multi:second");
1743	const observed = yield* runBackgroundItemScenario("bg-multi", (ids) => [
1744	backgroundTurnItemEvent(ids, "command_execution", "running", 1),
1745	backgroundTurnItemEvent(ids, "dynamic_tool", "running", 2, secondItemId),
1746	rootTerminalEvent(ids, "completed"),
1747	backgroundTurnItemEvent(ids, "command_execution", "completed", 3),
1748	backgroundTurnItemEvent(ids, "dynamic_tool", "completed", 4, secondItemId),
1749	]);
1750	assert.deepEqual(observed, [
1751	"turn_item:running",
1752	"turn_item:running",
1753	"root-finalized",
1754	"turn_item:completed",
1755	"turn_item:completed",
1756	]);
1757	}),
1758	);
1759
1760	it.effect("does not pin ingestion on background items when the root turn is interrupted", () =>
1761	Effect.gen(function* () {
1762	const observed = yield* runBackgroundItemScenario("bg-interrupted", (ids) => [
1763	backgroundTurnItemEvent(ids, "command_execution", "running", 1),
1764	rootTerminalEvent(ids, "interrupted"),
1765	backgroundTurnItemEvent(ids, "command_execution", "completed", 2),
1766	]);
1767	assert.deepEqual(observed, ["turn_item:running", "root-finalized"]);
1768	}),
1769	);
1770
1771	it.effect("seeds inherited background items before their next update", () =>
1772	Effect.gen(function* () {
1773	const key = "inherited-background-seeded";
1774	const priorRunId = RunId.make(`run:${key}:prior`);
1775	const observed = yield* runBackgroundItemScenario(
1776	key,
1777	(ids) => [
1778	rootTerminalEvent(ids, "completed"),
1779	backgroundTurnItemEventForRun(ids, priorRunId, "subagent", "completed", 1),
1780	],
1781	{
1782	loadInheritedBackgroundTurnItems: () =>
1783	Effect.succeed([{ id: TurnItemId.make(`turn-item:${key}`), runId: priorRunId }]),
1784	},
1785	);
1786
1787	assert.deepEqual(observed, ["root-finalized", "turn_item:completed"]);
1788	}),
1789	);
1790
1791	it.effect("releases the live run after an inherited background item terminalizes", () =>
1792	Effect.gen(function* () {
1793	const key = "inherited-background-terminal";
1794	const priorRunId = RunId.make(`run:${key}:prior`);
1795	const observed = yield* runBackgroundItemScenario(
1796	key,
1797	(ids) => [
1798	backgroundTurnItemEventForRun(ids, priorRunId, "subagent", "running", 1),
1799	rootTerminalEvent(ids, "completed"),
1800	backgroundTurnItemEventForRun(ids, priorRunId, "subagent", "completed", 2),
1801	],
1802	{
1803	loadInheritedBackgroundTurnItems: () =>
1804	Effect.succeed([{ id: TurnItemId.make(`turn-item:${key}`), runId: priorRunId }]),
1805	},
1806	);
1807
1808	assert.deepEqual(observed, ["turn_item:running", "root-finalized", "turn_item:completed"]);
1809	}),
1810	);
1811
1812	it.effect("does not hold the live stream open for a foreign provider's background item", () =>
1813	Effect.gen(function* () {
1814	const key = "inherited-background-foreign-provider";
1815	const ids = backgroundScenarioIds(key);
1816	const priorRunId = RunId.make(`run:${key}:prior`);
1817	const foreignProviderThreadId = ProviderThreadId.make(`provider-thread:${key}:foreign-session`);
1818	const foreignItem = {
1819	id: ids.itemId,
1820	threadId: ids.threadId,
1821	runId: priorRunId,
1822	providerThreadId: foreignProviderThreadId,
1823	type: "subagent",
1824	status: "running",
1825	} as OrchestrationV2TurnItem;
1826	const inherited = RunExecutionService.selectInheritedBackgroundTurnItems({
1827	threadId: ids.threadId,
1828	currentProviderThreadId: ids.providerThreadId,
1829	currentRunOrdinal: 2,
1830	runs: [
1831	{
1832	id: priorRunId,
1833	threadId: ids.threadId,
1834	ordinal: 1,
1835	status: "interrupted",
1836	} as OrchestrationV2Run,
1837	],
1838	turnItems: [foreignItem],
1839	});
1840
1841	const observed = yield* runBackgroundItemScenario(
1842	key,
1843	(scenarioIds) => [rootTerminalEvent(scenarioIds, "completed")],
1844	{
1845	keepEventStreamOpen: true,
1846	loadInheritedBackgroundTurnItems: () => Effect.succeed(inherited),
1847	},
1848	);
1849
1850	assert.deepEqual(inherited, []);
1851	assert.deepEqual(observed, ["root-finalized"]);
1852	}),
1853	);
1854
1855	it.effect("refreshes inherited background items after event subscription", () =>
1856	Effect.gen(function* () {
1857	const key = "inherited-background-subscription-refresh";
1858	const ids = backgroundScenarioIds(key);
1859	const priorRunId = RunId.make(`run:${key}:prior`);
1860	const itemStatus = yield* Ref.make<OrchestrationV2TurnItem["status"]>("running");
1861	const loadInheritedBackgroundTurnItems = () =>
1862	Ref.get(itemStatus).pipe(
1863	Effect.map((status) =>
1864	RunExecutionService.selectInheritedBackgroundTurnItems({
1865	threadId: ids.threadId,
1866	currentProviderThreadId: ids.providerThreadId,
1867	currentRunOrdinal: 2,
1868	runs: [
1869	{
1870	id: priorRunId,
1871	threadId: ids.threadId,
1872	ordinal: 1,
1873	status: "interrupted",
1874	} as OrchestrationV2Run,
1875	],
1876	turnItems: [
1877	{
1878	id: ids.itemId,
1879	threadId: ids.threadId,
1880	runId: priorRunId,
1881	providerThreadId: ids.providerThreadId,
1882	type: "subagent",
1883	status,
1884	} as OrchestrationV2TurnItem,
1885	],
1886	}),
1887	),
1888	);
1889
1890	const observed = yield* runBackgroundItemScenario(
1891	key,
1892	(scenarioIds) => [rootTerminalEvent(scenarioIds, "completed")],
1893	{
1894	keepEventStreamOpen: true,
1895	loadInheritedBackgroundTurnItems,
1896	onSubscribe: Ref.set(itemStatus, "completed"),
1897	},
1898	);
1899
1900	assert.deepEqual(yield* loadInheritedBackgroundTurnItems(), []);
1901	assert.deepEqual(observed, ["root-finalized"]);
1902	}),
1903	);
1904
1905	it.effect(
1906	"keeps ingesting a late empty provider-thread roster while thread-scoped pending work is true",
1907	() =>
1908	Effect.gen(function* () {
1909	const key = "bg-roster-pending-work";
1910	const ids = backgroundScenarioIds(key);
1911	const providerInstanceId = ProviderInstanceId.make("codex");
1912	const now = yield* DateTime.now;
1913	const observed = yield* Ref.make<ReadonlyArray<string>>([]);
1914	const pendingByProviderThreadId = yield* Ref.make(new Map([[ids.providerThreadId, true]]));
1915	const scopedProbeArgs = yield* Ref.make<ReadonlyArray<ProviderThreadId>>([]);
1916	const ingestCalls = yield* Ref.make<
1917	ReadonlyArray<{
1918	readonly activeAttemptId: RunAttemptId | null;
1919	readonly eventType: string;
1920	readonly hasWriteIfRunCurrent: boolean;
1921	readonly hasWriteIfProviderThreadOwner: boolean;
1922	readonly expectedLastRunOrdinal: number | null;
1923	readonly runId: RunId | null;
1924	readonly rosterLength: number | null;
1925	}>
1926	>([]);
1927	const ingestionDone = yield* Deferred.make<void>();
1928	const layerTest = RunExecutionService.layer.pipe(
1929	Layer.provide(
1930	Layer.mergeAll(
1931	McpAppModelContext.layerEmpty,
1932	Layer.mock(CheckpointService.CheckpointServiceV2)({
1933	captureBaseline: () => Effect.void,
1934	}),
1935	Layer.mock(EventSink.EventSinkV2)({
1936	write: () => Effect.succeed([]),
1937	writeWithEffects: (input) =>
1938	Effect.gen(function* () {
1939	if (
1940	input.events.some(
1941	(event) => event.type === "run.updated" && event.runId === ids.runId,
1942	)
1943	) {
1944	yield* Ref.update(observed, (current) => [...current, "root-finalized"]);
1945	}
1946	return [];
1947	}),
1948	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
1949	writeIfProviderThreadOwner: () =>
1950	Effect.succeed({ committed: true, storedEvents: [] }),
1951	}),
1952	IdAllocator.layer,
1953	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
1954	ingestNormalized: (input) =>
1955	Effect.gen(function* () {
1956	const event = input.event;
1957	const rosterLength =
1958	event.type === "provider_thread.updated"
1959	? (event.providerThread.pendingBackgroundTasks?.length ?? 0)
1960	: null;
1961	yield* Ref.update(ingestCalls, (current) => [
1962	...current,
1963	{
1964	activeAttemptId: input.writeIfProviderThreadOwner?.activeAttemptId ?? null,
1965	eventType: event.type,
1966	hasWriteIfRunCurrent: input.writeIfRunCurrent !== undefined,
1967	hasWriteIfProviderThreadOwner: input.writeIfProviderThreadOwner !== undefined,
1968	expectedLastRunOrdinal:
1969	input.writeIfProviderThreadOwner?.expectedLastRunOrdinal ?? null,
1970	runId: input.writeIfProviderThreadOwner?.runId ?? null,
1971	rosterLength,
1972	},
1973	]);
1974	if (event.type === "provider_thread.updated" && rosterLength === 0) {
1975	yield* Ref.update(pendingByProviderThreadId, (current) => {
1976	const next = new Map(current);
1977	next.set(event.providerThread.id, false);
1978	return next;
1979	});
1980	yield* Ref.update(observed, (current) => [...current, "roster-cleared"]);
1981	}
1982	if (event.type === "turn.terminal") {
1983	yield* Ref.update(observed, (current) => [...current, "terminal"]);
1984	}
1985	return [];
1986	}),
1987	}),
1988	ServerSettings.layerTest(),
1989	),
1990	),
1991	);
1992
1993	const providerThreadBase = {
1994	id: ids.providerThreadId,
1995	driver,
1996	providerInstanceId,
1997	providerSessionId: ProviderSessionId.make(`session:${key}`),
1998	appThreadId: ids.threadId,
1999	ownerNodeId: null,
2000	nativeThreadRef: null,
2001	nativeConversationHeadRef: null,
2002	status: "idle" as const,
2003	firstRunOrdinal: 1,
2004	lastRunOrdinal: 1,
2005	handoffIds: [],
2006	forkedFrom: null,
2007	createdAt: now,
2008	updatedAt: now,
2009	};
2010
2011	yield* Effect.gen(function* () {
2012	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
2013	yield* runExecution.startRootRun({
2014	commandId: CommandId.make(`command:${key}`),
2015	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
2016	providerSessionId: ProviderSessionId.make(`session:${key}`),
2017	session: {
2018	events: Stream.empty,
2019	// Session-wide stays true forever; the root must consult the
2020	// thread-scoped probe instead of being pinned by siblings.
2021	hasPendingBackgroundWork: Effect.succeed(true),
2022	hasPendingBackgroundWorkForThread: (providerThread: OrchestrationV2ProviderThread) =>
2023	Effect.gen(function* () {
2024	yield* Ref.update(scopedProbeArgs, (current) => [...current, providerThread.id]);
2025	return (yield* Ref.get(pendingByProviderThreadId)).get(providerThread.id) === true;
2026	}),
2027	subscribeEvents: Effect.succeed({
2028	events: Stream.fromIterable([
2029	{
2030	type: "provider_thread.updated",
2031	driver,
2032	providerThread: {
2033	...providerThreadBase,
2034	status: "active" as const,
2035	pendingBackgroundTasks: [
2036	{ taskId: "bg-1", description: "sleep 20", kind: "command" },
2037	],
2038	updatedAt: now,
2039	},
2040	} as ProviderAdapter.ProviderAdapterV2Event,
2041	rootTerminalEvent(ids, "completed"),
2042	{
2043	type: "provider_thread.updated",
2044	driver,
2045	providerThread: {
2046	...providerThreadBase,
2047	status: "idle" as const,
2048	pendingBackgroundTasks: [],
2049	updatedAt: now,
2050	},
2051	} as ProviderAdapter.ProviderAdapterV2Event,
2052	]),
2053	close: Deferred.succeed(ingestionDone, undefined),
2054	}),
2055	startTurn: () => Effect.void,
2056	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
2057	run: {
2058	id: ids.runId,
2059	threadId: ids.threadId,
2060	ordinal: 1,
2061	providerInstanceId,
2062	} as OrchestrationV2Run,
2063	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
2064	checkpointScope: {
2065	id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
2066	} as OrchestrationV2CheckpointScope,
2067	providerThread: providerThreadBase as OrchestrationV2ProviderThread,
2068	attempt: {
2069	id: ids.attemptId,
2070	providerTurnId: ids.rootProviderTurnId,
2071	} as OrchestrationV2RunAttempt,
2072	attemptId: ids.attemptId,
2073	providerTurnOrdinal: 1,
2074	message: {
2075	messageId: MessageId.make(`message:${key}:user`),
2076	text: "Start background work and settle.",
2077	attachments: [],
2078	createdBy: "user",
2079	creationSource: "web",
2080	},
2081	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
2082	runtimePolicy: {
2083	runtimeMode: "full-access",
2084	interactionMode: "default",
2085	cwd: process.cwd(),
2086	approvalPolicy: "never",
2087	sandboxPolicy: {
2088	type: "readOnly",
2089	access: { type: "fullAccess" },
2090	networkAccess: false,
2091	},
2092	},
2093	});
2094	}).pipe(Effect.provide(layerTest));
2095
2096	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
2097	assert.isTrue(Option.isSome(closed), "event subscription did not release");
2098	assert.deepEqual(yield* Ref.get(observed), ["terminal", "root-finalized", "roster-cleared"]);
2099	assert.isTrue((yield* Ref.get(scopedProbeArgs)).includes(ids.providerThreadId));
2100
2101	const calls = yield* Ref.get(ingestCalls);
2102	const preTerminalRoster = calls.find(
2103	(call) => call.eventType === "provider_thread.updated" && call.rosterLength === 1,
2104	);
2105	const lateClear = calls.find(
2106	(call) => call.eventType === "provider_thread.updated" && call.rosterLength === 0,
2107	);
2108	assert.isDefined(preTerminalRoster);
2109	assert.isTrue(preTerminalRoster?.hasWriteIfRunCurrent);
2110	assert.isFalse(preTerminalRoster?.hasWriteIfProviderThreadOwner);
2111	assert.isDefined(lateClear);
2112	assert.isFalse(
2113	lateClear?.hasWriteIfRunCurrent,
2114	"late empty roster must not use stale writeIfRunCurrent running-gate",
2115	);
2116	assert.isTrue(
2117	lateClear?.hasWriteIfProviderThreadOwner,
2118	"late empty roster must gate on provider-thread ownership",
2119	);
2120	assert.equal(lateClear?.expectedLastRunOrdinal, 1);
2121	assert.equal(lateClear?.runId, ids.runId);
2122	assert.equal(lateClear?.activeAttemptId, ids.attemptId);
2123	}),
2124	);
2125
2126	it.effect("drops late root provider-thread writes from a superseded attempt", () =>
2127	Effect.gen(function* () {
2128	const key = "bg-roster-attempt-owner-lost";
2129	const ids = backgroundScenarioIds(key);
2130	const replacementAttemptId = RunAttemptId.make(`attempt:${key}:replacement`);
2131	const providerInstanceId = ProviderInstanceId.make("codex");
2132	const now = yield* DateTime.now;
2133	const observed = yield* Ref.make<ReadonlyArray<string>>([]);
2134	// Probe stays true forever so only ownership-loss can release the stream.
2135	const ingestionDone = yield* Deferred.make<void>();
2136	const ingestCalls = yield* Ref.make<
2137	ReadonlyArray<{
2138	readonly activeAttemptId: RunAttemptId | null;
2139	readonly eventType: string;
2140	readonly hasWriteIfProviderThreadOwner: boolean;
2141	readonly expectedLastRunOrdinal: number | null;
2142	readonly rosterLength: number | null;
2143	readonly committed: boolean;
2144	}>
2145	>([]);
2146	const layerTest = RunExecutionService.layer.pipe(
2147	Layer.provide(
2148	Layer.mergeAll(
2149	McpAppModelContext.layerEmpty,
2150	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
2151	Layer.mock(EventSink.EventSinkV2)({
2152	write: () => Effect.succeed([]),
2153	writeWithEffects: (input) =>
2154	Effect.gen(function* () {
2155	if (
2156	input.events.some(
2157	(event) => event.type === "run.updated" && event.runId === ids.runId,
2158	)
2159	) {
2160	yield* Ref.update(observed, (current) => [...current, "root-finalized"]);
2161	}
2162	return [];
2163	}),
2164	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
2165	writeIfProviderThreadOwner: () =>
2166	Effect.succeed({ committed: false, storedEvents: [] }),
2167	}),
2168	IdAllocator.layer,
2169	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
2170	ingestNormalized: (input) =>
2171	Effect.gen(function* () {
2172	const event = input.event;
2173	const rosterLength =
2174	event.type === "provider_thread.updated"
2175	? (event.providerThread.pendingBackgroundTasks?.length ?? 0)
2176	: null;
2177	const ownerGate = input.writeIfProviderThreadOwner;
2178	// Simulate the EventSink reject after a same-run replacement
2179	// changes activeAttemptId without advancing lastRunOrdinal.
2180	const rejectAsStaleOwner =
2181	ownerGate !== undefined && ownerGate.activeAttemptId !== replacementAttemptId;
2182	yield* Ref.update(ingestCalls, (current) => [
2183	...current,
2184	{
2185	activeAttemptId: ownerGate?.activeAttemptId ?? null,
2186	eventType: event.type,
2187	hasWriteIfProviderThreadOwner: ownerGate !== undefined,
2188	expectedLastRunOrdinal: ownerGate?.expectedLastRunOrdinal ?? null,
2189	rosterLength,
2190	committed: !rejectAsStaleOwner,
2191	},
2192	]);
2193	if (event.type === "turn.terminal") {
2194	yield* Ref.update(observed, (current) => [...current, "terminal"]);
2195	}
2196	if (event.type === "provider_thread.updated" && rejectAsStaleOwner) {
2197	yield* Ref.update(observed, (current) => [...current, "stale-owner-rejected"]);
2198	return [];
2199	}
2200	if (event.type === "provider_thread.updated") {
2201	yield* Ref.update(observed, (current) => [...current, "roster-written"]);
2202	}
2203	return [];
2204	}),
2205	}),
2206	ServerSettings.layerTest(),
2207	),
2208	),
2209	);
2210
2211	const providerThreadBase = {
2212	id: ids.providerThreadId,
2213	driver,
2214	providerInstanceId,
2215	providerSessionId: ProviderSessionId.make(`session:${key}`),
2216	appThreadId: ids.threadId,
2217	ownerNodeId: null,
2218	nativeThreadRef: null,
2219	nativeConversationHeadRef: null,
2220	status: "idle" as const,
2221	firstRunOrdinal: 1,
2222	lastRunOrdinal: 1,
2223	handoffIds: [],
2224	forkedFrom: null,
2225	createdAt: now,
2226	updatedAt: now,
2227	};
2228
2229	yield* Effect.gen(function* () {
2230	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
2231	yield* runExecution.startRootRun({
2232	commandId: CommandId.make(`command:${key}`),
2233	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
2234	providerSessionId: ProviderSessionId.make(`session:${key}`),
2235	session: {
2236	events: Stream.empty,
2237	hasPendingBackgroundWork: Effect.succeed(true),
2238	hasPendingBackgroundWorkForThread: () => Effect.succeed(true),
2239	subscribeEvents: Effect.succeed({
2240	events: Stream.fromIterable([
2241	{
2242	type: "provider_thread.updated",
2243	driver,
2244	providerThread: {
2245	...providerThreadBase,
2246	status: "active" as const,
2247	pendingBackgroundTasks: [
2248	{ taskId: "bg-stale", description: "sleep 20", kind: "command" },
2249	],
2250	updatedAt: now,
2251	},
2252	} as ProviderAdapter.ProviderAdapterV2Event,
2253	rootTerminalEvent(ids, "completed"),
2254	// Late snapshot after a replacement attempt claimed the same
2255	// run ordinal. Without attempt gating this would clobber it.
2256	{
2257	type: "provider_thread.updated",
2258	driver,
2259	providerThread: {
2260	...providerThreadBase,
2261	status: "idle" as const,
2262	lastRunOrdinal: 1,
2263	pendingBackgroundTasks: [
2264	{ taskId: "bg-stale", description: "sleep 20", kind: "command" },
2265	],
2266	updatedAt: now,
2267	},
2268	} as ProviderAdapter.ProviderAdapterV2Event,
2269	]),
2270	close: Deferred.succeed(ingestionDone, undefined),
2271	}),
2272	startTurn: () => Effect.void,
2273	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
2274	run: {
2275	id: ids.runId,
2276	threadId: ids.threadId,
2277	ordinal: 1,
2278	providerInstanceId,
2279	} as OrchestrationV2Run,
2280	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
2281	checkpointScope: {
2282	id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
2283	} as OrchestrationV2CheckpointScope,
2284	providerThread: providerThreadBase as OrchestrationV2ProviderThread,
2285	attempt: {
2286	id: ids.attemptId,
2287	providerTurnId: ids.rootProviderTurnId,
2288	} as OrchestrationV2RunAttempt,
2289	attemptId: ids.attemptId,
2290	providerTurnOrdinal: 1,
2291	message: {
2292	messageId: MessageId.make(`message:${key}:user`),
2293	text: "Background work that loses ownership.",
2294	attachments: [],
2295	createdBy: "user",
2296	creationSource: "web",
2297	},
2298	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
2299	runtimePolicy: {
2300	runtimeMode: "full-access",
2301	interactionMode: "default",
2302	cwd: process.cwd(),
2303	approvalPolicy: "never",
2304	sandboxPolicy: {
2305	type: "readOnly",
2306	access: { type: "fullAccess" },
2307	networkAccess: false,
2308	},
2309	},
2310	});
2311	}).pipe(Effect.provide(layerTest));
2312
2313	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
2314	assert.isTrue(
2315	Option.isSome(closed),
2316	"subscription must release after ownership-loss reject even when probe stays true",
2317	);
2318	assert.deepEqual(yield* Ref.get(observed), [
2319	"roster-written",
2320	"terminal",
2321	"root-finalized",
2322	"stale-owner-rejected",
2323	]);
2324
2325	const calls = yield* Ref.get(ingestCalls);
2326	const lateStale = calls.find(
2327	(call) =>
2328	call.eventType === "provider_thread.updated" &&
2329	call.hasWriteIfProviderThreadOwner &&
2330	call.rosterLength === 1,
2331	);
2332	assert.isDefined(lateStale);
2333	assert.equal(lateStale?.expectedLastRunOrdinal, 1);
2334	assert.equal(lateStale?.activeAttemptId, ids.attemptId);
2335	assert.isFalse(lateStale?.committed);
2336	}),
2337	);
2338
2339	it.effect(
2340	"keeps ingesting a late background turn-item completion after ownership-loss rejects roster writes",
2341	() =>
2342	Effect.gen(function* () {
2343	const key = "bg-item-after-owner-lost";
2344	const ids = backgroundScenarioIds(key);
2345	const providerInstanceId = ProviderInstanceId.make("codex");
2346	const now = yield* DateTime.now;
2347	const observed = yield* Ref.make<ReadonlyArray<string>>([]);
2348	// Probe stays true forever; open background items must pin the stream
2349	// past ownership-loss so late turn_item completions still land.
2350	const ingestionDone = yield* Deferred.make<void>();
2351	const layerTest = RunExecutionService.layer.pipe(
2352	Layer.provide(
2353	Layer.mergeAll(
2354	McpAppModelContext.layerEmpty,
2355	Layer.mock(CheckpointService.CheckpointServiceV2)({
2356	captureBaseline: () => Effect.void,
2357	}),
2358	Layer.mock(EventSink.EventSinkV2)({
2359	write: () => Effect.succeed([]),
2360	writeWithEffects: (input) =>
2361	Effect.gen(function* () {
2362	if (
2363	input.events.some(
2364	(event) => event.type === "run.updated" && event.runId === ids.runId,
2365	)
2366	) {
2367	yield* Ref.update(observed, (current) => [...current, "root-finalized"]);
2368	}
2369	return [];
2370	}),
2371	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
2372	writeIfProviderThreadOwner: () =>
2373	Effect.succeed({ committed: false, storedEvents: [] }),
2374	}),
2375	IdAllocator.layer,
2376	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
2377	ingestNormalized: (input) =>
2378	Effect.gen(function* () {
2379	const event = input.event;
2380	if (event.type === "turn.terminal") {
2381	yield* Ref.update(observed, (current) => [...current, "terminal"]);
2382	}
2383	if (
2384	event.type === "provider_thread.updated" &&
2385	input.writeIfProviderThreadOwner !== undefined
2386	) {
2387	yield* Ref.update(observed, (current) => [...current, "stale-owner-rejected"]);
2388	return [];
2389	}
2390	if (event.type === "turn_item.updated") {
2391	yield* Ref.update(observed, (current) => [
2392	...current,
2393	`turn_item:${event.turnItem.status}`,
2394	]);
2395	}
2396	return [];
2397	}),
2398	}),
2399	ServerSettings.layerTest(),
2400	),
2401	),
2402	);
2403
2404	const providerThreadBase = {
2405	id: ids.providerThreadId,
2406	driver,
2407	providerInstanceId,
2408	providerSessionId: ProviderSessionId.make(`session:${key}`),
2409	appThreadId: ids.threadId,
2410	ownerNodeId: null,
2411	nativeThreadRef: null,
2412	nativeConversationHeadRef: null,
2413	status: "idle" as const,
2414	firstRunOrdinal: 1,
2415	lastRunOrdinal: 1,
2416	handoffIds: [],
2417	forkedFrom: null,
2418	createdAt: now,
2419	updatedAt: now,
2420	};
2421
2422	yield* Effect.gen(function* () {
2423	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
2424	yield* runExecution.startRootRun({
2425	commandId: CommandId.make(`command:${key}`),
2426	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
2427	providerSessionId: ProviderSessionId.make(`session:${key}`),
2428	session: {
2429	events: Stream.empty,
2430	hasPendingBackgroundWork: Effect.succeed(true),
2431	hasPendingBackgroundWorkForThread: () => Effect.succeed(true),
2432	subscribeEvents: Effect.succeed({
2433	events: Stream.fromIterable([
2434	backgroundTurnItemEvent(ids, "command_execution", "running", 1),
2435	rootTerminalEvent(ids, "completed"),
2436	// Ownership reject after a newer run claimed lastRunOrdinal.
2437	{
2438	type: "provider_thread.updated",
2439	driver,
2440	providerThread: {
2441	...providerThreadBase,
2442	status: "idle" as const,
2443	lastRunOrdinal: 1,
2444	pendingBackgroundTasks: [
2445	{ taskId: "bg-stale", description: "sleep 20", kind: "command" },
2446	],
2447	updatedAt: now,
2448	},
2449	} as ProviderAdapter.ProviderAdapterV2Event,
2450	// Late completion still writable (turn_item writes are not
2451	// ownership-gated); stream must stay open for it.
2452	backgroundTurnItemEvent(ids, "command_execution", "completed", 2),
2453	]),
2454	close: Deferred.succeed(ingestionDone, undefined),
2455	}),
2456	startTurn: () => Effect.void,
2457	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
2458	run: {
2459	id: ids.runId,
2460	threadId: ids.threadId,
2461	ordinal: 1,
2462	providerInstanceId,
2463	} as OrchestrationV2Run,
2464	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
2465	checkpointScope: {
2466	id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
2467	} as OrchestrationV2CheckpointScope,
2468	providerThread: providerThreadBase as OrchestrationV2ProviderThread,
2469	attempt: {
2470	id: ids.attemptId,
2471	providerTurnId: ids.rootProviderTurnId,
2472	} as OrchestrationV2RunAttempt,
2473	attemptId: ids.attemptId,
2474	providerTurnOrdinal: 1,
2475	message: {
2476	messageId: MessageId.make(`message:${key}:user`),
2477	text: "Background item after ownership loss.",
2478	attachments: [],
2479	createdBy: "user",
2480	creationSource: "web",
2481	},
2482	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
2483	runtimePolicy: {
2484	runtimeMode: "full-access",
2485	interactionMode: "default",
2486	cwd: process.cwd(),
2487	approvalPolicy: "never",
2488	sandboxPolicy: {
2489	type: "readOnly",
2490	access: { type: "fullAccess" },
2491	networkAccess: false,
2492	},
2493	},
2494	});
2495	}).pipe(Effect.provide(layerTest));
2496
2497	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
2498	assert.isTrue(
2499	Option.isSome(closed),
2500	"subscription must release after background item completes past ownership loss",
2501	);
2502	assert.deepEqual(yield* Ref.get(observed), [
2503	"turn_item:running",
2504	"terminal",
2505	"root-finalized",
2506	"stale-owner-rejected",
2507	"turn_item:completed",
2508	]);
2509	}),
2510	);
2511
2512	it.effect(
2513	"does not pin ingestion on a sibling session-wide pending state when this thread has no roster",
2514	() =>
2515	Effect.gen(function* () {
2516	const key = "bg-roster-sibling-not-pin";
2517	const ids = backgroundScenarioIds(key);
2518	const providerInstanceId = ProviderInstanceId.make("codex");
2519	const now = yield* DateTime.now;
2520	const observed = yield* Ref.make<ReadonlyArray<string>>([]);
2521	const ingestionDone = yield* Deferred.make<void>();
2522	const scopedProbeArgs = yield* Ref.make<ReadonlyArray<ProviderThreadId>>([]);
2523	const layerTest = RunExecutionService.layer.pipe(
2524	Layer.provide(
2525	Layer.mergeAll(
2526	McpAppModelContext.layerEmpty,
2527	Layer.mock(CheckpointService.CheckpointServiceV2)({
2528	captureBaseline: () => Effect.void,
2529	}),
2530	Layer.mock(EventSink.EventSinkV2)({
2531	write: () => Effect.succeed([]),
2532	writeWithEffects: (input) =>
2533	Effect.gen(function* () {
2534	if (
2535	input.events.some(
2536	(event) => event.type === "run.updated" && event.runId === ids.runId,
2537	)
2538	) {
2539	yield* Ref.update(observed, (current) => [...current, "root-finalized"]);
2540	}
2541	return [];
2542	}),
2543	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
2544	}),
2545	IdAllocator.layer,
2546	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
2547	ingestNormalized: (input) =>
2548	Effect.gen(function* () {
2549	if (input.event.type === "turn.terminal") {
2550	yield* Ref.update(observed, (current) => [...current, "terminal"]);
2551	}
2552	return [];
2553	}),
2554	}),
2555	ServerSettings.layerTest(),
2556	),
2557	),
2558	);
2559
2560	const providerThreadBase = {
2561	id: ids.providerThreadId,
2562	driver,
2563	providerInstanceId,
2564	providerSessionId: ProviderSessionId.make(`session:${key}`),
2565	appThreadId: ids.threadId,
2566	ownerNodeId: null,
2567	nativeThreadRef: {
2568	driver,
2569	nativeId: "native-self",
2570	strength: "strong" as const,
2571	},
2572	nativeConversationHeadRef: null,
2573	status: "idle" as const,
2574	firstRunOrdinal: 1,
2575	lastRunOrdinal: 1,
2576	handoffIds: [],
2577	forkedFrom: null,
2578	createdAt: now,
2579	updatedAt: now,
2580	};
2581	const siblingProviderThreadId = ProviderThreadId.make(`provider-thread:${key}:sibling`);
2582
2583	yield* Effect.gen(function* () {
2584	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
2585	yield* runExecution.startRootRun({
2586	commandId: CommandId.make(`command:${key}`),
2587	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
2588	providerSessionId: ProviderSessionId.make(`session:${key}`),
2589	session: {
2590	events: Stream.empty,
2591	// Session-wide stays true (sibling has work). Stop must use only
2592	// the scoped probe for this root's provider thread.
2593	hasPendingBackgroundWork: Effect.succeed(true),
2594	hasPendingBackgroundWorkForThread: (providerThread: OrchestrationV2ProviderThread) =>
2595	Effect.gen(function* () {
2596	yield* Ref.update(scopedProbeArgs, (current) => [...current, providerThread.id]);
2597	// Own thread has no roster; sibling would report true if probed.
2598	return providerThread.id !== ids.providerThreadId;
2599	}),
2600	subscribeEvents: Effect.succeed({
2601	events: Stream.fromIterable([
2602	rootTerminalEvent(ids, "completed"),
2603	// Sibling thread update after terminal. Own-thread scoped
2604	// pending is false, so this root must release without waiting
2605	// for sibling-driven session-wide pending work.
2606	{
2607	type: "provider_thread.updated",
2608	driver,
2609	providerThread: {
2610	...providerThreadBase,
2611	id: siblingProviderThreadId,
2612	appThreadId: ThreadId.make(`thread:${key}:sibling`),
2613	nativeThreadRef: {
2614	driver,
2615	nativeId: "native-sibling",
2616	strength: "strong" as const,
2617	},
2618	pendingBackgroundTasks: [
2619	{ taskId: "sibling-bg", description: "other thread", kind: "command" },
2620	],
2621	updatedAt: now,
2622	},
2623	} as ProviderAdapter.ProviderAdapterV2Event,
2624	]),
2625	close: Deferred.succeed(ingestionDone, undefined),
2626	}),
2627	startTurn: () => Effect.void,
2628	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
2629	run: {
2630	id: ids.runId,
2631	threadId: ids.threadId,
2632	ordinal: 1,
2633	providerInstanceId,
2634	} as OrchestrationV2Run,
2635	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
2636	checkpointScope: {
2637	id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
2638	} as OrchestrationV2CheckpointScope,
2639	providerThread: providerThreadBase as OrchestrationV2ProviderThread,
2640	attempt: {
2641	id: ids.attemptId,
2642	providerTurnId: ids.rootProviderTurnId,
2643	} as OrchestrationV2RunAttempt,
2644	attemptId: ids.attemptId,
2645	providerTurnOrdinal: 1,
2646	message: {
2647	messageId: MessageId.make(`message:${key}:user`),
2648	text: "Settle without local pending work.",
2649	attachments: [],
2650	createdBy: "user",
2651	creationSource: "web",
2652	},
2653	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
2654	runtimePolicy: {
2655	runtimeMode: "full-access",
2656	interactionMode: "default",
2657	cwd: process.cwd(),
2658	approvalPolicy: "never",
2659	sandboxPolicy: {
2660	type: "readOnly",
2661	access: { type: "fullAccess" },
2662	networkAccess: false,
2663	},
2664	},
2665	});
2666	}).pipe(Effect.provide(layerTest));
2667
2668	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
2669	assert.isTrue(Option.isSome(closed), "event subscription did not release");
2670	// Critical: release while session-wide hasPendingBackgroundWork stays true
2671	// and the scoped probe reports false for this root's provider thread.
2672	const observedEvents = [...(yield* Ref.get(observed))];
2673	assert.includeMembers(observedEvents, ["terminal", "root-finalized"]);
2674	const probedIds = yield* Ref.get(scopedProbeArgs);
2675	assert.isTrue(probedIds.includes(ids.providerThreadId));
2676	assert.isFalse(probedIds.includes(siblingProviderThreadId));
2677	}),
2678	);
2679
2680	it.effect(
2681	"cascade-terminalizes run-owned subagent rows on interrupt before root finalization",
2682	() =>
2683	Effect.gen(function* () {
2684	const ids = backgroundScenarioIds("subagent-interrupt-cascade");
2685	const childThreadId = ids.childThreadId;
2686	const unrelatedChildThreadId = ThreadId.make(
2687	"thread:subagent-interrupt-cascade:unrelated-child",
2688	);
2689	const providerInstanceId = ProviderInstanceId.make("codex");
2690	const written = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
2691	const ingested = yield* Ref.make<ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>>([]);
2692	const ingestionDone = yield* Deferred.make<void>();
2693	const layerTest = RunExecutionService.layer.pipe(
2694	Layer.provide(
2695	Layer.mergeAll(
2696	McpAppModelContext.layerEmpty,
2697	Layer.mock(CheckpointService.CheckpointServiceV2)({
2698	captureBaseline: () => Effect.void,
2699	}),
2700	Layer.mock(EventSink.EventSinkV2)({
2701	write: () => Effect.succeed([]),
2702	writeWithEffects: (input) =>
2703	Effect.gen(function* () {
2704	yield* Ref.update(written, (current) => [...current, ...input.events]);
2705	return [];
2706	}),
2707	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
2708	}),
2709	IdAllocator.layer,
2710	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
2711	ingestNormalized: (input) =>
2712	Ref.update(ingested, (current) => [...current, input.event]).pipe(Effect.as([])),
2713	}),
2714	ServerSettings.layerTest(),
2715	),
2716	),
2717	);
2718
2719	const runningSubagent = makeRunOwnedSubagentFixture({
2720	ids,
2721	providerInstanceId,
2722	childThreadId,
2723	driver,
2724	status: "running",
2725	});
2726	const runningTurnItem = makeRunOwnedSubagentTurnItemFixture({
2727	ids,
2728	providerInstanceId,
2729	childThreadId,
2730	driver,
2731	status: "running",
2732	});
2733	const runningNode = makeRunOwnedSubagentNodeFixture({
2734	ids,
2735	status: "running",
2736	});
2737	const runningChildNode = makeRunOwnedSubagentChildNodeFixture({
2738	ids,
2739	status: "running",
2740	});
2741	const memberThreadId = ThreadId.make("thread:subagent-interrupt-cascade:member");
2742	const runningMember = {
2743	...runningSubagent,
2744	id: NodeId.make("node:subagent-interrupt-cascade:member"),
2745	threadId: childThreadId,
2746	runId: null,
2747	parentNodeId: runningChildNode.id,
2748	childThreadId: memberThreadId,
2749	};
2750	const foreignSubagent = {
2751	...runningMember,
2752	id: NodeId.make("node:subagent-interrupt-cascade:foreign-subagent"),
2753	runId: RunId.make("run:subagent-interrupt-cascade:foreign"),
2754	childThreadId: unrelatedChildThreadId,
2755	};
2756	const runningMemberRoot = {
2757	...runningChildNode,
2758	id: NodeId.make("node:subagent-interrupt-cascade:member-root"),
2759	threadId: memberThreadId,
2760	rootNodeId: NodeId.make("node:subagent-interrupt-cascade:member-root"),
2761	};
2762	const foreignChildNode = {
2763	...runningChildNode,
2764	id: NodeId.make("node:subagent-interrupt-cascade:foreign-child"),
2765	runId: foreignSubagent.runId,
2766	};
2767	const runningChildTurnItem = makeLinkedChildTurnItemFixture({
2768	ids,
2769	driver,
2770	type: "command_execution",
2771	});
2772	const foreignChildTurnItem = {
2773	...runningChildTurnItem,
2774	id: TurnItemId.make("turn-item:subagent-interrupt-cascade:foreign-child"),
2775	runId: foreignSubagent.runId,
2776	};
2777	const suppressedChildAssistantTurnItem = makeLinkedChildTurnItemFixture({
2778	ids: {
2779	...ids,
2780	childItemId: TurnItemId.make("turn-item:subagent-interrupt-cascade:suppressed-assistant"),
2781	},
2782	driver,
2783	type: "assistant_message",
2784	});
2785	const unrelatedChildNode = {
2786	...runningChildNode,
2787	id: NodeId.make("node:subagent-interrupt-cascade:unrelated-child"),
2788	threadId: unrelatedChildThreadId,
2789	runId: ids.runId,
2790	rootNodeId: NodeId.make("node:subagent-interrupt-cascade:unrelated-child"),
2791	};
2792	const unrelatedChildTurnItem = {
2793	...runningChildTurnItem,
2794	id: TurnItemId.make("turn-item:subagent-interrupt-cascade:unrelated-child"),
2795	threadId: unrelatedChildThreadId,
2796	nodeId: unrelatedChildNode.id,
2797	};
2798
2799	yield* Effect.gen(function* () {
2800	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
2801	yield* runExecution.startRootRun({
2802	commandId: CommandId.make("command:subagent-interrupt-cascade"),
2803	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
2804	providerSessionId: ProviderSessionId.make("session:subagent-interrupt-cascade"),
2805	session: {
2806	events: Stream.empty,
2807	subscribeEvents: Effect.succeed({
2808	events: Stream.fromIterable([
2809	childThreadCreatedEvent(ids),
2810	{
2811	type: "subagent.updated",
2812	driver,
2813	subagent: runningSubagent,
2814	},
2815	{
2816	type: "node.updated",
2817	driver,
2818	node: runningNode,
2819	},
2820	{
2821	type: "node.updated",
2822	driver,
2823	node: runningChildNode,
2824	},
2825	{
2826	type: "node.updated",
2827	driver,
2828	node: unrelatedChildNode,
2829	},
2830	{
2831	type: "turn_item.updated",
2832	driver,
2833	turnItem: runningTurnItem,
2834	},
2835	{
2836	type: "turn_item.updated",
2837	driver,
2838	turnItem: runningChildTurnItem,
2839	},
2840	{
2841	type: "turn_item.updated",
2842	driver,
2843	turnItem: suppressedChildAssistantTurnItem,
2844	},
2845	{
2846	type: "turn_item.updated",
2847	driver,
2848	turnItem: unrelatedChildTurnItem,
2849	},
2850	{ type: "subagent.updated", driver, subagent: runningMember },
2851	{ type: "subagent.updated", driver, subagent: foreignSubagent },
2852	{ type: "node.updated", driver, node: runningMemberRoot },
2853	{ type: "node.updated", driver, node: foreignChildNode },
2854	{ type: "turn_item.updated", driver, turnItem: foreignChildTurnItem },
2855	rootTerminalEvent(ids, "interrupted"),
2856	// Late provider completion after interrupt must not be ingested.
2857	{
2858	type: "subagent.updated",
2859	driver,
2860	subagent: {
2861	...runningSubagent,
2862	status: "completed" as const,
2863	result: "should-not-apply",
2864	completedAt: runningSubagent.updatedAt,
2865	},
2866	},
2867	{
2868	type: "turn_item.updated",
2869	driver,
2870	turnItem: {
2871	...runningTurnItem,
2872	status: "completed" as const,
2873	result: "should-not-apply",
2874	completedAt: runningTurnItem.updatedAt,
2875	updatedAt: runningTurnItem.updatedAt,
2876	},
2877	},
2878	{
2879	type: "turn_item.updated",
2880	driver,
2881	turnItem: {
2882	...runningChildTurnItem,
2883	status: "completed" as const,
2884	completedAt: runningChildTurnItem.updatedAt,
2885	updatedAt: runningChildTurnItem.updatedAt,
2886	},
2887	},
2888	{
2889	type: "node.updated",
2890	driver,
2891	node: {
2892	...runningChildNode,
2893	status: "completed" as const,
2894	completedAt: runningChildNode.startedAt,
2895	},
2896	},
2897	] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>),
2898	close: Deferred.succeed(ingestionDone, undefined),
2899	}),
2900	startTurn: () => Effect.void,
2901	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
2902	run: {
2903	id: ids.runId,
2904	threadId: ids.threadId,
2905	ordinal: 1,
2906	providerInstanceId,
2907	} as OrchestrationV2Run,
2908	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
2909	checkpointScope: {
2910	id: CheckpointScopeId.make("checkpoint-scope:subagent-interrupt-cascade"),
2911	} as OrchestrationV2CheckpointScope,
2912	providerThread: {
2913	id: ids.providerThreadId,
2914	driver,
2915	} as OrchestrationV2ProviderThread,
2916	attempt: {
2917	id: ids.attemptId,
2918	providerTurnId: ids.rootProviderTurnId,
2919	} as OrchestrationV2RunAttempt,
2920	attemptId: ids.attemptId,
2921	relatedThreadIds: [unrelatedChildThreadId, memberThreadId],
2922	providerTurnOrdinal: 1,
2923	message: {
2924	messageId: MessageId.make("message:subagent-interrupt-cascade:user"),
2925	text: "Spawn a subagent then stop.",
2926	attachments: [],
2927	createdBy: "user",
2928	creationSource: "web",
2929	},
2930	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
2931	runtimePolicy: {
2932	runtimeMode: "full-access",
2933	interactionMode: "default",
2934	cwd: process.cwd(),
2935	approvalPolicy: "never",
2936	sandboxPolicy: {
2937	type: "readOnly",
2938	access: { type: "fullAccess" },
2939	networkAccess: false,
2940	},
2941	},
2942	});
2943	}).pipe(Effect.provide(layerTest));
2944
2945	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
2946	assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish");
2947
2948	const events = yield* Ref.get(written);
2949	const subagentEvents = events.flatMap((event) =>
2950	event.type === "subagent.updated" ? [event] : [],
2951	);
2952	const turnItemEvents = events.flatMap((event) => {
2953	if (event.type !== "turn-item.updated" || event.payload.type !== "subagent") {
2954	return [];
2955	}
2956	return [
2957	{
2958	...event,
2959	payload: event.payload,
2960	},
2961	];
2962	});
2963	const nodeEvents = events.flatMap((event) =>
2964	event.type === "node.updated" &&
2965	event.payload.status === "interrupted" &&
2966	(event.payload.id === runningNode.id || event.payload.id === runningChildNode.id)
2967	? [event]
2968	: [],
2969	);
2970	const runUpdatedIndex = events.findIndex((event) => event.type === "run.updated");
2971	assert.isAtLeast(runUpdatedIndex, 0, "root run.updated must be written");
2972
2973	assert.lengthOf(subagentEvents, 2);
2974	const terminalSubagent = subagentEvents.find(
2975	(event) => event.payload.id === runningSubagent.id,
2976	);
2977	const terminalMember = subagentEvents.find((event) => event.payload.id === runningMember.id);
2978	assert.isDefined(terminalMember);
2979	assert.equal(terminalMember.payload.status, "interrupted");
2980	assert.equal(terminalMember.payload.runId, null);
2981	assert.equal(terminalMember.payload.childThreadId, memberThreadId);
2982	assert.isTrue(
2983	events.some(
2984	(event) =>
2985	event.type === "node.updated" &&
2986	event.payload.id === runningMemberRoot.id &&
2987	event.payload.status === "interrupted",
2988	),
2989	);
2990	assert.isFalse(
2991	events.some(
2992	(event) => event.type === "subagent.updated" && event.payload.id === foreignSubagent.id,
2993	),
2994	);
2995	assert.isFalse(
2996	events.some(
2997	(event) => event.type === "node.updated" && event.payload.id === foreignChildNode.id,
2998	),
2999	);
3000	assert.isFalse(
3001	events.some(
3002	(event) =>
3003	event.type === "turn-item.updated" && event.payload.id === foreignChildTurnItem.id,
3004	),
3005	);
3006	assert.isDefined(terminalSubagent);
3007	assert.equal(terminalSubagent.payload.status, "interrupted");
3008	assert.equal(terminalSubagent.payload.childThreadId, childThreadId);
3009	assert.equal(terminalSubagent.payload.result, null);
3010	assert.isNotNull(terminalSubagent.payload.completedAt);
3011
3012	assert.lengthOf(turnItemEvents, 1);
3013	const terminalTurnItem = turnItemEvents[0];
3014	assert.isDefined(terminalTurnItem);
3015	assert.equal(terminalTurnItem.payload.status, "interrupted");
3016	assert.equal(terminalTurnItem.payload.childThreadId, childThreadId);
3017	assert.equal(terminalTurnItem.payload.result, null);
3018
3019	const terminalChildTurnItems = events.flatMap((event) =>
3020	event.type === "turn-item.updated" &&
3021	event.payload.id === runningChildTurnItem.id &&
3022	event.payload.status === "interrupted"
3023	? [event]
3024	: [],
3025	);
3026	assert.lengthOf(terminalChildTurnItems, 1);
3027	const terminalChildTurnItem = terminalChildTurnItems[0];
3028	assert.isDefined(terminalChildTurnItem);
3029	assert.equal(terminalChildTurnItem.threadId, childThreadId);
3030	assert.equal(terminalChildTurnItem.payload.threadId, childThreadId);
3031	assert.equal(terminalChildTurnItem.payload.runId, null);
3032	assert.equal(terminalChildTurnItem.payload.type, "command_execution");
3033	assert.isFalse(
3034	events.some(
3035	(event) =>
3036	event.type === "turn-item.updated" &&
3037	event.payload.id === unrelatedChildTurnItem.id &&
3038	event.payload.status === "interrupted",
3039	),
3040	"related but unlinked child turn items must not cascade",
3041	);
3042	assert.isFalse(
3043	events.some(
3044	(event) =>
3045	event.type === "turn-item.updated" &&
3046	event.payload.id === suppressedChildAssistantTurnItem.id,
3047	),
3048	"suppressed streaming child items must not be created by the cascade",
3049	);
3050
3051	assert.lengthOf(nodeEvents, 2);
3052	assert.isTrue(
3053	nodeEvents.some(
3054	(event) => event.payload.id === runningNode.id && event.payload.kind === "subagent",
3055	),
3056	);
3057	assert.isFalse(
3058	events.some(
3059	(event) =>
3060	event.type === "node.updated" &&
3061	event.payload.id === unrelatedChildNode.id &&
3062	event.payload.status === "interrupted",
3063	),
3064	"owned child threads without a live run-owned subagent link must not cascade",
3065	);
3066	assert.isTrue(
3067	nodeEvents.some(
3068	(event) =>
3069	event.payload.id === runningChildNode.id &&
3070	event.runId === (runningChildNode.runId ?? ids.runId) &&
3071	event.payload.threadId === childThreadId &&
3072	event.payload.kind === "root_turn",
3073	),
3074	);
3075
3076	const cascadeIndexes = events.flatMap((event, index) => {
3077	if (event.type === "subagent.updated" && event.payload.status === "interrupted") {
3078	return [index];
3079	}
3080	if (
3081	event.type === "turn-item.updated" &&
3082	event.payload.status === "interrupted" &&
3083	(event.payload.type === "subagent" || event.payload.id === runningChildTurnItem.id)
3084	) {
3085	return [index];
3086	}
3087	if (event.type === "node.updated" && event.payload.status === "interrupted") {
3088	return event.payload.id === runningNode.id || event.payload.id === runningChildNode.id
3089	? [index]
3090	: [];
3091	}
3092	return [];
3093	});
3094	assert.isTrue(
3095	cascadeIndexes.every((index) => index < runUpdatedIndex),
3096	"subagent cascade must precede run.updated",
3097	);
3098	assert.isFalse(
3099	events.some((event) => {
3100	if (event.type === "subagent.updated") {
3101	return event.payload.status === "completed";
3102	}
3103	if (event.type === "turn-item.updated" && event.payload.type === "subagent") {
3104	return event.payload.status === "completed";
3105	}
3106	return false;
3107	}),
3108	"late provider completion must not reopen cascaded subagent rows",
3109	);
3110	assert.isFalse(
3111	(yield* Ref.get(ingested)).some(
3112	(event) =>
3113	(event.type === "subagent.updated" && event.subagent.status === "completed") ||
3114	(event.type === "turn_item.updated" && event.turnItem.status === "completed") ||
3115	(event.type === "node.updated" && event.node.status === "completed"),
3116	),
3117	"late provider completion must not be ingested after interrupt",
3118	);
3119	}),
3120	);
3121
3122	it.effect(
3123	"cascades linked child-thread nodes after run-owned subagent and turn-item terminalize",
3124	() =>
3125	Effect.gen(function* () {
3126	const ids = backgroundScenarioIds("subagent-link-survives-terminal");
3127	const childThreadId = ids.childThreadId;
3128	const unrelatedChildThreadId = ThreadId.make(
3129	"thread:subagent-link-survives-terminal:unrelated-child",
3130	);
3131	const providerInstanceId = ProviderInstanceId.make("codex");
3132	const written = yield* Ref.make<ReadonlyArray<OrchestrationV2DomainEvent>>([]);
3133	const ingestionDone = yield* Deferred.make<void>();
3134	const layerTest = RunExecutionService.layer.pipe(
3135	Layer.provide(
3136	Layer.mergeAll(
3137	McpAppModelContext.layerEmpty,
3138	Layer.mock(CheckpointService.CheckpointServiceV2)({
3139	captureBaseline: () => Effect.void,
3140	}),
3141	Layer.mock(EventSink.EventSinkV2)({
3142	write: () => Effect.succeed([]),
3143	writeWithEffects: (input) =>
3144	Effect.gen(function* () {
3145	yield* Ref.update(written, (current) => [...current, ...input.events]);
3146	return [];
3147	}),
3148	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
3149	}),
3150	IdAllocator.layer,
3151	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
3152	ingestNormalized: () => Effect.succeed([]),
3153	}),
3154	ServerSettings.layerTest(),
3155	),
3156	),
3157	);
3158
3159	const runningSubagent = makeRunOwnedSubagentFixture({
3160	ids,
3161	providerInstanceId,
3162	childThreadId,
3163	driver,
3164	status: "running",
3165	});
3166	const runningTurnItem = makeRunOwnedSubagentTurnItemFixture({
3167	ids,
3168	providerInstanceId,
3169	childThreadId,
3170	driver,
3171	status: "running",
3172	});
3173	const runningNode = makeRunOwnedSubagentNodeFixture({
3174	ids,
3175	status: "running",
3176	});
3177	const runningChildNode = makeRunOwnedSubagentChildNodeFixture({
3178	ids,
3179	status: "running",
3180	});
3181	const unrelatedChildNode = {
3182	...runningChildNode,
3183	id: NodeId.make("node:subagent-link-survives-terminal:unrelated-child"),
3184	threadId: unrelatedChildThreadId,
3185	runId: ids.runId,
3186	rootNodeId: NodeId.make("node:subagent-link-survives-terminal:unrelated-child"),
3187	};
3188	const completedAt = runningSubagent.updatedAt;
3189
3190	yield* Effect.gen(function* () {
3191	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
3192	yield* runExecution.startRootRun({
3193	commandId: CommandId.make("command:subagent-link-survives-terminal"),
3194	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
3195	providerSessionId: ProviderSessionId.make("session:subagent-link-survives-terminal"),
3196	session: {
3197	events: Stream.empty,
3198	subscribeEvents: Effect.succeed({
3199	events: Stream.fromIterable([
3200	childThreadCreatedEvent(ids),
3201	{
3202	type: "subagent.updated",
3203	driver,
3204	subagent: runningSubagent,
3205	},
3206	{
3207	type: "node.updated",
3208	driver,
3209	node: runningNode,
3210	},
3211	{
3212	type: "node.updated",
3213	driver,
3214	node: runningChildNode,
3215	},
3216	{
3217	type: "node.updated",
3218	driver,
3219	node: unrelatedChildNode,
3220	},
3221	{
3222	type: "turn_item.updated",
3223	driver,
3224	turnItem: runningTurnItem,
3225	},
3226	// Subagent + turn-item settle before root interrupt; linkage
3227	// must still prove the open child-thread node is cascadeable.
3228	{
3229	type: "subagent.updated",
3230	driver,
3231	subagent: {
3232	...runningSubagent,
3233	status: "completed" as const,
3234	result: "subagent finished first",
3235	completedAt,
3236	},
3237	},
3238	{
3239	type: "turn_item.updated",
3240	driver,
3241	turnItem: {
3242	...runningTurnItem,
3243	status: "completed" as const,
3244	result: "turn item finished first",
3245	completedAt,
3246	updatedAt: completedAt,
3247	},
3248	},
3249	{
3250	type: "node.updated",
3251	driver,
3252	node: {
3253	...runningNode,
3254	status: "completed" as const,
3255	completedAt,
3256	},
3257	},
3258	rootTerminalEvent(ids, "interrupted"),
3259	] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>),
3260	close: Deferred.succeed(ingestionDone, undefined),
3261	}),
3262	startTurn: () => Effect.void,
3263	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
3264	run: {
3265	id: ids.runId,
3266	threadId: ids.threadId,
3267	ordinal: 1,
3268	providerInstanceId,
3269	} as OrchestrationV2Run,
3270	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
3271	checkpointScope: {
3272	id: CheckpointScopeId.make("checkpoint-scope:subagent-link-survives-terminal"),
3273	} as OrchestrationV2CheckpointScope,
3274	providerThread: {
3275	id: ids.providerThreadId,
3276	driver,
3277	} as OrchestrationV2ProviderThread,
3278	attempt: {
3279	id: ids.attemptId,
3280	providerTurnId: ids.rootProviderTurnId,
3281	} as OrchestrationV2RunAttempt,
3282	attemptId: ids.attemptId,
3283	relatedThreadIds: [unrelatedChildThreadId],
3284	providerTurnOrdinal: 1,
3285	message: {
3286	messageId: MessageId.make("message:subagent-link-survives-terminal:user"),
3287	text: "Subagent settles before root interrupt.",
3288	attachments: [],
3289	createdBy: "user",
3290	creationSource: "web",
3291	},
3292	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
3293	runtimePolicy: {
3294	runtimeMode: "full-access",
3295	interactionMode: "default",
3296	cwd: process.cwd(),
3297	approvalPolicy: "never",
3298	sandboxPolicy: {
3299	type: "readOnly",
3300	access: { type: "fullAccess" },
3301	networkAccess: false,
3302	},
3303	},
3304	});
3305	}).pipe(Effect.provide(layerTest));
3306
3307	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
3308	assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish");
3309
3310	const events = yield* Ref.get(written);
3311	const runUpdatedIndex = events.findIndex((event) => event.type === "run.updated");
3312	assert.isAtLeast(runUpdatedIndex, 0, "root run.updated must be written");
3313
3314	const cascadedChildNodeEvents = events.flatMap((event, index) =>
3315	event.type === "node.updated" &&
3316	event.payload.id === runningChildNode.id &&
3317	event.payload.status === "interrupted"
3318	? [{ event, index }]
3319	: [],
3320	);
3321	assert.lengthOf(
3322	cascadedChildNodeEvents,
3323	1,
3324	"open linked child-thread node must cascade after subagent/turn-item terminalize",
3325	);
3326	const cascadedChild = cascadedChildNodeEvents[0];
3327	assert.isDefined(cascadedChild);
3328	assert.isTrue(
3329	cascadedChild.index < runUpdatedIndex,
3330	"child-thread cascade must precede run.updated",
3331	);
3332	assert.equal(cascadedChild.event.payload.threadId, childThreadId);
3333	assert.equal(cascadedChild.event.payload.kind, "root_turn");
3334	assert.isFalse(
3335	events.some(
3336	(event) =>
3337	event.type === "node.updated" &&
3338	event.payload.id === unrelatedChildNode.id &&
3339	event.payload.status === "interrupted",
3340	),
3341	"related but unlinked child threads must not cascade",
3342	);
3343	assert.isFalse(
3344	events.some(
3345	(event) => event.type === "subagent.updated" && event.payload.status === "interrupted",
3346	),
3347	"already-terminal subagent rows must not be re-cascaded",
3348	);
3349	assert.isFalse(
3350	events.some(
3351	(event) =>
3352	event.type === "turn-item.updated" &&
3353	event.payload.type === "subagent" &&
3354	event.payload.status === "interrupted",
3355	),
3356	"already-terminal subagent turn items must not be re-cascaded",
3357	);
3358	}),
3359	);
3360
3361	it.effect("cascade helper is provider-neutral for Claude and Codex-shaped child projections", () =>
3362	Effect.gen(function* () {
3363	const now = yield* DateTime.now;
3364	let nextId = 0;
3365	const allocateEventId = () =>
3366	Effect.sync(() => EventId.make(`event:cascade-helper:${nextId++}`));
3367
3368	for (const driverKind of [
3369	ProviderDriverKind.make("claudeAgent"),
3370	ProviderDriverKind.make("codex"),
3371	] as const) {
3372	const runId = RunId.make(`run:cascade-helper:${driverKind}`);
3373	const threadId = ThreadId.make(`thread:cascade-helper:${driverKind}`);
3374	const childThreadId = ThreadId.make(`thread:cascade-helper:${driverKind}:child`);
3375	const subagentId = NodeId.make(`node:cascade-helper:${driverKind}:subagent`);
3376	const childNodeId = NodeId.make(`node:cascade-helper:${driverKind}:child-root`);
3377	const providerInstanceId = ProviderInstanceId.make(String(driverKind));
3378	const terminalStatus = driverKind === "claudeAgent" ? "failed" : "cancelled";
3379	const subagent: OrchestrationV2Subagent = {
3380	id: subagentId,
3381	threadId,
3382	runId,
3383	parentNodeId: NodeId.make(`node:cascade-helper:${driverKind}:root`),
3384	origin: "provider_native",
3385	createdBy: "agent",
3386	driver: driverKind,
3387	providerInstanceId,
3388	providerThreadId: null,
3389	childThreadId,
3390	nativeTaskRef: {
3391	driver: driverKind,
3392	nativeId: `native-${driverKind}`,
3393	strength: "strong",
3394	},
3395	prompt: "hold",
3396	title: "hold",
3397	model: null,
3398	status: "running",
3399	progress: "partial progress",
3400	result: "partial result",
3401	startedAt: now,
3402	completedAt: null,
3403	updatedAt: now,
3404	};
3405	const turnItem = {
3406	id: TurnItemId.make(`turn-item:cascade-helper:${driverKind}`),
3407	threadId,
3408	runId,
3409	nodeId: subagentId,
3410	providerThreadId: null,
3411	providerTurnId: null,
3412	nativeItemRef: subagent.nativeTaskRef,
3413	parentItemId: null,
3414	ordinal: 3,
3415	status: "running" as const,
3416	title: "hold",
3417	startedAt: now,
3418	completedAt: null,
3419	updatedAt: now,
3420	type: "subagent" as const,
3421	subagentId,
3422	origin: "provider_native" as const,
3423	driver: driverKind,
3424	providerInstanceId,
3425	childThreadId,
3426	prompt: "hold",
3427	progress: "partial progress",
3428	result: "partial result",
3429	} satisfies Extract<OrchestrationV2TurnItem, { type: "subagent" }>;
3430	const node: OrchestrationV2ExecutionNode = {
3431	id: subagentId,
3432	threadId,
3433	runId,
3434	parentNodeId: NodeId.make(`node:cascade-helper:${driverKind}:root`),
3435	rootNodeId: NodeId.make(`node:cascade-helper:${driverKind}:root`),
3436	kind: "subagent",
3437	status: "running",
3438	countsForRun: false,
3439	providerThreadId: null,
3440	providerTurnId: null,
3441	nativeItemRef: subagent.nativeTaskRef,
3442	runtimeRequestId: null,
3443	checkpointScopeId: null,
3444	startedAt: now,
3445	completedAt: null,
3446	};
3447	const openChildNode: OrchestrationV2ExecutionNode = {
3448	id: childNodeId,
3449	threadId: childThreadId,
3450	runId: null,
3451	parentNodeId: null,
3452	rootNodeId: childNodeId,
3453	kind: "root_turn",
3454	status: "running",
3455	countsForRun: false,
3456	providerThreadId: null,
3457	providerTurnId: null,
3458	nativeItemRef: null,
3459	runtimeRequestId: null,
3460	checkpointScopeId: null,
3461	startedAt: now,
3462	completedAt: null,
3463	};
3464	const childTurnItem: OrchestrationV2TurnItem =
3465	driverKind === "claudeAgent"
3466	? {
3467	id: TurnItemId.make(`turn-item:cascade-helper:${driverKind}:child-reasoning`),
3468	threadId: childThreadId,
3469	runId: null,
3470	nodeId: childNodeId,
3471	providerThreadId: null,
3472	providerTurnId: null,
3473	nativeItemRef: null,
3474	parentItemId: null,
3475	ordinal: 1,
3476	status: "running",
3477	title: "Working",
3478	startedAt: now,
3479	completedAt: null,
3480	updatedAt: now,
3481	type: "reasoning",
3482	text: "partial progress",
3483	streaming: true,
3484	}
3485	: {
3486	id: TurnItemId.make(`turn-item:cascade-helper:${driverKind}:child-command`),
3487	threadId: childThreadId,
3488	runId: null,
3489	nodeId: childNodeId,
3490	providerThreadId: null,
3491	providerTurnId: null,
3492	nativeItemRef: null,
3493	parentItemId: null,
3494	ordinal: 1,
3495	status: "running",
3496	title: "sleep 300",
3497	startedAt: now,
3498	completedAt: null,
3499	updatedAt: now,
3500	type: "command_execution",
3501	input: "sleep 300",
3502	};
3503
3504	const events = yield* RunExecutionService.cascadeTerminalizeRunOwnedSubagents({
3505	run: {
3506	id: runId,
3507	threadId,
3508	ordinal: 1,
3509	providerInstanceId,
3510	} as OrchestrationV2Run,
3511	open: {
3512	subagents: new Map([[subagentId, subagent]]),
3513	turnItems: new Map([[subagentId, turnItem]]),
3514	childTurnItems: new Map(),
3515	nodes: new Map([[subagentId, node]]),
3516	linkedChildThreadIds: new Set([childThreadId]),
3517	},
3518	status: terminalStatus,
3519	completedAt: now,
3520	allocateEventId,
3521	});
3522
3523	assert.equal(events.length, 3, `${driverKind}: subagent + node + turn item`);
3524	const terminalSubagent = events.find((event) => event.type === "subagent.updated");
3525	assert.isDefined(terminalSubagent);
3526	if (terminalSubagent?.type !== "subagent.updated") {
3527	assert.fail("expected subagent.updated event");
3528	return;
3529	}
3530	assert.equal(terminalSubagent.payload.status, terminalStatus);
3531	assert.equal(terminalSubagent.payload.childThreadId, childThreadId);
3532	assert.equal(terminalSubagent.payload.progress, "partial progress");
3533	assert.equal(terminalSubagent.payload.result, "partial result");
3534	assert.equal(terminalSubagent.payload.driver, driverKind);
3535
3536	const terminalItem = events.find(
3537	(event) => event.type === "turn-item.updated" && event.payload.type === "subagent",
3538	);
3539	assert.isDefined(terminalItem);
3540	if (terminalItem?.type !== "turn-item.updated" || terminalItem.payload.type !== "subagent") {
3541	assert.fail("expected subagent turn-item.updated event");
3542	return;
3543	}
3544	assert.equal(terminalItem.payload.status, terminalStatus);
3545	assert.equal(terminalItem.payload.childThreadId, childThreadId);
3546	assert.equal(terminalItem.payload.progress, "partial progress");
3547	assert.equal(terminalItem.payload.result, "partial result");
3548
3549	// Shared cascade path: after subagent/turn-item rows are gone, only the
3550	// preserved linkage may prove an open child-thread node is cascadeable.
3551	const afterTerminalLinkEvents =
3552	yield* RunExecutionService.cascadeTerminalizeRunOwnedSubagents({
3553	run: {
3554	id: runId,
3555	threadId,
3556	ordinal: 1,
3557	providerInstanceId,
3558	} as OrchestrationV2Run,
3559	open: {
3560	subagents: new Map(),
3561	turnItems: new Map(),
3562	childTurnItems: new Map([[childTurnItem.id, childTurnItem]]),
3563	nodes: new Map([[childNodeId, openChildNode]]),
3564	linkedChildThreadIds: new Set([childThreadId]),
3565	},
3566	status: terminalStatus,
3567	completedAt: now,
3568	allocateEventId,
3569	});
3570	assert.equal(
3571	afterTerminalLinkEvents.length,
3572	2,
3573	`${driverKind}: linked child node and turn item cascade after link rows terminalize`,
3574	);
3575	const cascadedChild = afterTerminalLinkEvents.find((event) => event.type === "node.updated");
3576	assert.isDefined(cascadedChild);
3577	if (cascadedChild?.type !== "node.updated") {
3578	assert.fail("expected node.updated for linked child thread");
3579	return;
3580	}
3581	assert.equal(cascadedChild.payload.id, childNodeId);
3582	assert.equal(cascadedChild.payload.threadId, childThreadId);
3583	assert.equal(cascadedChild.payload.status, terminalStatus);
3584	assert.equal(cascadedChild.payload.kind, "root_turn");
3585	const cascadedChildTurnItem = afterTerminalLinkEvents.find(
3586	(event) => event.type === "turn-item.updated",
3587	);
3588	assert.isDefined(cascadedChildTurnItem);
3589	if (cascadedChildTurnItem?.type !== "turn-item.updated") {
3590	assert.fail("expected turn-item.updated for linked child thread");
3591	return;
3592	}
3593	assert.equal(cascadedChildTurnItem.threadId, childThreadId);
3594	assert.equal(cascadedChildTurnItem.payload.id, childTurnItem.id);
3595	assert.equal(cascadedChildTurnItem.payload.status, terminalStatus);
3596	assert.equal(cascadedChildTurnItem.payload.runId, null);
3597	assert.equal(cascadedChildTurnItem.payload.type, childTurnItem.type);
3598	if (cascadedChildTurnItem.payload.type === "reasoning") {
3599	assert.isFalse(cascadedChildTurnItem.payload.streaming);
3600	}
3601	}
3602	}),
3603	);
3604
3605	it.effect("omits interrupt results and subagent cascade for a superseded attempt", () =>
3606	Effect.gen(function* () {
3607	const { written, observed } = yield* captureRootRunTermination({
3608	key: "steer-supersede",
3609	shouldFinalizeRun: () => Effect.succeed(false),
3610	seedOpenSubagent: true,
3611	});
3612	assert.deepEqual(
3613	written.map((item) => item.type),
3614	[],
3615	);
3616	assert.deepEqual(observed, []);
3617	}),
3618	);
3619
3620	it.effect("emits run_interrupt_result when superseded attempt still has a hard-stop request", () =>
3621	Effect.gen(function* () {
3622	const { written, observed } = yield* captureRootRunTermination({
3623	key: "stop-then-steer-supersede",
3624	shouldFinalizeRun: () => Effect.succeed(false),
3625	hasUnpairedRunInterruptRequest: () => Effect.succeed(true),
3626	});
3627	assert.deepEqual(
3628	written.map((item) => item.type),
3629	["run_interrupt_result"],
3630	);
3631	assert.deepEqual(observed, ["pull-requests-refreshed"]);
3632	const ids = backgroundScenarioIds("stop-then-steer-supersede");
3633	const expectedRequestId = yield* Effect.gen(function* () {
3634	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3635	return idAllocator.derive.runSignalTurnItem({
3636	runId: ids.runId,
3637	signal: "interrupt-request",
3638	});
3639	}).pipe(Effect.provide(IdAllocator.layer));
3640	assert.equal(written[0]?.parentItemId, expectedRequestId);
3641	}),
3642	);
3643
3644	it.effect("omits run_interrupt_result when superseded attempt request is already paired", () =>
3645	Effect.gen(function* () {
3646	const { written, observed } = yield* captureRootRunTermination({
3647	key: "stop-then-steer-already-paired",
3648	shouldFinalizeRun: () => Effect.succeed(false),
3649	hasUnpairedRunInterruptRequest: () => Effect.succeed(false),
3650	});
3651	assert.deepEqual(
3652	written.map((item) => item.type),
3653	[],
3654	);
3655	assert.deepEqual(observed, []);
3656	}),
3657	);
3658
3659	it.effect("does not overwrite Stop when ownership changes after the finalization read", () =>
3660	Effect.gen(function* () {
3661	const { written, observed, submittedEffects, committedEffects } =
3662	yield* captureRootRunTermination({
3663	key: "stop-wins-finalization-gap",
3664	shouldFinalizeRun: () => Effect.succeed(true),
3665	rejectTerminalWrite: true,
3666	});
3667	assert.deepEqual(written, []);
3668	assert.deepEqual(observed, []);
3669	assert.deepEqual(
3670	submittedEffects.map((effect) => effect.request.type),
3671	["checkpoint.capture"],
3672	);
3673	assert.deepEqual(committedEffects, []);
3674	}),
3675	);
3676
3677	it.effect("emits run_interrupt_result when hard-stop finalizes the active attempt", () =>
3678	Effect.gen(function* () {
3679	const { written, observed, committedEffects } = yield* captureRootRunTermination({
3680	key: "hard-stop",
3681	shouldFinalizeRun: () => Effect.succeed(true),
3682	});
3683	assert.deepEqual(
3684	written.map((item) => item.type),
3685	["run_interrupt_result"],
3686	);
3687	assert.deepEqual(observed, ["run:interrupted", "pull-requests-refreshed"]);
3688	assert.deepEqual(
3689	committedEffects.map((effect) => effect.request.type),
3690	["checkpoint.capture"],
3691	);
3692	}),
3693	);
3694
3695	it.effect.each(["completed", "interrupted", "cancelled", "failed"] as const)(
3696	"refreshes pull requests after the current root run %s",
3697	(status) =>
3698	Effect.gen(function* () {
3699	const { observed } = yield* captureRootRunTermination({
3700	key: `pull-request-refresh:${status}`,
3701	shouldFinalizeRun: () => Effect.succeed(true),
3702	events: (ids) => Stream.make(rootTerminalEvent(ids, status)),
3703	});
3704	assert.deepEqual(observed, [
3705	`run:${status === "completed" ? "waiting" : status}`,
3706	"pull-requests-refreshed",
3707	]);
3708	}),
3709	);
3710
3711	it.effect("records a finished run as failed when its ownership check cannot be read", () =>
3712	Effect.gen(function* () {
3713	const { observed } = yield* captureRootRunTermination({
3714	key: "finalize-guard-read-failure",
3715	shouldFinalizeRun: () =>
3716	Effect.fail(
3717	new ProjectionStore.ProjectionStoreReadError({
3718	threadId: ThreadId.make("thread:finalize-guard-read-failure"),
3719	cause: "database unavailable",
3720	}),
3721	),
3722	events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")),
3723	});
3724	// The fallback settles through the guarded write instead of the same
3725	// failing read, so the run does not stay running.
3726	assert.include(observed, "run:failed");
3727	}),
3728	);
3729
3730	it.effect("does not refresh pull requests for auxiliary or stale provider terminals", () =>
3731	Effect.gen(function* () {
3732	const { observed } = yield* captureRootRunTermination({
3733	key: "pull-request-refresh:auxiliary",
3734	shouldFinalizeRun: () => Effect.succeed(true),
3735	events: (ids) =>
3736	Stream.fromIterable([
3737	rootTerminalEvent(
3738	{ ...ids, rootProviderTurnId: ProviderTurnId.make("turn:child") },
3739	"completed",
3740	),
3741	rootTerminalEvent(
3742	{ ...ids, rootProviderTurnId: ProviderTurnId.make("turn:previous") },
3743	"interrupted",
3744	),
3745	]),
3746	});
3747	assert.deepEqual(observed, []);
3748	}),
3749	);
3750
3751	it.effect("refreshes pull requests after a provider stream exits with an error", () =>
3752	Effect.gen(function* () {
3753	const { written, observed } = yield* captureRootRunTermination({
3754	key: "pull-request-refresh:stream-error",
3755	shouldFinalizeRun: () => Effect.succeed(true),
3756	events: () =>
3757	Stream.fail(
3758	new ProviderAdapter.ProviderAdapterEventStreamError({
3759	driver,
3760	providerSessionId: ProviderSessionId.make("session:exited"),
3761	cause: "provider process exited",
3762	}),
3763	),
3764	});
3765	assert.deepEqual(observed, ["run:failed", "pull-requests-refreshed"]);
3766	assert.deepEqual(
3767	written.map((item) => item.type),
3768	["error"],
3769	);
3770	const error = written.find((item) => item.type === "error");
3771	assert.include(error?.failure.message ?? "", "provider event stream closed unexpectedly");
3772	}),
3773	);
3774
3775	it.effect("refreshes pull requests only once when startup failure closes its event stream", () =>
3776	Effect.gen(function* () {
3777	const ingestionStarted = yield* Deferred.make<void>();
3778	const { observed, written } = yield* captureRootRunTermination({
3779	key: "pull-request-refresh:startup-error",
3780	shouldFinalizeRun: () => Effect.succeed(true),
3781	events: () =>
3782	Stream.unwrap(Deferred.succeed(ingestionStarted, undefined).pipe(Effect.as(Stream.never))),
3783	startTurn: (input) =>
3784	Deferred.await(ingestionStarted).pipe(
3785	Effect.andThen(
3786	Effect.fail(
3787	new ProviderAdapter.ProviderAdapterTurnStartError({
3788	driver,
3789	threadId: input.threadId,
3790	providerThreadId: input.providerThread.id,
3791	runId: input.runId,
3792	cause: "provider rejected the turn",
3793	}),
3794	),
3795	),
3796	),
3797	});
3798	assert.equal(observed.filter((item) => item === "pull-requests-refreshed").length, 1);
3799	assert.equal(observed[0], "run:failed");
3800	const error = written.find((item) => item.type === "error");
3801	assert.include(error?.failure.message ?? "", "provider could not start this turn");
3802	}),
3803	);
3804
3805	it.effect("keeps completed runs completed when pull request refresh fails", () =>
3806	Effect.gen(function* () {
3807	const { observed } = yield* captureRootRunTermination({
3808	key: "pull-request-refresh:refresh-failure",
3809	shouldFinalizeRun: () => Effect.succeed(true),
3810	events: (ids) => Stream.make(rootTerminalEvent(ids, "completed")),
3811	refreshAfterTurn: Effect.die("refresh failed"),
3812	});
3813	assert.deepEqual(observed, ["run:waiting", "pull-requests-refreshed"]);
3814	}),
3815	);
3816
3817	function captureRootRunTermination(input: {
3818	readonly key: string;
3819	readonly shouldFinalizeRun: () => Effect.Effect<boolean, ProjectionStore.ProjectionStoreV2Error>;
3820	readonly rejectTerminalWrite?: boolean;
3821	readonly hasUnpairedRunInterruptRequest?: () => Effect.Effect<boolean, never>;
3822	readonly seedOpenSubagent?: boolean;
3823	readonly events?: (
3824	ids: BackgroundScenarioIds,
3825	) => Stream.Stream<
3826	ProviderAdapter.ProviderAdapterV2Event,
3827	ProviderAdapter.ProviderAdapterV2Error
3828	>;
3829	readonly startTurn?: ProviderAdapter.ProviderAdapterV2SessionRuntime["startTurn"];
3830	readonly refreshAfterTurn?: Effect.Effect<void>;
3831	}) {
3832	return Effect.gen(function* () {
3833	const ids = backgroundScenarioIds(input.key);
3834	const providerInstanceId = ProviderInstanceId.make("codex");
3835	const runningSubagent = makeRunOwnedSubagentFixture({
3836	ids,
3837	providerInstanceId,
3838	childThreadId: ids.childThreadId,
3839	driver,
3840	status: "running",
3841	});
3842	const writtenItems = yield* Ref.make<ReadonlyArray<OrchestrationV2TurnItem>>([]);
3843	const observed = yield* Ref.make<ReadonlyArray<string>>([]);
3844	const submittedEffects = yield* Ref.make<ReadonlyArray<PendingOrchestrationEffectV2>>([]);
3845	const committedEffects = yield* Ref.make<ReadonlyArray<PendingOrchestrationEffectV2>>([]);
3846	const ingestionDone = yield* Deferred.make<void>();
3847	const captureTurnItem = (payload: OrchestrationV2TurnItem) =>
3848	Ref.update(writtenItems, (current) => [...current, payload]);
3849	const captureFinalEvents = (events: ReadonlyArray<OrchestrationV2DomainEvent>) =>
3850	Effect.gen(function* () {
3851	for (const event of events) {
3852	if (event.type === "turn-item.updated") {
3853	yield* captureTurnItem(event.payload);
3854	}
3855	if (event.type === "run.updated") {
3856	yield* Ref.update(observed, (current) => [...current, `run:${event.payload.status}`]);
3857	}
3858	}
3859	});
3860	const layerTest = RunExecutionService.layer.pipe(
3861	Layer.provide(
3862	Layer.mergeAll(
3863	McpAppModelContext.layerEmpty,
3864	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
3865	Layer.mock(EventSink.EventSinkV2)({
3866	write: (payload) =>
3867	Effect.gen(function* () {
3868	for (const event of payload.events) {
3869	if (event.type === "turn-item.updated") {
3870	yield* captureTurnItem(event.payload);
3871	}
3872	}
3873	return [];
3874	}),
3875	writeWithEffects: (payload) =>
3876	Effect.gen(function* () {
3877	yield* Ref.update(submittedEffects, (current) => [...current, ...payload.effects]);
3878	yield* Ref.update(committedEffects, (current) => [...current, ...payload.effects]);
3879	yield* captureFinalEvents(payload.events);
3880	return [];
3881	}),
3882	writeIfRunCurrent: (payload) =>
3883	Effect.gen(function* () {
3884	const effects = payload.effects ?? [];
3885	yield* Ref.update(submittedEffects, (current) => [...current, ...effects]);
3886	if (input.rejectTerminalWrite === true) {
3887	return { committed: false, storedEvents: [] };
3888	}
3889	yield* Ref.update(committedEffects, (current) => [...current, ...effects]);
3890	yield* captureFinalEvents(payload.events);
3891	return { committed: true, storedEvents: [] };
3892	}),
3893	}),
3894	IdAllocator.layer,
3895	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
3896	ingestNormalized: () => Effect.succeed([]),
3897	}),
3898	ServerSettings.layerTest(),
3899	Layer.succeed(RunFinalizationService.RunFinalizationObserver, {
3900	refresh: () => Effect.void,
3901	refreshAfterTurn: () =>
3902	Ref.update(observed, (current) => [...current, "pull-requests-refreshed"]).pipe(
3903	Effect.andThen(input.refreshAfterTurn ?? Effect.void),
3904	),
3905	}),
3906	),
3907	),
3908	);
3909
3910	yield* Effect.gen(function* () {
3911	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
3912	yield* runExecution.startRootRun({
3913	commandId: CommandId.make(`command:${input.key}`),
3914	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
3915	providerSessionId: ProviderSessionId.make(`session:${input.key}`),
3916	session: {
3917	events: Stream.empty,
3918	subscribeEvents: Effect.succeed({
3919	events:
3920	input.events?.(ids) ??
3921	Stream.fromIterable([
3922	...(input.seedOpenSubagent
3923	? [
3924	{ type: "subagent.updated", driver, subagent: runningSubagent } as const,
3925	{
3926	type: "node.updated",
3927	driver,
3928	node: makeRunOwnedSubagentNodeFixture({ ids, status: "running" }),
3929	} as const,
3930	{
3931	type: "turn_item.updated",
3932	driver,
3933	turnItem: makeRunOwnedSubagentTurnItemFixture({
3934	ids,
3935	providerInstanceId,
3936	childThreadId: ids.childThreadId,
3937	driver,
3938	status: "running",
3939	}),
3940	} as const,
3941	]
3942	: []),
3943	rootTerminalEvent(ids, "interrupted"),
3944	] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>),
3945	close: Deferred.succeed(ingestionDone, undefined),
3946	}),
3947	startTurn: input.startTurn ?? (() => Effect.void),
3948	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
3949	run: {
3950	id: ids.runId,
3951	threadId: ids.threadId,
3952	ordinal: 1,
3953	providerInstanceId,
3954	} as OrchestrationV2Run,
3955	rootNode: {
3956	id: ids.rootNodeId,
3957	providerTurnId: ids.rootProviderTurnId,
3958	} as OrchestrationV2ExecutionNode,
3959	checkpointScope: {
3960	id: CheckpointScopeId.make(`checkpoint-scope:${input.key}`),
3961	} as OrchestrationV2CheckpointScope,
3962	providerThread: {
3963	id: ids.providerThreadId,
3964	driver,
3965	} as OrchestrationV2ProviderThread,
3966	attempt: {
3967	id: ids.attemptId,
3968	providerTurnId: ids.rootProviderTurnId,
3969	} as OrchestrationV2RunAttempt,
3970	attemptId: ids.attemptId,
3971	providerTurnOrdinal: 1,
3972	shouldFinalizeRun: input.shouldFinalizeRun,
3973	...(input.hasUnpairedRunInterruptRequest === undefined
3974	? {}
3975	: {
3976	hasUnpairedRunInterruptRequest: input.hasUnpairedRunInterruptRequest,
3977	}),
3978	message: {
3979	messageId: MessageId.make(`message:${input.key}`),
3980	text: "interrupt projection",
3981	attachments: [],
3982	createdBy: "user",
3983	creationSource: "web",
3984	},
3985	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
3986	runtimePolicy: {
3987	runtimeMode: "full-access",
3988	interactionMode: "default",
3989	cwd: process.cwd(),
3990	approvalPolicy: "never",
3991	sandboxPolicy: {
3992	type: "readOnly",
3993	access: { type: "fullAccess" },
3994	networkAccess: false,
3995	},
3996	},
3997	});
3998	}).pipe(Effect.provide(layerTest));
3999
4000	yield* Deferred.await(ingestionDone);
4001	return {
4002	written: yield* Ref.get(writtenItems),
4003	observed: yield* Ref.get(observed),
4004	submittedEffects: yield* Ref.get(submittedEffects),
4005	committedEffects: yield* Ref.get(committedEffects),
4006	};
4007	});
4008	}
4009
4010	interface BackgroundScenarioIds {
4011	readonly threadId: ThreadId;
4012	readonly childThreadId: ThreadId;
4013	readonly runId: RunId;
4014	readonly attemptId: RunAttemptId;
4015	readonly providerThreadId: ProviderThreadId;
4016	readonly rootProviderTurnId: ProviderTurnId;
4017	readonly rootNodeId: NodeId;
4018	readonly itemId: TurnItemId;
4019	readonly childItemId: TurnItemId;
4020	readonly subagentNodeId: NodeId;
4021	}
4022
4023	function backgroundScenarioIds(key: string): BackgroundScenarioIds {
4024	return {
4025	threadId: ThreadId.make(`thread:${key}`),
4026	childThreadId: ThreadId.make(`thread:${key}:child`),
4027	runId: RunId.make(`run:${key}`),
4028	attemptId: RunAttemptId.make(`attempt:${key}`),
4029	providerThreadId: ProviderThreadId.make(`provider-thread:${key}`),
4030	rootProviderTurnId: ProviderTurnId.make(`provider-turn:${key}`),
4031	rootNodeId: NodeId.make(`node:${key}`),
4032	itemId: TurnItemId.make(`turn-item:${key}`),
4033	childItemId: TurnItemId.make(`turn-item:${key}:child`),
4034	subagentNodeId: NodeId.make(`node:${key}:subagent`),
4035	};
4036	}
4037
4038	function childThreadCreatedEvent(
4039	ids: BackgroundScenarioIds,
4040	): ProviderAdapter.ProviderAdapterV2Event {
4041	return {
4042	type: "app_thread.created",
4043	driver,
4044	appThread: {
4045	id: ids.childThreadId,
4046	lineage: {
4047	parentThreadId: ids.threadId,
4048	relationshipToParent: "subagent",
4049	rootThreadId: ids.threadId,
4050	},
4051	},
4052	} as ProviderAdapter.ProviderAdapterV2Event;
4053	}
4054
4055	function childBackgroundTurnItemEvent(
4056	ids: BackgroundScenarioIds,
4057	status: "running" | "completed" | "idle",
4058	ordinal: number,
4059	): ProviderAdapter.ProviderAdapterV2Event {
4060	return {
4061	type: "turn_item.updated",
4062	driver,
4063	turnItem: {
4064	id: ids.childItemId,
4065	threadId: ids.childThreadId,
4066	runId: null,
4067	providerTurnId: null,
4068	ordinal,
4069	type: "command_execution",
4070	status,
4071	},
4072	} as ProviderAdapter.ProviderAdapterV2Event;
4073	}
4074
4075	function backgroundTurnItemEvent(
4076	ids: BackgroundScenarioIds,
4077	type: "command_execution" | "dynamic_tool" | "subagent",
4078	status: "running" | "completed" | "idle",
4079	ordinal: number,
4080	itemId?: TurnItemId,
4081	): ProviderAdapter.ProviderAdapterV2Event {
4082	return {
4083	type: "turn_item.updated",
4084	driver,
4085	turnItem: {
4086	id: itemId ?? ids.itemId,
4087	threadId: ids.threadId,
4088	runId: ids.runId,
4089	providerTurnId: ids.rootProviderTurnId,
4090	ordinal,
4091	type,
4092	status,
4093	},
4094	} as ProviderAdapter.ProviderAdapterV2Event;
4095	}
4096
4097	function backgroundTurnItemEventForRun(
4098	ids: BackgroundScenarioIds,
4099	runId: RunId,
4100	type: "command_execution" | "dynamic_tool" | "subagent",
4101	status: "running" | "completed" | "idle",
4102	ordinal: number,
4103	): ProviderAdapter.ProviderAdapterV2Event {
4104	const event = backgroundTurnItemEvent(ids, type, status, ordinal);
4105	if (event.type !== "turn_item.updated") {
4106	return event;
4107	}
4108	return { ...event, turnItem: { ...event.turnItem, runId } };
4109	}
4110
4111	function subagentEvent(
4112	ids: BackgroundScenarioIds,
4113	status: "running" | "completed" | "idle",
4114	): ProviderAdapter.ProviderAdapterV2Event {
4115	return {
4116	type: "subagent.updated",
4117	driver,
4118	subagent: {
4119	id: ids.subagentNodeId,
4120	threadId: ids.threadId,
4121	runId: ids.runId,
4122	status,
4123	},
4124	} as ProviderAdapter.ProviderAdapterV2Event;
4125	}
4126
4127	function makeRunOwnedSubagentFixture(input: {
4128	readonly ids: BackgroundScenarioIds;
4129	readonly providerInstanceId: ProviderInstanceId;
4130	readonly childThreadId: ThreadId;
4131	readonly driver: typeof driver;
4132	readonly status: "running" | "interrupted";
4133	}): OrchestrationV2Subagent {
4134	const now = DateTime.makeUnsafe("2026-07-21T12:00:00.000Z");
4135	return {
4136	id: input.ids.subagentNodeId,
4137	threadId: input.ids.threadId,
4138	runId: input.ids.runId,
4139	parentNodeId: input.ids.rootNodeId,
4140	origin: "provider_native",
4141	createdBy: "agent",
4142	driver: input.driver,
4143	providerInstanceId: input.providerInstanceId,
4144	providerThreadId: null,
4145	childThreadId: input.childThreadId,
4146	nativeTaskRef: {
4147	driver: input.driver,
4148	nativeId: `task:${input.ids.subagentNodeId}`,
4149	strength: "strong",
4150	},
4151	prompt: "hold",
4152	title: "Live-test subagent hold",
4153	model: null,
4154	status: input.status,
4155	result: null,
4156	startedAt: now,
4157	completedAt: null,
4158	updatedAt: now,
4159	};
4160	}
4161
4162	function makeRunOwnedSubagentTurnItemFixture(input: {
4163	readonly ids: BackgroundScenarioIds;
4164	readonly providerInstanceId: ProviderInstanceId;
4165	readonly childThreadId: ThreadId;
4166	readonly driver: typeof driver;
4167	readonly status: "running" | "interrupted";
4168	}): Extract<OrchestrationV2TurnItem, { type: "subagent" }> {
4169	const now = DateTime.makeUnsafe("2026-07-21T12:00:00.000Z");
4170	return {
4171	id: input.ids.itemId,
4172	threadId: input.ids.threadId,
4173	runId: input.ids.runId,
4174	nodeId: input.ids.subagentNodeId,
4175	providerThreadId: input.ids.providerThreadId,
4176	providerTurnId: input.ids.rootProviderTurnId,
4177	nativeItemRef: {
4178	driver: input.driver,
4179	nativeId: `task:${input.ids.subagentNodeId}`,
4180	strength: "strong",
4181	},
4182	parentItemId: null,
4183	ordinal: 3,
4184	status: input.status,
4185	title: "Live-test subagent hold",
4186	startedAt: now,
4187	completedAt: null,
4188	updatedAt: now,
4189	type: "subagent",
4190	subagentId: input.ids.subagentNodeId,
4191	origin: "provider_native",
4192	driver: input.driver,
4193	providerInstanceId: input.providerInstanceId,
4194	childThreadId: input.childThreadId,
4195	prompt: "hold",
4196	result: null,
4197	};
4198	}
4199
4200	function makeLinkedChildTurnItemFixture(input: {
4201	readonly ids: BackgroundScenarioIds;
4202	readonly driver: typeof driver;
4203	readonly type: "assistant_message" | "command_execution" | "reasoning";
4204	}): OrchestrationV2TurnItem {
4205	const now = DateTime.makeUnsafe("2026-07-21T12:00:00.000Z");
4206	const base = {
4207	id: input.ids.childItemId,
4208	threadId: input.ids.childThreadId,
4209	runId: null,
4210	nodeId: NodeId.make(`${input.ids.subagentNodeId}:child-root`),
4211	providerThreadId: null,
4212	providerTurnId: null,
4213	nativeItemRef: null,
4214	parentItemId: null,
4215	ordinal: 1,
4216	status: "running" as const,
4217	title: "Working",
4218	startedAt: now,
4219	completedAt: null,
4220	updatedAt: now,
4221	};
4222	if (input.type === "assistant_message") {
4223	return {
4224	...base,
4225	type: "assistant_message",
4226	messageId: MessageId.make(`${input.ids.childItemId}:message`),
4227	text: "partial response",
4228	streaming: true,
4229	};
4230	}
4231	if (input.type === "reasoning") {
4232	return {
4233	...base,
4234	type: "reasoning",
4235	text: "partial progress",
4236	streaming: true,
4237	};
4238	}
4239	return {
4240	...base,
4241	type: "command_execution",
4242	input: "sleep 300",
4243	};
4244	}
4245
4246	function makeRunOwnedSubagentNodeFixture(input: {
4247	readonly ids: BackgroundScenarioIds;
4248	readonly status: "running" | "interrupted";
4249	}): OrchestrationV2ExecutionNode {
4250	const now = DateTime.makeUnsafe("2026-07-21T12:00:00.000Z");
4251	return {
4252	id: input.ids.subagentNodeId,
4253	threadId: input.ids.threadId,
4254	runId: input.ids.runId,
4255	parentNodeId: input.ids.rootNodeId,
4256	rootNodeId: input.ids.rootNodeId,
4257	kind: "subagent",
4258	status: input.status,
4259	countsForRun: false,
4260	providerThreadId: input.ids.providerThreadId,
4261	providerTurnId: input.ids.rootProviderTurnId,
4262	nativeItemRef: {
4263	driver,
4264	nativeId: `task:${input.ids.subagentNodeId}`,
4265	strength: "strong",
4266	},
4267	runtimeRequestId: null,
4268	checkpointScopeId: null,
4269	startedAt: now,
4270	completedAt: null,
4271	};
4272	}
4273
4274	function makeRunOwnedSubagentChildNodeFixture(input: {
4275	readonly ids: BackgroundScenarioIds;
4276	readonly status: "running" | "interrupted";
4277	}): OrchestrationV2ExecutionNode {
4278	const now = DateTime.makeUnsafe("2026-07-21T12:00:00.000Z");
4279	return {
4280	id: NodeId.make(`${input.ids.subagentNodeId}:child-root`),
4281	threadId: input.ids.childThreadId,
4282	runId: null,
4283	parentNodeId: null,
4284	rootNodeId: NodeId.make(`${input.ids.subagentNodeId}:child-root`),
4285	kind: "root_turn",
4286	status: input.status,
4287	countsForRun: false,
4288	providerThreadId: null,
4289	providerTurnId: null,
4290	nativeItemRef: null,
4291	runtimeRequestId: null,
4292	checkpointScopeId: null,
4293	startedAt: now,
4294	completedAt: null,
4295	};
4296	}
4297
4298	function rootTerminalEvent(
4299	ids: BackgroundScenarioIds,
4300	status: "completed" | "interrupted" | "cancelled" | "failed",
4301	): ProviderAdapter.ProviderAdapterV2Event {
4302	const common = {
4303	type: "turn.terminal" as const,
4304	driver,
4305	providerThreadId: ids.providerThreadId,
4306	providerTurnId: ids.rootProviderTurnId,
4307	runOrdinal: 1,
4308	threadDisposition: "reusable" as const,
4309	};
4310	return status === "failed"
4311	? {
4312	...common,
4313	status,
4314	failureItemOrdinal: 101,
4315	failure: {
4316	class: "provider_error",
4317	message: "Provider failed",
4318	code: null,
4319	retryable: null,
4320	},
4321	}
4322	: { ...common, status, failure: null };
4323	}
4324
4325	function runBackgroundItemScenario(
4326	key: string,
4327	makeEvents: (ids: BackgroundScenarioIds) => ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>,
4328	options?: {
4329	readonly keepEventStreamOpen?: boolean;
4330	readonly loadInheritedBackgroundTurnItems?: () => Effect.Effect<
4331	ReadonlyArray<{ readonly id: TurnItemId; readonly runId: RunId }>
4332	>;
4333	readonly onSubscribe?: Effect.Effect<void>;
4334	},
4335	) {
4336	return Effect.gen(function* () {
4337	const ids = backgroundScenarioIds(key);
4338	const providerInstanceId = ProviderInstanceId.make("codex");
4339	const observed = yield* Ref.make<ReadonlyArray<string>>([]);
4340	const ingestionDone = yield* Deferred.make<void>();
4341	const layerTest = RunExecutionService.layer.pipe(
4342	Layer.provide(
4343	Layer.mergeAll(
4344	McpAppModelContext.layerEmpty,
4345	Layer.mock(CheckpointService.CheckpointServiceV2)({ captureBaseline: () => Effect.void }),
4346	Layer.mock(EventSink.EventSinkV2)({
4347	write: () => Effect.succeed([]),
4348	writeWithEffects: (input) =>
4349	Effect.gen(function* () {
4350	if (
4351	input.events.some(
4352	(event) => event.type === "run.updated" && event.runId === ids.runId,
4353	)
4354	) {
4355	yield* Ref.update(observed, (current) => [...current, "root-finalized"]);
4356	}
4357	return [];
4358	}),
4359	writeIfRunCurrent: () => Effect.succeed({ committed: true, storedEvents: [] }),
4360	}),
4361	IdAllocator.layer,
4362	Layer.mock(ProviderEventIngestor.ProviderEventIngestorV2)({
4363	ingestNormalized: (input) =>
4364	Effect.gen(function* () {
4365	const event = input.event;
4366	if (event.type === "turn_item.updated") {
4367	yield* Ref.update(observed, (current) => [
4368	...current,
4369	`turn_item:${event.turnItem.status}`,
4370	]);
4371	}
4372	if (event.type === "subagent.updated") {
4373	yield* Ref.update(observed, (current) => [
4374	...current,
4375	`subagent:${event.subagent.status}`,
4376	]);
4377	}
4378	return [];
4379	}),
4380	}),
4381	ServerSettings.layerTest(),
4382	),
4383	),
4384	);
4385
4386	yield* Effect.gen(function* () {
4387	const runExecution = yield* RunExecutionService.RunExecutionServiceV2;
4388	yield* runExecution.startRootRun({
4389	commandId: CommandId.make(`command:${key}`),
4390	appThread: { id: ids.threadId } as OrchestrationV2AppThread,
4391	providerSessionId: ProviderSessionId.make(`session:${key}`),
4392	session: {
4393	events: Stream.empty,
4394	subscribeEvents: Effect.gen(function* () {
4395	yield* options?.onSubscribe ?? Effect.void;
4396	const events = Stream.fromIterable(makeEvents(ids));
4397	return {
4398	events:
4399	options?.keepEventStreamOpen === true
4400	? events.pipe(Stream.concat(Stream.never))
4401	: events,
4402	close: Deferred.succeed(ingestionDone, undefined),
4403	};
4404	}),
4405	startTurn: () => Effect.void,
4406	} as unknown as ProviderAdapter.ProviderAdapterV2SessionRuntime,
4407	run: {
4408	id: ids.runId,
4409	threadId: ids.threadId,
4410	ordinal: 1,
4411	providerInstanceId,
4412	} as OrchestrationV2Run,
4413	rootNode: { id: ids.rootNodeId } as OrchestrationV2ExecutionNode,
4414	checkpointScope: {
4415	id: CheckpointScopeId.make(`checkpoint-scope:${key}`),
4416	} as OrchestrationV2CheckpointScope,
4417	providerThread: {
4418	id: ids.providerThreadId,
4419	driver,
4420	} as OrchestrationV2ProviderThread,
4421	attempt: {
4422	id: ids.attemptId,
4423	providerTurnId: ids.rootProviderTurnId,
4424	} as OrchestrationV2RunAttempt,
4425	attemptId: ids.attemptId,
4426	...(options?.loadInheritedBackgroundTurnItems === undefined
4427	? {}
4428	: {
4429	loadInheritedBackgroundTurnItems: options.loadInheritedBackgroundTurnItems,
4430	}),
4431	providerTurnOrdinal: 1,
4432	message: {
4433	messageId: MessageId.make(`message:${key}:user`),
4434	text: "Start a background item and finish.",
4435	attachments: [],
4436	createdBy: "user",
4437	creationSource: "web",
4438	},
4439	modelSelection: { instanceId: providerInstanceId, model: "gpt-5.4" },
4440	runtimePolicy: {
4441	runtimeMode: "full-access",
4442	interactionMode: "default",
4443	cwd: process.cwd(),
4444	approvalPolicy: "never",
4445	sandboxPolicy: {
4446	type: "readOnly",
4447	access: { type: "fullAccess" },
4448	networkAccess: false,
4449	},
4450	},
4451	});
4452	}).pipe(Effect.provide(layerTest));
4453
4454	const closed = yield* Deferred.await(ingestionDone).pipe(Effect.timeoutOption("2 seconds"));
4455	assert.isTrue(Option.isSome(closed), "event ingestion fiber did not finish");
4456	return yield* Ref.get(observed);
4457	});
4458	}
4459
4460	it.effect("releases ingestion after idle subagent rows and items settle", () =>
4461	Effect.gen(function* () {
4462	const observed = yield* runBackgroundItemScenario(
4463	"subagent-idle",
4464	(ids) => [
4465	subagentEvent(ids, "running"),
4466	backgroundTurnItemEvent(ids, "subagent", "running", 1),
4467	subagentEvent(ids, "idle"),
4468	backgroundTurnItemEvent(ids, "subagent", "idle", 2),
4469	rootTerminalEvent(ids, "completed"),
4470	],
4471	{ keepEventStreamOpen: true },
4472	);
4473	assert.deepEqual(observed, [
4474	"subagent:running",
4475	"turn_item:running",
4476	"subagent:idle",
4477	"turn_item:idle",
4478	"root-finalized",
4479	]);
4480	}),
4481	);