clover's git

the-knife

1	import * as ScriptProgress from "@t3tools/provider-core/server/ScriptProgress";
2	import * as NetAddress from "effect/net/NetAddress";
3	import * as NodeServices from "@effect/platform-node/NodeServices";
4	import { assert, it } from "@effect/vitest";
5	import { AgentScope } from "@t3tools/shared/AgentScope";
6	import {
7	EnvironmentId,
8	type ModelSelection,
9	type OrchestrationV2AppThread,
10	type OrchestrationV2DomainEvent,
11	type OrchestrationV2ProviderCapabilities,
12	type OrchestrationV2ProviderSession,
13	type OrchestrationV2ProviderThread,
14	type Project,
15	ProjectId,
16	ProviderDriverKind,
17	ProviderInstanceId,
18	type ProviderSessionId,
19	ThreadId,
20	} from "@t3tools/contracts";
21	import * as Cause from "effect/Cause";
22	import * as DateTime from "effect/DateTime";
23	import * as Deferred from "effect/Deferred";
24	import * as Effect from "effect/Effect";
25	import * as Exit from "effect/Exit";
26	import * as Fiber from "effect/Fiber";
27	import * as FileSystem from "effect/FileSystem";
28	import * as Layer from "effect/Layer";
29	import * as Metric from "effect/Metric";
30	import * as Option from "effect/Option";
31	import * as Queue from "effect/Queue";
32	import * as Ref from "effect/Ref";
33	import * as Stream from "effect/Stream";
34	import { TestClock } from "effect/testing";
35	import { HttpServer } from "effect/http";
36
37	import { ProviderWorkspaceMissingError } from "../provider/Errors.ts";
38	import * as ServerEnvironment from "../environment/ServerEnvironment.ts";
39	import * as ProjectService from "../project/ProjectService.ts";
40	import * as McpProviderSession from "@t3tools/provider-core/server/mcpSession";
41	import * as McpProviderSessions from "@t3tools/provider-core/server/McpProviderSessions";
42	import * as McpSessionRegistry from "../mcp/McpSessionRegistry.ts";
43	import * as SqlitePersistence from "../persistence/Sqlite.ts";
44	import * as ServerSettings from "../serverSettings.ts";
45	import { CodexProviderCapabilitiesV2 } from "./Adapters/CodexAdapterV2.ts";
46	import * as EventSink from "./EventSink.ts";
47	import * as EventStore from "./EventStore.ts";
48	import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
49	import * as ProjectionStore from "./ProjectionStore.ts";
50	import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts";
51	import * as ProviderEventIngestor from "./ProviderEventIngestor.ts";
52	import * as ThreadCommandExecutor from "./ThreadCommandExecutor.ts";
53	import * as ProviderSessionManager from "./ProviderSessionManager.ts";
54	import * as ProviderAdapter from "@t3tools/provider-core/server/ProviderAdapter";
55
56	const layerTestDatabase = SqlitePersistence.layerMemory;
57	const layerTestStores = Layer.merge(EventStore.layer, ProjectionStore.layer).pipe(
58	Layer.provide(layerTestDatabase),
59	);
60	const layerTestEventSink = EventSink.layer.pipe(
61	Layer.provide(Layer.mergeAll(layerTestStores, layerTestDatabase)),
62	);
63	const layerFailingReleaseEventSink = Layer.effect(
64	EventSink.EventSinkV2,
65	Effect.gen(function* () {
66	const delegate = yield* EventSink.EventSinkV2;
67	return EventSink.EventSinkV2.of({
68	...delegate,
69	write: (input) =>
70	input.events.some(
71	(event) =>
72	event.type === "provider-session.updated" &&
73	(event.payload.status === "stopped" || event.payload.status === "error"),
74	)
75	? Effect.fail(new EventSink.EventSinkWriteError({ eventCount: input.events.length }))
76	: delegate.write(input),
77	});
78	}),
79	).pipe(Layer.provide(layerTestEventSink));
80
81	interface FlakyReleaseWrites {
82	/** Which release writes fail right now. */
83	readonly failing: Ref.Ref<"none" | "session" | "session-and-requests">;
84	/** Receives one item per failed write. */
85	readonly failures: Queue.Queue<void>;
86	/** Holds runtime request writes: completes `paused`, then waits for `resume`. */
87	readonly pauseRequestWrites?: {
88	readonly paused: Deferred.Deferred<void>;
89	readonly resume: Deferred.Deferred<void>;
90	};
91	}
92
93	// Fails release writes with a defect, the way a failed SQL commit surfaces.
94	const layerFlakyReleaseEventSink = (flaky: FlakyReleaseWrites) =>
95	Layer.effect(
96	EventSink.EventSinkV2,
97	Effect.gen(function* () {
98	const delegate = yield* EventSink.EventSinkV2;
99	return EventSink.EventSinkV2.of({
100	...delegate,
101	write: (input) =>
102	Effect.gen(function* () {
103	const failing = yield* Ref.get(flaky.failing);
104	const fails = input.events.some(
105	(event) =>
106	(failing !== "none" &&
107	event.type === "provider-session.updated" &&
108	(event.payload.status === "stopped" || event.payload.status === "error")) ||
109	(failing === "session-and-requests" && event.type === "runtime-request.updated"),
110	);
111	const pause = flaky.pauseRequestWrites;
112	if (
113	pause !== undefined &&
114	input.events.some((event) => event.type === "runtime-request.updated")
115	) {
116	yield* Deferred.succeed(pause.paused, undefined);
117	yield* Deferred.await(pause.resume);
118	}
119	if (!fails) return yield* delegate.write(input);
120	yield* Queue.offer(flaky.failures, undefined);
121	return yield* Effect.die(new Error("simulated commit failure"));
122	}),
123	});
124	}),
125	).pipe(Layer.provide(layerTestEventSink));
126
127	// Once armed, holds the next attach write until the writer is interrupted.
128	const layerPausingAttachEventSink = (pause: {
129	readonly armed: Ref.Ref<boolean>;
130	readonly paused: Deferred.Deferred<void>;
131	}) =>
132	Layer.effect(
133	EventSink.EventSinkV2,
134	Effect.gen(function* () {
135	const delegate = yield* EventSink.EventSinkV2;
136	return EventSink.EventSinkV2.of({
137	...delegate,
138	write: (input) =>
139	Effect.gen(function* () {
140	const attach = input.events.some((event) => event.type === "provider-session.attached");
141	if (attach && (yield* Ref.getAndSet(pause.armed, false))) {
142	yield* Deferred.succeed(pause.paused, undefined);
143	return yield* Effect.never;
144	}
145	return yield* delegate.write(input);
146	}),
147	});
148	}),
149	).pipe(Layer.provide(layerTestEventSink));
150
151	const layerBeforeUnchangedWriteEventSink = (before: Ref.Ref<Effect.Effect<void>>) =>
152	Layer.effect(
153	EventSink.EventSinkV2,
154	Effect.gen(function* () {
155	const delegate = yield* EventSink.EventSinkV2;
156	return EventSink.EventSinkV2.of({
157	...delegate,
158	writeIfProviderThreadUnchanged: (input) =>
159	Ref.getAndSet(before, Effect.void).pipe(
160	Effect.flatten,
161	Effect.andThen(delegate.writeIfProviderThreadUnchanged(input)),
162	),
163	});
164	}),
165	).pipe(Layer.provide(layerTestEventSink));
166
167	const CodexCapabilities: OrchestrationV2ProviderCapabilities = CodexProviderCapabilitiesV2;
168	const ExclusiveCapabilities: OrchestrationV2ProviderCapabilities = {
169	...CodexCapabilities,
170	sessions: {
171	...CodexCapabilities.sessions,
172	supportsMultipleProviderThreadsPerSession: false,
173	},
174	};
175
176	interface TestProviderRuntimeState {
177	readonly openCount: number;
178	readonly closeCount: number;
179	readonly interruptCount: number;
180	readonly resumeCount: number;
181	readonly unloadedNativeThreadIds: ReadonlyArray<string>;
182	readonly eventQueues: ReadonlyMap<
183	string,
184	Queue.Queue<ProviderAdapter.ProviderAdapterV2Event, Cause.Done>
185	>;
186	}
187
188	const emptyState: TestProviderRuntimeState = {
189	openCount: 0,
190	closeCount: 0,
191	interruptCount: 0,
192	resumeCount: 0,
193	unloadedNativeThreadIds: [],
194	eventQueues: new Map(),
195	};
196
197	const modelSelection = {
198	instanceId: ProviderInstanceId.make("codex"),
199	model: "gpt-5.4",
200	} satisfies ModelSelection;
201	const CODEX_DRIVER = ProviderDriverKind.make("codex");
202
203	const runtimePolicy = {
204	runtimeMode: "full-access",
205	interactionMode: "default",
206	cwd: process.cwd(),
207	} satisfies ProviderAdapter.ProviderAdapterV2RuntimePolicy;
208
209	function makeProviderSession(input: {
210	readonly providerSessionId: ProviderSessionId;
211	readonly now: DateTime.Utc;
212	readonly capabilities?: OrchestrationV2ProviderCapabilities;
213	}): OrchestrationV2ProviderSession {
214	return {
215	id: input.providerSessionId,
216	driver: CODEX_DRIVER,
217	providerInstanceId: modelSelection.instanceId,
218	status: "ready",
219	cwd: process.cwd(),
220	model: "gpt-5.4",
221	capabilities: input.capabilities ?? CodexCapabilities,
222	createdAt: input.now,
223	updatedAt: input.now,
224	lastError: null,
225	};
226	}
227
228	function makeThreadCreatedEvent(input: {
229	readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
230	readonly threadId: ThreadId;
231	readonly now: DateTime.Utc;
232	readonly projectId?: ProjectId;
233	}) {
234	return Effect.gen(function* () {
235	const projectId =
236	input.projectId ??
237	(yield* input.idAllocator.allocate.project({
238	fixtureName: "provider-session-manager",
239	}));
240	const providerThreadId = input.idAllocator.derive.providerThread({
241	driver: CODEX_DRIVER,
242	nativeThreadId: "native-thread",
243	});
244	const thread: OrchestrationV2AppThread = {
245	createdBy: "user",
246	creationSource: "web",
247	id: input.threadId,
248	projectId,
249	title: "Provider session manager",
250	providerInstanceId: modelSelection.instanceId,
251	modelSelection,
252	runtimeMode: "full-access",
253	interactionMode: "default",
254	branch: null,
255	worktreePath: null,
256	activeProviderThreadId: providerThreadId,
257	lineage: {
258	parentThreadId: null,
259	relationshipToParent: null,
260	rootThreadId: input.threadId,
261	},
262	forkedFrom: null,
263	createdAt: input.now,
264	updatedAt: input.now,
265	archivedAt: null,
266	settledOverride: null,
267	settledAt: null,
268	lastVisitedAt: null,
269	deletedAt: null,
270	};
271	return {
272	id: yield* input.idAllocator.allocate.event({ threadId: input.threadId }),
273	type: "thread.created" as const,
274	threadId: input.threadId,
275	occurredAt: input.now,
276	payload: thread,
277	};
278	});
279	}
280
281	function makeProviderThread(input: {
282	readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
283	readonly threadId: ThreadId;
284	readonly providerSessionId: ProviderSessionId;
285	readonly now: DateTime.Utc;
286	readonly nativeThreadId?: string;
287	}): OrchestrationV2ProviderThread {
288	const nativeThreadId = input.nativeThreadId ?? "native-thread";
289	return {
290	id: input.idAllocator.derive.providerThread({
291	driver: CODEX_DRIVER,
292	nativeThreadId,
293	}),
294	driver: CODEX_DRIVER,
295	providerInstanceId: modelSelection.instanceId,
296	providerSessionId: input.providerSessionId,
297	appThreadId: input.threadId,
298	ownerNodeId: null,
299	nativeThreadRef: {
300	driver: CODEX_DRIVER,
301	nativeId: nativeThreadId,
302	strength: "strong",
303	},
304	nativeConversationHeadRef: null,
305	status: "idle",
306	firstRunOrdinal: null,
307	lastRunOrdinal: null,
308	handoffIds: [],
309	forkedFrom: null,
310	createdAt: input.now,
311	updatedAt: input.now,
312	};
313	}
314
315	function unimplemented(detail: string) {
316	return Effect.fail(
317	new ProviderAdapter.ProviderAdapterProtocolError({
318	driver: CODEX_DRIVER,
319	detail,
320	}),
321	);
322	}
323
324	const makeProviderAdapter = Effect.fnUntraced(function* (
325	state: Ref.Ref<TestProviderRuntimeState>,
326	options: {
327	readonly failEventStream?: boolean;
328	readonly capabilities?: OrchestrationV2ProviderCapabilities;
329	readonly mcpConfigs?: Ref.Ref<
330	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
331	>;
332	readonly beforeOpen?: (input: {
333	readonly providerSessionId: ProviderSessionId;
334	readonly initialProviderItemIdentityVersion?: 2;
335	}) => Effect.Effect<void>;
336	readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
337	readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
338	/** False models a runtime whose background tasks can outlive its process. */
339	readonly backgroundTasksEndWithProcess?: boolean;
340	readonly hangSessionScopeClose?: boolean;
341	readonly startTurn?: Effect.Effect<void>;
342	readonly beforeUnload?: Effect.Effect<void>;
343	/** Registers the process's closeCount finalizer before `beforeOpen` runs. */
344	readonly spawnBeforeOpen?: boolean;
345	/** Completed when a hanging scope close reaches its wedged finalizer. */
346	readonly scopeCloseReached?: Deferred.Deferred<void>;
347	} = {},
348	) {
349	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
350	const countClose = Effect.addFinalizer(() =>
351	Ref.update(state, (current) => ({
352	...current,
353	closeCount: current.closeCount + 1,
354	})),
355	);
356	// Registered after countClose so it runs first on scope close, wedging the
357	// close before the closeCount finalizer, like a provider process that never
358	// yields its message stream.
359	const hangClose = Effect.addFinalizer(() =>
360	(options.scopeCloseReached === undefined
361	? Effect.void
362	: Deferred.succeed(options.scopeCloseReached, undefined)
363	).pipe(Effect.andThen(Effect.never)),
364	);
365	return ProviderAdapter.ProviderAdapterV2.of({
366	instanceId: ProviderInstanceId.make("codex"),
367	driver: CODEX_DRIVER,
368	getCapabilities: () => Effect.succeed(options.capabilities ?? CodexCapabilities),
369	planSelectionTransition: () => Effect.succeed({ type: "apply_on_next_turn" }),
370	openSession: (input) =>
371	Effect.gen(function* () {
372	if (options.spawnBeforeOpen === true) {
373	yield* countClose;
374	if (options.hangSessionScopeClose === true) yield* hangClose;
375	}
376	if (options.mcpConfigs !== undefined && options.spawnBeforeOpen === true) {
377	const mcpConfig = yield* mcpSessions.read(input.threadId);
378	yield* Ref.update(options.mcpConfigs, (configs) => [...configs, mcpConfig]);
379	}
380	if (options.beforeOpen !== undefined) {
381	yield* options.beforeOpen(input);
382	}
383	if (options.mcpConfigs !== undefined && options.spawnBeforeOpen !== true) {
384	const mcpConfig = yield* mcpSessions.read(input.threadId);
385	yield* Ref.update(options.mcpConfigs, (configs) => [...configs, mcpConfig]);
386	}
387	const now = yield* DateTime.now;
388	const events = yield* Queue.unbounded<ProviderAdapter.ProviderAdapterV2Event, Cause.Done>();
389	const session = makeProviderSession({
390	providerSessionId: input.providerSessionId,
391	now,
392	...(options.capabilities === undefined ? {} : { capabilities: options.capabilities }),
393	});
394	yield* Ref.update(state, (current) => {
395	const eventQueues = new Map(current.eventQueues);
396	eventQueues.set(String(input.providerSessionId), events);
397	return {
398	...current,
399	openCount: current.openCount + 1,
400	eventQueues,
401	};
402	});
403	if (options.spawnBeforeOpen !== true) {
404	yield* countClose;
405	if (options.hangSessionScopeClose === true) yield* hangClose;
406	}
407
408	return {
409	instanceId: ProviderInstanceId.make("codex"),
410	driver: CODEX_DRIVER,
411	providerSessionId: input.providerSessionId,
412	providerSession: session,
413	events: options.failEventStream
414	? Stream.fail(
415	new ProviderAdapter.ProviderAdapterEventStreamError({
416	driver: CODEX_DRIVER,
417	providerSessionId: input.providerSessionId,
418	cause: "process exited",
419	}),
420	)
421	: Stream.fromQueue(events),
422	...(options.backgroundTasksEndWithProcess === false
423	? {}
424	: { backgroundTasksEndWithProcess: true as const }),
425	...(options.hasPendingBackgroundWork === undefined
426	? {}
427	: { hasPendingBackgroundWork: options.hasPendingBackgroundWork }),
428	...(options.hasPendingBackgroundWorkForThread === undefined
429	? {}
430	: {
431	hasPendingBackgroundWorkForThread: () => options.hasPendingBackgroundWorkForThread!,
432	}),
433	ensureThread: () => unimplemented("ensureThread unused in test"),
434	resumeThread: (threadInput) =>
435	Ref.update(state, (current) => ({
436	...current,
437	resumeCount: current.resumeCount + 1,
438	})).pipe(Effect.as(threadInput.providerThread)),
439	startTurn: () => options.startTurn ?? Effect.void,
440	steerTurn: () => Effect.void,
441	interruptTurn: () =>
442	Ref.update(state, (current) => ({
443	...current,
444	interruptCount: current.interruptCount + 1,
445	})),
446	unloadThread: ({ providerThread }) =>
447	(options.beforeUnload ?? Effect.void).pipe(
448	Effect.andThen(
449	Ref.update(state, (current) => ({
450	...current,
451	unloadedNativeThreadIds: [
452	...current.unloadedNativeThreadIds,
453	providerThread.nativeThreadRef?.nativeId ?? "",
454	],
455	})),
456	),
457	),
458	respondToRuntimeRequest: () => Effect.void,
459	readThreadSnapshot: () => unimplemented("readThreadSnapshot unused in test"),
460	rollbackThread: () => unimplemented("rollbackThread unused in test"),
461	forkThread: () => unimplemented("forkThread unused in test"),
462	} satisfies ProviderAdapter.ProviderAdapterV2SessionRuntime;
463	}),
464	});
465	});
466
467	function layerTest(input: {
468	readonly state: Ref.Ref<TestProviderRuntimeState>;
469	readonly idleTimeoutMs: number;
470	readonly maxIdlePinMs?: number;
471	readonly failEventStream?: boolean;
472	readonly capabilities?: OrchestrationV2ProviderCapabilities;
473	readonly mcpConfigs?: Ref.Ref<
474	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
475	>;
476	readonly beforeOpen?: (input: {
477	readonly providerSessionId: ProviderSessionId;
478	readonly initialProviderItemIdentityVersion?: 2;
479	}) => Effect.Effect<void>;
480	readonly failReleaseEventWrites?: boolean;
481	readonly flakyReleaseWrites?: FlakyReleaseWrites;
482	readonly pauseAttachWrite?: Parameters<typeof layerPausingAttachEventSink>[0];
483	/** Runs once, right before the next provider-thread compare-and-set write. */
484	readonly beforeUnchangedWrite?: Ref.Ref<Effect.Effect<void>>;
485	/** Once armed, holds the next credential lookup until it is interrupted. */
486	readonly pauseResolve?: {
487	readonly armed: Ref.Ref<boolean>;
488	readonly paused: Deferred.Deferred<void>;
489	};
490	readonly hasPendingBackgroundWork?: Effect.Effect<boolean>;
491	readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean>;
492	readonly backgroundTasksEndWithProcess?: boolean;
493	readonly hangSessionScopeClose?: boolean;
494	readonly startTurn?: Effect.Effect<void>;
495	readonly beforeUnload?: Effect.Effect<void>;
496	readonly spawnBeforeOpen?: boolean;
497	readonly scopeCloseReached?: Deferred.Deferred<void>;
498	readonly serverSettingsLayer?: ReturnType<typeof ServerSettings.layerTest>;
499	readonly projectServiceLayer?: Layer.Layer<ProjectService.ProjectService>;
500	}) {
501	const layerConfiguredEventSink =
502	input.flakyReleaseWrites !== undefined
503	? layerFlakyReleaseEventSink(input.flakyReleaseWrites)
504	: input.pauseAttachWrite !== undefined
505	? layerPausingAttachEventSink(input.pauseAttachWrite)
506	: input.beforeUnchangedWrite !== undefined
507	? layerBeforeUnchangedWriteEventSink(input.beforeUnchangedWrite)
508	: input.failReleaseEventWrites
509	? layerFailingReleaseEventSink
510	: layerTestEventSink;
511	const layerRegistry = Layer.unwrap(
512	makeProviderAdapter(input.state, {
513	failEventStream: input.failEventStream ?? false,
514	...(input.capabilities === undefined ? {} : { capabilities: input.capabilities }),
515	...(input.mcpConfigs === undefined ? {} : { mcpConfigs: input.mcpConfigs }),
516	...(input.beforeOpen === undefined ? {} : { beforeOpen: input.beforeOpen }),
517	...(input.hasPendingBackgroundWork === undefined
518	? {}
519	: { hasPendingBackgroundWork: input.hasPendingBackgroundWork }),
520	...(input.hasPendingBackgroundWorkForThread === undefined
521	? {}
522	: { hasPendingBackgroundWorkForThread: input.hasPendingBackgroundWorkForThread }),
523	...(input.backgroundTasksEndWithProcess === undefined
524	? {}
525	: { backgroundTasksEndWithProcess: input.backgroundTasksEndWithProcess }),
526	...(input.hangSessionScopeClose === undefined
527	? {}
528	: { hangSessionScopeClose: input.hangSessionScopeClose }),
529	...(input.startTurn === undefined ? {} : { startTurn: input.startTurn }),
530	...(input.beforeUnload === undefined ? {} : { beforeUnload: input.beforeUnload }),
531	...(input.spawnBeforeOpen === undefined ? {} : { spawnBeforeOpen: input.spawnBeforeOpen }),
532	...(input.scopeCloseReached === undefined
533	? {}
534	: { scopeCloseReached: input.scopeCloseReached }),
535	}).pipe(Effect.map(ProviderAdapterRegistry.layerSingle)),
536	);
537	const layerConfiguredMcpRegistry =
538	input.pauseResolve === undefined
539	? layerTestMcpRegistry
540	: layerPausingMcpRegistry(input.pauseResolve);
541	const layerProviderEventIngestorTest = ProviderEventIngestor.layer.pipe(
542	Layer.provide(
543	Layer.mergeAll(
544	layerConfiguredEventSink,
545	IdAllocator.layer,
546	layerTestStores,
547	ThreadCommandExecutor.layer,
548	),
549	),
550	);
551	return Layer.mergeAll(
552	layerTestStores,
553	layerConfiguredEventSink,
554	IdAllocator.layer,
555	layerConfiguredMcpRegistry,
556	McpProviderSessions.layer,
557	ScriptProgress.layerUnavailable,
558	ProviderSessionManager.layerWithOptions({
559	idleTimeoutMs: input.idleTimeoutMs,
560	...(input.maxIdlePinMs === undefined ? {} : { maxIdlePinMs: input.maxIdlePinMs }),
561	}).pipe(
562	Layer.provide(
563	Layer.mergeAll(
564	layerRegistry.pipe(
565	Layer.provideMerge(
566	Layer.merge(McpProviderSessions.layer, ScriptProgress.layerUnavailable),
567	),
568	),
569	layerConfiguredEventSink,
570	IdAllocator.layer,
571	layerProviderEventIngestorTest,
572	layerConfiguredMcpRegistry,
573	layerTestStores,
574	...(input.serverSettingsLayer === undefined ? [] : [input.serverSettingsLayer]),
575	...(input.projectServiceLayer === undefined ? [] : [input.projectServiceLayer]),
576	),
577	),
578	),
579	).pipe(Layer.provide(NodeServices.layer));
580	}
581
582	const fakeHttpServer = HttpServer.HttpServer.of({
583	address: NetAddress.inetAddressFromIpStringUnsafe("127.0.0.1", 43123),
584	serve: (() => Effect.void) as HttpServer.HttpServer["Service"]["serve"],
585	});
586
587	const fakeEnvironment = ServerEnvironment.ServerEnvironment.of({
588	getEnvironmentId: Effect.succeed(EnvironmentId.make("environment-provider-session-manager")),
589	getDescriptor: Effect.die("unused"),
590	});
591
592	const layerTestMcpRegistry = Layer.effect(
593	McpSessionRegistry.McpSessionRegistry,
594	McpSessionRegistry.__testing.make(),
595	).pipe(
596	Layer.provide(Layer.succeed(HttpServer.HttpServer, fakeHttpServer)),
597	Layer.provide(Layer.succeed(ServerEnvironment.ServerEnvironment, fakeEnvironment)),
598	Layer.provide(NodeServices.layer),
599	);
600
601	const layerPausingMcpRegistry = (pause: {
602	readonly armed: Ref.Ref<boolean>;
603	readonly paused: Deferred.Deferred<void>;
604	}) =>
605	Layer.effect(
606	McpSessionRegistry.McpSessionRegistry,
607	Effect.gen(function* () {
608	const delegate = yield* McpSessionRegistry.McpSessionRegistry;
609	return McpSessionRegistry.McpSessionRegistry.of({
610	...delegate,
611	resolve: (rawToken) =>
612	Ref.getAndSet(pause.armed, false).pipe(
613	Effect.flatMap((armed) =>
614	armed
615	? Deferred.succeed(pause.paused, undefined).pipe(Effect.andThen(Effect.never))
616	: delegate.resolve(rawToken),
617	),
618	),
619	});
620	}),
621	).pipe(Layer.provide(layerTestMcpRegistry));
622
623	function makeBrowserAccessProject(projectId: ProjectId): Project {
624	return {
625	id: projectId,
626	title: "Browser access project",
627	workspaceRoot: process.cwd(),
628	repositoryIdentity: null,
629	faviconPath: null,
630	projectIcon: null,
631	defaultModelSelection: null,
632	defaultThreadEnvMode: null,
633	autoPull: false,
634	scripts: [],
635	createdAt: "2026-01-01T00:00:00.000Z",
636	updatedAt: "2026-01-01T00:00:00.000Z",
637	deletedAt: null,
638	};
639	}
640
641	function runBrowserAccessScenario(input: {
642	readonly enableAgentBrowserAccess: boolean;
643	readonly projectOverride: boolean;
644	readonly deviceOverride?: boolean;
645	readonly createThread?: boolean;
646	readonly projectExists?: boolean;
647	}) {
648	return Effect.gen(function* () {
649	const state = yield* Ref.make(emptyState);
650	const mcpConfigs = yield* Ref.make<
651	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
652	>([]);
653	const projectId = ProjectId.make("project-provider-session-manager-browser-access");
654	const threadId = ThreadId.make("thread-provider-session-manager-browser-access");
655	const layerProjectService = Layer.mock(ProjectService.ProjectService)({
656	getById: (requestedProjectId) =>
657	Effect.succeed(
658	input.projectExists === false
659	? Option.none()
660	: Option.some(makeBrowserAccessProject(requestedProjectId)),
661	),
662	});
663
664	yield* Effect.gen(function* () {
665	const eventSink = yield* EventSink.EventSinkV2;
666	const idAllocator = yield* IdAllocator.IdAllocatorV2;
667	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
668	const now = yield* DateTime.now;
669	const providerSessionId = yield* idAllocator.allocate.providerSession({
670	providerInstanceId: modelSelection.instanceId,
671	threadId,
672	});
673	if (input.createThread !== false) {
674	yield* eventSink.write({
675	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now, projectId })],
676	});
677	}
678	yield* manager
679	.open({ threadId, providerSessionId, modelSelection, runtimePolicy })
680	.pipe(Effect.ignore);
681	}).pipe(
682	Effect.provide(
683	layerTest({
684	state,
685	idleTimeoutMs: 1_000,
686	mcpConfigs,
687	projectServiceLayer: layerProjectService,
688	serverSettingsLayer: ServerSettings.layerTest({
689	enableAgentBrowserAccess: input.enableAgentBrowserAccess,
690	projectSettingsOverrides: {
691	[projectId]: {
692	enableAgentBrowserAccess: input.projectOverride,
693	...(input.deviceOverride === undefined
694	? {}
695	: { enableAgentDeviceAccess: input.deviceOverride }),
696	},
697	},
698	}),
699	}),
700	),
701	);
702
703	return (yield* Ref.get(mcpConfigs))[0];
704	});
705	}
706
707	function makePendingRuntimeRequestEvents(input: {
708	readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
709	readonly threadId: ThreadId;
710	readonly providerSessionId: ProviderSessionId;
711	readonly providerThread: OrchestrationV2ProviderThread;
712	readonly now: DateTime.Utc;
713	}) {
714	return Effect.gen(function* () {
715	const requestId = yield* input.idAllocator.allocate.runtimeRequest({
716	driver: CODEX_DRIVER,
717	nativeRequestId: "pending-approval",
718	});
719	const nodeId = input.idAllocator.derive.approvalNode({ requestId });
720	const node = {
721	id: nodeId,
722	threadId: input.threadId,
723	runId: null,
724	parentNodeId: null,
725	rootNodeId: nodeId,
726	kind: "approval_request" as const,
727	status: "waiting" as const,
728	countsForRun: false,
729	providerThreadId: input.providerThread.id,
730	providerTurnId: null,
731	nativeItemRef: null,
732	runtimeRequestId: requestId,
733	checkpointScopeId: null,
734	startedAt: input.now,
735	completedAt: null,
736	};
737	const request = {
738	id: requestId,
739	nodeId,
740	providerTurnId: null,
741	nativeRequestRef: {
742	driver: CODEX_DRIVER,
743	nativeId: "pending-approval",
744	strength: "strong" as const,
745	},
746	kind: "command" as const,
747	status: "pending" as const,
748	responseCapability: {
749	type: "live" as const,
750	providerSessionId: input.providerSessionId,
751	},
752	createdAt: input.now,
753	resolvedAt: null,
754	};
755	const turnItem = {
756	id: input.idAllocator.derive.approvalTurnItem({ requestId }),
757	threadId: input.threadId,
758	runId: null,
759	nodeId,
760	providerThreadId: input.providerThread.id,
761	providerTurnId: null,
762	nativeItemRef: null,
763	parentItemId: null,
764	ordinal: 1,
765	status: "waiting" as const,
766	title: null,
767	startedAt: input.now,
768	completedAt: null,
769	updatedAt: input.now,
770	type: "approval_request" as const,
771	requestId,
772	requestKind: "command" as const,
773	};
774	const events = [
775	{
776	id: yield* input.idAllocator.allocate.event({
777	threadId: input.threadId,
778	providerSessionId: input.providerSessionId,
779	}),
780	type: "node.updated" as const,
781	threadId: input.threadId,
782	nodeId,
783	driver: CODEX_DRIVER,
784	occurredAt: input.now,
785	payload: node,
786	},
787	{
788	id: yield* input.idAllocator.allocate.event({
789	threadId: input.threadId,
790	providerSessionId: input.providerSessionId,
791	}),
792	type: "runtime-request.updated" as const,
793	threadId: input.threadId,
794	nodeId,
795	driver: CODEX_DRIVER,
796	occurredAt: input.now,
797	payload: request,
798	},
799	{
800	id: yield* input.idAllocator.allocate.event({
801	threadId: input.threadId,
802	providerSessionId: input.providerSessionId,
803	}),
804	type: "turn-item.updated" as const,
805	threadId: input.threadId,
806	nodeId,
807	driver: CODEX_DRIVER,
808	occurredAt: input.now,
809	payload: turnItem,
810	},
811	] satisfies ReadonlyArray<OrchestrationV2DomainEvent>;
812	const providerEvents = [
813	{
814	type: "runtime_request.updated" as const,
815	driver: CODEX_DRIVER,
816	threadId: input.threadId,
817	runtimeRequest: request,
818	},
819	{
820	type: "node.updated" as const,
821	driver: CODEX_DRIVER,
822	node,
823	},
824	{
825	type: "turn_item.updated" as const,
826	driver: CODEX_DRIVER,
827	turnItem,
828	},
829	] satisfies ReadonlyArray<ProviderAdapter.ProviderAdapterV2Event>;
830	return { events, providerEvents, requestId, nodeId };
831	});
832	}
833
834	it.effect("ProviderSessionManagerV2 opens independent sessions concurrently", () =>
835	Effect.gen(function* () {
836	const state = yield* Ref.make(emptyState);
837	const openStartedCount = yield* Ref.make(0);
838	const firstOpenStarted = yield* Deferred.make<void>();
839	const secondOpenStarted = yield* Deferred.make<void>();
840	const releaseOpens = yield* Deferred.make<void>();
841	const beforeOpen = () =>
842	Effect.gen(function* () {
843	const openNumber = yield* Ref.modify(openStartedCount, (count) => [count + 1, count + 1]);
844	yield* Deferred.succeed(openNumber === 1 ? firstOpenStarted : secondOpenStarted, undefined);
845	yield* Deferred.await(releaseOpens);
846	});
847
848	const effect = Effect.gen(function* () {
849	const eventSink = yield* EventSink.EventSinkV2;
850	const idAllocator = yield* IdAllocator.IdAllocatorV2;
851	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
852	const now = yield* DateTime.now;
853	const firstThreadId = ThreadId.make("thread-provider-session-manager-concurrent-a");
854	const secondThreadId = ThreadId.make("thread-provider-session-manager-concurrent-b");
855	const firstProviderSessionId = yield* idAllocator.allocate.providerSession({
856	providerInstanceId: modelSelection.instanceId,
857	threadId: firstThreadId,
858	});
859	const secondProviderSessionId = yield* idAllocator.allocate.providerSession({
860	providerInstanceId: modelSelection.instanceId,
861	threadId: secondThreadId,
862	});
863
864	yield* eventSink.write({
865	events: [
866	yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
867	yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
868	],
869	});
870	const firstFiber = yield* manager
871	.open({
872	threadId: firstThreadId,
873	providerSessionId: firstProviderSessionId,
874	modelSelection,
875	runtimePolicy,
876	})
877	.pipe(Effect.forkScoped);
878	yield* Deferred.await(firstOpenStarted);
879	const secondFiber = yield* manager
880	.open({
881	threadId: secondThreadId,
882	providerSessionId: secondProviderSessionId,
883	modelSelection,
884	runtimePolicy,
885	})
886	.pipe(Effect.forkScoped);
887
888	yield* Deferred.await(secondOpenStarted);
889	assert.equal(yield* Ref.get(openStartedCount), 2);
890	yield* Deferred.succeed(releaseOpens, undefined);
891	const [firstRuntime, secondRuntime] = yield* Effect.all([
892	Fiber.join(firstFiber),
893	Fiber.join(secondFiber),
894	]);
895	assert.notStrictEqual(firstRuntime, secondRuntime);
896	assert.equal((yield* Ref.get(state)).openCount, 2);
897	});
898
899	yield* effect.pipe(
900	Effect.provide(
901	layerTest({
902	state,
903	idleTimeoutMs: 60_000,
904	beforeOpen,
905	}),
906	),
907	);
908	}),
909	);
910
911	it.effect("ProviderSessionManagerV2 closes every live session for a provider instance", () =>
912	Effect.gen(function* () {
913	const state = yield* Ref.make(emptyState);
914	const effect = Effect.gen(function* () {
915	const eventSink = yield* EventSink.EventSinkV2;
916	const idAllocator = yield* IdAllocator.IdAllocatorV2;
917	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
918	const now = yield* DateTime.now;
919	const firstThreadId = ThreadId.make("thread-provider-session-manager-logout-a");
920	const secondThreadId = ThreadId.make("thread-provider-session-manager-logout-b");
921	const firstProviderSessionId = yield* idAllocator.allocate.providerSession({
922	providerInstanceId: modelSelection.instanceId,
923	threadId: firstThreadId,
924	});
925	const secondProviderSessionId = yield* idAllocator.allocate.providerSession({
926	providerInstanceId: modelSelection.instanceId,
927	threadId: secondThreadId,
928	});
929
930	yield* eventSink.write({
931	events: [
932	yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
933	yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
934	],
935	});
936	yield* manager.open({
937	threadId: firstThreadId,
938	providerSessionId: firstProviderSessionId,
939	modelSelection,
940	runtimePolicy,
941	});
942	yield* manager.open({
943	threadId: secondThreadId,
944	providerSessionId: secondProviderSessionId,
945	modelSelection,
946	runtimePolicy,
947	});
948
949	yield* manager.closeInstance(modelSelection.instanceId);
950
951	assert.isTrue(Option.isNone(yield* manager.get(firstProviderSessionId)));
952	assert.isTrue(Option.isNone(yield* manager.get(secondProviderSessionId)));
953	assert.equal((yield* Ref.get(state)).closeCount, 2);
954	});
955
956	yield* effect.pipe(
957	Effect.provide(
958	layerTest({
959	state,
960	idleTimeoutMs: 60_000,
961	}),
962	),
963	);
964	}),
965	);
966
967	it.effect("ProviderSessionManagerV2 records provider session and turn metrics", () =>
968	Effect.gen(function* () {
969	const state = yield* Ref.make(emptyState);
970	const effect = Effect.gen(function* () {
971	const eventSink = yield* EventSink.EventSinkV2;
972	const idAllocator = yield* IdAllocator.IdAllocatorV2;
973	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
974	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
975	const now = yield* DateTime.now;
976	const threadId = ThreadId.make("thread-provider-session-manager-metrics");
977	const providerSessionId = yield* idAllocator.allocate.providerSession({
978	providerInstanceId: modelSelection.instanceId,
979	threadId,
980	});
981	const providerThread = makeProviderThread({ idAllocator, threadId, providerSessionId, now });
982	const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
983	yield* eventSink.write({
984	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
985	});
986
987	const runtime = yield* manager.open({
988	threadId,
989	providerSessionId,
990	modelSelection,
991	runtimePolicy,
992	});
993	yield* runtime.startTurn({
994	appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
995	threadId,
996	runId,
997	runOrdinal: 1,
998	providerTurnOrdinal: 1,
999	attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
1000	rootNodeId: idAllocator.derive.rootNode({ runId }),
1001	providerThread,
1002	message: {
1003	createdBy: "user",
1004	creationSource: "web",
1005	messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
1006	text: "hello",
1007	attachments: [],
1008	},
1009	modelSelection,
1010	runtimePolicy,
1011	});
1012	yield* runtime.interruptTurn({
1013	providerThread,
1014	providerTurnId: idAllocator.derive.providerTurn({
1015	driver: CODEX_DRIVER,
1016	nativeTurnId: "native-turn",
1017	}),
1018	});
1019	yield* manager.close(providerSessionId);
1020
1021	const snapshots = yield* Metric.snapshot;
1022	const has = (id: string, attributes: Readonly<Record<string, string>>) =>
1023	snapshots.some(
1024	(snapshot) =>
1025	snapshot.id === id &&
1026	Object.entries(attributes).every(
1027	([key, value]) => snapshot.attributes?.[key] === value,
1028	),
1029	);
1030	assert.isTrue(
1031	has("t3_provider_sessions_total", {
1032	provider: "codex",
1033	operation: "open",
1034	outcome: "success",
1035	}),
1036	);
1037	assert.isTrue(
1038	has("t3_provider_sessions_total", {
1039	provider: "codex",
1040	operation: "release",
1041	reason: "manual_shutdown",
1042	outcome: "success",
1043	}),
1044	);
1045	assert.isTrue(
1046	has("t3_provider_turns_total", {
1047	provider: "codex",
1048	operation: "send",
1049	modelFamily: "gpt",
1050	outcome: "success",
1051	}),
1052	);
1053	assert.isTrue(has("t3_provider_turn_duration", { provider: "codex", operation: "send" }));
1054	assert.isTrue(
1055	has("t3_provider_turns_total", {
1056	provider: "codex",
1057	operation: "interrupt",
1058	outcome: "success",
1059	}),
1060	);
1061	});
1062
1063	yield* effect.pipe(
1064	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })),
1065	// A private registry keeps other tests' provider metrics out of the assertions.
1066	Effect.provideService(Metric.MetricRegistry, new Map()),
1067	);
1068	}),
1069	);
1070
1071	it.effect("ProviderSessionManagerV2 opens a duplicate session only once", () =>
1072	Effect.gen(function* () {
1073	const state = yield* Ref.make(emptyState);
1074	const openStartedCount = yield* Ref.make(0);
1075	const firstOpenStarted = yield* Deferred.make<void>();
1076	const releaseOpen = yield* Deferred.make<void>();
1077	const beforeOpen = () =>
1078	Ref.updateAndGet(openStartedCount, (count) => count + 1).pipe(
1079	Effect.tap(() => Deferred.succeed(firstOpenStarted, undefined)),
1080	Effect.andThen(Deferred.await(releaseOpen)),
1081	);
1082
1083	const effect = Effect.gen(function* () {
1084	const eventSink = yield* EventSink.EventSinkV2;
1085	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1086	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1087	const now = yield* DateTime.now;
1088	const threadId = ThreadId.make("thread-provider-session-manager-single-flight");
1089	const providerSessionId = yield* idAllocator.allocate.providerSession({
1090	providerInstanceId: modelSelection.instanceId,
1091	threadId,
1092	});
1093	const open = manager.open({
1094	threadId,
1095	providerSessionId,
1096	modelSelection,
1097	runtimePolicy,
1098	});
1099
1100	yield* eventSink.write({
1101	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1102	});
1103	const firstFiber = yield* open.pipe(Effect.forkScoped);
1104	yield* Deferred.await(firstOpenStarted);
1105	const secondFiber = yield* open.pipe(Effect.forkScoped);
1106	yield* Effect.yieldNow;
1107	yield* Effect.yieldNow;
1108	assert.equal(yield* Ref.get(openStartedCount), 1);
1109
1110	yield* Deferred.succeed(releaseOpen, undefined);
1111	const [firstRuntime, secondRuntime] = yield* Effect.all([
1112	Fiber.join(firstFiber),
1113	Fiber.join(secondFiber),
1114	]);
1115	assert.strictEqual(firstRuntime, secondRuntime);
1116	assert.equal((yield* Ref.get(state)).openCount, 1);
1117	});
1118
1119	yield* effect.pipe(
1120	Effect.provide(
1121	layerTest({
1122	state,
1123	idleTimeoutMs: 60_000,
1124	beforeOpen,
1125	}),
1126	),
1127	);
1128	}),
1129	);
1130
1131	it.effect(
1132	"ProviderSessionManagerV2 starts a thread's session without its last agent's OOM kill",
1133	() =>
1134	Effect.gen(function* () {
1135	const state = yield* Ref.make(emptyState);
1136	const owner = ThreadId.make("thread-provider-session-manager-oom-owner");
1137	const joiner = ThreadId.make("thread-provider-session-manager-oom-joiner");
1138	// Both threads' previous agents were OOM-killed. A new session may run
1139	// without a scope (Cursor, OpenCode, a shared Codex session), so it must
1140	// not inherit that answer.
1141	const oomKilledThreads = new Set<string>([owner, joiner]);
1142	const agentScope = {
1143	...AgentScope.defaultValue(),
1144	oomKilled: (id: string) => Effect.sync(() => oomKilledThreads.has(id)),
1145	clear: (id: string) => Effect.sync(() => void oomKilledThreads.delete(id)),
1146	};
1147
1148	const effect = Effect.gen(function* () {
1149	const eventSink = yield* EventSink.EventSinkV2;
1150	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1151	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1152	const now = yield* DateTime.now;
1153	const providerSessionId = yield* idAllocator.allocate.providerSession({
1154	providerInstanceId: modelSelection.instanceId,
1155	threadId: owner,
1156	});
1157	yield* eventSink.write({
1158	events: [
1159	yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }),
1160	yield* makeThreadCreatedEvent({ idAllocator, threadId: joiner, now }),
1161	],
1162	});
1163	yield* manager.open({ threadId: owner, providerSessionId, modelSelection, runtimePolicy });
1164	assert.isFalse(yield* agentScope.oomKilled(owner));
1165
1166	yield* manager.open({ threadId: joiner, providerSessionId, modelSelection, runtimePolicy });
1167	assert.isFalse(yield* agentScope.oomKilled(joiner));
1168
1169	// Reopening a session the thread already uses keeps its answer.
1170	oomKilledThreads.add(owner);
1171	yield* manager.open({ threadId: owner, providerSessionId, modelSelection, runtimePolicy });
1172	assert.isTrue(yield* agentScope.oomKilled(owner));
1173	assert.equal((yield* Ref.get(state)).openCount, 1);
1174	});
1175
1176	yield* effect.pipe(
1177	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })),
1178	Effect.provideService(AgentScope, agentScope),
1179	);
1180	}),
1181	);
1182
1183	it.effect("ProviderSessionManagerV2 cleans up an open interrupted mid-handshake", () =>
1184	Effect.gen(function* () {
1185	const state = yield* Ref.make(emptyState);
1186	const mcpConfigs = yield* Ref.make<
1187	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
1188	>([]);
1189	const handshakeStarted = yield* Deferred.make<void>();
1190	const holdHandshake = yield* Ref.make(true);
1191	const effect = Effect.gen(function* () {
1192	const eventSink = yield* EventSink.EventSinkV2;
1193	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1194	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1195	const registry = yield* McpSessionRegistry.McpSessionRegistry;
1196	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
1197	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1198	const now = yield* DateTime.now;
1199	const threadId = ThreadId.make("thread-provider-session-manager-interrupted-open");
1200	const providerSessionId = yield* idAllocator.allocate.providerSession({
1201	providerInstanceId: modelSelection.instanceId,
1202	threadId,
1203	});
1204	yield* eventSink.write({
1205	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1206	});
1207
1208	const opening = yield* manager
1209	.open({ threadId, providerSessionId, modelSelection, runtimePolicy })
1210	.pipe(Effect.forkScoped);
1211	yield* Deferred.await(handshakeStarted);
1212	const issued = (yield* Ref.get(mcpConfigs)).at(-1);
1213	const token = issued?.authorizationHeader.replace(/^Bearer\s+/, "");
1214	assert.isDefined(token);
1215	assert.isDefined(yield* registry.resolve(token!));
1216
1217	// A Stop while the provider is still starting.
1218	yield* Fiber.interrupt(opening);
1219
1220	// The process started for this open is stopped, and the credential minted
1221	// for it revoked.
1222	assert.equal((yield* Ref.get(state)).closeCount, 1);
1223	assert.isUndefined(yield* registry.resolve(token!));
1224	assert.isUndefined(yield* mcpSessions.read(threadId));
1225	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
1226
1227	// Nothing of the interrupted open is left behind: the next open starts a
1228	// fresh process with a fresh credential that a later release revokes,
1229	// which a leaked reservation would prevent.
1230	yield* Ref.set(holdHandshake, false);
1231	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
1232	const replacement = (yield* Ref.get(mcpConfigs)).at(-1);
1233	const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, "");
1234	assert.isDefined(replacementToken);
1235	assert.notEqual(replacementToken, token);
1236	const projection = yield* projectionStore.getThreadProjection(threadId);
1237	assert.equal(projection.providerSessions.at(-1)?.status, "ready");
1238
1239	yield* manager.close(providerSessionId);
1240	assert.isUndefined(yield* registry.resolve(replacementToken!));
1241	});
1242
1243	yield* effect.pipe(
1244	Effect.provide(
1245	layerTest({
1246	state,
1247	idleTimeoutMs: 60_000,
1248	mcpConfigs,
1249	beforeOpen: () =>
1250	Ref.get(holdHandshake).pipe(
1251	Effect.flatMap((hold) =>
1252	hold
1253	? Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never))
1254	: Effect.void,
1255	),
1256	),
1257	// The process is spawned before the handshake that is interrupted.
1258	spawnBeforeOpen: true,
1259	}),
1260	),
1261	);
1262	}),
1263	);
1264
1265	it.effect("ProviderSessionManagerV2 cleans up an interrupted open whose scope close hangs", () =>
1266	Effect.gen(function* () {
1267	const state = yield* Ref.make(emptyState);
1268	const mcpConfigs = yield* Ref.make<
1269	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
1270	>([]);
1271	const handshakeStarted = yield* Deferred.make<void>();
1272	const scopeCloseReached = yield* Deferred.make<void>();
1273	const releaseRetry = yield* Deferred.make<void>();
1274	yield* Effect.gen(function* () {
1275	const eventSink = yield* EventSink.EventSinkV2;
1276	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1277	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1278	const registry = yield* McpSessionRegistry.McpSessionRegistry;
1279	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
1280	const now = yield* DateTime.now;
1281	const threadId = ThreadId.make("thread-provider-session-manager-interrupted-hung-open");
1282	const providerSessionId = yield* idAllocator.allocate.providerSession({
1283	providerInstanceId: modelSelection.instanceId,
1284	threadId,
1285	});
1286	yield* eventSink.write({
1287	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1288	});
1289
1290	const opening = yield* manager
1291	.open({ threadId, providerSessionId, modelSelection, runtimePolicy })
1292	.pipe(Effect.forkChild);
1293	yield* Deferred.await(handshakeStarted);
1294	const issued = (yield* Ref.get(mcpConfigs)).at(-1);
1295	const token = issued?.authorizationHeader.replace(/^Bearer\s+/, "");
1296	assert.isDefined(token);
1297
1298	// The interrupt starts cleanup; the scope close then never finishes.
1299	const interrupter = yield* Fiber.interrupt(opening).pipe(Effect.forkChild);
1300	yield* Deferred.await(scopeCloseReached);
1301
1302	// The session cleanup already ran, ahead of the stuck close.
1303	assert.isUndefined(yield* registry.resolve(token!));
1304	assert.isUndefined(yield* mcpSessions.read(threadId));
1305	assert.equal((yield* Ref.get(state)).closeCount, 0);
1306
1307	// The close is time-boxed, so the interrupter returns and the session's
1308	// open lock is free for the next open.
1309	yield* TestClock.adjust("30 seconds");
1310	yield* Fiber.join(interrupter);
1311	yield* Deferred.succeed(releaseRetry, undefined);
1312	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
1313	assert.equal((yield* Ref.get(state)).openCount, 1);
1314
1315	// Its close hangs as well; release it while the test clock can still move.
1316	const stopping = yield* manager.shutdown.pipe(Effect.forkChild);
1317	yield* TestClock.adjust("30 seconds");
1318	yield* Fiber.join(stopping);
1319	}).pipe(
1320	Effect.provide(
1321	layerTest({
1322	state,
1323	idleTimeoutMs: 60_000,
1324	mcpConfigs,
1325	// The first open hangs in its handshake; the retry completes.
1326	beforeOpen: () =>
1327	Deferred.isDone(handshakeStarted).pipe(
1328	Effect.flatMap((retry) =>
1329	retry
1330	? Deferred.await(releaseRetry)
1331	: Deferred.succeed(handshakeStarted, undefined).pipe(
1332	Effect.andThen(Effect.never),
1333	),
1334	),
1335	),
1336	spawnBeforeOpen: true,
1337	hangSessionScopeClose: true,
1338	scopeCloseReached,
1339	}),
1340	),
1341	);
1342	}),
1343	);
1344
1345	it.effect("ProviderSessionManagerV2 releases an idle session whose turn start was stopped", () =>
1346	Effect.gen(function* () {
1347	const state = yield* Ref.make(emptyState);
1348	const startTurnReached = yield* Deferred.make<void>();
1349	yield* Effect.gen(function* () {
1350	const eventSink = yield* EventSink.EventSinkV2;
1351	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1352	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1353	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1354	const now = yield* DateTime.now;
1355	const threadId = ThreadId.make("thread-provider-session-manager-stopped-turn-start");
1356	const providerSessionId = yield* idAllocator.allocate.providerSession({
1357	providerInstanceId: modelSelection.instanceId,
1358	threadId,
1359	});
1360	yield* eventSink.write({
1361	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1362	});
1363	const runtime = yield* manager.open({
1364	threadId,
1365	providerSessionId,
1366	modelSelection,
1367	runtimePolicy,
1368	});
1369	const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
1370	const starting = yield* runtime
1371	.startTurn({
1372	appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
1373	threadId,
1374	runId,
1375	runOrdinal: 1,
1376	providerTurnOrdinal: 1,
1377	attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
1378	rootNodeId: idAllocator.derive.rootNode({ runId }),
1379	providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }),
1380	message: {
1381	createdBy: "user",
1382	creationSource: "web",
1383	messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
1384	text: "stopped before the provider accepted it",
1385	attachments: [],
1386	},
1387	modelSelection,
1388	runtimePolicy,
1389	})
1390	.pipe(Effect.forkChild);
1391	yield* Deferred.await(startTurnReached);
1392
1393	// Stop lands while the provider is still accepting the turn. No terminal
1394	// event follows, so the session must count itself idle again.
1395	yield* Fiber.interrupt(starting);
1396	yield* TestClock.adjust("2 seconds");
1397	yield* Effect.yieldNow;
1398	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
1399	assert.equal((yield* Ref.get(state)).closeCount, 1);
1400	}).pipe(
1401	Effect.provide(
1402	layerTest({
1403	state,
1404	idleTimeoutMs: 1000,
1405	startTurn: Deferred.succeed(startTurnReached, undefined).pipe(
1406	Effect.andThen(Effect.never),
1407	),
1408	}),
1409	),
1410	);
1411	}),
1412	);
1413
1414	it.effect(
1415	"ProviderSessionManagerV2 keeps a shared session busy when another thread's start is stopped early",
1416	() =>
1417	Effect.gen(function* () {
1418	const state = yield* Ref.make(emptyState);
1419	const projectId = ProjectId.make("project-provider-session-manager-shared-stopped-start");
1420	// Blocks the next project read, so a re-attach can be stopped before its
1421	// start marks the session busy.
1422	const holdProjectRead = yield* Ref.make(false);
1423	const projectReadHeld = yield* Deferred.make<void>();
1424	const projectServiceLayer = Layer.mock(ProjectService.ProjectService)({
1425	getById: (requestedProjectId) =>
1426	Ref.get(holdProjectRead).pipe(
1427	Effect.flatMap((hold) =>
1428	hold
1429	? Deferred.succeed(projectReadHeld, undefined).pipe(Effect.andThen(Effect.never))
1430	: Effect.succeed(Option.some(makeBrowserAccessProject(requestedProjectId))),
1431	),
1432	),
1433	});
1434	yield* Effect.gen(function* () {
1435	const eventSink = yield* EventSink.EventSinkV2;
1436	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1437	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1438	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1439	const now = yield* DateTime.now;
1440	const threadA = ThreadId.make("thread-provider-session-manager-shared-stopped-a");
1441	const threadB = ThreadId.make("thread-provider-session-manager-shared-stopped-b");
1442	const providerSessionId = idAllocator.derive.providerSession({
1443	providerInstanceId: modelSelection.instanceId,
1444	});
1445	yield* eventSink.write({
1446	events: [
1447	yield* makeThreadCreatedEvent({ idAllocator, threadId: threadA, now, projectId }),
1448	yield* makeThreadCreatedEvent({ idAllocator, threadId: threadB, now, projectId }),
1449	],
1450	});
1451	const runtime = yield* manager.open({
1452	threadId: threadA,
1453	providerSessionId,
1454	modelSelection,
1455	runtimePolicy,
1456	});
1457	yield* manager.open({
1458	threadId: threadB,
1459	providerSessionId,
1460	modelSelection,
1461	runtimePolicy,
1462	});
1463	const startTurn = (threadId: ThreadId) =>
1464	Effect.gen(function* () {
1465	const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
1466	return yield* runtime.startTurn({
1467	appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
1468	threadId,
1469	runId,
1470	runOrdinal: 1,
1471	providerTurnOrdinal: 1,
1472	attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
1473	rootNodeId: idAllocator.derive.rootNode({ runId }),
1474	providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }),
1475	message: {
1476	createdBy: "user",
1477	creationSource: "web",
1478	messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
1479	text: "turn",
1480	attachments: [],
1481	},
1482	modelSelection,
1483	runtimePolicy,
1484	});
1485	});
1486
1487	// B's turn is accepted and still running.
1488	yield* startTurn(threadB);
1489
1490	// A detaches, and its next start is stopped while re-attaching, before
1491	// it marks the session busy.
1492	yield* manager.detach({ providerSessionId, threadId: threadA });
1493	yield* Ref.set(holdProjectRead, true);
1494	const startingA = yield* startTurn(threadA).pipe(Effect.forkChild);
1495	yield* Deferred.await(projectReadHeld);
1496	yield* Fiber.interrupt(startingA);
1497
1498	// B's running turn keeps the session busy past the idle timeout.
1499	yield* TestClock.adjust("2 seconds");
1500	yield* Effect.yieldNow;
1501	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
1502	assert.equal((yield* Ref.get(state)).closeCount, 0);
1503	}).pipe(
1504	Effect.provide(
1505	layerTest({
1506	state,
1507	idleTimeoutMs: 1000,
1508	projectServiceLayer,
1509	serverSettingsLayer: ServerSettings.layerTest({
1510	projectSettingsOverrides: { [projectId]: { enableAgentBrowserAccess: true } },
1511	}),
1512	}),
1513	),
1514	);
1515	}),
1516	);
1517
1518	it.effect(
1519	"ProviderSessionManagerV2 keeps a shared session busy when a stopped start still ends its turn",
1520	() =>
1521	Effect.gen(function* () {
1522	const state = yield* Ref.make(emptyState);
1523	// A's adapter start installs its turn, then is stopped mid-request; like
1524	// OpenCode2, the adapter still ends that turn with turn.terminal later.
1525	const holdStart = yield* Ref.make(false);
1526	const startHeld = yield* Deferred.make<void>();
1527	yield* Effect.gen(function* () {
1528	const eventSink = yield* EventSink.EventSinkV2;
1529	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1530	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1531	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
1532	const now = yield* DateTime.now;
1533	const threadA = ThreadId.make("thread-provider-session-manager-stopped-ended-a");
1534	const threadB = ThreadId.make("thread-provider-session-manager-stopped-ended-b");
1535	const providerSessionId = idAllocator.derive.providerSession({
1536	providerInstanceId: modelSelection.instanceId,
1537	});
1538	yield* eventSink.write({
1539	events: [
1540	yield* makeThreadCreatedEvent({ idAllocator, threadId: threadA, now }),
1541	yield* makeThreadCreatedEvent({ idAllocator, threadId: threadB, now }),
1542	],
1543	});
1544	const runtime = yield* manager.open({
1545	threadId: threadA,
1546	providerSessionId,
1547	modelSelection,
1548	runtimePolicy,
1549	});
1550	yield* manager.open({
1551	threadId: threadB,
1552	providerSessionId,
1553	modelSelection,
1554	runtimePolicy,
1555	});
1556	const providerThreadOf = (threadId: ThreadId) =>
1557	makeProviderThread({
1558	idAllocator,
1559	threadId,
1560	providerSessionId,
1561	now,
1562	nativeThreadId: `native-${threadId}`,
1563	});
1564	const startTurn = (threadId: ThreadId) =>
1565	Effect.gen(function* () {
1566	const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
1567	return yield* runtime.startTurn({
1568	appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
1569	threadId,
1570	runId,
1571	runOrdinal: 1,
1572	providerTurnOrdinal: 1,
1573	attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
1574	rootNodeId: idAllocator.derive.rootNode({ runId }),
1575	providerThread: providerThreadOf(threadId),
1576	message: {
1577	createdBy: "user",
1578	creationSource: "web",
1579	messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
1580	text: "turn",
1581	attachments: [],
1582	},
1583	modelSelection,
1584	runtimePolicy,
1585	});
1586	});
1587
1588	// B's turn is accepted and still running.
1589	yield* startTurn(threadB);
1590
1591	// A's start is stopped after the adapter began the turn.
1592	yield* Ref.set(holdStart, true);
1593	const startingA = yield* startTurn(threadA).pipe(Effect.forkChild);
1594	yield* Deferred.await(startHeld);
1595	yield* Fiber.interrupt(startingA);
1596
1597	// The adapter ends A's turn anyway.
1598	const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
1599	assert.isDefined(queue);
1600	yield* Queue.offer(queue!, {
1601	type: "turn.terminal",
1602	driver: CODEX_DRIVER,
1603	providerThreadId: providerThreadOf(threadA).id,
1604	providerTurnId: idAllocator.derive.providerTurn({
1605	driver: CODEX_DRIVER,
1606	nativeTurnId: "native-turn-stopped-a",
1607	}),
1608	runOrdinal: 1,
1609	status: "completed",
1610	failure: null,
1611	threadDisposition: "reusable",
1612	});
1613
1614	// B's running turn keeps the session busy past the idle timeout.
1615	yield* TestClock.adjust("2 seconds");
1616	yield* Effect.yieldNow;
1617	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
1618	assert.equal((yield* Ref.get(state)).closeCount, 0);
1619	}).pipe(
1620	Effect.provide(
1621	layerTest({
1622	state,
1623	idleTimeoutMs: 1000,
1624	startTurn: Ref.get(holdStart).pipe(
1625	Effect.flatMap((hold) =>
1626	hold
1627	? Deferred.succeed(startHeld, undefined).pipe(Effect.andThen(Effect.never))
1628	: Effect.void,
1629	),
1630	),
1631	}),
1632	),
1633	);
1634	}),
1635	);
1636
1637	it.effect("ProviderSessionManagerV2 stops a session still opening when its layer shuts down", () =>
1638	Effect.gen(function* () {
1639	const state = yield* Ref.make(emptyState);
1640	const handshakeStarted = yield* Deferred.make<void>();
1641	yield* Effect.gen(function* () {
1642	const eventSink = yield* EventSink.EventSinkV2;
1643	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1644	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1645	const now = yield* DateTime.now;
1646	const threadId = ThreadId.make("thread-provider-session-manager-shutdown-during-open");
1647	const providerSessionId = yield* idAllocator.allocate.providerSession({
1648	providerInstanceId: modelSelection.instanceId,
1649	threadId,
1650	});
1651	yield* eventSink.write({
1652	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1653	});
1654	// Detached, like an open the layer does not own: only the session
1655	// scope's parent can stop its process when the layer closes.
1656	yield* manager
1657	.open({ threadId, providerSessionId, modelSelection, runtimePolicy })
1658	.pipe(Effect.forkDetach);
1659	yield* Deferred.await(handshakeStarted);
1660	assert.equal((yield* Ref.get(state)).closeCount, 0);
1661	}).pipe(
1662	Effect.provide(
1663	layerTest({
1664	state,
1665	idleTimeoutMs: 60_000,
1666	beforeOpen: () =>
1667	Deferred.succeed(handshakeStarted, undefined).pipe(Effect.andThen(Effect.never)),
1668	spawnBeforeOpen: true,
1669	}),
1670	),
1671	);
1672
1673	assert.equal((yield* Ref.get(state)).closeCount, 1);
1674	}),
1675	);
1676
1677	it.effect("ProviderSessionManagerV2 releases live sessions when its layer shuts down", () =>
1678	Effect.gen(function* () {
1679	const state = yield* Ref.make(emptyState);
1680	const effect = Effect.gen(function* () {
1681	const eventSink = yield* EventSink.EventSinkV2;
1682	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1683	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1684	const now = yield* DateTime.now;
1685	const threadId = ThreadId.make("thread-provider-session-manager-shutdown");
1686	const providerSessionId = yield* idAllocator.allocate.providerSession({
1687	providerInstanceId: modelSelection.instanceId,
1688	threadId,
1689	});
1690
1691	yield* eventSink.write({
1692	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1693	});
1694	yield* manager.open({
1695	threadId,
1696	providerSessionId,
1697	modelSelection,
1698	runtimePolicy,
1699	});
1700
1701	const liveState = yield* Ref.get(state);
1702	assert.equal(liveState.openCount, 1);
1703	assert.equal(liveState.closeCount, 0);
1704	});
1705
1706	yield* effect.pipe(
1707	Effect.provide(
1708	layerTest({
1709	state,
1710	idleTimeoutMs: 60_000,
1711	}),
1712	),
1713	);
1714
1715	assert.equal((yield* Ref.get(state)).closeCount, 1);
1716	}),
1717	);
1718
1719	it.effect("ProviderSessionManagerV2 releases sessions in parallel on server shutdown", () =>
1720	Effect.gen(function* () {
1721	const state = yield* Ref.make(emptyState);
1722	yield* Effect.gen(function* () {
1723	const eventSink = yield* EventSink.EventSinkV2;
1724	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1725	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1726	const now = yield* DateTime.now;
1727	for (const name of ["a", "b", "c"]) {
1728	const threadId = ThreadId.make(`thread-provider-session-manager-parallel-shutdown-${name}`);
1729	yield* eventSink.write({
1730	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1731	});
1732	yield* manager.open({
1733	threadId,
1734	providerSessionId: yield* idAllocator.allocate.providerSession({
1735	providerInstanceId: modelSelection.instanceId,
1736	threadId,
1737	}),
1738	modelSelection,
1739	runtimePolicy,
1740	});
1741	}
1742
1743	// Every close hangs until its time box ends. One time box must cover all
1744	// three sessions, or a busy server outlives the launcher's kill timer.
1745	const stopping = yield* manager.shutdown.pipe(Effect.forkChild);
1746	yield* TestClock.adjust("30 seconds");
1747	yield* Effect.yieldNow;
1748	assert.isDefined(stopping.pollUnsafe());
1749	}).pipe(
1750	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, hangSessionScopeClose: true })),
1751	);
1752	}),
1753	);
1754
1755	it.effect("ProviderSessionManagerV2 closes event subscriptions normally on server shutdown", () =>
1756	Effect.gen(function* () {
1757	const state = yield* Ref.make(emptyState);
1758	const effect = Effect.gen(function* () {
1759	const eventSink = yield* EventSink.EventSinkV2;
1760	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1761	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1762	const now = yield* DateTime.now;
1763	const threadId = ThreadId.make("thread-provider-session-manager-shutdown-subscription");
1764	const providerSessionId = yield* idAllocator.allocate.providerSession({
1765	providerInstanceId: modelSelection.instanceId,
1766	threadId,
1767	});
1768	yield* eventSink.write({
1769	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1770	});
1771	const runtime = yield* manager.open({
1772	threadId,
1773	providerSessionId,
1774	modelSelection,
1775	runtimePolicy,
1776	});
1777	const bufferedSubscription = yield* runtime.subscribeEvents!;
1778	const activeSubscription = yield* runtime.subscribeEvents!;
1779	const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
1780	assert.isDefined(adapterQueue);
1781	yield* Queue.offer(adapterQueue!, {
1782	type: "provider_session.updated",
1783	driver: CODEX_DRIVER,
1784	providerSession: runtime.providerSession,
1785	});
1786	assert.isTrue(Option.isSome(yield* activeSubscription.events.pipe(Stream.runHead)));
1787
1788	yield* manager.shutdown;
1789
1790	assert.isEmpty(yield* bufferedSubscription.events.pipe(Stream.runCollect));
1791	});
1792
1793	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })));
1794	}),
1795	);
1796
1797	it.effect("ProviderSessionManagerV2 drains subscribers when the provider stops", () =>
1798	Effect.gen(function* () {
1799	const state = yield* Ref.make(emptyState);
1800	const effect = Effect.gen(function* () {
1801	const eventSink = yield* EventSink.EventSinkV2;
1802	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1803	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1804	const now = yield* DateTime.now;
1805	const threadId = ThreadId.make("thread-provider-session-manager-provider-stop");
1806	const providerSessionId = yield* idAllocator.allocate.providerSession({
1807	providerInstanceId: modelSelection.instanceId,
1808	threadId,
1809	});
1810	yield* eventSink.write({
1811	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1812	});
1813	const runtime = yield* manager.open({
1814	threadId,
1815	providerSessionId,
1816	modelSelection,
1817	runtimePolicy,
1818	});
1819	const subscription = yield* runtime.subscribeEvents!;
1820	const collected = yield* subscription.events.pipe(Stream.runCollect, Effect.forkScoped);
1821	const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
1822	assert.isDefined(adapterQueue);
1823	const providerThreadId = idAllocator.derive.providerThread({
1824	driver: CODEX_DRIVER,
1825	nativeThreadId: "provider-stop-thread",
1826	});
1827	const providerTurnId = idAllocator.derive.providerTurn({
1828	driver: CODEX_DRIVER,
1829	nativeTurnId: "provider-stop-turn",
1830	});
1831	yield* Queue.offer(adapterQueue!, {
1832	type: "turn.terminal",
1833	driver: CODEX_DRIVER,
1834	providerThreadId,
1835	providerTurnId,
1836	runOrdinal: 1,
1837	status: "completed",
1838	failure: null,
1839	threadDisposition: "reusable",
1840	});
1841	yield* Queue.offer(adapterQueue!, {
1842	type: "provider_session.updated",
1843	driver: CODEX_DRIVER,
1844	providerSession: {
1845	...runtime.providerSession,
1846	status: "stopped",
1847	updatedAt: now,
1848	},
1849	});
1850	yield* Queue.end(adapterQueue!);
1851
1852	const events = Array.from(yield* Fiber.join(collected));
1853	assert.deepEqual(
1854	events.map((event) => event.type),
1855	["turn.terminal", "provider_session.updated"],
1856	);
1857	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
1858	assert.equal((yield* Ref.get(state)).closeCount, 1);
1859	});
1860
1861	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })));
1862	}),
1863	);
1864
1865	it.effect(
1866	"ProviderSessionManagerV2 issues MCP credentials before opening and revokes them on close",
1867	() =>
1868	Effect.gen(function* () {
1869	const state = yield* Ref.make(emptyState);
1870	const mcpConfigs = yield* Ref.make<
1871	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
1872	>([]);
1873	const effect = Effect.gen(function* () {
1874	const eventSink = yield* EventSink.EventSinkV2;
1875	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1876	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1877	const registry = yield* McpSessionRegistry.McpSessionRegistry;
1878	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
1879	const now = yield* DateTime.now;
1880	const threadId = ThreadId.make("thread-provider-session-manager-mcp");
1881	const providerSessionId = yield* idAllocator.allocate.providerSession({
1882	providerInstanceId: modelSelection.instanceId,
1883	threadId,
1884	});
1885
1886	yield* eventSink.write({
1887	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1888	});
1889	yield* manager.open({
1890	threadId,
1891	providerSessionId,
1892	modelSelection,
1893	runtimePolicy,
1894	});
1895
1896	const captured = (yield* Ref.get(mcpConfigs))[0];
1897	assert.isDefined(captured);
1898	assert.equal(captured?.threadId, threadId);
1899	assert.equal(captured?.providerInstanceId, modelSelection.instanceId);
1900	assert.equal(captured?.endpoint, "http://127.0.0.1:43123/mcp");
1901	const token = captured?.authorizationHeader.replace(/^Bearer\s+/, "");
1902	assert.isDefined(token);
1903	const resolved = yield* registry.resolve(token!);
1904	assert.equal(resolved?.thread.threadId, threadId);
1905	assert.deepEqual(
1906	resolved?.capabilities,
1907	new Set(["preview", "orchestration", "worktree", "pull-requests"]),
1908	);
1909
1910	yield* manager.close(providerSessionId);
1911	assert.isUndefined(yield* mcpSessions.read(threadId));
1912	assert.isUndefined(yield* registry.resolve(token!));
1913	});
1914
1915	yield* effect.pipe(
1916	Effect.provide(
1917	layerTest({
1918	state,
1919	idleTimeoutMs: 1_000,
1920	mcpConfigs,
1921	}),
1922	),
1923	);
1924	}),
1925	);
1926
1927	it.effect(
1928	"ProviderSessionManagerV2 withholds the preview capability when agent browser access is off",
1929	() =>
1930	Effect.gen(function* () {
1931	const state = yield* Ref.make(emptyState);
1932	const mcpConfigs = yield* Ref.make<
1933	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
1934	>([]);
1935	const effect = Effect.gen(function* () {
1936	const eventSink = yield* EventSink.EventSinkV2;
1937	const idAllocator = yield* IdAllocator.IdAllocatorV2;
1938	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
1939	const registry = yield* McpSessionRegistry.McpSessionRegistry;
1940	const now = yield* DateTime.now;
1941	const threadId = ThreadId.make("thread-provider-session-manager-no-browser");
1942	const providerSessionId = yield* idAllocator.allocate.providerSession({
1943	providerInstanceId: modelSelection.instanceId,
1944	threadId,
1945	});
1946
1947	yield* eventSink.write({
1948	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
1949	});
1950	yield* manager.open({
1951	threadId,
1952	providerSessionId,
1953	modelSelection,
1954	runtimePolicy,
1955	});
1956
1957	const captured = (yield* Ref.get(mcpConfigs))[0];
1958	assert.isDefined(captured);
1959	assert.equal(captured?.browserToolsAvailable, false);
1960	const token = captured?.authorizationHeader.replace(/^Bearer\s+/, "");
1961	const resolved = yield* registry.resolve(token!);
1962	assert.deepEqual(
1963	resolved?.capabilities,
1964	new Set(["orchestration", "worktree", "pull-requests"]),
1965	);
1966
1967	yield* manager.close(providerSessionId);
1968	});
1969
1970	yield* effect.pipe(
1971	Effect.provide(
1972	layerTest({
1973	state,
1974	idleTimeoutMs: 1_000,
1975	mcpConfigs,
1976	// orDie: the test layer's settings-normalization error cannot
1977	// occur for a literal override and the slot requires error never.
1978	serverSettingsLayer: ServerSettings.layerTest({
1979	enableAgentBrowserAccess: false,
1980	}).pipe(Layer.orDie),
1981	}),
1982	),
1983	);
1984	}),
1985	);
1986
1987	it.effect("ProviderSessionManagerV2 honors a project browser-access opt-out", () =>
1988	Effect.gen(function* () {
1989	const captured = yield* runBrowserAccessScenario({
1990	enableAgentBrowserAccess: true,
1991	projectOverride: false,
1992	});
1993	assert.isDefined(captured);
1994	assert.equal(captured?.browserToolsAvailable, false);
1995	}),
1996	);
1997
1998	it.effect("ProviderSessionManagerV2 honors a project browser-access opt-in", () =>
1999	Effect.gen(function* () {
2000	const captured = yield* runBrowserAccessScenario({
2001	enableAgentBrowserAccess: false,
2002	projectOverride: true,
2003	});
2004	assert.isDefined(captured);
2005	assert.equal(captured?.browserToolsAvailable, true);
2006	}),
2007	);
2008
2009	it.effect("ProviderSessionManagerV2 fails browser access closed for a missing project", () =>
2010	Effect.gen(function* () {
2011	const captured = yield* runBrowserAccessScenario({
2012	enableAgentBrowserAccess: true,
2013	projectOverride: true,
2014	projectExists: false,
2015	});
2016	assert.isDefined(captured);
2017	assert.equal(captured?.browserToolsAvailable, false);
2018	}),
2019	);
2020
2021	it.effect("ProviderSessionManagerV2 fails browser access closed for a missing thread", () =>
2022	Effect.gen(function* () {
2023	const captured = yield* runBrowserAccessScenario({
2024	enableAgentBrowserAccess: true,
2025	projectOverride: true,
2026	createThread: false,
2027	});
2028	assert.isDefined(captured);
2029	assert.equal(captured?.browserToolsAvailable, false);
2030	}),
2031	);
2032
2033	it.effect("ProviderSessionManagerV2 revokes MCP credentials when release persistence fails", () =>
2034	Effect.gen(function* () {
2035	const state = yield* Ref.make(emptyState);
2036	const mcpConfigs = yield* Ref.make<
2037	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2038	>([]);
2039	const effect = Effect.gen(function* () {
2040	const eventSink = yield* EventSink.EventSinkV2;
2041	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2042	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2043	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2044	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2045	const now = yield* DateTime.now;
2046	const threadId = ThreadId.make("thread-provider-session-manager-mcp-release-failure");
2047	const providerSessionId = yield* idAllocator.allocate.providerSession({
2048	providerInstanceId: modelSelection.instanceId,
2049	threadId,
2050	});
2051
2052	yield* eventSink.write({
2053	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2054	});
2055	yield* manager.open({
2056	threadId,
2057	providerSessionId,
2058	modelSelection,
2059	runtimePolicy,
2060	});
2061
2062	const captured = (yield* Ref.get(mcpConfigs))[0];
2063	const token = captured?.authorizationHeader.replace(/^Bearer\s+/, "");
2064	assert.isDefined(token);
2065	assert.isDefined(yield* registry.resolve(token!));
2066
2067	const closeError = yield* manager.close(providerSessionId).pipe(Effect.flip);
2068	assert.equal(closeError._tag, "ProviderSessionCloseError");
2069	assert.isUndefined(yield* mcpSessions.read(threadId));
2070	assert.isUndefined(yield* registry.resolve(token!));
2071	});
2072
2073	yield* effect.pipe(
2074	Effect.provide(
2075	layerTest({
2076	state,
2077	idleTimeoutMs: 1_000,
2078	mcpConfigs,
2079	failReleaseEventWrites: true,
2080	}),
2081	),
2082	);
2083	}),
2084	);
2085
2086	it.effect(
2087	"ProviderSessionManagerV2 keeps a thread attached when an earlier attach of it is interrupted",
2088	() =>
2089	Effect.gen(function* () {
2090	const state = yield* Ref.make(emptyState);
2091	const armed = yield* Ref.make(false);
2092	const paused = yield* Deferred.make<void>();
2093	yield* Effect.gen(function* () {
2094	const eventSink = yield* EventSink.EventSinkV2;
2095	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2096	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2097	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2098	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2099	const now = yield* DateTime.now;
2100	const owner = ThreadId.make("thread-provider-session-manager-attach-race-owner");
2101	const threadId = ThreadId.make("thread-provider-session-manager-attach-race");
2102	const providerSessionId = idAllocator.derive.providerSession({
2103	providerInstanceId: modelSelection.instanceId,
2104	});
2105	yield* eventSink.write({
2106	events: [
2107	yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }),
2108	yield* makeThreadCreatedEvent({ idAllocator, threadId, now }),
2109	],
2110	});
2111	const runtime = yield* manager.open({
2112	threadId: owner,
2113	providerSessionId,
2114	modelSelection,
2115	runtimePolicy,
2116	});
2117	const resume = runtime.resumeThread({
2118	threadId,
2119	providerThread: makeProviderThread({
2120	idAllocator,
2121	threadId,
2122	providerSessionId,
2123	now,
2124	nativeThreadId: "native-attach-race",
2125	}),
2126	});
2127
2128	// The first attach of the thread is stopped after attaching it, while
2129	// a second attach of the same thread is already on its way.
2130	yield* Ref.set(armed, true);
2131	const first = yield* resume.pipe(Effect.forkChild({ startImmediately: true }));
2132	yield* Deferred.await(paused);
2133	const second = yield* resume.pipe(Effect.forkChild({ startImmediately: true }));
2134	yield* Fiber.interrupt(first);
2135	yield* Fiber.join(second);
2136
2137	// The second attach owns the thread's attachment and credential.
2138	const config = yield* mcpSessions.read(threadId);
2139	assert.isDefined(config);
2140	const token = config!.authorizationHeader.replace(/^Bearer\s+/, "");
2141	assert.equal((yield* registry.resolve(token))?.thread.threadId, threadId);
2142	}).pipe(
2143	Effect.provide(
2144	layerTest({ state, idleTimeoutMs: 60_000, pauseAttachWrite: { armed, paused } }),
2145	),
2146	);
2147	}),
2148	);
2149
2150	it.effect(
2151	"ProviderSessionManagerV2 keeps a replacement session's attachment when a stale attach is interrupted",
2152	() =>
2153	Effect.gen(function* () {
2154	const state = yield* Ref.make(emptyState);
2155	const armed = yield* Ref.make(false);
2156	const paused = yield* Deferred.make<void>();
2157	yield* Effect.gen(function* () {
2158	const eventSink = yield* EventSink.EventSinkV2;
2159	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2160	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2161	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2162	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2163	const eventStore = yield* EventStore.EventStoreV2;
2164	const now = yield* DateTime.now;
2165	const owner = ThreadId.make("thread-provider-session-manager-stale-attach-owner");
2166	const threadId = ThreadId.make("thread-provider-session-manager-stale-attach");
2167	const providerSessionId = idAllocator.derive.providerSession({
2168	providerInstanceId: modelSelection.instanceId,
2169	});
2170	yield* eventSink.write({
2171	events: [
2172	yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }),
2173	yield* makeThreadCreatedEvent({ idAllocator, threadId, now }),
2174	],
2175	});
2176	const runtime = yield* manager.open({
2177	threadId: owner,
2178	providerSessionId,
2179	modelSelection,
2180	runtimePolicy,
2181	});
2182
2183	// An attach of the thread to this session stalls mid-write...
2184	yield* Ref.set(armed, true);
2185	const stale = yield* runtime
2186	.resumeThread({
2187	threadId,
2188	providerThread: makeProviderThread({
2189	idAllocator,
2190	threadId,
2191	providerSessionId,
2192	now,
2193	nativeThreadId: "native-stale-attach",
2194	}),
2195	})
2196	.pipe(Effect.forkChild({ startImmediately: true }));
2197	yield* Deferred.await(paused);
2198
2199	// ...while the session is replaced and the thread opens the new one.
2200	yield* manager.release({ providerSessionId, reason: "runtime_error" });
2201	const replacement = yield* manager.open({
2202	threadId,
2203	providerSessionId,
2204	modelSelection,
2205	runtimePolicy,
2206	});
2207	const config = yield* mcpSessions.read(threadId);
2208	assert.isDefined(config);
2209
2210	yield* Fiber.interrupt(stale);
2211
2212	// The replacement keeps the thread and its credential.
2213	assert.equal(
2214	(yield* mcpSessions.read(threadId))?.providerSessionId,
2215	config!.providerSessionId,
2216	);
2217	const token = config!.authorizationHeader.replace(/^Bearer\s+/, "");
2218	assert.equal((yield* registry.resolve(token))?.thread.threadId, threadId);
2219	// Still attached: resuming on the replacement does not attach the thread again.
2220	const attachedEvents = eventStore
2221	.read({ threadId, eventType: "provider-session.attached" })
2222	.pipe(
2223	Stream.runCollect,
2224	Effect.map((events) => events.length),
2225	);
2226	const attachedBefore = yield* attachedEvents;
2227	yield* replacement.resumeThread({
2228	threadId,
2229	providerThread: makeProviderThread({
2230	idAllocator,
2231	threadId,
2232	providerSessionId,
2233	now,
2234	nativeThreadId: "native-stale-attach",
2235	}),
2236	});
2237	assert.equal(yield* attachedEvents, attachedBefore);
2238	}).pipe(
2239	Effect.provide(
2240	layerTest({ state, idleTimeoutMs: 60_000, pauseAttachWrite: { armed, paused } }),
2241	),
2242	);
2243	}),
2244	);
2245
2246	it.effect(
2247	"ProviderSessionManagerV2 revokes a reused credential after a resume stopped while checking it",
2248	() =>
2249	Effect.gen(function* () {
2250	const state = yield* Ref.make(emptyState);
2251	const armed = yield* Ref.make(false);
2252	const paused = yield* Deferred.make<void>();
2253	yield* Effect.gen(function* () {
2254	const eventSink = yield* EventSink.EventSinkV2;
2255	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2256	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2257	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2258	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2259	const now = yield* DateTime.now;
2260	const owner = ThreadId.make("thread-provider-session-manager-resolve-stop-owner");
2261	const threadId = ThreadId.make("thread-provider-session-manager-resolve-stop");
2262	const providerSessionId = idAllocator.derive.providerSession({
2263	providerInstanceId: modelSelection.instanceId,
2264	});
2265	yield* eventSink.write({
2266	events: [
2267	yield* makeThreadCreatedEvent({ idAllocator, threadId: owner, now }),
2268	yield* makeThreadCreatedEvent({ idAllocator, threadId, now }),
2269	],
2270	});
2271	const runtime = yield* manager.open({
2272	threadId: owner,
2273	providerSessionId,
2274	modelSelection,
2275	runtimePolicy,
2276	});
2277	const resume = runtime.resumeThread({
2278	threadId,
2279	providerThread: makeProviderThread({
2280	idAllocator,
2281	threadId,
2282	providerSessionId,
2283	now,
2284	nativeThreadId: "native-resolve-stop",
2285	}),
2286	});
2287	// The thread gets a credential, then detaches and keeps it for a re-attach.
2288	yield* resume;
2289	yield* manager.detach({ providerSessionId, threadId });
2290	const config = yield* mcpSessions.read(threadId);
2291	assert.isDefined(config);
2292	const token = config!.authorizationHeader.replace(/^Bearer\s+/, "");
2293
2294	// A re-attach is stopped while it checks whether that credential is reusable.
2295	yield* Ref.set(armed, true);
2296	const stopped = yield* resume.pipe(Effect.forkChild({ startImmediately: true }));
2297	yield* Deferred.await(paused);
2298	yield* Fiber.interrupt(stopped);
2299
2300	// Nothing holds the credential now, so a terminal release revokes it.
2301	yield* manager.release({ providerSessionId, reason: "manual_shutdown" });
2302	assert.isUndefined(yield* registry.resolve(token));
2303	}).pipe(
2304	Effect.provide(
2305	layerTest({ state, idleTimeoutMs: 60_000, pauseResolve: { armed, paused } }),
2306	),
2307	);
2308	}),
2309	);
2310
2311	it.effect("ProviderSessionManagerV2 duplicate detach preserves replacement MCP credentials", () =>
2312	Effect.gen(function* () {
2313	const state = yield* Ref.make(emptyState);
2314	const mcpConfigs = yield* Ref.make<
2315	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2316	>([]);
2317	const effect = Effect.gen(function* () {
2318	const eventSink = yield* EventSink.EventSinkV2;
2319	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2320	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2321	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2322	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2323	const now = yield* DateTime.now;
2324	const threadId = ThreadId.make("thread-provider-session-manager-replacement-mcp");
2325	const oldSessionId = yield* idAllocator.allocate.providerSession({
2326	providerInstanceId: modelSelection.instanceId,
2327	threadId,
2328	});
2329	const replacementSessionId = yield* idAllocator.allocate.providerSession({
2330	providerInstanceId: modelSelection.instanceId,
2331	threadId,
2332	});
2333
2334	yield* eventSink.write({
2335	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2336	});
2337	yield* manager.open({
2338	threadId,
2339	providerSessionId: oldSessionId,
2340	modelSelection,
2341	runtimePolicy,
2342	});
2343	yield* manager.detach({ providerSessionId: oldSessionId, threadId });
2344	yield* manager.open({
2345	threadId,
2346	providerSessionId: replacementSessionId,
2347	modelSelection,
2348	runtimePolicy,
2349	});
2350
2351	const replacement = (yield* Ref.get(mcpConfigs)).at(-1);
2352	assert.isDefined(replacement);
2353	const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, "");
2354	assert.isDefined(replacementToken);
2355	assert.equal(
2356	(yield* mcpSessions.read(threadId))?.providerSessionId,
2357	replacement?.providerSessionId,
2358	);
2359
2360	yield* manager.detach({ providerSessionId: oldSessionId, threadId });
2361
2362	assert.equal(
2363	(yield* mcpSessions.read(threadId))?.providerSessionId,
2364	replacement?.providerSessionId,
2365	);
2366	assert.equal((yield* registry.resolve(replacementToken!))?.thread.threadId, threadId);
2367	});
2368
2369	yield* effect.pipe(
2370	Effect.provide(
2371	layerTest({
2372	state,
2373	idleTimeoutMs: 1_000,
2374	capabilities: ExclusiveCapabilities,
2375	mcpConfigs,
2376	}),
2377	),
2378	);
2379	}),
2380	);
2381
2382	it.effect(
2383	"ProviderSessionManagerV2 detach of a superseded live session preserves replacement MCP credentials",
2384	() =>
2385	Effect.gen(function* () {
2386	const state = yield* Ref.make(emptyState);
2387	const mcpConfigs = yield* Ref.make<
2388	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2389	>([]);
2390	const effect = Effect.gen(function* () {
2391	const eventSink = yield* EventSink.EventSinkV2;
2392	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2393	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2394	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2395	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2396	const now = yield* DateTime.now;
2397	const threadId = ThreadId.make("thread-provider-session-manager-superseded-mcp");
2398	const oldSessionId = yield* idAllocator.allocate.providerSession({
2399	providerInstanceId: modelSelection.instanceId,
2400	threadId,
2401	});
2402	const replacementSessionId = yield* idAllocator.allocate.providerSession({
2403	providerInstanceId: modelSelection.instanceId,
2404	threadId,
2405	});
2406
2407	yield* eventSink.write({
2408	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2409	});
2410	yield* manager.open({
2411	threadId,
2412	providerSessionId: oldSessionId,
2413	modelSelection,
2414	runtimePolicy,
2415	});
2416	// The replacement opens while the old session is still attached: this is
2417	// the workspace-handoff sequence, where the queued continuation run can
2418	// start its session before the outbox executes the old session's detach.
2419	yield* manager.open({
2420	threadId,
2421	providerSessionId: replacementSessionId,
2422	modelSelection,
2423	runtimePolicy,
2424	});
2425
2426	const replacement = (yield* Ref.get(mcpConfigs)).at(-1);
2427	assert.isDefined(replacement);
2428	const replacementToken = replacement?.authorizationHeader.replace(/^Bearer\s+/, "");
2429	assert.isDefined(replacementToken);
2430	assert.equal(
2431	(yield* mcpSessions.read(threadId))?.providerSessionId,
2432	replacement?.providerSessionId,
2433	);
2434
2435	// First (non-duplicate) detach of the superseded session must not revoke
2436	// the replacement's credential or clear its config slot.
2437	yield* manager.detach({ providerSessionId: oldSessionId, threadId });
2438
2439	assert.equal(
2440	(yield* mcpSessions.read(threadId))?.providerSessionId,
2441	replacement?.providerSessionId,
2442	);
2443	assert.equal((yield* registry.resolve(replacementToken!))?.thread.threadId, threadId);
2444	});
2445
2446	yield* effect.pipe(
2447	Effect.provide(
2448	layerTest({
2449	state,
2450	idleTimeoutMs: 1_000,
2451	capabilities: ExclusiveCapabilities,
2452	mcpConfigs,
2453	}),
2454	),
2455	);
2456	}),
2457	);
2458
2459	it.effect(
2460	"ProviderSessionManagerV2 keeps a thread's MCP credential stable across detach and re-attach on a shared session",
2461	() =>
2462	Effect.gen(function* () {
2463	const state = yield* Ref.make(emptyState);
2464	const mcpConfigs = yield* Ref.make<
2465	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2466	>([]);
2467	const effect = Effect.gen(function* () {
2468	const eventSink = yield* EventSink.EventSinkV2;
2469	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2470	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2471	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2472	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2473	const now = yield* DateTime.now;
2474	const threadId = ThreadId.make("thread-provider-session-manager-stable-mcp");
2475	const providerSessionId = yield* idAllocator.allocate.providerSession({
2476	providerInstanceId: modelSelection.instanceId,
2477	threadId,
2478	});
2479
2480	yield* eventSink.write({
2481	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2482	});
2483	yield* manager.open({
2484	threadId,
2485	providerSessionId,
2486	modelSelection,
2487	runtimePolicy,
2488	});
2489
2490	const original = (yield* Ref.get(mcpConfigs)).at(-1);
2491	assert.isDefined(original);
2492	const originalToken = original?.authorizationHeader.replace(/^Bearer\s+/, "");
2493	assert.isDefined(originalToken);
2494
2495	// Workspace-change handoff on a shared multi-thread session (codex):
2496	// the thread detaches while the provider process keeps running, and the
2497	// process's MCP client keeps using the credential it was started with.
2498	yield* manager.detach({ providerSessionId, threadId, detail: "Workspace changed." });
2499	assert.equal(
2500	(yield* registry.resolve(originalToken!))?.thread.threadId,
2501	threadId,
2502	"detach must not revoke the credential the live provider process still holds",
2503	);
2504
2505	// The continuation run re-attaches the same thread to the same session;
2506	// the credential must be reused, not rotated, so the provider process's
2507	// long-lived MCP client stays authorized.
2508	yield* manager.open({
2509	threadId,
2510	providerSessionId,
2511	modelSelection,
2512	runtimePolicy,
2513	});
2514	assert.equal(
2515	(yield* mcpSessions.read(threadId))?.providerSessionId,
2516	original?.providerSessionId,
2517	"re-attach must reuse the existing credential, not rotate it",
2518	);
2519	assert.equal((yield* registry.resolve(originalToken!))?.thread.threadId, threadId);
2520
2521	// Releasing the session (provider process gone) still revokes.
2522	yield* manager.close(providerSessionId);
2523	assert.isUndefined(yield* registry.resolve(originalToken!));
2524	});
2525
2526	yield* effect.pipe(
2527	Effect.provide(
2528	layerTest({
2529	state,
2530	idleTimeoutMs: 1_000,
2531	mcpConfigs,
2532	}),
2533	),
2534	);
2535	}),
2536	);
2537
2538	it.effect(
2539	"ProviderSessionManagerV2 revokes a rotated credential despite a stale record on another live session",
2540	() =>
2541	Effect.gen(function* () {
2542	const state = yield* Ref.make(emptyState);
2543	const mcpConfigs = yield* Ref.make<
2544	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2545	>([]);
2546	const effect = Effect.gen(function* () {
2547	const eventSink = yield* EventSink.EventSinkV2;
2548	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2549	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2550	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2551	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2552	const now = yield* DateTime.now;
2553	const threadId = ThreadId.make("thread-provider-session-manager-stale-record");
2554	const s1 = yield* idAllocator.allocate.providerSession({
2555	providerInstanceId: modelSelection.instanceId,
2556	threadId,
2557	});
2558	const s2 = yield* idAllocator.allocate.providerSession({
2559	providerInstanceId: modelSelection.instanceId,
2560	threadId,
2561	});
2562
2563	yield* eventSink.write({
2564	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2565	});
2566	// S1 (shared session) records credential C1 for the thread, then the
2567	// thread detaches; S1 stays alive with the stale record.
2568	yield* manager.open({ threadId, providerSessionId: s1, modelSelection, runtimePolicy });
2569	yield* manager.detach({ providerSessionId: s1, threadId });
2570
2571	// The credential dies externally, so S2's attach must rotate to C2.
2572	yield* registry.revokeThread(threadId);
2573	yield* manager.open({ threadId, providerSessionId: s2, modelSelection, runtimePolicy });
2574	const rotated = yield* mcpSessions.read(threadId);
2575	assert.isDefined(rotated);
2576	const rotatedToken = rotated?.authorizationHeader.replace(/^Bearer\s+/, "");
2577	assert.isDefined(yield* registry.resolve(rotatedToken!));
2578
2579	// Releasing S2 must revoke C2 even though S1 still carries a stale
2580	// record (of dead C1) for the same thread.
2581	yield* manager.close(s2);
2582	assert.isUndefined(
2583	yield* registry.resolve(rotatedToken!),
2584	"stale record on S1 must not veto revoking S2's rotated credential",
2585	);
2586	yield* manager.close(s1);
2587	});
2588
2589	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1_000, mcpConfigs })));
2590	}),
2591	);
2592
2593	it.effect(
2594	"ProviderSessionManagerV2 protects a reused credential from a predecessor release during open",
2595	() =>
2596	Effect.gen(function* () {
2597	const state = yield* Ref.make(emptyState);
2598	const mcpConfigs = yield* Ref.make<
2599	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2600	>([]);
2601	const duringOpen = yield* Ref.make<Effect.Effect<void>>(Effect.void);
2602	const effect = Effect.gen(function* () {
2603	const eventSink = yield* EventSink.EventSinkV2;
2604	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2605	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2606	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2607	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2608	const now = yield* DateTime.now;
2609	const threadId = ThreadId.make("thread-provider-session-manager-open-race");
2610	const s1 = yield* idAllocator.allocate.providerSession({
2611	providerInstanceId: modelSelection.instanceId,
2612	threadId,
2613	});
2614	const s2 = yield* idAllocator.allocate.providerSession({
2615	providerInstanceId: modelSelection.instanceId,
2616	threadId,
2617	});
2618
2619	yield* eventSink.write({
2620	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2621	});
2622	yield* manager.open({ threadId, providerSessionId: s1, modelSelection, runtimePolicy });
2623	const original = (yield* Ref.get(mcpConfigs)).at(-1);
2624	const originalToken = original?.authorizationHeader.replace(/^Bearer\s+/, "");
2625	assert.isDefined(originalToken);
2626	yield* manager.detach({ providerSessionId: s1, threadId });
2627
2628	// While S2's provider process is spawning (after prepare reused the
2629	// credential, before the entry is visible), the predecessor session
2630	// releases. Eager adapters (ACP, OpenCode) bake the credential into
2631	// the process during openSession, so the release must not revoke it;
2632	// rotating afterwards cannot repair those adapters.
2633	yield* Ref.set(duringOpen, manager.close(s1).pipe(Effect.orDie));
2634	yield* manager.open({ threadId, providerSessionId: s2, modelSelection, runtimePolicy });
2635
2636	const slot = yield* mcpSessions.read(threadId);
2637	assert.equal(
2638	slot?.providerSessionId,
2639	original?.providerSessionId,
2640	"the credential the adapter was configured with must remain current",
2641	);
2642	assert.equal(
2643	(yield* registry.resolve(originalToken!))?.thread.threadId,
2644	threadId,
2645	"the predecessor release must not revoke a credential reserved by an in-flight open",
2646	);
2647	yield* manager.close(s2);
2648	});
2649
2650	yield* effect.pipe(
2651	Effect.provide(
2652	layerTest({
2653	state,
2654	idleTimeoutMs: 1_000,
2655	mcpConfigs,
2656	beforeOpen: (input) =>
2657	input.providerSessionId === undefined
2658	? Effect.void
2659	: Ref.get(duringOpen).pipe(
2660	Effect.flatten,
2661	Effect.tap(() => Ref.set(duringOpen, Effect.void)),
2662	),
2663	}),
2664	),
2665	);
2666	}),
2667	);
2668
2669	it.effect("ProviderSessionManagerV2 terminal detach revokes the thread's MCP credential", () =>
2670	Effect.gen(function* () {
2671	const state = yield* Ref.make(emptyState);
2672	const mcpConfigs = yield* Ref.make<
2673	ReadonlyArray<McpProviderSession.McpProviderSessionConfig | undefined>
2674	>([]);
2675	const effect = Effect.gen(function* () {
2676	const eventSink = yield* EventSink.EventSinkV2;
2677	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2678	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2679	const registry = yield* McpSessionRegistry.McpSessionRegistry;
2680	const mcpSessions = yield* McpProviderSessions.McpProviderSessions;
2681	const now = yield* DateTime.now;
2682	const threadId = ThreadId.make("thread-provider-session-manager-terminal-detach");
2683	const providerSessionId = yield* idAllocator.allocate.providerSession({
2684	providerInstanceId: modelSelection.instanceId,
2685	threadId,
2686	});
2687
2688	yield* eventSink.write({
2689	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2690	});
2691	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2692	const issued = (yield* Ref.get(mcpConfigs)).at(-1);
2693	const token = issued?.authorizationHeader.replace(/^Bearer\s+/, "");
2694	assert.isDefined(yield* registry.resolve(token!));
2695
2696	// Archive/delete detaches carry revokeMcpCredential: the token must die
2697	// with the thread even though the shared provider process lives on.
2698	yield* manager.detach({
2699	providerSessionId,
2700	threadId,
2701	detail: "Thread deleted.",
2702	revokeMcpCredential: true,
2703	});
2704	assert.isUndefined(yield* registry.resolve(token!));
2705	assert.isUndefined(yield* mcpSessions.read(threadId));
2706	});
2707
2708	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1_000, mcpConfigs })));
2709	}),
2710	);
2711
2712	it.effect("ProviderSessionManagerV2 releases idle sessions without sweeping all sessions", () =>
2713	Effect.gen(function* () {
2714	const state = yield* Ref.make(emptyState);
2715	const effect = Effect.gen(function* () {
2716	const eventSink = yield* EventSink.EventSinkV2;
2717	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2718	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2719	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2720	const now = yield* DateTime.now;
2721	const projectId = yield* idAllocator.allocate.project({
2722	fixtureName: "provider-session-manager-idle",
2723	});
2724	const threadId = yield* idAllocator.allocate.thread({
2725	fixtureName: "provider-session-manager-idle",
2726	projectId,
2727	});
2728	const providerSessionId = yield* idAllocator.allocate.providerSession({
2729	providerInstanceId: modelSelection.instanceId,
2730	threadId,
2731	});
2732
2733	yield* eventSink.write({
2734	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2735	});
2736	yield* manager.open({
2737	threadId,
2738	providerSessionId,
2739	modelSelection,
2740	runtimePolicy,
2741	});
2742
2743	yield* TestClock.adjust("1 second");
2744	yield* Effect.yieldNow;
2745
2746	const liveSession = yield* manager.get(providerSessionId);
2747	const runtimeState = yield* Ref.get(state);
2748	const projection = yield* projectionStore.getThreadProjection(threadId);
2749
2750	assert.isTrue(Option.isNone(liveSession));
2751	assert.equal(runtimeState.openCount, 1);
2752	assert.equal(runtimeState.closeCount, 1);
2753	assert.equal(projection.providerSessions.at(-1)?.status, "stopped");
2754	});
2755
2756	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
2757	}),
2758	);
2759
2760	it.effect("ProviderSessionManagerV2 persists release when session scope close hangs", () =>
2761	Effect.gen(function* () {
2762	const state = yield* Ref.make(emptyState);
2763	const effect = Effect.gen(function* () {
2764	const eventSink = yield* EventSink.EventSinkV2;
2765	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2766	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2767	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2768	const now = yield* DateTime.now;
2769	const threadId = yield* idAllocator.allocate.thread({
2770	fixtureName: "provider-session-manager-hung-close",
2771	projectId: yield* idAllocator.allocate.project({
2772	fixtureName: "provider-session-manager-hung-close",
2773	}),
2774	});
2775	const providerSessionId = yield* idAllocator.allocate.providerSession({
2776	providerInstanceId: modelSelection.instanceId,
2777	threadId,
2778	});
2779
2780	yield* eventSink.write({
2781	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2782	});
2783	yield* manager.open({
2784	threadId,
2785	providerSessionId,
2786	modelSelection,
2787	runtimePolicy,
2788	});
2789
2790	yield* TestClock.adjust("1 second");
2791	yield* Effect.yieldNow;
2792	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
2793
2794	yield* TestClock.adjust("30 seconds");
2795	yield* Effect.yieldNow;
2796	const projection = yield* projectionStore.getThreadProjection(threadId);
2797	assert.equal(projection.providerSessions.at(-1)?.status, "stopped");
2798	assert.equal((yield* Ref.get(state)).closeCount, 0);
2799	});
2800
2801	yield* effect.pipe(
2802	Effect.provide(layerTest({ state, idleTimeoutMs: 1000, hangSessionScopeClose: true })),
2803	);
2804	}),
2805	);
2806
2807	// Seeds a thread with a background command on a provider thread of the
2808	// session it then opens, and one on a provider thread of another session.
2809	const openSessionWithBackgroundCommand = (fixtureName: string) =>
2810	Effect.gen(function* () {
2811	const eventSink = yield* EventSink.EventSinkV2;
2812	const idAllocator = yield* IdAllocator.IdAllocatorV2;
2813	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2814	const now = yield* DateTime.now;
2815	const threadId = yield* idAllocator.allocate.thread({
2816	fixtureName,
2817	projectId: yield* idAllocator.allocate.project({ fixtureName }),
2818	});
2819	const providerSessionId = yield* idAllocator.allocate.providerSession({
2820	providerInstanceId: modelSelection.instanceId,
2821	threadId,
2822	});
2823	const otherProviderSessionId = yield* idAllocator.allocate.providerSession({
2824	providerInstanceId: modelSelection.instanceId,
2825	threadId,
2826	});
2827	const pendingBackgroundTasks = [
2828	{ taskId: "dev-server", kind: "command", description: "python3 -m http.server" },
2829	] as const;
2830	const providerThreads = [
2831	{
2832	...makeProviderThread({ idAllocator, threadId, providerSessionId, now }),
2833	pendingBackgroundTasks,
2834	},
2835	{
2836	...makeProviderThread({
2837	idAllocator,
2838	threadId,
2839	providerSessionId: otherProviderSessionId,
2840	now,
2841	nativeThreadId: "native-thread-other-session",
2842	}),
2843	pendingBackgroundTasks,
2844	},
2845	];
2846	const writeProviderThread = (payload: OrchestrationV2ProviderThread) =>
2847	Effect.gen(function* () {
2848	yield* eventSink.write({
2849	events: [
2850	{
2851	id: yield* idAllocator.allocate.event({ threadId }),
2852	type: "provider-thread.updated",
2853	threadId,
2854	driver: CODEX_DRIVER,
2855	occurredAt: now,
2856	payload,
2857	},
2858	],
2859	});
2860	});
2861	yield* eventSink.write({
2862	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
2863	});
2864	yield* Effect.forEach(providerThreads, writeProviderThread);
2865	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
2866	// The task ids each seeded provider thread lists now.
2867	const rosters = Effect.gen(function* () {
2868	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
2869	const projection = yield* projectionStore.getThreadProjection(threadId);
2870	return providerThreads.map((seeded) =>
2871	(
2872	projection.providerThreads.find((thread) => thread.id === seeded.id)
2873	?.pendingBackgroundTasks ?? []
2874	).map((task) => task.taskId),
2875	);
2876	});
2877	return { threadId, providerSessionId, providerThreads, writeProviderThread, rosters };
2878	});
2879
2880	it.effect(
2881	"ProviderSessionManagerV2 clears the background roster of a session a detach releases",
2882	() =>
2883	Effect.gen(function* () {
2884	const state = yield* Ref.make(emptyState);
2885	const effect = Effect.gen(function* () {
2886	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2887	const { threadId, providerSessionId, rosters } = yield* openSessionWithBackgroundCommand(
2888	"provider-session-manager-detach-roster",
2889	);
2890	assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]);
2891
2892	// Settle and archive detach the thread, which releases an exclusive
2893	// (Claude-like) session.
2894	yield* manager.detach({ providerSessionId, threadId, detail: "Thread settled." });
2895
2896	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
2897	assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]);
2898	});
2899
2900	yield* effect.pipe(
2901	Effect.provide(
2902	layerTest({ state, idleTimeoutMs: 60_000, capabilities: ExclusiveCapabilities }),
2903	),
2904	);
2905	}),
2906	);
2907
2908	it.effect("ProviderSessionManagerV2 checks liveness without delaying idle release", () =>
2909	Effect.gen(function* () {
2910	const state = yield* Ref.make(emptyState);
2911	const effect = Effect.gen(function* () {
2912	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2913	const { providerSessionId } = yield* openSessionWithBackgroundCommand(
2914	"provider-session-manager-is-live",
2915	);
2916	yield* TestClock.adjust("600 millis");
2917	assert.isTrue(yield* manager.isLive(providerSessionId));
2918	yield* TestClock.adjust("600 millis");
2919	yield* Effect.yieldNow;
2920	assert.isFalse(yield* manager.isLive(providerSessionId));
2921	});
2922
2923	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
2924	}),
2925	);
2926
2927	it.effect("ProviderSessionManagerV2 clears the background roster of an idle-released session", () =>
2928	Effect.gen(function* () {
2929	const state = yield* Ref.make(emptyState);
2930	const effect = Effect.gen(function* () {
2931	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2932	const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand(
2933	"provider-session-manager-idle-roster",
2934	);
2935	assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]);
2936
2937	yield* TestClock.adjust("1 second");
2938	yield* Effect.yieldNow;
2939
2940	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
2941	assert.deepStrictEqual(yield* rosters, [[], ["dev-server"]]);
2942	});
2943
2944	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
2945	}),
2946	);
2947
2948	it.effect("ProviderSessionManagerV2 keeps the roster of a runtime whose tasks can outlive it", () =>
2949	Effect.gen(function* () {
2950	const state = yield* Ref.make(emptyState);
2951	const effect = Effect.gen(function* () {
2952	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2953	const { providerSessionId, rosters } = yield* openSessionWithBackgroundCommand(
2954	"provider-session-manager-outliving-roster",
2955	);
2956
2957	// Like OpenCode, whose background subagents run on a shared server.
2958	yield* manager.close(providerSessionId);
2959
2960	assert.deepStrictEqual(yield* rosters, [["dev-server"], ["dev-server"]]);
2961	});
2962
2963	yield* effect.pipe(
2964	Effect.provide(
2965	layerTest({ state, idleTimeoutMs: 60_000, backgroundTasksEndWithProcess: false }),
2966	),
2967	);
2968	}),
2969	);
2970
2971	it.effect(
2972	"ProviderSessionManagerV2 keeps background tasks a replacement reports during release",
2973	() =>
2974	Effect.gen(function* () {
2975	const state = yield* Ref.make(emptyState);
2976	const scopeCloseReached = yield* Deferred.make<void>();
2977	const effect = Effect.gen(function* () {
2978	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
2979	const { providerSessionId, providerThreads, writeProviderThread, rosters } =
2980	yield* openSessionWithBackgroundCommand("provider-session-manager-replacement-roster");
2981	const release = yield* manager.close(providerSessionId).pipe(Effect.forkChild);
2982
2983	// While the old process closes, a replacement session with the same id
2984	// reports a task of its own next to the dead one.
2985	yield* Deferred.await(scopeCloseReached);
2986	yield* writeProviderThread({
2987	...providerThreads[0]!,
2988	pendingBackgroundTasks: [
2989	...providerThreads[0]!.pendingBackgroundTasks,
2990	{ taskId: "replacement-task", kind: "command" },
2991	],
2992	});
2993	yield* TestClock.adjust("30 seconds");
2994	yield* Fiber.join(release);
2995
2996	assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]);
2997	});
2998
2999	yield* effect.pipe(
3000	Effect.provide(
3001	layerTest({
3002	state,
3003	idleTimeoutMs: 60_000,
3004	hangSessionScopeClose: true,
3005	scopeCloseReached,
3006	}),
3007	),
3008	);
3009	}),
3010	);
3011
3012	it.effect("ProviderSessionManagerV2 keeps a provider update that lands while release clears", () =>
3013	Effect.gen(function* () {
3014	const state = yield* Ref.make(emptyState);
3015	const beforeUnchangedWrite = yield* Ref.make<Effect.Effect<void>>(Effect.void);
3016	const effect = Effect.gen(function* () {
3017	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3018	const { providerSessionId, providerThreads, writeProviderThread, rosters } =
3019	yield* openSessionWithBackgroundCommand("provider-session-manager-raced-roster");
3020	// A replacement reports a task after cleanup read the row, before it
3021	// writes, without moving updatedAt (not every writer does).
3022	const seeded = providerThreads[0]!;
3023	yield* Ref.set(
3024	beforeUnchangedWrite,
3025	writeProviderThread({
3026	...seeded,
3027	pendingBackgroundTasks: [
3028	...seeded.pendingBackgroundTasks,
3029	{ taskId: "replacement-task", kind: "command" },
3030	],
3031	}).pipe(Effect.orDie),
3032	);
3033
3034	yield* manager.close(providerSessionId);
3035
3036	assert.deepStrictEqual(yield* rosters, [["replacement-task"], ["dev-server"]]);
3037	});
3038
3039	yield* effect.pipe(
3040	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, beforeUnchangedWrite })),
3041	);
3042	}),
3043	);
3044
3045	it.effect("ProviderSessionManagerV2 defers idle release while background work is pending", () =>
3046	Effect.gen(function* () {
3047	const state = yield* Ref.make(emptyState);
3048	const pendingWork = yield* Ref.make(true);
3049	const effect = Effect.gen(function* () {
3050	const eventSink = yield* EventSink.EventSinkV2;
3051	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3052	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3053	const now = yield* DateTime.now;
3054	const threadId = yield* idAllocator.allocate.thread({
3055	fixtureName: "provider-session-manager-idle-pin",
3056	projectId: yield* idAllocator.allocate.project({
3057	fixtureName: "provider-session-manager-idle-pin",
3058	}),
3059	});
3060	const providerSessionId = yield* idAllocator.allocate.providerSession({
3061	providerInstanceId: modelSelection.instanceId,
3062	threadId,
3063	});
3064
3065	yield* eventSink.write({
3066	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3067	});
3068	yield* manager.open({
3069	threadId,
3070	providerSessionId,
3071	modelSelection,
3072	runtimePolicy,
3073	});
3074
3075	yield* TestClock.adjust("3 seconds");
3076	yield* Effect.yieldNow;
3077	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
3078	assert.equal((yield* Ref.get(state)).closeCount, 0);
3079
3080	yield* Ref.set(pendingWork, false);
3081	yield* TestClock.adjust("1 second");
3082	yield* Effect.yieldNow;
3083	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
3084	assert.equal((yield* Ref.get(state)).closeCount, 1);
3085	});
3086
3087	yield* effect.pipe(
3088	Effect.provide(
3089	layerTest({
3090	state,
3091	idleTimeoutMs: 1000,
3092	hasPendingBackgroundWork: Ref.get(pendingWork),
3093	}),
3094	),
3095	);
3096	}),
3097	);
3098
3099	it.effect("ProviderSessionManagerV2 releases pinned idle sessions once the pin cap expires", () =>
3100	Effect.gen(function* () {
3101	const state = yield* Ref.make(emptyState);
3102	const effect = Effect.gen(function* () {
3103	const eventSink = yield* EventSink.EventSinkV2;
3104	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3105	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3106	const now = yield* DateTime.now;
3107	const threadId = yield* idAllocator.allocate.thread({
3108	fixtureName: "provider-session-manager-pin-cap",
3109	projectId: yield* idAllocator.allocate.project({
3110	fixtureName: "provider-session-manager-pin-cap",
3111	}),
3112	});
3113	const providerSessionId = yield* idAllocator.allocate.providerSession({
3114	providerInstanceId: modelSelection.instanceId,
3115	threadId,
3116	});
3117
3118	yield* eventSink.write({
3119	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3120	});
3121	yield* manager.open({
3122	threadId,
3123	providerSessionId,
3124	modelSelection,
3125	runtimePolicy,
3126	});
3127
3128	yield* TestClock.adjust("3 seconds");
3129	yield* Effect.yieldNow;
3130	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
3131
3132	yield* TestClock.adjust("1 second");
3133	yield* Effect.yieldNow;
3134	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
3135	assert.equal((yield* Ref.get(state)).closeCount, 1);
3136	});
3137
3138	yield* effect.pipe(
3139	Effect.provide(
3140	layerTest({
3141	state,
3142	idleTimeoutMs: 1000,
3143	maxIdlePinMs: 3000,
3144	hasPendingBackgroundWork: Effect.succeed(true),
3145	}),
3146	),
3147	);
3148	}),
3149	);
3150
3151	it.effect("ProviderSessionManagerV2 keeps pinned work that reports progress", () =>
3152	Effect.gen(function* () {
3153	const state = yield* Ref.make(emptyState);
3154	const effect = Effect.gen(function* () {
3155	const eventSink = yield* EventSink.EventSinkV2;
3156	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3157	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3158	const now = yield* DateTime.now;
3159	const threadId = yield* idAllocator.allocate.thread({
3160	fixtureName: "provider-session-manager-pin-progress",
3161	projectId: yield* idAllocator.allocate.project({
3162	fixtureName: "provider-session-manager-pin-progress",
3163	}),
3164	});
3165	const providerSessionId = yield* idAllocator.allocate.providerSession({
3166	providerInstanceId: modelSelection.instanceId,
3167	threadId,
3168	});
3169	yield* eventSink.write({
3170	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3171	});
3172	const runtime = yield* manager.open({
3173	threadId,
3174	providerSessionId,
3175	modelSelection,
3176	runtimePolicy,
3177	});
3178	const adapterQueue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
3179	assert.isDefined(adapterQueue);
3180	const step = (duration: Parameters<typeof TestClock.adjust>[0]) =>
3181	Effect.gen(function* () {
3182	yield* TestClock.adjust(duration);
3183	for (let i = 0; i < 20; i += 1) yield* Effect.yieldNow;
3184	});
3185
3186	// A long workflow keeps reporting progress well past the pin cap.
3187	for (let tick = 0; tick < 12; tick += 1) {
3188	yield* step("1500 millis");
3189	yield* Queue.offer(adapterQueue!, {
3190	type: "provider_session.updated",
3191	driver: CODEX_DRIVER,
3192	providerSession: runtime.providerSession,
3193	});
3194	for (let i = 0; i < 20; i += 1) yield* Effect.yieldNow;
3195	}
3196	assert.equal((yield* Ref.get(state)).closeCount, 0);
3197
3198	// Work that goes silent past the cap is still stopped.
3199	for (let tick = 0; tick < 6; tick += 1) {
3200	yield* step("1 second");
3201	}
3202	assert.equal((yield* Ref.get(state)).closeCount, 1);
3203	});
3204
3205	yield* effect.pipe(
3206	Effect.provide(
3207	layerTest({
3208	state,
3209	idleTimeoutMs: 1000,
3210	maxIdlePinMs: 3000,
3211	hasPendingBackgroundWork: Effect.succeed(true),
3212	}),
3213	),
3214	);
3215	}),
3216	);
3217
3218	it.effect(
3219	"ProviderSessionManagerV2 does not idle-release a session that turns busy during the pending-work check",
3220	() =>
3221	Effect.gen(function* () {
3222	const state = yield* Ref.make(emptyState);
3223	const firstCheck = yield* Ref.make(true);
3224	const checkEntered = yield* Deferred.make<void>();
3225	const checkGate = yield* Deferred.make<void>();
3226	const effect = Effect.gen(function* () {
3227	const eventSink = yield* EventSink.EventSinkV2;
3228	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3229	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3230	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3231	const now = yield* DateTime.now;
3232	const projectId = yield* idAllocator.allocate.project({
3233	fixtureName: "provider-session-manager-busy-during-check",
3234	});
3235	const threadId = yield* idAllocator.allocate.thread({
3236	fixtureName: "provider-session-manager-busy-during-check",
3237	projectId,
3238	});
3239	const providerSessionId = yield* idAllocator.allocate.providerSession({
3240	providerInstanceId: modelSelection.instanceId,
3241	threadId,
3242	});
3243	const providerThread = makeProviderThread({
3244	idAllocator,
3245	threadId,
3246	providerSessionId,
3247	now,
3248	});
3249	const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
3250	const attemptId = idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 });
3251	const rootNodeId = idAllocator.derive.rootNode({ runId });
3252	const providerTurnId = idAllocator.derive.providerTurn({
3253	driver: CODEX_DRIVER,
3254	nativeTurnId: "native-turn-busy-during-check",
3255	});
3256
3257	yield* eventSink.write({
3258	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3259	});
3260	const runtime = yield* manager.open({
3261	threadId,
3262	providerSessionId,
3263	modelSelection,
3264	runtimePolicy,
3265	});
3266	yield* runtime.events.pipe(Stream.runDrain, Effect.forkScoped);
3267	const appThread = (yield* projectionStore.getThreadProjection(threadId)).thread;
3268
3269	yield* TestClock.adjust("1 second");
3270	yield* Deferred.await(checkEntered);
3271
3272	// The release fiber is parked inside the pending-work check, so the
3273	// idle decision it already made is stale once this turn marks the
3274	// session busy.
3275	const turnFiber = yield* runtime
3276	.startTurn({
3277	appThread,
3278	threadId,
3279	runId,
3280	runOrdinal: 1,
3281	providerTurnOrdinal: 1,
3282	attemptId,
3283	rootNodeId,
3284	providerThread,
3285	message: {
3286	createdBy: "user",
3287	creationSource: "web",
3288	messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
3289	text: "hello",
3290	attachments: [],
3291	},
3292	modelSelection,
3293	runtimePolicy,
3294	})
3295	.pipe(Effect.forkDetach);
3296	for (let i = 0; i < 10; i += 1) {
3297	yield* Effect.yieldNow;
3298	}
3299	yield* Deferred.succeed(checkGate, undefined);
3300	yield* Fiber.join(turnFiber);
3301	yield* Effect.yieldNow;
3302
3303	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
3304	assert.equal((yield* Ref.get(state)).closeCount, 0);
3305
3306	const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
3307	assert.isDefined(queue);
3308	yield* Queue.offer(queue!, {
3309	type: "turn.terminal",
3310	driver: CODEX_DRIVER,
3311	providerThreadId: providerThread.id,
3312	providerTurnId,
3313	runOrdinal: 1,
3314	status: "completed",
3315	failure: null,
3316	threadDisposition: "reusable",
3317	});
3318	yield* TestClock.adjust("1 second");
3319	yield* Effect.yieldNow;
3320	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
3321	assert.equal((yield* Ref.get(state)).closeCount, 1);
3322	});
3323
3324	yield* effect.pipe(
3325	Effect.provide(
3326	layerTest({
3327	state,
3328	idleTimeoutMs: 1000,
3329	// Uninterruptible so the markBusy-triggered interrupt cannot land
3330	// inside the check, mirroring an adapter that masks interruption
3331	// while inspecting its own state.
3332	hasPendingBackgroundWork: Effect.uninterruptible(
3333	Effect.gen(function* () {
3334	if (yield* Ref.getAndSet(firstCheck, false)) {
3335	yield* Deferred.succeed(checkEntered, undefined);
3336	yield* Deferred.await(checkGate);
3337	}
3338	return false;
3339	}),
3340	),
3341	}),
3342	),
3343	);
3344	}),
3345	);
3346
3347	it.effect("ProviderSessionManagerV2 does not apply a stale idle pin to a replacement session", () =>
3348	Effect.gen(function* () {
3349	const state = yield* Ref.make(emptyState);
3350	const firstCheck = yield* Ref.make(true);
3351	const checkEntered = yield* Deferred.make<void>();
3352	const checkGate = yield* Deferred.make<void>();
3353	const effect = Effect.gen(function* () {
3354	const eventSink = yield* EventSink.EventSinkV2;
3355	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3356	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3357	const now = yield* DateTime.now;
3358	const threadId = yield* idAllocator.allocate.thread({
3359	fixtureName: "provider-session-manager-stale-pin",
3360	projectId: yield* idAllocator.allocate.project({
3361	fixtureName: "provider-session-manager-stale-pin",
3362	}),
3363	});
3364	const providerSessionId = yield* idAllocator.allocate.providerSession({
3365	providerInstanceId: modelSelection.instanceId,
3366	threadId,
3367	});
3368
3369	yield* eventSink.write({
3370	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3371	});
3372	yield* manager.open({
3373	threadId,
3374	providerSessionId,
3375	modelSelection,
3376	runtimePolicy,
3377	});
3378
3379	// Park the first idle fiber inside an uninterruptible pending-work probe.
3380	yield* TestClock.adjust("1 second");
3381	yield* Deferred.await(checkEntered);
3382
3383	// close removes the map entry first, then waits to interrupt the idle
3384	// fiber (still uninterruptible). That window lets a replacement open
3385	// under the same providerSessionId before the stale probe finishes.
3386	const closeFiber = yield* manager.close(providerSessionId).pipe(Effect.forkDetach);
3387	for (let i = 0; i < 20; i += 1) {
3388	yield* Effect.yieldNow;
3389	}
3390	yield* manager.open({
3391	threadId,
3392	providerSessionId,
3393	modelSelection,
3394	runtimePolicy,
3395	});
3396	assert.equal((yield* Ref.get(state)).openCount, 2);
3397
3398	// Stale probe reports pending work against the old runtime; the pin
3399	// stamp must no-op on the replacement (runtime / generation mismatch).
3400	yield* Deferred.succeed(checkGate, undefined);
3401	yield* Fiber.join(closeFiber);
3402	for (let i = 0; i < 10; i += 1) {
3403	yield* Effect.yieldNow;
3404	}
3405
3406	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
3407	assert.equal((yield* Ref.get(state)).closeCount, 1);
3408
3409	// Replacement has no pending background work. After one idle window it
3410	// must release. A stale pin stamp would have deferred release until
3411	// maxIdlePinMs.
3412	yield* TestClock.adjust("1 second");
3413	yield* Effect.yieldNow;
3414	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
3415	assert.equal((yield* Ref.get(state)).closeCount, 2);
3416	});
3417
3418	yield* effect.pipe(
3419	Effect.provide(
3420	layerTest({
3421	state,
3422	idleTimeoutMs: 1000,
3423	maxIdlePinMs: 60_000,
3424	hasPendingBackgroundWork: Effect.uninterruptible(
3425	Effect.gen(function* () {
3426	if (yield* Ref.getAndSet(firstCheck, false)) {
3427	yield* Deferred.succeed(checkEntered, undefined);
3428	yield* Deferred.await(checkGate);
3429	return true;
3430	}
3431	return false;
3432	}),
3433	),
3434	}),
3435	),
3436	);
3437	}),
3438	);
3439
3440	it.effect(
3441	"ProviderSessionManagerV2 keeps active sessions alive until the provider turn terminates",
3442	() =>
3443	Effect.gen(function* () {
3444	const state = yield* Ref.make(emptyState);
3445	const effect = Effect.gen(function* () {
3446	const eventSink = yield* EventSink.EventSinkV2;
3447	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3448	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3449	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3450	const now = yield* DateTime.now;
3451	const projectId = yield* idAllocator.allocate.project({
3452	fixtureName: "provider-session-manager-active",
3453	});
3454	const threadId = yield* idAllocator.allocate.thread({
3455	fixtureName: "provider-session-manager-active",
3456	projectId,
3457	});
3458	const providerSessionId = yield* idAllocator.allocate.providerSession({
3459	providerInstanceId: modelSelection.instanceId,
3460	threadId,
3461	});
3462	const providerThread = makeProviderThread({
3463	idAllocator,
3464	threadId,
3465	providerSessionId,
3466	now,
3467	});
3468	const runId = idAllocator.derive.run({ threadId, ordinal: 1 });
3469	const attemptId = idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 });
3470	const rootNodeId = idAllocator.derive.rootNode({ runId });
3471	const providerTurnId = idAllocator.derive.providerTurn({
3472	driver: CODEX_DRIVER,
3473	nativeTurnId: "native-turn",
3474	});
3475
3476	yield* eventSink.write({
3477	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3478	});
3479	const runtime = yield* manager.open({
3480	threadId,
3481	providerSessionId,
3482	modelSelection,
3483	runtimePolicy,
3484	});
3485	yield* runtime.events.pipe(Stream.runDrain, Effect.forkScoped);
3486	const appThread = (yield* projectionStore.getThreadProjection(threadId)).thread;
3487	yield* runtime.startTurn({
3488	appThread,
3489	threadId,
3490	runId,
3491	runOrdinal: 1,
3492	providerTurnOrdinal: 1,
3493	attemptId,
3494	rootNodeId,
3495	providerThread,
3496	message: {
3497	createdBy: "user",
3498	creationSource: "web",
3499	messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }),
3500	text: "hello",
3501	attachments: [],
3502	},
3503	modelSelection,
3504	runtimePolicy,
3505	});
3506
3507	yield* TestClock.adjust("2 seconds");
3508	yield* Effect.yieldNow;
3509	assert.equal((yield* Ref.get(state)).closeCount, 0);
3510
3511	const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
3512	assert.isDefined(queue);
3513	yield* Queue.offer(queue!, {
3514	type: "turn.terminal",
3515	driver: CODEX_DRIVER,
3516	providerThreadId: providerThread.id,
3517	providerTurnId,
3518	runOrdinal: 1,
3519	status: "completed",
3520	failure: null,
3521	threadDisposition: "reusable",
3522	});
3523	yield* TestClock.adjust("1 second");
3524	yield* Effect.yieldNow;
3525
3526	const liveSession = yield* manager.get(providerSessionId);
3527	const projection = yield* projectionStore.getThreadProjection(threadId);
3528	assert.isTrue(Option.isNone(liveSession));
3529	assert.equal((yield* Ref.get(state)).closeCount, 1);
3530	assert.equal(projection.providerSessions.at(-1)?.status, "stopped");
3531	});
3532
3533	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
3534	}),
3535	);
3536
3537	it.effect("ProviderSessionManagerV2 uses the same release path for runtime failures", () =>
3538	Effect.gen(function* () {
3539	const state = yield* Ref.make(emptyState);
3540	const effect = Effect.gen(function* () {
3541	const eventSink = yield* EventSink.EventSinkV2;
3542	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3543	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3544	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3545	const now = yield* DateTime.now;
3546	const projectId = yield* idAllocator.allocate.project({
3547	fixtureName: "provider-session-manager-runtime-error",
3548	});
3549	const threadId = yield* idAllocator.allocate.thread({
3550	fixtureName: "provider-session-manager-runtime-error",
3551	projectId,
3552	});
3553	const providerSessionId = yield* idAllocator.allocate.providerSession({
3554	providerInstanceId: modelSelection.instanceId,
3555	threadId,
3556	});
3557
3558	yield* eventSink.write({
3559	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3560	});
3561	yield* manager.open({
3562	threadId,
3563	providerSessionId,
3564	modelSelection,
3565	runtimePolicy,
3566	});
3567	yield* manager.release({
3568	providerSessionId,
3569	reason: "runtime_error",
3570	detail: "process exited",
3571	});
3572
3573	const liveSession = yield* manager.get(providerSessionId);
3574	const runtimeState = yield* Ref.get(state);
3575	const projection = yield* projectionStore.getThreadProjection(threadId);
3576
3577	assert.isTrue(Option.isNone(liveSession));
3578	assert.equal(runtimeState.closeCount, 1);
3579	assert.equal(projection.providerSessions.at(-1)?.status, "error");
3580	assert.equal(projection.providerSessions.at(-1)?.lastError, "process exited");
3581	});
3582
3583	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
3584	}),
3585	);
3586
3587	it.effect("ProviderSessionManagerV2 releases sessions when provider event streams fail", () =>
3588	Effect.gen(function* () {
3589	const state = yield* Ref.make(emptyState);
3590	const effect = Effect.gen(function* () {
3591	const eventSink = yield* EventSink.EventSinkV2;
3592	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3593	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3594	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3595	const now = yield* DateTime.now;
3596	const projectId = yield* idAllocator.allocate.project({
3597	fixtureName: "provider-session-manager-stream-error",
3598	});
3599	const threadId = yield* idAllocator.allocate.thread({
3600	fixtureName: "provider-session-manager-stream-error",
3601	projectId,
3602	});
3603	const providerSessionId = yield* idAllocator.allocate.providerSession({
3604	providerInstanceId: modelSelection.instanceId,
3605	threadId,
3606	});
3607
3608	yield* eventSink.write({
3609	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3610	});
3611	const runtime = yield* manager.open({
3612	threadId,
3613	providerSessionId,
3614	modelSelection,
3615	runtimePolicy,
3616	});
3617	yield* runtime.events.pipe(Stream.runDrain, Effect.ignore, Effect.forkScoped);
3618	yield* Effect.yieldNow;
3619
3620	const liveSession = yield* manager.get(providerSessionId);
3621	const runtimeState = yield* Ref.get(state);
3622	const projection = yield* projectionStore.getThreadProjection(threadId);
3623
3624	assert.isTrue(Option.isNone(liveSession));
3625	assert.equal(runtimeState.closeCount, 1);
3626	assert.equal(projection.providerSessions.at(-1)?.status, "error");
3627	});
3628
3629	yield* effect.pipe(
3630	Effect.provide(
3631	layerTest({
3632	state,
3633	idleTimeoutMs: 1000,
3634	failEventStream: true,
3635	}),
3636	),
3637	);
3638	}),
3639	);
3640
3641	it.effect("ProviderSessionManagerV2 marks pending runtime requests non-live on release", () =>
3642	Effect.gen(function* () {
3643	const state = yield* Ref.make(emptyState);
3644	const effect = Effect.gen(function* () {
3645	const eventSink = yield* EventSink.EventSinkV2;
3646	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3647	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3648	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3649	const now = yield* DateTime.now;
3650	const projectId = yield* idAllocator.allocate.project({
3651	fixtureName: "provider-session-manager-request-expire",
3652	});
3653	const threadId = yield* idAllocator.allocate.thread({
3654	fixtureName: "provider-session-manager-request-expire",
3655	projectId,
3656	});
3657	const providerSessionId = yield* idAllocator.allocate.providerSession({
3658	providerInstanceId: modelSelection.instanceId,
3659	threadId,
3660	});
3661	const providerThread = makeProviderThread({
3662	idAllocator,
3663	threadId,
3664	providerSessionId,
3665	now,
3666	});
3667
3668	yield* eventSink.write({
3669	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3670	});
3671	const pendingRequest = yield* makePendingRuntimeRequestEvents({
3672	idAllocator,
3673	threadId,
3674	providerSessionId,
3675	providerThread,
3676	now,
3677	});
3678	yield* eventSink.write({ events: pendingRequest.events });
3679	yield* manager.open({
3680	threadId,
3681	providerSessionId,
3682	modelSelection,
3683	runtimePolicy,
3684	});
3685	yield* manager.release({
3686	providerSessionId,
3687	reason: "runtime_error",
3688	detail: "process exited",
3689	});
3690
3691	const projection = yield* projectionStore.getThreadProjection(threadId);
3692	const request = projection.runtimeRequests.at(-1);
3693	const requestNode = projection.nodes.find((node) => node.id === request?.nodeId);
3694	const requestTurnItem = projection.turnItems.find(
3695	(item) => item.type === "approval_request" && item.requestId === request?.id,
3696	);
3697
3698	assert.equal(request?.status, "expired");
3699	assert.equal(request?.responseCapability.type, "not_resumable");
3700	assert.equal(requestNode?.status, "failed");
3701	assert.equal(requestTurnItem?.status, "failed");
3702	});
3703
3704	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
3705	}),
3706	);
3707	it.effect("ProviderSessionManagerV2 retries release records that failed to persist", () =>
3708	Effect.gen(function* () {
3709	const state = yield* Ref.make(emptyState);
3710	const flaky: FlakyReleaseWrites = {
3711	failing: yield* Ref.make<"none" | "session" | "session-and-requests">("session"),
3712	failures: yield* Queue.unbounded<void>(),
3713	};
3714	const effect = Effect.gen(function* () {
3715	const eventSink = yield* EventSink.EventSinkV2;
3716	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3717	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3718	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3719	const now = yield* DateTime.now;
3720	const threadId = ThreadId.make("thread-provider-session-manager-release-retry");
3721	const providerSessionId = yield* idAllocator.allocate.providerSession({
3722	providerInstanceId: modelSelection.instanceId,
3723	threadId,
3724	});
3725	const providerThread = makeProviderThread({ idAllocator, threadId, providerSessionId, now });
3726	yield* eventSink.write({
3727	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3728	});
3729	const pendingRequest = yield* makePendingRuntimeRequestEvents({
3730	idAllocator,
3731	threadId,
3732	providerSessionId,
3733	providerThread,
3734	now,
3735	});
3736	yield* eventSink.write({ events: pendingRequest.events });
3737	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
3738
3739	assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId))));
3740	yield* Queue.take(flaky.failures);
3741	// The failed session write does not keep the approval answerable.
3742	const afterClose = yield* projectionStore.getThreadProjection(threadId);
3743	assert.equal(afterClose.runtimeRequests.at(-1)?.responseCapability.type, "not_resumable");
3744	assert.equal(afterClose.providerSessions.at(-1)?.status, "ready");
3745
3746	// The first retry fails as well, and the retries continue.
3747	yield* TestClock.adjust("1 second");
3748	yield* Queue.take(flaky.failures);
3749	yield* Ref.set(flaky.failing, "none");
3750	const stopped = yield* eventSink
3751	.stream({
3752	threadId,
3753	afterSequence: yield* eventSink.latestSequence({ threadId }),
3754	eventType: "provider-session.updated",
3755	})
3756	.pipe(Stream.runHead, Effect.forkScoped);
3757	yield* TestClock.adjust("1 second");
3758	yield* Fiber.join(stopped);
3759
3760	const afterRetry = yield* projectionStore.getThreadProjection(threadId);
3761	assert.equal(afterRetry.providerSessions.at(-1)?.status, "stopped");
3762	});
3763
3764	yield* effect.pipe(
3765	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
3766	);
3767	}),
3768	);
3769
3770	it.effect("ProviderSessionManagerV2 release retries leave a replacement session alone", () =>
3771	Effect.gen(function* () {
3772	const state = yield* Ref.make(emptyState);
3773	const flaky: FlakyReleaseWrites = {
3774	failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"),
3775	failures: yield* Queue.unbounded<void>(),
3776	};
3777	const effect = Effect.gen(function* () {
3778	const eventSink = yield* EventSink.EventSinkV2;
3779	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3780	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3781	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3782	const threadId = ThreadId.make("thread-provider-session-manager-release-replacement");
3783	const providerSessionId = yield* idAllocator.allocate.providerSession({
3784	providerInstanceId: modelSelection.instanceId,
3785	threadId,
3786	});
3787	const writePendingRequest = Effect.gen(function* () {
3788	const now = yield* DateTime.now;
3789	const request = yield* makePendingRuntimeRequestEvents({
3790	idAllocator,
3791	threadId,
3792	providerSessionId,
3793	providerThread: makeProviderThread({ idAllocator, threadId, providerSessionId, now }),
3794	now,
3795	});
3796	yield* eventSink.write({ events: request.events });
3797	return request.requestId;
3798	});
3799	yield* eventSink.write({
3800	events: [
3801	yield* makeThreadCreatedEvent({ idAllocator, threadId, now: yield* DateTime.now }),
3802	],
3803	});
3804	const oldRequestId = yield* writePendingRequest;
3805	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
3806	yield* Ref.set(flaky.failing, "session-and-requests");
3807	assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId))));
3808	yield* Queue.take(flaky.failures);
3809	yield* Queue.take(flaky.failures);
3810	yield* Ref.set(flaky.failing, "none");
3811
3812	// A replacement opens with the same id before the retry runs.
3813	yield* TestClock.adjust("500 millis");
3814	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
3815	const newRequestId = yield* writePendingRequest;
3816	const replacementStatus = (yield* projectionStore.getThreadProjection(
3817	threadId,
3818	)).providerSessions.at(-1)?.status;
3819	const settled = yield* eventSink
3820	.stream({
3821	threadId,
3822	afterSequence: yield* eventSink.latestSequence({ threadId }),
3823	eventType: "runtime-request.updated",
3824	})
3825	.pipe(Stream.runHead, Effect.forkScoped);
3826	yield* TestClock.adjust("500 millis");
3827	yield* Fiber.join(settled);
3828
3829	const projection = yield* projectionStore.getThreadProjection(threadId);
3830	const request = (id: typeof oldRequestId) =>
3831	projection.runtimeRequests.find((candidate) => candidate.id === id);
3832	assert.equal(request(oldRequestId)?.responseCapability.type, "not_resumable");
3833	assert.equal(request(newRequestId)?.responseCapability.type, "live");
3834	assert.equal(projection.providerSessions.at(-1)?.status, replacementStatus);
3835	});
3836
3837	yield* effect.pipe(
3838	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
3839	);
3840	}),
3841	);
3842
3843	it.effect("ProviderSessionManagerV2 keeps each failed release's cleanup", () =>
3844	Effect.gen(function* () {
3845	const state = yield* Ref.make(emptyState);
3846	const flaky: FlakyReleaseWrites = {
3847	failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"),
3848	failures: yield* Queue.unbounded<void>(),
3849	};
3850	const effect = Effect.gen(function* () {
3851	const eventSink = yield* EventSink.EventSinkV2;
3852	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3853	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3854	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3855	const now = yield* DateTime.now;
3856	const firstThreadId = ThreadId.make("thread-provider-session-manager-release-each-a");
3857	const secondThreadId = ThreadId.make("thread-provider-session-manager-release-each-b");
3858	const providerSessionId = yield* idAllocator.allocate.providerSession({
3859	providerInstanceId: modelSelection.instanceId,
3860	threadId: firstThreadId,
3861	});
3862	yield* eventSink.write({
3863	events: [
3864	yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
3865	yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
3866	],
3867	});
3868	const secondThreadRequest = yield* makePendingRuntimeRequestEvents({
3869	idAllocator,
3870	threadId: secondThreadId,
3871	providerSessionId,
3872	providerThread: makeProviderThread({
3873	idAllocator,
3874	threadId: secondThreadId,
3875	providerSessionId,
3876	now,
3877	}),
3878	now,
3879	});
3880	yield* eventSink.write({ events: secondThreadRequest.events });
3881	const failRelease = (failedWrites: number) =>
3882	Effect.gen(function* () {
3883	yield* Ref.set(flaky.failing, "session-and-requests");
3884	assert.isTrue(Exit.isFailure(yield* Effect.exit(manager.close(providerSessionId))));
3885	yield* Effect.repeat(Queue.take(flaky.failures), { times: failedWrites - 1 });
3886	yield* Ref.set(flaky.failing, "none");
3887	});
3888
3889	// The first session serves both threads. Its replacement serves one.
3890	yield* manager.open({
3891	threadId: firstThreadId,
3892	providerSessionId,
3893	modelSelection,
3894	runtimePolicy,
3895	});
3896	yield* manager.open({
3897	threadId: secondThreadId,
3898	providerSessionId,
3899	modelSelection,
3900	runtimePolicy,
3901	});
3902	// Both the session write and the second thread's request write fail.
3903	yield* failRelease(2);
3904	yield* TestClock.adjust("500 millis");
3905	yield* manager.open({
3906	threadId: firstThreadId,
3907	providerSessionId,
3908	modelSelection,
3909	runtimePolicy,
3910	});
3911	// Only the session write fails: the first thread has no requests.
3912	yield* failRelease(1);
3913
3914	const settled = yield* eventSink
3915	.stream({
3916	threadId: secondThreadId,
3917	afterSequence: yield* eventSink.latestSequence({ threadId: secondThreadId }),
3918	eventType: "runtime-request.updated",
3919	})
3920	.pipe(Stream.runHead, Effect.forkScoped);
3921	yield* TestClock.adjust("500 millis");
3922	yield* Fiber.join(settled);
3923
3924	const projection = yield* projectionStore.getThreadProjection(secondThreadId);
3925	assert.equal(projection.runtimeRequests.at(-1)?.responseCapability.type, "not_resumable");
3926	});
3927
3928	yield* effect.pipe(
3929	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
3930	);
3931	}),
3932	);
3933
3934	it.effect("ProviderSessionManagerV2 settles a request the event pump persists during release", () =>
3935	Effect.gen(function* () {
3936	const state = yield* Ref.make(emptyState);
3937	const flaky: FlakyReleaseWrites = {
3938	failing: yield* Ref.make<"none" | "session" | "session-and-requests">("none"),
3939	failures: yield* Queue.unbounded<void>(),
3940	pauseRequestWrites: {
3941	paused: yield* Deferred.make<void>(),
3942	resume: yield* Deferred.make<void>(),
3943	},
3944	};
3945	const pause = flaky.pauseRequestWrites!;
3946	const effect = Effect.gen(function* () {
3947	const eventSink = yield* EventSink.EventSinkV2;
3948	const idAllocator = yield* IdAllocator.IdAllocatorV2;
3949	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
3950	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
3951	const now = yield* DateTime.now;
3952	const threadId = ThreadId.make("thread-provider-session-manager-release-pump");
3953	const providerSessionId = yield* idAllocator.allocate.providerSession({
3954	providerInstanceId: modelSelection.instanceId,
3955	threadId,
3956	});
3957	yield* eventSink.write({
3958	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
3959	});
3960	yield* manager.open({ threadId, providerSessionId, modelSelection, runtimePolicy });
3961	// The runtime creates the request a moment after the release starts.
3962	const createdAt = DateTime.add(now, { seconds: 1 });
3963	const pendingRequest = yield* makePendingRuntimeRequestEvents({
3964	idAllocator,
3965	threadId,
3966	providerSessionId,
3967	providerThread: makeProviderThread({
3968	idAllocator,
3969	threadId,
3970	providerSessionId,
3971	now: createdAt,
3972	}),
3973	now: createdAt,
3974	});
3975	const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
3976	assert.isDefined(adapterEvents);
3977	yield* Queue.offerAll(adapterEvents!, pendingRequest.providerEvents);
3978	// The event pump holds the request permit while it persists the request.
3979	yield* Deferred.await(pause.paused);
3980	const closed = yield* manager
3981	.close(providerSessionId)
3982	.pipe(Effect.forkScoped({ startImmediately: true }));
3983	yield* TestClock.adjust("1 second");
3984	yield* Deferred.succeed(pause.resume, undefined);
3985	yield* Fiber.join(closed);
3986
3987	const projection = yield* projectionStore.getThreadProjection(threadId);
3988	const request = projection.runtimeRequests.find(
3989	(candidate) => candidate.id === pendingRequest.requestId,
3990	);
3991	assert.equal(request?.responseCapability.type, "not_resumable");
3992	});
3993
3994	yield* effect.pipe(
3995	Effect.provide(layerTest({ state, idleTimeoutMs: 60_000, flakyReleaseWrites: flaky })),
3996	);
3997	}),
3998	);
3999
4000	it.effect("ProviderSessionManagerV2 terminalizes a pending input transcript item on release", () =>
4001	Effect.gen(function* () {
4002	const state = yield* Ref.make(emptyState);
4003	const effect = Effect.gen(function* () {
4004	const eventSink = yield* EventSink.EventSinkV2;
4005	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4006	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4007	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4008	const now = yield* DateTime.now;
4009	const projectId = yield* idAllocator.allocate.project({
4010	fixtureName: "provider-session-manager-request-expire",
4011	});
4012	const threadId = yield* idAllocator.allocate.thread({
4013	fixtureName: "provider-session-manager-request-expire",
4014	projectId,
4015	});
4016	const providerSessionId = yield* idAllocator.allocate.providerSession({
4017	providerInstanceId: modelSelection.instanceId,
4018	threadId,
4019	});
4020	const providerThread = makeProviderThread({
4021	idAllocator,
4022	threadId,
4023	providerSessionId,
4024	now,
4025	});
4026
4027	yield* eventSink.write({
4028	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
4029	});
4030	const pendingRequest = yield* makePendingRuntimeRequestEvents({
4031	idAllocator,
4032	threadId,
4033	providerSessionId,
4034	providerThread,
4035	now,
4036	});
4037	yield* eventSink.write({
4038	events: pendingRequest.events.map((event) =>
4039	event.type === "turn-item.updated"
4040	? { ...event, payload: { ...event.payload, type: "user_input_request", questions: [] } }
4041	: event,
4042	),
4043	});
4044	yield* manager.open({
4045	threadId,
4046	providerSessionId,
4047	modelSelection,
4048	runtimePolicy,
4049	});
4050	yield* manager.release({
4051	providerSessionId,
4052	reason: "runtime_error",
4053	detail: "process exited",
4054	});
4055
4056	const projection = yield* projectionStore.getThreadProjection(threadId);
4057	const request = projection.runtimeRequests.at(-1);
4058	const requestNode = projection.nodes.find((node) => node.id === request?.nodeId);
4059	const requestTurnItem = projection.turnItems.find(
4060	(item) => item.type === "user_input_request" && item.requestId === request?.id,
4061	);
4062
4063	assert.equal(request?.status, "expired");
4064	assert.equal(request?.responseCapability.type, "not_resumable");
4065	assert.equal(requestNode?.status, "failed");
4066	assert.equal(requestTurnItem?.status, "failed");
4067	});
4068
4069	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
4070	}),
4071	);
4072
4073	it.effect("ProviderSessionManagerV2 persists session-scoped runtime requests without a run", () =>
4074	Effect.gen(function* () {
4075	const state = yield* Ref.make(emptyState);
4076	const effect = Effect.gen(function* () {
4077	const eventSink = yield* EventSink.EventSinkV2;
4078	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4079	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4080	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4081	const now = yield* DateTime.now;
4082	const projectId = yield* idAllocator.allocate.project({
4083	fixtureName: "provider-session-manager-session-request",
4084	});
4085	const threadId = yield* idAllocator.allocate.thread({
4086	fixtureName: "provider-session-manager-session-request",
4087	projectId,
4088	});
4089	const providerSessionId = yield* idAllocator.allocate.providerSession({
4090	providerInstanceId: modelSelection.instanceId,
4091	threadId,
4092	});
4093	const providerThread = makeProviderThread({
4094	idAllocator,
4095	threadId,
4096	providerSessionId,
4097	now,
4098	});
4099
4100	yield* eventSink.write({
4101	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
4102	});
4103	yield* manager.open({
4104	threadId,
4105	providerSessionId,
4106	modelSelection,
4107	runtimePolicy,
4108	});
4109	const pendingRequest = yield* makePendingRuntimeRequestEvents({
4110	idAllocator,
4111	threadId,
4112	providerSessionId,
4113	providerThread,
4114	now,
4115	});
4116	const afterSequence = yield* eventSink.latestSequence({ threadId });
4117	const persistedFiber = yield* eventSink.stream({ threadId, afterSequence }).pipe(
4118	Stream.filter(
4119	(stored) =>
4120	stored.event.type === "runtime-request.updated" ||
4121	stored.event.type === "node.updated" ||
4122	stored.event.type === "turn-item.updated",
4123	),
4124	Stream.take(3),
4125	Stream.runCollect,
4126	Effect.forkScoped,
4127	);
4128	const adapterEvents = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
4129	assert.isDefined(adapterEvents);
4130	yield* Queue.offerAll(adapterEvents!, pendingRequest.providerEvents);
4131	const persisted = Array.from(yield* Fiber.join(persistedFiber));
4132
4133	assert.sameMembers(
4134	persisted.map((stored) => stored.event.type),
4135	["runtime-request.updated", "node.updated", "turn-item.updated"],
4136	);
4137	const projection = yield* projectionStore.getThreadProjection(threadId);
4138	const request = projection.runtimeRequests.find(
4139	(candidate) => candidate.id === pendingRequest.requestId,
4140	);
4141	const node = projection.nodes.find((candidate) => candidate.id === pendingRequest.nodeId);
4142	const turnItem = projection.turnItems.find(
4143	(candidate) =>
4144	candidate.type === "approval_request" && candidate.requestId === pendingRequest.requestId,
4145	);
4146	assert.equal(request?.status, "pending");
4147	assert.equal(request?.providerTurnId, null);
4148	assert.equal(node?.runId, null);
4149	assert.equal(node?.status, "waiting");
4150	assert.equal(turnItem?.runId, null);
4151	assert.equal(turnItem?.status, "waiting");
4152	});
4153
4154	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
4155	}),
4156	);
4157
4158	it.effect(
4159	"ProviderSessionManagerV2 preserves item identity during eager native session activation",
4160	() =>
4161	Effect.gen(function* () {
4162	const state = yield* Ref.make(emptyState);
4163	const effect = Effect.gen(function* () {
4164	const eventSink = yield* EventSink.EventSinkV2;
4165	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4166	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4167	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4168	const now = yield* DateTime.now;
4169	const projectId = yield* idAllocator.allocate.project({
4170	fixtureName: "provider-session-manager-request-expire",
4171	});
4172	const threadId = yield* idAllocator.allocate.thread({
4173	fixtureName: "provider-session-manager-request-expire",
4174	projectId,
4175	});
4176	const providerSessionId = yield* idAllocator.allocate.providerSession({
4177	providerInstanceId: modelSelection.instanceId,
4178	threadId,
4179	});
4180	const providerThread = makeProviderThread({
4181	idAllocator,
4182	threadId,
4183	providerSessionId,
4184	now,
4185	});
4186
4187	yield* eventSink.write({
4188	events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })],
4189	});
4190	yield* eventSink.write({
4191	events: (yield* makePendingRuntimeRequestEvents({
4192	idAllocator,
4193	threadId,
4194	providerSessionId,
4195	providerThread,
4196	now,
4197	})).events,
4198	});
4199	yield* manager.open({
4200	threadId,
4201	providerSessionId,
4202	modelSelection,
4203	runtimePolicy,
4204	initialNativeThreadId: "native-import",
4205	initialProviderItemIdentityVersion: 2,
4206	});
4207	yield* manager.release({
4208	providerSessionId,
4209	reason: "runtime_error",
4210	detail: "process exited",
4211	});
4212
4213	const projection = yield* projectionStore.getThreadProjection(threadId);
4214	const request = projection.runtimeRequests.at(-1);
4215	const requestNode = projection.nodes.find((node) => node.id === request?.nodeId);
4216	const requestTurnItem = projection.turnItems.find(
4217	(item) => item.type === "approval_request" && item.requestId === request?.id,
4218	);
4219
4220	assert.equal(request?.status, "expired");
4221	assert.equal(request?.responseCapability.type, "not_resumable");
4222	assert.equal(requestNode?.status, "failed");
4223	assert.equal(requestTurnItem?.status, "failed");
4224	});
4225
4226	yield* effect.pipe(
4227	Effect.provide(
4228	layerTest({
4229	state,
4230	idleTimeoutMs: 1000,
4231	beforeOpen: (input) =>
4232	Effect.sync(() => assert.equal(input.initialProviderItemIdentityVersion, 2)),
4233	}),
4234	),
4235	);
4236	}),
4237	);
4238
4239	it.effect(
4240	"ProviderSessionManagerV2 keeps a multi-thread session alive until all turns finish",
4241	() =>
4242	Effect.gen(function* () {
4243	const state = yield* Ref.make(emptyState);
4244	const effect = Effect.gen(function* () {
4245	const eventSink = yield* EventSink.EventSinkV2;
4246	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4247	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4248	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4249	const now = yield* DateTime.now;
4250	const projectId = yield* idAllocator.allocate.project({
4251	fixtureName: "provider-session-manager-multi-thread-active",
4252	});
4253	const firstThreadId = yield* idAllocator.allocate.thread({
4254	fixtureName: "provider-session-manager-multi-thread-active-a",
4255	projectId,
4256	});
4257	const secondThreadId = yield* idAllocator.allocate.thread({
4258	fixtureName: "provider-session-manager-multi-thread-active-b",
4259	projectId,
4260	});
4261	const providerSessionId = yield* idAllocator.allocate.providerSession({
4262	providerInstanceId: modelSelection.instanceId,
4263	threadId: firstThreadId,
4264	});
4265	// Each thread has its own native thread on the shared session.
4266	const firstProviderThread = makeProviderThread({
4267	idAllocator,
4268	threadId: firstThreadId,
4269	providerSessionId,
4270	now,
4271	nativeThreadId: "native-thread-a",
4272	});
4273	const secondProviderThread = makeProviderThread({
4274	idAllocator,
4275	threadId: secondThreadId,
4276	providerSessionId,
4277	now,
4278	nativeThreadId: "native-thread-b",
4279	});
4280	const firstRunId = idAllocator.derive.run({ threadId: firstThreadId, ordinal: 1 });
4281	const secondRunId = idAllocator.derive.run({ threadId: secondThreadId, ordinal: 1 });
4282	const firstProviderTurnId = idAllocator.derive.providerTurn({
4283	driver: CODEX_DRIVER,
4284	nativeTurnId: "native-turn-a",
4285	});
4286	const secondProviderTurnId = idAllocator.derive.providerTurn({
4287	driver: CODEX_DRIVER,
4288	nativeTurnId: "native-turn-b",
4289	});
4290
4291	yield* eventSink.write({
4292	events: [
4293	yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
4294	yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
4295	],
4296	});
4297	const runtime = yield* manager.open({
4298	threadId: firstThreadId,
4299	providerSessionId,
4300	modelSelection,
4301	runtimePolicy,
4302	});
4303	yield* manager.open({
4304	threadId: secondThreadId,
4305	providerSessionId,
4306	modelSelection,
4307	runtimePolicy,
4308	});
4309	yield* runtime.events.pipe(Stream.runDrain, Effect.forkScoped);
4310	const firstAppThread = (yield* projectionStore.getThreadProjection(firstThreadId)).thread;
4311	const secondAppThread = (yield* projectionStore.getThreadProjection(secondThreadId)).thread;
4312	yield* runtime.startTurn({
4313	appThread: firstAppThread,
4314	threadId: firstThreadId,
4315	runId: firstRunId,
4316	runOrdinal: 1,
4317	providerTurnOrdinal: 1,
4318	attemptId: idAllocator.derive.runAttempt({ runId: firstRunId, attemptOrdinal: 1 }),
4319	rootNodeId: idAllocator.derive.rootNode({ runId: firstRunId }),
4320	providerThread: firstProviderThread,
4321	message: {
4322	createdBy: "user",
4323	creationSource: "web",
4324	messageId: yield* idAllocator.allocate.message({ threadId: firstThreadId, ordinal: 1 }),
4325	text: "first",
4326	attachments: [],
4327	},
4328	modelSelection,
4329	runtimePolicy,
4330	});
4331	yield* runtime.startTurn({
4332	appThread: secondAppThread,
4333	threadId: secondThreadId,
4334	runId: secondRunId,
4335	runOrdinal: 1,
4336	providerTurnOrdinal: 1,
4337	attemptId: idAllocator.derive.runAttempt({ runId: secondRunId, attemptOrdinal: 1 }),
4338	rootNodeId: idAllocator.derive.rootNode({ runId: secondRunId }),
4339	providerThread: secondProviderThread,
4340	message: {
4341	createdBy: "user",
4342	creationSource: "web",
4343	messageId: yield* idAllocator.allocate.message({
4344	threadId: secondThreadId,
4345	ordinal: 1,
4346	}),
4347	text: "second",
4348	attachments: [],
4349	},
4350	modelSelection,
4351	runtimePolicy,
4352	});
4353
4354	const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
4355	assert.isDefined(queue);
4356	yield* Queue.offer(queue!, {
4357	type: "turn.terminal",
4358	driver: CODEX_DRIVER,
4359	providerThreadId: firstProviderThread.id,
4360	providerTurnId: firstProviderTurnId,
4361	runOrdinal: 1,
4362	status: "completed",
4363	failure: null,
4364	threadDisposition: "reusable",
4365	});
4366	yield* TestClock.adjust("2 seconds");
4367	yield* Effect.yieldNow;
4368	assert.equal((yield* Ref.get(state)).closeCount, 0);
4369
4370	yield* Queue.offer(queue!, {
4371	type: "turn.terminal",
4372	driver: CODEX_DRIVER,
4373	providerThreadId: secondProviderThread.id,
4374	providerTurnId: secondProviderTurnId,
4375	runOrdinal: 1,
4376	status: "completed",
4377	failure: null,
4378	threadDisposition: "reusable",
4379	});
4380	yield* TestClock.adjust("1 second");
4381	yield* Effect.yieldNow;
4382	assert.equal((yield* Ref.get(state)).closeCount, 1);
4383	});
4384
4385	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
4386	}),
4387	);
4388
4389	it.effect(
4390	"ProviderSessionManagerV2 opens one shared runtime, broadcasts events, and detaches threads independently",
4391	() =>
4392	Effect.gen(function* () {
4393	const state = yield* Ref.make(emptyState);
4394	const effect = Effect.gen(function* () {
4395	const eventSink = yield* EventSink.EventSinkV2;
4396	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4397	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4398	const now = yield* DateTime.now;
4399	const projectId = yield* idAllocator.allocate.project({
4400	fixtureName: "provider-session-manager-shared-runtime",
4401	});
4402	const firstThreadId = yield* idAllocator.allocate.thread({
4403	fixtureName: "provider-session-manager-shared-runtime-a",
4404	projectId,
4405	});
4406	const secondThreadId = yield* idAllocator.allocate.thread({
4407	fixtureName: "provider-session-manager-shared-runtime-b",
4408	projectId,
4409	});
4410	const providerSessionId = idAllocator.derive.providerSession({
4411	providerInstanceId: modelSelection.instanceId,
4412	});
4413
4414	yield* eventSink.write({
4415	events: [
4416	yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
4417	yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
4418	],
4419	});
4420	const firstProviderThread = makeProviderThread({
4421	idAllocator,
4422	threadId: firstThreadId,
4423	providerSessionId,
4424	now,
4425	});
4426	const secondProviderThread = makeProviderThread({
4427	idAllocator,
4428	threadId: secondThreadId,
4429	providerSessionId,
4430	now,
4431	});
4432	const firstRunId = idAllocator.derive.run({ threadId: firstThreadId, ordinal: 1 });
4433	yield* eventSink.write({
4434	events: [
4435	{
4436	id: yield* idAllocator.allocate.event({ threadId: firstThreadId }),
4437	type: "provider-thread.updated",
4438	threadId: firstThreadId,
4439	driver: CODEX_DRIVER,
4440	occurredAt: now,
4441	payload: firstProviderThread,
4442	},
4443	{
4444	id: yield* idAllocator.allocate.event({ threadId: firstThreadId }),
4445	type: "provider-turn.updated",
4446	threadId: firstThreadId,
4447	runId: firstRunId,
4448	driver: CODEX_DRIVER,
4449	occurredAt: now,
4450	payload: {
4451	id: idAllocator.derive.providerTurn({
4452	driver: CODEX_DRIVER,
4453	nativeTurnId: "native-turn-shared-runtime-a",
4454	}),
4455	providerThreadId: firstProviderThread.id,
4456	nodeId: idAllocator.derive.rootNode({ runId: firstRunId }),
4457	runAttemptId: null,
4458	nativeTurnRef: null,
4459	ordinal: 1,
4460	status: "running",
4461	startedAt: now,
4462	completedAt: null,
4463	},
4464	},
4465	],
4466	});
4467	const firstRuntime = yield* manager.open({
4468	threadId: firstThreadId,
4469	providerSessionId,
4470	modelSelection,
4471	runtimePolicy,
4472	});
4473	const secondRuntime = yield* manager.open({
4474	threadId: secondThreadId,
4475	providerSessionId,
4476	modelSelection,
4477	runtimePolicy,
4478	});
4479
4480	assert.strictEqual(firstRuntime, secondRuntime);
4481	assert.equal((yield* Ref.get(state)).openCount, 1);
4482	const resumeSecondThread = secondRuntime.resumeThread({
4483	providerThread: secondProviderThread,
4484	threadId: secondThreadId,
4485	modelSelection,
4486	runtimePolicy,
4487	});
4488	yield* resumeSecondThread;
4489	yield* resumeSecondThread;
4490	assert.equal((yield* Ref.get(state)).resumeCount, 1);
4491	yield* secondRuntime.resumeThread({
4492	providerThread: secondProviderThread,
4493	threadId: secondThreadId,
4494	modelSelection: { ...modelSelection, model: "gpt-5.4-mini" },
4495	runtimePolicy,
4496	});
4497	assert.equal((yield* Ref.get(state)).resumeCount, 2);
4498	yield* resumeSecondThread;
4499	assert.equal((yield* Ref.get(state)).resumeCount, 3);
4500	const subscribe = firstRuntime.subscribeEvents;
4501	assert.isDefined(subscribe);
4502	if (subscribe === undefined) return;
4503	const firstSubscription = yield* subscribe;
4504	const secondSubscription = yield* subscribe;
4505	const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
4506	assert.isDefined(queue);
4507	yield* Queue.offer(queue!, {
4508	type: "provider_session.updated",
4509	driver: CODEX_DRIVER,
4510	providerSession: firstRuntime.providerSession,
4511	});
4512	const received = yield* Effect.all([
4513	firstSubscription.events.pipe(Stream.runHead),
4514	secondSubscription.events.pipe(Stream.runHead),
4515	]);
4516	assert.isTrue(received.every(Option.isSome));
4517	assert.isTrue(
4518	received.every(
4519	(event) => Option.isSome(event) && event.value.type === "provider_session.updated",
4520	),
4521	);
4522
4523	yield* manager.detach({ providerSessionId, threadId: secondThreadId });
4524	yield* manager.open({
4525	threadId: secondThreadId,
4526	providerSessionId,
4527	modelSelection,
4528	runtimePolicy,
4529	});
4530	yield* resumeSecondThread;
4531	assert.equal((yield* Ref.get(state)).resumeCount, 4);
4532
4533	// The second thread has no persisted provider thread, so nothing is unloaded.
4534	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, []);
4535
4536	yield* manager.detach({ providerSessionId, threadId: firstThreadId });
4537	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
4538	assert.equal((yield* Ref.get(state)).closeCount, 0);
4539	assert.equal((yield* Ref.get(state)).interruptCount, 1);
4540	// The runtime stays up for the second thread; the first thread's
4541	// native state is unloaded after its turn is interrupted.
4542	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]);
4543
4544	yield* manager.detach({ providerSessionId, threadId: secondThreadId });
4545	yield* TestClock.adjust("1 second");
4546	yield* Effect.yieldNow;
4547	assert.equal((yield* Ref.get(state)).closeCount, 1);
4548	});
4549
4550	yield* effect.pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 1000 })));
4551	}),
4552	);
4553
4554	it.effect(
4555	"ProviderSessionManagerV2 re-attaching a thread waits for its in-flight unload, then reloads it",
4556	() =>
4557	Effect.gen(function* () {
4558	const state = yield* Ref.make(emptyState);
4559	const unloadStarted = yield* Deferred.make<void>();
4560	const releaseUnload = yield* Deferred.make<void>();
4561	// Resumes the provider had served when the unload actually reached it.
4562	let resumesBeforeUnload: number | undefined;
4563	// The unload parks after detach removed the attachment, leaving the
4564	// window in which the same thread's next turn re-attaches it.
4565	const beforeUnload = Effect.gen(function* () {
4566	yield* Deferred.succeed(unloadStarted, undefined);
4567	yield* Deferred.await(releaseUnload);
4568	resumesBeforeUnload = (yield* Ref.get(state)).resumeCount;
4569	});
4570	const effect = Effect.gen(function* () {
4571	const eventSink = yield* EventSink.EventSinkV2;
4572	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4573	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4574	const now = yield* DateTime.now;
4575	const projectId = yield* idAllocator.allocate.project({
4576	fixtureName: "provider-session-manager-unload-race",
4577	});
4578	const threadId = yield* idAllocator.allocate.thread({
4579	fixtureName: "provider-session-manager-unload-race-a",
4580	projectId,
4581	});
4582	const otherThreadId = yield* idAllocator.allocate.thread({
4583	fixtureName: "provider-session-manager-unload-race-b",
4584	projectId,
4585	});
4586	const providerSessionId = idAllocator.derive.providerSession({
4587	providerInstanceId: modelSelection.instanceId,
4588	});
4589	const providerThread = makeProviderThread({
4590	idAllocator,
4591	threadId,
4592	providerSessionId,
4593	now,
4594	});
4595	yield* eventSink.write({
4596	events: [
4597	yield* makeThreadCreatedEvent({ idAllocator, threadId, now }),
4598	yield* makeThreadCreatedEvent({ idAllocator, threadId: otherThreadId, now }),
4599	{
4600	id: yield* idAllocator.allocate.event({ threadId }),
4601	type: "provider-thread.updated",
4602	threadId,
4603	driver: CODEX_DRIVER,
4604	occurredAt: now,
4605	payload: providerThread,
4606	},
4607	],
4608	});
4609	const runtime = yield* manager.open({
4610	threadId,
4611	providerSessionId,
4612	modelSelection,
4613	runtimePolicy,
4614	});
4615	// A second thread keeps the shared runtime up after the detach.
4616	yield* manager.open({
4617	threadId: otherThreadId,
4618	providerSessionId,
4619	modelSelection,
4620	runtimePolicy,
4621	});
4622	// Resuming re-attaches the thread to the shared runtime.
4623	const resume = runtime.resumeThread({
4624	providerThread,
4625	threadId,
4626	modelSelection,
4627	runtimePolicy,
4628	});
4629	yield* resume;
4630
4631	const detach = yield* manager
4632	.detach({ providerSessionId, threadId })
4633	.pipe(Effect.forkScoped);
4634	yield* Deferred.await(unloadStarted);
4635	// The same thread's next turn re-attaches while the unload is parked.
4636	// Give it room to run: unfixed, it reaches the provider's resume
4637	// here; serialized, it waits for the unload.
4638	const reattach = yield* resume.pipe(Effect.forkScoped);
4639	yield* Effect.yieldNow;
4640	yield* Effect.yieldNow;
4641	yield* Deferred.succeed(releaseUnload, undefined);
4642	yield* Fiber.join(detach);
4643	yield* Fiber.join(reattach);
4644
4645	// The unload reached the provider before the re-attached resume, so
4646	// that resume reloads the thread instead of being torn down after it.
4647	assert.equal(resumesBeforeUnload, 1);
4648	assert.equal((yield* Ref.get(state)).resumeCount, 2);
4649	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, ["native-thread"]);
4650	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
4651	});
4652
4653	yield* effect.pipe(
4654	Effect.provide(layerTest({ state, idleTimeoutMs: 1000, beforeUnload })),
4655	Effect.scoped,
4656	);
4657	}),
4658	);
4659
4660	function runIdleThreadUnloadScenario(
4661	name: string,
4662	scenario: (input: {
4663	readonly state: Ref.Ref<TestProviderRuntimeState>;
4664	readonly manager: ProviderSessionManager.ProviderSessionManagerV2Shape;
4665	readonly providerSessionId: ProviderSessionId;
4666	readonly threadA: ThreadId;
4667	/** Starts run `ordinal` on a thread and returns once the provider accepted it. */
4668	readonly startTurn: (threadId: ThreadId, ordinal: number) => Effect.Effect<void>;
4669	/** Ends run `ordinal` on a thread and waits for the session to process it. */
4670	readonly endTurn: (threadId: ThreadId, ordinal: number) => Effect.Effect<void>;
4671	readonly resume: (threadId: ThreadId) => Effect.Effect<void>;
4672	}) => Effect.Effect<void, ProviderSessionManager.ProviderSessionManagerV2Error>,
4673	options: { readonly hasPendingBackgroundWorkForThread?: Effect.Effect<boolean> } = {},
4674	) {
4675	return Effect.gen(function* () {
4676	const state = yield* Ref.make(emptyState);
4677	yield* Effect.gen(function* () {
4678	const eventSink = yield* EventSink.EventSinkV2;
4679	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4680	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4681	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4682	const now = yield* DateTime.now;
4683	const threadA = ThreadId.make(`thread-provider-session-manager-${name}-a`);
4684	const threadB = ThreadId.make(`thread-provider-session-manager-${name}-b`);
4685	const providerSessionId = idAllocator.derive.providerSession({
4686	providerInstanceId: modelSelection.instanceId,
4687	});
4688	yield* eventSink.write({
4689	events: [
4690	yield* makeThreadCreatedEvent({ idAllocator, threadId: threadA, now }),
4691	yield* makeThreadCreatedEvent({ idAllocator, threadId: threadB, now }),
4692	],
4693	});
4694	const runtime = yield* manager.open({
4695	threadId: threadA,
4696	providerSessionId,
4697	modelSelection,
4698	runtimePolicy,
4699	});
4700	yield* manager.open({ threadId: threadB, providerSessionId, modelSelection, runtimePolicy });
4701	const providerThreadOf = (threadId: ThreadId) =>
4702	makeProviderThread({
4703	idAllocator,
4704	threadId,
4705	providerSessionId,
4706	now,
4707	nativeThreadId: `native-${threadId}`,
4708	});
4709	const resume = (threadId: ThreadId) =>
4710	runtime
4711	.resumeThread({
4712	providerThread: providerThreadOf(threadId),
4713	threadId,
4714	modelSelection,
4715	runtimePolicy,
4716	})
4717	.pipe(Effect.asVoid, Effect.orDie);
4718	const startTurn = (threadId: ThreadId, ordinal: number) =>
4719	Effect.gen(function* () {
4720	yield* resume(threadId);
4721	const runId = idAllocator.derive.run({ threadId, ordinal });
4722	yield* runtime.startTurn({
4723	appThread: (yield* projectionStore.getThreadProjection(threadId)).thread,
4724	threadId,
4725	runId,
4726	runOrdinal: ordinal,
4727	providerTurnOrdinal: ordinal,
4728	attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }),
4729	rootNodeId: idAllocator.derive.rootNode({ runId }),
4730	providerThread: providerThreadOf(threadId),
4731	message: {
4732	createdBy: "user",
4733	creationSource: "web",
4734	messageId: yield* idAllocator.allocate.message({ threadId, ordinal }),
4735	text: "turn",
4736	attachments: [],
4737	},
4738	modelSelection,
4739	runtimePolicy,
4740	});
4741	}).pipe(Effect.orDie);
4742	const subscribe = runtime.subscribeEvents;
4743	assert.isDefined(subscribe);
4744	const endTurn = (threadId: ThreadId, ordinal: number) =>
4745	Effect.scoped(
4746	Effect.gen(function* () {
4747	// The pump hands an event to subscribers only after the session
4748	// has processed it, so receiving it is the receipt.
4749	const subscription = yield* Effect.acquireRelease(subscribe!, (sub) => sub.close);
4750	const received = yield* subscription.events.pipe(
4751	Stream.filter((event) => event.type === "turn.terminal"),
4752	Stream.runHead,
4753	Effect.forkScoped,
4754	);
4755	const queue = (yield* Ref.get(state)).eventQueues.get(String(providerSessionId));
4756	assert.isDefined(queue);
4757	yield* Queue.offer(queue!, {
4758	type: "turn.terminal",
4759	driver: CODEX_DRIVER,
4760	providerThreadId: providerThreadOf(threadId).id,
4761	providerTurnId: idAllocator.derive.providerTurn({
4762	driver: CODEX_DRIVER,
4763	nativeTurnId: `native-turn-${threadId}-${ordinal}`,
4764	}),
4765	runOrdinal: ordinal,
4766	status: "completed",
4767	failure: null,
4768	threadDisposition: "reusable",
4769	});
4770	yield* Fiber.join(received);
4771	}),
4772	).pipe(Effect.orDie);
4773
4774	// B's running turn keeps the shared runtime itself busy throughout.
4775	yield* startTurn(threadB, 1);
4776	yield* scenario({
4777	state,
4778	manager,
4779	providerSessionId,
4780	threadA,
4781	startTurn,
4782	endTurn,
4783	resume,
4784	});
4785	}).pipe(
4786	Effect.provide(
4787	layerTest({
4788	state,
4789	idleTimeoutMs: 1000,
4790	...(options.hasPendingBackgroundWorkForThread === undefined
4791	? {}
4792	: { hasPendingBackgroundWorkForThread: options.hasPendingBackgroundWorkForThread }),
4793	}),
4794	),
4795	);
4796	});
4797	}
4798
4799	it.effect(
4800	"ProviderSessionManagerV2 unloads a shared-runtime thread left idle, and reloads it on its next turn",
4801	() =>
4802	runIdleThreadUnloadScenario(
4803	"idle-unload",
4804	({ state, manager, providerSessionId, threadA, startTurn, endTurn, resume }) =>
4805	Effect.gen(function* () {
4806	yield* startTurn(threadA, 1);
4807	assert.equal((yield* Ref.get(state)).resumeCount, 2);
4808	yield* endTurn(threadA, 1);
4809
4810	// A follow-up before the timeout keeps the thread loaded.
4811	yield* TestClock.adjust("500 millis");
4812	yield* startTurn(threadA, 2);
4813	yield* TestClock.adjust("1 second");
4814	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, []);
4815
4816	// Idle for the full timeout after its last turn, A is unloaded while
4817	// the runtime stays up for B.
4818	yield* endTurn(threadA, 2);
4819	yield* TestClock.adjust("1 second");
4820	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [`native-${threadA}`]);
4821	assert.isTrue(Option.isSome(yield* manager.get(providerSessionId)));
4822	assert.equal((yield* Ref.get(state)).closeCount, 0);
4823
4824	// The next turn's resume reaches the provider and reloads A.
4825	const resumes = (yield* Ref.get(state)).resumeCount;
4826	yield* resume(threadA);
4827	assert.equal((yield* Ref.get(state)).resumeCount, resumes + 1);
4828	}),
4829	),
4830	);
4831
4832	it.effect(
4833	"ProviderSessionManagerV2 keeps an idle shared-runtime thread loaded while its background work runs",
4834	() =>
4835	Effect.gen(function* () {
4836	const pendingWork = yield* Ref.make(true);
4837	yield* runIdleThreadUnloadScenario(
4838	"idle-unload-pinned",
4839	({ state, threadA, startTurn, endTurn }) =>
4840	Effect.gen(function* () {
4841	yield* startTurn(threadA, 1);
4842	yield* endTurn(threadA, 1);
4843	yield* TestClock.adjust("3 seconds");
4844	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, []);
4845
4846	yield* Ref.set(pendingWork, false);
4847	yield* TestClock.adjust("1 second");
4848	assert.deepEqual((yield* Ref.get(state)).unloadedNativeThreadIds, [
4849	`native-${threadA}`,
4850	]);
4851	}),
4852	{ hasPendingBackgroundWorkForThread: Ref.get(pendingWork) },
4853	);
4854	}),
4855	);
4856
4857	it.effect(
4858	"ProviderSessionManagerV2 rejects a second thread when the provider runtime is exclusive",
4859	() =>
4860	Effect.gen(function* () {
4861	const state = yield* Ref.make(emptyState);
4862	const effect = Effect.gen(function* () {
4863	const eventSink = yield* EventSink.EventSinkV2;
4864	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4865	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4866	const now = yield* DateTime.now;
4867	const projectId = yield* idAllocator.allocate.project({
4868	fixtureName: "provider-session-manager-exclusive-runtime",
4869	});
4870	const firstThreadId = yield* idAllocator.allocate.thread({
4871	fixtureName: "provider-session-manager-exclusive-runtime-a",
4872	projectId,
4873	});
4874	const secondThreadId = yield* idAllocator.allocate.thread({
4875	fixtureName: "provider-session-manager-exclusive-runtime-b",
4876	projectId,
4877	});
4878	const providerSessionId = yield* idAllocator.allocate.providerSession({
4879	providerInstanceId: modelSelection.instanceId,
4880	threadId: firstThreadId,
4881	});
4882	yield* eventSink.write({
4883	events: [
4884	yield* makeThreadCreatedEvent({ idAllocator, threadId: firstThreadId, now }),
4885	yield* makeThreadCreatedEvent({ idAllocator, threadId: secondThreadId, now }),
4886	],
4887	});
4888
4889	yield* manager.open({
4890	threadId: firstThreadId,
4891	providerSessionId,
4892	modelSelection,
4893	runtimePolicy,
4894	});
4895	const error = yield* manager
4896	.open({
4897	threadId: secondThreadId,
4898	providerSessionId,
4899	modelSelection,
4900	runtimePolicy,
4901	})
4902	.pipe(Effect.flip);
4903
4904	assert.equal(error._tag, "ProviderSessionOpenError");
4905	assert.equal((yield* Ref.get(state)).openCount, 1);
4906	});
4907
4908	yield* effect.pipe(
4909	Effect.provide(
4910	layerTest({ state, idleTimeoutMs: 1000, capabilities: ExclusiveCapabilities }),
4911	),
4912	);
4913	}),
4914	);
4915
4916	it.effect.each(["missing", "file"] as const)(
4917	"rejects a %s workspace before opening a provider session",
4918	(workspaceState) =>
4919	Effect.gen(function* () {
4920	const fileSystem = yield* FileSystem.FileSystem;
4921	const root = yield* fileSystem.makeTempDirectoryScoped();
4922	const cwd = `${root}/workspace`;
4923	if (workspaceState === "file") yield* fileSystem.writeFileString(cwd, "not a directory");
4924	const state = yield* Ref.make(emptyState);
4925	yield* Effect.gen(function* () {
4926	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4927	const eventSink = yield* EventSink.EventSinkV2;
4928	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4929	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4930	const threadId = ThreadId.make(`thread-${workspaceState}-workspace`);
4931	const providerSessionId = yield* idAllocator.allocate.providerSession({
4932	providerInstanceId: modelSelection.instanceId,
4933	threadId,
4934	});
4935	yield* eventSink.write({
4936	events: [
4937	yield* makeThreadCreatedEvent({
4938	idAllocator,
4939	threadId,
4940	now: yield* DateTime.now,
4941	}),
4942	],
4943	});
4944	const error = yield* manager
4945	.open({
4946	threadId,
4947	providerSessionId,
4948	modelSelection,
4949	runtimePolicy: { ...runtimePolicy, cwd },
4950	})
4951	.pipe(Effect.flip);
4952	assert.instanceOf(error, ProviderWorkspaceMissingError);
4953	assert.include(error.message, cwd);
4954	assert.include(error.message, "Restore the folder at this path before retrying.");
4955	assert.equal((yield* Ref.get(state)).openCount, 0);
4956	assert.isTrue(Option.isNone(yield* manager.get(providerSessionId)));
4957	assert.deepEqual(
4958	(yield* projectionStore.getThreadProjection(threadId)).providerSessions,
4959	[],
4960	);
4961	}).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })));
4962	}).pipe(Effect.provide(NodeServices.layer)),
4963	);
4964
4965	it.effect(
4966	"rejects a deleted workspace before reusing a live session without changing its state",
4967	() =>
4968	Effect.gen(function* () {
4969	const fileSystem = yield* FileSystem.FileSystem;
4970	const root = yield* fileSystem.makeTempDirectoryScoped();
4971	const cwd = `${root}/workspace`;
4972	yield* fileSystem.makeDirectory(cwd);
4973	const state = yield* Ref.make(emptyState);
4974	yield* Effect.gen(function* () {
4975	const manager = yield* ProviderSessionManager.ProviderSessionManagerV2;
4976	const eventSink = yield* EventSink.EventSinkV2;
4977	const projectionStore = yield* ProjectionStore.ProjectionStoreV2;
4978	const idAllocator = yield* IdAllocator.IdAllocatorV2;
4979	const threadId = ThreadId.make("thread-deleted-live-workspace");
4980	const providerSessionId = yield* idAllocator.allocate.providerSession({
4981	providerInstanceId: modelSelection.instanceId,
4982	threadId,
4983	});
4984	yield* eventSink.write({
4985	events: [
4986	yield* makeThreadCreatedEvent({
4987	idAllocator,
4988	threadId,
4989	now: yield* DateTime.now,
4990	}),
4991	],
4992	});
4993	const input = {
4994	threadId,
4995	providerSessionId,
4996	modelSelection,
4997	runtimePolicy: { ...runtimePolicy, cwd },
4998	};
4999	const runtime = yield* manager.open(input);
5000	const before = yield* projectionStore.getThreadProjection(threadId);
5001	yield* fileSystem.remove(cwd, { recursive: true });
5002	const error = yield* manager.open(input).pipe(Effect.flip);
5003	assert.instanceOf(error, ProviderWorkspaceMissingError);
5004	assert.equal((yield* Ref.get(state)).openCount, 1);
5005	assert.equal((yield* Ref.get(state)).closeCount, 0);
5006	assert.strictEqual(Option.getOrThrow(yield* manager.get(providerSessionId)), runtime);
5007	assert.deepEqual(
5008	(yield* projectionStore.getThreadProjection(threadId)).providerSessions,
5009	before.providerSessions,
5010	);
5011	}).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60_000 })));
5012	}).pipe(Effect.provide(NodeServices.layer)),
5013	);
5014
5015	it.effect(
5016	"ProviderSessionManagerV2 applies project device access independently of browser access",
5017	() =>
5018	Effect.gen(function* () {
5019	const enabled = yield* runBrowserAccessScenario({
5020	enableAgentBrowserAccess: false,
5021	projectOverride: false,
5022	deviceOverride: true,
5023	});
5024	assert.isTrue(enabled?.capabilities?.has("device"));
5025	assert.isFalse(enabled?.browserToolsAvailable);
5026	const denied = yield* runBrowserAccessScenario({
5027	enableAgentBrowserAccess: false,
5028	projectOverride: false,
5029	deviceOverride: true,
5030	projectExists: false,
5031	});
5032	assert.isFalse(denied?.capabilities?.has("device"));
5033	}),
5034	);